Implementing a pipe within C/C++ program

Question

I have the following problem:

I have written a simple unzipper, which unzippes .gz files which are, per line (/n), processed by another process. So, in shell, I can type:

unzipper file | program

Both programs are C/C++ coded.

Does anybody know if and how I can implement this 'pipe' (|) within one C/C++ program, so that I can make something like this multi-threaded for instance...

In my particular case, it is important to keep the new-line structure intact, that's why I'm using a pipe. The gz files are too big to keep in memory as a whole.

Answer 1

In programming in general, there is something called generators; in C++ we tend to think of them as input iterators however the concern remains identical: much like a pipe, it's about pull-driven production.

As such, you can restructure your program around the idea of a Producer (preferably with the interface of an input iterator) and a Consumer, and the Consumer will ask for input one line at the time, that the Producer will lazily come up with.

For a good guide on the necessary interface I recommend the venerable SGI STL website: here it is for the InputIterator concept.

For a simpler example, let's suppose we don't have to deal with unzipping and just read a file on a line-per-line basis:

class LineIterator: public std::iterator<std::input_iterator_tag,
                                         std::string const>
{
public:
    // Default Constructible
    LineIterator(): stream(nullptr) {}

    explicit LineIterator(std::istream& is): stream(&is) { this->advance(); }

    // Equality Comparable
    friend bool operator==(LineIterator const& left, LineIterator const& right) {
        return left.stream == right.stream
           and left.buffer == right.buffer
           and left.currentLine == right.currentLine;
    }

    friend bool operator!=(LineIterator const& left, LineIterator const& right) {
        return not (left == right);
    }

    // Trivial Iterator (non mutable)
    pointer operator->() const { return &currentLine; }

    reference operator*() const { return currentLine; }

    // Input Iterator
    LineIterator& operator++() {
        this->advance();
        return *this;
    } // operator++

    LineIterator operator++(int) {
        LineIterator tmp(*this);
        ++*this;
        return tmp;
    } // operator++

private:
    void advance() {
        // Advance a valid iterator to fetch the next line from the source stream.
        static LineIterator const SingularValue;

        assert(*this != SingularValue and "Cannot advance singular iterator");
        // Note: in real life, I would use std::getline...
        // ... but it would not showcase the double-buffering model
        // required to solve the OP problem (because of decoding)

        // We use double-buffering, so clear current and swap buffers
        currentLine.clear();
        swap(buffer, currentLine);

        // Check if we found some new line or not
        size_t const nl = currentLine.find('\n');

        // If we found one already, preserve what's after in the buffer
        // as we only want to expose one line worth of material.
        if (nl != std::string::npos) {
            if (nl == currentLine.size()) { return; } // nothing to preserve

            buffer.assign(currentLine.begin() + nl + 1, currentLine.end());
            currentLine.erase(currentLine.begin() + nl + 1, currentLine.end());
            return;
        }

        // If we did not, then we need to pump more data into the buffer.
        if (not stream) { return; } // Nothing to pump...

        static size_t const ReadBufferSize = 256;
        char input[ReadBufferSize];

        while (stream->read(input, ReadBufferSize)) {
            if (this->splitBuffer(input, ReadBufferSize)) { break; }
        }

        // We end up here either if we found a new line or if some read failed.
        // If the stream is still good, we successfully found a new line!
        if (*stream) { return; }

        // Otherwise, the stream is no good any longer (it dried up!)
        // but we may still have read some little things from it.
        this->splitBuffer(input, stream->gcount());

        stream = SingularValue.stream; // stream dried up,
                                       // so reset it to match singular value.
    } // advance

    bool splitBuffer(char const* input, size_t const size) {
        // Split input at the newline character, the first chunk ends
        // up in currentLine, the second chunk in buffer.
        // Returns true if a newline character was found, false otherwise.

        // Check if we finally found a new line
        char const* const newLine = std::find(input, input + size, '\n');

        // If we did not, copy everything into currentLine and signal it.
        if (newLine == input + size) {
            currentLine.append(input, size);
            return false;
        }

        // If we did, copy everything up to it (including it) into currentLine
        // and then bufferize the rest for the next iteration.
        currentLine.append(input, newLine + 1);
        buffer.assign(newLine + 1, input + size);
        return true;
    } // splitBuffer

    std::istream* stream;
    std::string buffer;

    std::string currentLine;
}; // class LineIterator

It's a bit of a mouthful (and is probably buggy...) still, it has the interface we need to compose it with STL algorithms, such as:

std::ifstream file("someFile.txt");
std::copy(LineIterator(file), LineIterator(), std::ostream_iterator(std::cout));

which will echo the file on the terminal one line at a time (demo here).

Now, all you have to do is replace the fetch portion (stream.read) to a block by block read & unzip :)