BufferedIPCChannel Class Reference

Socket-based implementation of IPCChannel that uses intermediate buffer for data exchange between client and server. More...

#include <BufferedIPCChannel.h>

Inheritance diagram for BufferedIPCChannel:

Collaboration diagram for BufferedIPCChannel:

Public Member Functions
	BufferedIPCChannel ()
	~BufferedIPCChannel () override
	Destroys channel and stops any data exchange. More...
void	Send (const void *bytes, size_t length) override
	Thread-safe. More...
void	StartConversation (SOCKET socket, IPCChannelStatusCallback &callback)
	Allowed to be called only once during object lifetime. Takes ownership over a socket. Callback should be guaranteed to be alive between IPCChannelStatusCallback::OnConnect and IPCChannelStatusCallback::OnDisconnect, and will be accessed from multiple threads. More...
Public Member Functions inherited from IPCChannel
virtual	~IPCChannel ()
virtual void	Send (const void *bytes, size_t length)=0
	Write data to the channel. More...

Private Attributes
bool	mAlive {true}
std::mutex	mSocketSync
std::condition_variable	mSendCondition
std::unique_ptr< std::thread >	mRecvRoutine
std::unique_ptr< std::thread >	mSendRoutine
SOCKET	mSocket {INVALID_SOCKET}
std::vector< char >	mOutputBuffer

Static Private Attributes
static constexpr int	DefaultOutputBufferCapacity { 2048 }
static constexpr int	DefaultInputBufferSize { 2048 }

Detailed Description

Socket-based implementation of IPCChannel that uses intermediate buffer for data exchange between client and server.

Definition at line 30 of file BufferedIPCChannel.h.

Constructor & Destructor Documentation

BufferedIPCChannel()

BufferedIPCChannel::BufferedIPCChannel

(

)

Definition at line 21 of file BufferedIPCChannel.cpp.

{
   mOutputBuffer.reserve(DefaultOutputBufferCapacity);
}

References DefaultOutputBufferCapacity, and mOutputBuffer.

~BufferedIPCChannel()

BufferedIPCChannel::~BufferedIPCChannel ( )

override

Destroys channel and stops any data exchange.

Definition at line 26 of file BufferedIPCChannel.cpp.

{
   if(mSocket != INVALID_SOCKET)
   {
      //Shut down connection and wake up select
      //No need to check possible error codes set by this call
#ifdef _WIN32
      shutdown(mSocket, SD_BOTH);
#else
      shutdown(mSocket, SHUT_RDWR);
#endif
      //Make sure all socket IO operations complete before close
      if(mSendRoutine)
         mSendRoutine->join();
      if(mRecvRoutine)
         mRecvRoutine->join();
 
      CLOSE_SOCKET(mSocket);
   }
}

References CLOSE_SOCKET, INVALID_SOCKET, mRecvRoutine, mSendRoutine, and mSocket.

Member Function Documentation

Send()

void BufferedIPCChannel::Send ( const void *  bytes, size_t  length  )

overridevirtual

Thread-safe.

Implements IPCChannel.

Definition at line 47 of file BufferedIPCChannel.cpp.

{
   assert(length > 0);
   if(length == 0)
      return;
 
   {
      std::lock_guard lck(mSocketSync);
 
      auto offset = mOutputBuffer.size();
      mOutputBuffer.resize(offset + length);
      std::memcpy(mOutputBuffer.data() + offset, bytes, length);
   }
   mSendCondition.notify_one();
}

References mOutputBuffer, mSendCondition, and mSocketSync.

StartConversation()

void BufferedIPCChannel::StartConversation	(	SOCKET	socket,
		IPCChannelStatusCallback &	callback
	)

Allowed to be called only once during object lifetime. Takes ownership over a socket. Callback should be guaranteed to be alive between IPCChannelStatusCallback::OnConnect and IPCChannelStatusCallback::OnDisconnect, and will be accessed from multiple threads.

Parameters

socket	A valid socket on which data exchange should happen
callback	Used to send status updates

Definition at line 63 of file BufferedIPCChannel.cpp.

{
   assert(socket != INVALID_SOCKET);
   assert(mSocket == INVALID_SOCKET && !mSendRoutine && !mRecvRoutine);
   mSocket = socket;
 
   //create "sending" thread first, it should be blocked
   //on condition until both IPCChannelStatusCallback::OnConnect
   //and IPCChannel::Send are called (in that sequence)
   mSendRoutine = std::make_unique<std::thread>([this]
   {
      std::vector<char> secondaryOutputBuffer;
      secondaryOutputBuffer.reserve(DefaultOutputBufferCapacity);
      
      while(true)
      {
         std::unique_lock lck(mSocketSync);
         mSendCondition.wait(lck, [this]{ return !mAlive || !mOutputBuffer.empty(); });
 
         if(!mAlive)
            return;
         
         std::swap(secondaryOutputBuffer, mOutputBuffer);
         mOutputBuffer.clear();
 
         lck.unlock();
 
         {
            int offset = 0;
            while(offset < static_cast<int>(secondaryOutputBuffer.size()))
            {
               fd_set writefds, exceptfds;
               FD_ZERO(&writefds);
               FD_ZERO(&exceptfds);
               FD_SET(mSocket, &writefds);
               FD_SET(mSocket, &exceptfds);
               auto ret = select(NFDS(mSocket), nullptr, &writefds, &exceptfds, nullptr);
               if(ret == 1)
               {
                  //try to send data...
                  ret = send(
                     mSocket,
                     secondaryOutputBuffer.data() + offset,
                     static_cast<int>(secondaryOutputBuffer.size()) - offset,
                     0);
                  if(ret > 0)
                     offset += ret;
                  else if (ret == SOCKET_ERROR)
                     break;//error
               }
               else
                  break;//error
            }
         }
      }
   });
 
   mRecvRoutine = std::make_unique<std::thread>([this, &callback]{
      //such order guarantees that IPCStatusCallback::OnConnect will be called
      //only if both routines have started successfully
      callback.OnConnect(*this);
 
      auto terminate = finally([this, &callback]
      {
         {
            //Let "sending" thread know that we're done
            std::lock_guard lck(mSocketSync);
            mAlive = false;
         }
         mSendCondition.notify_one();
 
         //It may happen so, that "sending" thread sends some data while
         //"reading" thread notifies callback about disconnection. It's
         //not a big deal since a) sending may fail b) from the user code
         //perspective IPCChannel::Send was called before it receives OnDisconnect
         callback.OnDisconnect();
      });
 
      std::vector<char> buffer(DefaultInputBufferSize);
      while(true)
      {
         fd_set readfds, exceptfds;
         FD_ZERO(&readfds);
         FD_ZERO(&exceptfds);
         FD_SET(mSocket, &readfds);
         FD_SET(mSocket, &exceptfds);
 
         auto ret = select(NFDS(mSocket), &readfds, nullptr, &exceptfds, nullptr);
         if(ret == 1)
         {
            //try fetch data...
            ret = recv(
               mSocket,
               buffer.data(),
               static_cast<int>(buffer.size()),
               0);
            if(ret > 0)
               //success
               callback.OnDataAvailable(buffer.data(), ret);
            else if (ret == SOCKET_ERROR)
            {
#ifdef _WIN32
               auto err = WSAGetLastError();
               if(err != WSAEWOULDBLOCK && err != EAGAIN)
                  break;
#else
               if(errno != EWOULDBLOCK && errno != EAGAIN)
                  break;
#endif
            }
            else
               break;//closed by remote(0)
         }
         else//SOCKET_ERROR
            break;
      }
      
   });
}

References DefaultInputBufferSize, DefaultOutputBufferCapacity, INVALID_SOCKET, mAlive, mOutputBuffer, mRecvRoutine, mSendCondition, mSendRoutine, mSocket, mSocketSync, NFDS, IPCChannelStatusCallback::OnConnect(), IPCChannelStatusCallback::OnDataAvailable(), IPCChannelStatusCallback::OnDisconnect(), SOCKET_ERROR, and anonymous_namespace{NoteTrack.cpp}::swap().

Here is the call graph for this function:

Member Data Documentation

DefaultInputBufferSize

constexpr int BufferedIPCChannel::DefaultInputBufferSize { 2048 }

staticconstexprprivate

Definition at line 33 of file BufferedIPCChannel.h.

Referenced by StartConversation().

DefaultOutputBufferCapacity

constexpr int BufferedIPCChannel::DefaultOutputBufferCapacity { 2048 }

staticconstexprprivate

Definition at line 32 of file BufferedIPCChannel.h.

Referenced by BufferedIPCChannel(), and StartConversation().

mAlive

bool BufferedIPCChannel::mAlive {true}

private

Definition at line 35 of file BufferedIPCChannel.h.

Referenced by StartConversation().

mOutputBuffer

std::vector<char> BufferedIPCChannel::mOutputBuffer

private

Definition at line 44 of file BufferedIPCChannel.h.

Referenced by BufferedIPCChannel(), Send(), and StartConversation().

mRecvRoutine

std::unique_ptr<std::thread> BufferedIPCChannel::mRecvRoutine

private

Definition at line 39 of file BufferedIPCChannel.h.

Referenced by StartConversation(), and ~BufferedIPCChannel().

mSendCondition

std::condition_variable BufferedIPCChannel::mSendCondition

private

Definition at line 37 of file BufferedIPCChannel.h.

Referenced by Send(), and StartConversation().

mSendRoutine

std::unique_ptr<std::thread> BufferedIPCChannel::mSendRoutine

private

Definition at line 40 of file BufferedIPCChannel.h.

Referenced by StartConversation(), and ~BufferedIPCChannel().

mSocket

SOCKET BufferedIPCChannel::mSocket {INVALID_SOCKET}

private

Definition at line 42 of file BufferedIPCChannel.h.

Referenced by StartConversation(), and ~BufferedIPCChannel().

mSocketSync

std::mutex BufferedIPCChannel::mSocketSync

private

Definition at line 36 of file BufferedIPCChannel.h.

Referenced by Send(), and StartConversation().

---

The documentation for this class was generated from the following files:

• BufferedIPCChannel.h
• BufferedIPCChannel.cpp