4
\$\begingroup\$

The first code review of this code can be found here:

Nisse Server: Part 3 Stream Layer

After the bugs and points made by @Deduplicator (thank you) the code has changed significantly enough that a subsequent review is needed. Hopefully more people can help me out by providing input.

SocketStream.h

#ifndef THORSANVIL_SIMPLE_STREAM_THOR_STREAM_H
#define THORSANVIL_SIMPLE_STREAM_THOR_STREAM_H

#include "Socket.h"
#include <istream>
#include <vector>

namespace ThorsAnvil
{
    namespace Socket
    {

using Notifier = std::function<void()>;

class SocketStreamBuffer: public std::streambuf
{
    private:
        typedef std::streambuf::traits_type traits;
        typedef traits::int_type            int_type;
        typedef traits::char_type           char_type;

        DataSocket&             stream;
        Notifier                noAvailableData;
        Notifier                flushing;
        std::vector<char>       buffer;
        bool                    closeSocketOnDestruction;

    public:
        virtual ~SocketStreamBuffer() override;
        SocketStreamBuffer(DataSocket& stream,
                           Notifier noAvailableData, Notifier flushing, bool closeSocketOnDestruction = true,
                           std::vector<char>&& bufData = std::vector<char>(4000),
                           char const* currentStart = nullptr, char const* currentEnd = nullptr);
        SocketStreamBuffer(SocketStreamBuffer&& move) noexcept;

    protected:
        virtual int_type        underflow() override;
        virtual std::streamsize xsgetn(char_type* s, std::streamsize count) override;

        virtual int_type        overflow(int_type ch = traits::eof()) override;
        virtual std::streamsize xsputn(char_type const* s,std::streamsize count) override;
    private:
        std::streamsize writeToStream(char_type const* source, std::streamsize count);
        std::streamsize readFromStream(char_type* dest, std::streamsize count, bool fill = true);
};

class ISocketStream: public std::istream
{
    SocketStreamBuffer buffer;

    public:
        ISocketStream(DataSocket& stream,
                      Notifier noAvailableData, Notifier flushing, bool closeSocketOnDestruction,
                      std::vector<char>&& bufData, char const* currentStart, char const* currentEnd);
        ISocketStream(DataSocket& stream,
                      Notifier noAvailableData, Notifier flushing, bool closeSocketOnDestruction = true);
        ISocketStream(ISocketStream&& move) noexcept;
};

class OSocketStream: public std::ostream
{
    SocketStreamBuffer buffer;

    public:
        OSocketStream(DataSocket& stream,
                      Notifier noAvailableData, Notifier flushing, bool closeSocketOnDestruction = true);
        OSocketStream(OSocketStream&& move) noexcept;
};

    }
}

SocketStream.cpp (constructors/destructors etc)

#include "SocketStream.h"

using namespace ThorsAnvil::Socket;

SocketStreamBuffer::SocketStreamBuffer(DataSocket& stream,
                                       Notifier noAvailableData, Notifier flushing, bool closeSocketOnDestruction,
                                       std::vector<char>&& bufData, char const* currentStart, char const* currentEnd)
    : stream(stream)
    , noAvailableData(noAvailableData)
    , flushing(flushing)
    , buffer(std::move(bufData))
    , closeSocketOnDestruction(closeSocketOnDestruction)
{
    char* newStart = const_cast<char*>(currentStart);
    char* newEnd   = const_cast<char*>(currentEnd);
    if (newStart == nullptr || newEnd == nullptr)
    {
        newStart = &buffer[0];
        newEnd   = &buffer[0];
    }

    setg(&buffer[0], newStart, newEnd);
    setp(&buffer[0], &buffer[buffer.size() - 1]);
}

SocketStreamBuffer::SocketStreamBuffer(SocketStreamBuffer&& move) noexcept
    : stream(move.stream)
    , noAvailableData(move.noAvailableData)
    , buffer(std::move(move.buffer))
    , closeSocketOnDestruction(move.closeSocketOnDestruction)
{
    move.setg(nullptr, nullptr, nullptr);
    move.setp(nullptr, nullptr);
}

SocketStreamBuffer::~SocketStreamBuffer()
{
    // Force the buffer to be output to the socket
    overflow();
    if (closeSocketOnDestruction)
    {
        stream.putMessageClose();
    }
}

SocketStream.cpp (Input)

SocketStreamBuffer::int_type SocketStreamBuffer::underflow()
{
    /*
     * Ensures that at least one character is available in the input area by updating the pointers
     * to the input area (if needed) * and reading more data in from the input sequence
     * (if applicable).
     *
     * Returns the value of that character (converted to int_type with Traits::to_int_type(c)) on success
     * or Traits::eof() on failure.
     *
     * The function may update gptr, egptr and eback pointers to define the location of newly
     * loaded data (if any).
     *
     * On failure, the function ensures that either gptr() == nullptr or gptr() == egptr.
     * The base class version of the function does nothing. The derived classes may override this function
     * to allow updates to the get area in the case of exhaustion.
     */    

    std::streamsize retrievedData = readFromStream(&buffer[0], buffer.size(), false);
    setg(&buffer[0], &buffer[0], &buffer[retrievedData]);
    return (retrievedData == 0) ? traits::eof() : traits::to_int_type(*gptr());
}

std::streamsize SocketStreamBuffer::xsgetn(char_type* dest, std::streamsize count)
{
    /*
     * Reads count characters from the input sequence and stores them into a character array pointed to by s.
     * The characters are read as if by repeated calls to sbumpc().
     * That is, if less than count characters are immediately available, the function calls uflow() to
     * provide more until traits::eof() is returned. Classes derived from std::basic_streambuf are permitted
     * to provide more efficient implementations of this function.
     */    
    std::streamsize currentBufferSize = egptr() - gptr();
    std::streamsize nextChunkSize    = std::min(count, currentBufferSize);
    std::copy_n(gptr(), nextChunkSize, dest);
    gbump(nextChunkSize);

    std::streamsize       retrieved  = nextChunkSize;
    std::streamsize const bufferSize = static_cast<std::streamsize>(buffer.size());

    while (retrieved != count)
    {
        nextChunkSize    = std::min((count - retrieved), bufferSize);

        // A significant chunk
        if (nextChunkSize > (bufferSize / 2))
        {
            std::streamsize read = readFromStream(dest + retrieved, count - retrieved);
            retrieved += read;
        }
        else
        {
            underflow();
            nextChunkSize    = std::min(nextChunkSize, egptr() - gptr());
            std::copy_n(gptr(), nextChunkSize, dest);
            gbump(nextChunkSize);
            retrieved += nextChunkSize;
        }
    }
    return retrieved;
}

std::streamsize SocketStreamBuffer::readFromStream(char_type* dest, std::streamsize count, bool fill)
{
    std::streamsize read = 0;
    while (read != count)
    {
        bool    moreData;
        size_t  dataRead;
        std::tie(moreData, dataRead) = stream.getMessageData(dest, count, read);
        if (dataRead != 0)
        {
            read += dataRead;
            if (!fill)
            {
                break;
            }
        }
        else if (moreData)
        {
            noAvailableData();
        }
        else
        {
            break;
        }
    }
    return read;
}  

SocketStream.cpp (Output)

SocketStreamBuffer::int_type SocketStreamBuffer::overflow(int_type ch)
{
    /*
     * Ensures that there is space at the put area for at least one character by saving some initial subsequence of
     * characters starting at pbase() to the output sequence and updating the pointers to the put area (if needed).
     * If ch is not Traits::eof() (i.e. Traits::eq_int_type(ch, Traits::eof()) != true),
     *     it is either put to the put area or directly saved to the output sequence.
     * The function may update pptr, epptr and pbase pointers to define the location to write more data.
     * On failure, the function ensures that either pptr() == nullptr or pptr() == epptr.
     * The base class version of the function does nothing. The derived classes may override this function to allow
     * updates to the put area in the case of exhaustion.
     */

    if (ch != traits::eof())
    {
        /* Note: When we set the "put" pointers we deliberately leave an extra space that is not buffer.
         * When overflow is called the normal buffer is used up, but there is an extra space in the real
         * underlying buffer that we can use.
         *
         * So: *pptr = ch; // will never fail.
         */
        *pptr() = ch;
        pbump(1);
    }

    flushing();
    std::streamsize written = writeToStream(pbase(), pptr() - pbase());
    if (written != (pptr() - pbase()))
    {
        setp(&buffer[0], &buffer[0]);
    }
    else
    {
        setp(&buffer[0], &buffer[buffer.size() - 1]);
    }
    return written;
}

std::streamsize SocketStreamBuffer::xsputn(char_type const* source, std::streamsize count)
{
    /*
     * Writes count characters to the output sequence from the character array whose first element is pointed to by s.
     * The characters are written as if by repeated calls to sputc().
     * Writing stops when either count characters are written or a call to sputc() would have returned Traits::eof().
     * If the put area becomes full (pptr() == epptr()), this function may call overflow(),
     * or achieve the effect of calling overflow() by some other, unspecified, means.
     */
    std::streamsize spaceInBuffer = epptr() - pptr();
    if (spaceInBuffer > count)
    {
        // If we have space in the internal buffer then just place it there.
        // We want a lot of little writtes to be buffered so we only talk to the stream
        // chunks of a resonable size.
        std::copy_n(source, count, pptr());
        pbump(count);
        return count;
    }

    // Not enough room in the internal buffer.
    // So write everything to the output stream.
    if (overflow() == traits::eof())
    {
        return 0;
    }
    std::streamsize       exported   = 0;
    std::streamsize const bufferSize = static_cast<std::streamsize>(buffer.size());
    while (exported != count)
    {
        std::streamsize nextChunk = count - exported;
        if (nextChunk > (bufferSize / 2))
        {
            std::streamsize written = writeToStream(source, nextChunk);
            exported += written;
        }
        else
        {
            std::copy_n(source + exported, nextChunk, pptr());
            pbump(nextChunk);
            exported += nextChunk;
        }
    }
    return exported;
}

std::streamsize SocketStreamBuffer::writeToStream(char_type const* source, std::streamsize count)
{
    std::streamsize written = 0;
    while (written != count)
    {
        bool        moreSpace;
        std::size_t dataWritten;
        std::tie(moreSpace, dataWritten) = stream.putMessageData(source, count, written);
        if (dataWritten != 0)
        {
            written += dataWritten;
        }
        else if (moreSpace)
        {
            noAvailableData();
        }
        else
        {
            break;
        }
    }
    return written;
}

SocketStream.cpp (Stream Wrappers)

// ------------------------

ISocketStream::ISocketStream(DataSocket& stream,
                             Notifier noAvailableData, Notifier flushing, bool closeSocketOnDestruction)
    : std::istream(nullptr)
    , buffer(stream, noAvailableData, flushing, closeSocketOnDestruction)
{
    std::istream::rdbuf(&buffer);
}

ISocketStream::ISocketStream(DataSocket& stream,
                             Notifier noAvailableData, Notifier flushing, bool closeSocketOnDestruction,
                             std::vector<char>&& bufData, char const* currentStart, char const* currentEnd)
    : std::istream(nullptr)
    , buffer(stream,
             noAvailableData, flushing, closeSocketOnDestruction,
             std::move(bufData), currentStart, currentEnd)
{
    rdbuf(&buffer);
}

ISocketStream::ISocketStream(ISocketStream&& move) noexcept
    : std::istream(nullptr)
    , buffer(std::move(move.buffer))
{
    rdbuf(&buffer);
}

// ------------------------

OSocketStream::OSocketStream(DataSocket& stream,
                             Notifier noAvailableData, Notifier flushing, bool closeSocketOnDestruction)
    : std::ostream(nullptr)
    , buffer(stream,
             noAvailableData, flushing, closeSocketOnDestruction)
{
    rdbuf(&buffer);
}

OSocketStream::OSocketStream(OSocketStream&& move) noexcept
    : std::ostream(nullptr)
    , buffer(std::move(move.buffer))
{
    rdbuf(&buffer);
}
\$\endgroup\$
  • 1
    \$\begingroup\$ Writing it here and going through it I already found two bugs :-( Good news I have two more unit tests. :-) \$\endgroup\$ – Martin York Jul 14 '17 at 5:11
5
\$\begingroup\$
  1. SocketStreamBuffer is really an input-streambuffer and an output-streambuffer which use the same members in a mutually incompatible way. It works if any instance is consistently only used in one of the two modes, though it's not that clean, though it safes a modest amount of source and compiled code.

  2. std::function<void()> should be moved instead of copied whenever you can. It's potentially significantly more efficient and less likely to throw.

  3. There's a reason for having both .overflow() and .sync(). The first just makes space in the buffer, a partial write is enough for that, the other flushes the whole buffer, and has to complete partial writes by looping.
    Aside from fixing the semantics of the one you implement, you should implement the other and fix the dtor, as well as re-consider all other call-sites.

  4. Your streambuffer's move-constructor doesn't seem to work. If you want to make it work, replace the member-reference with a non-owning pointer, and make sure to transfer the information about the current buffer-state too.

  5. .xsgetn() must only loop if and only if .underflow() was called and didn't return an error.

  6. The fact that you do a full flush where you only need a partial flush has far-reaching consequences for the design and implementation of .xsputn(). It makes it a bit simpler, but less efficient and more prone to blocking.

| improve this answer | |
\$\endgroup\$
0
\$\begingroup\$

After adding a couple of unit tests I found some more issues:

The destructor of SocketStreamBuffer calls methods that can potentially throw. Since destructors are noexcept(true) by default this will cause a catastrophic failure. I decided that it should act like other streams. The destructor swallows any exceptions.

| improve this answer | |
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.