Boost Mini‑Program Install Speed by 21% with Streaming Download Pipeline

Problem Background

Mini‑program installation originally performs four steps—download, save, signature verification, and decompression—in a strictly sequential order. The total installation time equals the sum of each step, and the download stage dominates because it monopolises network I/O while local I/O and CPU remain under‑utilised.

Solution Overview

The original workflow consists of three phases:

Download package: network I/O busiest, CPU idle, local I/O moderate.

Verify package: no network I/O, CPU (signature calculation) busiest, local I/O (file read) moderate.

Extract files: no network I/O, CPU (decryption & decompression) busiest, local I/O (read/write) busiest.

Network I/O dominates overall latency.

To reduce the post‑download overhead, a streaming installation scheme is introduced. It reads the download stream while simultaneously performing signature verification and decryption/decompression, thus overlapping network, CPU, and local I/O work.

MultiPipe Design

MultiPipe is a utility that splits a single input channel into multiple consumer pipelines, similar to an intake manifold. Each consumer receives the same byte stream and can process it independently (e.g., signature verification, decryption, decompression).

Usage Example

ReadableByteChannel srcChannel = ... // okhttp3.ResponseBody#source result
ExecutorService threadPool = Executors.newFixedThreadPool(2);
MultiPipe multiplePipe = new MultiPipe(
    new Consumer<ReadableByteChannel>() {
        @Override
        public void accept(ReadableByteChannel source) {
            // Compute MD5 for signature verification (CPU busy)
        }
    },
    new Consumer<ReadableByteChannel>() {
        @Override
        public void accept(ReadableByteChannel source) {
            // Decrypt, decompress, write to disk (CPU & I/O busy)
        }
    }) {
    @Override
    protected ExecutorService onCreateExecutor(int consumerSize) {
        return threadPool;
    }
};
multiplePipe.setTmpBufferCapacity(MultiPipe.TMP_BUFFER_CAPACITY);
multiplePipe.connect(srcChannel);

Implementation Details

1. Connect and Create Pipeline List

public final void connect(ReadableByteChannel source) {
    onStart(source);
    List<PipeLine> pipeLineList = createPipeLineList();
    CountDownLatch latch = new CountDownLatch(pipeLineList.size());
    ExecutorService executorService = launchPipeLineList(pipeLineList, latch);
    try {
        transfer(source, pipeLineList);
        onTransferComplete(latch);
    } catch (IOException e) {
        onException(e);
    } finally {
        onFinish(source, executorService);
    }
}

2. Launch Pipelines

private ExecutorService launchPipeLineList(List<PipeLine> pipeLineList, CountDownLatch latch) {
    ExecutorService executorService = onCreateExecutor(pipeLineList.size());
    for (PipeLine pipeLine : pipeLineList) {
        pipeLine.setLaunch(latch);
        executorService.submit(pipeLine);
    }
    return executorService;
}

protected ExecutorService onCreateExecutor(int consumerSize) {
    return Executors.newFixedThreadPool(consumerSize);
}

3. Transfer Data to Pipelines

private void transfer(ReadableByteChannel source, List<PipeLine> pipeLineList) throws IOException {
    long writeBytes = 0;
    onUpdateProgress(writeBytes);
    ByteBuffer buf = ByteBuffer.allocate(mTmpBufferCapacity);
    long reads;
    while ((reads = source.read(buf)) != -1) {
        buf.flip();
        for (PipeLine pipeLine : pipeLineList) {
            if (pipeLine.mSink.isOpen() && pipeLine.mSource.isOpen()) {
                buf.rewind();
                pipeLine.mSink.write(buf);
            }
        }
        buf.clear();
        writeBytes += reads;
        onUpdateProgress(writeBytes);
    }
    finally {
        for (PipeLine pipeLine : pipeLineList) {
            closeChannel(pipeLine.mSink);
        }
    }
}

4. PipeLine Implementation

private static class PipeLine implements Runnable {
    final Consumer<ReadableByteChannel> mConsumer;
    final ReadableByteChannel mSource;
    final WritableByteChannel mSink;
    transient CountDownLatch mLatch;

    PipeLine(Consumer<ReadableByteChannel> consumer, boolean hasBuffer) {
        mConsumer = consumer;
        if (hasBuffer) {
            okio.Pipe okioPipe = new okio.Pipe(getPipeMaxBufferBytes());
            mSink = okio.Okio.buffer(okioPipe.sink());
            mSource = okio.Okio.buffer(okioPipe.source());
        } else {
            try {
                java.nio.channels.Pipe pipe = java.nio.channels.Pipe.open();
                mSource = pipe.source();
                mSink = pipe.sink();
            } catch (IOException e) {
                throw new IllegalStateException(e);
            }
        }
    }

    @Override
    public void run() {
        try {
            mConsumer.accept(mSource);
        } finally {
            closeChannel(mSink);
            closeChannel(mSource);
            if (mLatch != null) {
                mLatch.countDown();
            }
        }
    }
}

5. Buffer Size Calculation

private static final float FACTOR = 0.75F;
private static long getPipeMaxBufferBytes() {
    Runtime r = Runtime.getRuntime();
    long available = r.maxMemory() - r.totalMemory() + r.freeMemory();
    return (long) (available * FACTOR);
}

Result

The streaming download scheme reduces overall installation time by approximately 21 % by overlapping network download with CPU‑intensive verification and decompression.

Future Work

Potential extensions include progressive resource loading, allowing low‑priority assets to be fetched asynchronously while sharing the same input stream.