using System;

using System.Threading;

public class Program

{

    struct Grain<T> where T : class {

        public T Data;

        public Exception Failure;

    }

    class Channel<T> where T : class {

        public Channel(int bufferSize) { throw new Exception("not implemented"); }

        public bool Send(Grain<T> grain) { throw new Exception("not implemented"); }

        public Grain<T> Receive() { throw new Exception("not implemented"); }

        public void Close() { throw new Exception("not implemented"); }

        public void Wait() /* throws Exception */ { throw new Exception("not implemented"); }

    }

    static void Start<T>(Channel<T> dst, int threadsNumber, Func<Grain<T>> provider) where T : class {

        int threadsLeft = threadsNumber;

        for (var i = 0; i < threadsNumber; ++i) {

            new Thread(() => {

                while (true) {

                    var output = provider();

                    if (output.Data == null && output.Failure == null) {

                        if (Interlocked.Decrement(ref threadsLeft) == 0)

                            dst.Close();

                        break;

                    }

                    if (!dst.Send(output) || output.Failure != null)

                        break;

                }

            }).Start();

        }

    }

    static void Pipe<I, O>(Channel<I> src, Channel<O> dst, int threadsNumber, Func<I, Grain<O>> processor)

        where I : class where O : class {

        int threadsLeft = threadsNumber;

        for (var i = 0; i < threadsNumber; ++i) {

            new Thread(() => {

                while (true) {

                    var input = src.Receive();

                    if (input.Failure != null) {

                        dst.Send(new Grain<O>() {Data = null, Failure = input.Failure});

                    } else if (input.Data == null) {

                        if (Interlocked.Decrement(ref threadsLeft) == 0)

                            dst.Close();

                        break;

                    } else {

                        var output = processor(input.Data);

                        if (!dst.Send(output) || output.Failure != null) {

                            src.Close();

                            break;

                        }

                    }

                }

            }).Start();

        }

    }

    static void End<T>(Channel<T> src, int threadsNumber, Action<T> consumer) where T : class {

        int threadsLeft = threadsNumber;

        for (var i = 0; i < threadsNumber; ++i) {

            new Thread(() => {

                while (true) {

                    var input = src.Receive();

                    if (input.Data != null)

                        consumer(input.Data);

                    else

                        break;

                }

            }).Start();

        }

    }

    interface Bytes {}

    public static void Main()

    {

        var compressors = 5; // use CPUs count

        // Channels.

        var reader2compressors = new Channel<Bytes>(compressors * 2);

        var compressors2writer = new Channel<Bytes>(compressors * 2);

        // Functions.

        Func<Grain<Bytes>> reader = () => {

            // Read next chunk from input stream.

            throw new Exception("not implemented");

        };

        Func<Bytes, Grain<Bytes>> compressor = (input) => {

            // Compress the chunk.

            throw new Exception("not implemented");

        };

        Action<Bytes> writer = (intput) => {

            // Write the chunk into the output stream.

            throw new Exception("not implemented");

        };

        // Write file header here.

        // Pipeline.

        Start(reader2compressors, 1 /* reader */, reader);

        Pipe(reader2compressors, compressors2writer, compressors, compressor);

        End(compressors2writer, 1 /* writer */, writer);

        // Wait for completion.

        compressors2writer.Wait();

        // Write file footer here.

    }

}

​