63 RpcCaller

We want the RpcCaller to instead use the new messaging approach in cxMessage2. There is no longer a concept of a MultiplexedMsgConnection, or interfaces IMessageWriter and IMessageReader. There isn't even a concept of a MessageId.

Consider that $adt+ RpcCaller also implements the following functions: 

    void Destroy();
    ssize_t WriteMessage(octet_t* payload, ssize_t payloadSizeAvail, ssize_t& payloadSize);
    std::error_code ReadMessage(const octet_t* payload, ssize_t payloadSize);

(i.e. we add these functions to the $adt+ RpcCaller definition)

Note that WriteMessage/ReadMessage are compatible with the same named functions in the ITcpMsgSessionHandler interface.

There is no concept of a MessageId in this interface.

In practise WriteMessage is often passed a buffer of at least a few kilobytes, and this tends to be a lot bigger than the serialised size of an individual method invocation. Therefore it makes sense to allow the RpcCaller to serialise multiple method invocations or flush calls in a single message! This batching effect increases the space efficiency (because each TcpMsg message has a 4 byte header for the message size), and performance.

This design is similar to the previous approach using CEDA_FAST_TCP_MSG_WRITES and makes quite a big difference to the performance of the RMI messages (E.g. one test showed 11MHz versus 6.8MHz - which is a 60% faster).

The previous approach to multiplexing was to try to offset message ids into one range of integers. This makes it very difficult to make the set of message handler objects dynamic, or to allow alternative identifiers, such as strings or oids. E.g. we might use some kind of object id to identify the object receiving the message, and the set of objects which can receive messages changes over time.

The above WriteMessage/ReadMessage functions can possibly represent the serialisation of only a part of some larger message

Infinite recording of the method invocations

A BufferedMessageWriter is being used to infinite buffer all the method invocations. However there are some issues:

  • BufferedMessageWriter uses a MultiplexedMsgConnection which we're not using anymore in cxMessage2
  • BufferedMessageWriter uses a ForwardLinkedPagedBuffer to provide the basis for concurrent reading and writing of a queue of messages. This is great except ForwardLinkedPagedBuffer constantly allocates and frees 4kB pages.
  • There is a mutex, do we need it?
  • There are three queues of message sizes, seems very complex
  • There is a ThrottleProducer
  • Shouldn't we support infinite buffering of all kinds of messages, rather than just one RpcCaller?
  • This breaks the whole idea of separating CPU work from IO work. An RpcCaller should only be concerned with serialisation method invocations to a buffer in memory, and not with IO using sockets.

So it's fundamentally wrong! Instead we should:

  • RpcCaller has no dependency on cxMessage at all.
  • RpcCaller serialises method invocations to/from memory buffers
  • RpcCaller has no mutexes / condition variables / throttling / threads
  • RpcCaller is deterministic, and can be easily tested without needing sockets etc
  • RpcCaller doesn't try to implement infinite buffering

However, we currently support synchronous messages over the wire (i.e. where the thread blocks, waiting for a response). Messaging like this is an antipattern, but we have supported it, so we have to expect to break the above principles.

Proposal:

  • We drop support for synchronous IPC methods. So there are no out parameters of IPC methods. There is no need to block a thread (see ThreadBlocker). There is no need to queue responses (see ResponseQueue). There is no need for RpcCaller methods Mark(), Wait(), MarkAndWait(), ReachedMark(), AbortWaitProducer(), WaitProducer(ssize_t bufferSize)
  • It is assumed the calls on the stub(s) are serialised without need for a mutex in the IPC framework. For example the stubs may already be protected by a CSpace mutex.
  • All the stubs share the same Archive. There is a generic mechanism for writing the id of the object/interface on which the susequent method calls are being made. This only needs to be done when the id changes. The Archive is is typically serialising all the method calls into memory with infinite buffering, and memory pooling of pages to avoid heap allocations
  • Sending the memory buffer down a socket is a separate concern. There is no need for a queue supporting reading at the front concurrently with writing at the back. Instead, we have automatic and manual flushing. It is assumed Flush() is serialised with respect to all the stub calls. Flush() flushes the Archive buffer, then swaps out the PagedBuffer with an empty one. There is no need for a mutex. The effect is to clear the write archive, and to retrieve a buffer of known size for writing to a socket.
  • Similarly, and completely independently we can do all this for deserialising and issuing invocations using a skeleton. We read a buffer from a socket, then deserialise the invocations from the buffer.

RpcCaller.h

// RpcCaller.h
//
// Author David Barrett-Lennard
// (C)opyright Cedanet Pty Ltd 2020

@import "Rpc.h"
@import "Stub.h"
#include "Ceda/cxUtils/PagedBuffer.h"
#include "Ceda/cxUtils/PagedBufferAsArchive.h"
#include <memory>

namespace ceda
{
$adt+ implement RpcCaller final
{
    cxNotCloneable(RpcCaller)

public:
    explicit RpcCaller(Stub* stub);
    $implementing adt RpcCaller;

private:
    std::unique_ptr<Stub> stub_;
    PagedBuffer pb_;
    OutputArchiveOnPagedBuffer ar_;
};

} // namespace ceda

RpcCaller.cpp

// RpcCaller.cpp
//
// Author David Barrett-Lennard
// (C)opyright Cedanet Pty Ltd 2020

@import "RpcCaller.h"

namespace ceda
{
@api RpcCaller* CreateRpcCaller(ConstStringZ qualifiedInterfaceName)
{
    if (Stub* stub = CreateStub(qualifiedInterfaceName))
    {
        return new RpcCaller(stub);
    }
    else
    {
        return nullptr;
    }
}

///////////////////////////////////////////////////////////////////////////////////////////////////
// RpcCaller

RpcCaller::RpcCaller(Stub* stub) :
    stub_(stub),
    ar_(pb_)
{
    cxAssert(stub != nullptr);
    cxAssert(0 <= stub->numMethods);
    stub->m_ar = &ar_;
}

void RpcCaller::Close()
{
    delete this;
}

void RpcCaller::GetBuffer(xvector<octet_t>& buffer)
{
    // It might be good to just swap the PagedBuffer with one provided in this call.
    // That would work even though pb_ is associated with ar_!
    // But in any case we probably want a contiguous buffer anyway

    ar_.Flush();
    pb_.WriteTo(buffer);

    // Note that PagedBuffer keeps the allocated pages when it is cleared so it can shrink and regrow efficiently.
    // For that reason we don't need to implement a memory pool for the pages of the PagedBuffer.
    pb_.Clear();
}

AnyInterface RpcCaller::GetStub()
{
    return stub_->m_stub;
}

} // namespace ceda