4
\$\begingroup\$

Getting close to a release of generic server.

Nisse Server: Part 1 Helper Functions
Nisse Server: Part 2 Socket Layer

Here is the stream layer code.

This basically provides a std::istream and std::ostream wrapper around a DataSocket. This allowing you to use normal C++ stream operations on a socket.

One notable non-standard part are the two callbacks that are passed to the constructors.

    Notifier                noAvailableData;
    Notifier                flushing;

These are used by higher level code. So that when flushing or no data is available we get a callback to the higher level code. This is what allows the current code to yield back to main loop if there was a writing/read operation that would block.

i.e. If any write/read operation would block we go and find another socket that is currently available and execute its code.

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;

    public:
        virtual ~SocketStreamBuffer() override;
        SocketStreamBuffer(DataSocket& stream, Notifier noAvailableData, Notifier flushing, 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;
};

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

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

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

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

    }
}

#endif

SocketStream.cpp

#include "SocketStream.h"

using namespace ThorsAnvil::Socket;

SocketStreamBuffer::SocketStreamBuffer(DataSocket& stream, Notifier noAvailableData, Notifier flushing, std::vector<char>&& bufData, char const* currentStart, char const* currentEnd)
    : stream(stream)
    , noAvailableData(noAvailableData)
    , flushing(flushing)
    , buffer(std::move(bufData))
{
    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))
{
    move.setg(nullptr, nullptr, nullptr);
    move.setp(nullptr, nullptr);
}

SocketStreamBuffer::~SocketStreamBuffer()
{
    overflow();
}

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.
     */

    while (gptr() == egptr())
    {
        bool    moreData;
        size_t  count;
        std::tie(moreData, count) = stream.getMessageData(&buffer[0], buffer.size());
        if (moreData && count == 0)
        {
            noAvailableData();
        }
        else if (count != 0)
        {
            setg(&buffer[0], &buffer[0], &buffer[count]);
        }
        else if (!moreData)
        {
            break;
        }
    }
    return (gptr() == egptr()) ? 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 movedCharacter    = std::min(count, currentBufferSize);
    std::copy(gptr(), gptr() + movedCharacter, dest);

    dest  += movedCharacter;
    count -= movedCharacter;

    while (count > 0)
    {
        bool    moreData;
        size_t  dataRead;
        std::tie(moreData, dataRead) = stream.getMessageData(dest, count);
        if (moreData && dataRead == 0)
        {
            noAvailableData();
        }
        else if (!moreData)
        {
            break;
        }
        dest  += dataRead;
        count -= dataRead;
        movedCharacter += dataRead;
    }
    return movedCharacter;
}

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 delibrately 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 toWrite = pptr() - pbase();
    std::streamsize written = 0;
    while (toWrite != written)
    {
        bool        moreSpace;
        std::size_t count;
        std::tie(moreSpace, count) = stream.putMessageData(pbase(), pptr() - pbase(), written);
        if (moreSpace && count == 0)
        {
            noAvailableData();
        }
        else if (moreSpace)
        {
            written += count;
        }
        else
        {
            return traits::eof();
        }
    }
    setp(&buffer[0], &buffer[buffer.size() - 1]);
    return written;
}

std::streamsize SocketStreamBuffer::xsputn(char_type const* source, std::streamsize count)
{
    std::streamsize written = 0;
    if (epptr() - pptr() > count)
    {
        // If we have space in the internal buffer then just place it there.
        std::copy(source, source + count, pptr());
        pbump(count);
        written = count;
    }
    else
    {
        // Not enough room in the internal buffer.
        // So write everything to the output stream.
        if (overflow() != traits::eof())
        {
            while (count != written)
            {
                bool        moreSpace;
                std::size_t dataWritten;
                std::tie(moreSpace, dataWritten) = stream.putMessageData(source, count, written);
                if (moreSpace && dataWritten == 0)
                {
                    noAvailableData();
                }
                else if (moreSpace)
                {
                    written += dataWritten;
                }
                else
                {
                    break;
                }
            }
        }
    }
    return written;
}
// ------------------------

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

ISocketStream::ISocketStream(DataSocket& stream, Notifier noAvailableData, Notifier flushing, std::vector<char>&& bufData, char const* currentStart, char const* currentEnd)
    : std::istream(nullptr)
    , buffer(stream, noAvailableData, flushing, 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)
    : std::ostream(nullptr)
    , buffer(stream, noAvailableData, flushing)
{
    rdbuf(&buffer);
}

OSocketStream::OSocketStream(OSocketStream&& move) noexcept
    : std::ostream(nullptr)
    , buffer(std::move(move.buffer))
{
    rdbuf(&buffer);
}
\$\endgroup\$
2
\$\begingroup\$
  1. Looks like Single Entry, Single Exit got you good.

    Not only does that lead to more complicated source-code, but also to possibly slightly bigger and less efficient compiled code.

  2. Try to handle the most likely case first in an if-else if-else-chain.

    It's often marginally more efficient, and in SocketStreamBuffer::underflow() even shorter.

  3. Using control-flow-statements often allows you to reduce nesting.

while (gptr() == egptr())
{
    bool moreData;
    size_t count;
    std::tie(moreData, count) = stream.getMessageData(&buffer[0], buffer.size());
    if (count != 0)
    {
        setg(&buffer[0], &buffer[0], &buffer[count]);
        break;
    }
    else if (!moreData)
    {
        return traits::eof();
    }
    noAvailableData();
}
return traits::to_int_type(*gptr());
  1. In SocketStreamBuffer::xsgetn() you call std::copy() using InputIt, Count, and OutputIt. Why do that contortion instead of using std::copy_n()? At least the compiler should fix the compiled code.

  2. You forgot to modify the get-area in SocketStreamBuffer::xsgetn().

  3. You mishandle !moreData && dataRead by discarding the last read bytes.

  4. If SocketStreamBuffer::overflow() is only ever called when the buffer is full, it works. You also call it in the dtor and the class could be subclassed, so please re-read the requirements for overflow() in the failure-case, for example here: http://en.cppreference.com/w/cpp/io/basic_streambuf/overflow

  5. Your .overflow() is more appropriate for .sync(). And you should call .sync() in the dtor instead.

  6. You should re-read how your DataSocket::putMessageData uses its third argument. You are misusing it, leading to a bug.

  7. You should only get directly into / put directly from a user-supplied buffer in xsgetn() / xsputn() if the free space in it is a very significant fraction of your stream-buffers internal buffer, or you use vectored IO. Unbuffered small-request IO is really quite inefficient.

  8. Why do you only allow setting a custom buffer for ISocketStream but not for OSocketStream?

| improve this answer | |
\$\endgroup\$
  • \$\begingroup\$ Can't figure out this: You mishandle !moreData && dataRead by discarding the last read bytes. \$\endgroup\$ – Martin York Jul 12 '17 at 7:31
  • \$\begingroup\$ I can't quite spot the bug you mention: You should re-read how your DataSocket::putMessageData uses its third argument. You are misusing it, leading to a bug. \$\endgroup\$ – Martin York Jul 12 '17 at 7:41
  • \$\begingroup\$ See xsgetn() for the case !moreData && dataRead. See xsputn() or overflow(), you are adding the combined read amount to the amount read before the call to putMessageData(), so in effect written = 2 * written + written_now. \$\endgroup\$ – Deduplicator Jul 12 '17 at 9:49

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.