61 cxRpc

Recording and playing invocations of interface methods

RPC stands for Remote Procedure Call.

The cxRpc library allows for recording the invocations of the methods of an interface implemented by a "proxy" object called a stub, and for subsequently "playing" those recorded invocations on some other object that implements that same interface.

The invocations are recorded in a buffer in memory, using an efficient binary representation. This representation can for example be persisted on a hard-drive, to be played back later. It can be played back any number of times on objects that implements the interface. The buffer can be sent over the wire (e.g. using TCP/IP) and played on a remote object. So it provides a basis for remote procedure call.

Only interface methods that return void and have no out parameters can have their invocations recorded in this manner.

The invocations are asynchronous - in the sense that the calling thread doesn't block waiting for the call to be made on the remote object.

These invocations may for example represent events.

In the Pascal programming language, there is a distinction made between procedures and functions. A procedure has no return value. In that sense this library only supports remote invocation of interface methods which represent procedures.

Performance

The performance is very good, invocations can be both recorded and played at rates of a few hundred million per second

Stub/skeleton

                           .
                           .
    +---------+         messages           +------------+          +----------+
    |  Stub   |  ----------->-----------   |  Skeleton  |----->----| Receiver |
    +---------+            .               +------------+          +----------+
                           .
    RpcCaller              .                 RpcCallee
                           .

A 'stub' acts as a local proxy of the remote object, enabling clients to make calls on the remote object (the 'receiver') as if it were local.

A 'skeleton' is responsible for translating incoming messages to calls on the receiver object.

Thread safety

The RpcCaller is not thread-safe. It is assumed the calls on the stub(s) are already serialised without need for a mutex in the RpcCaller. For example the stub may already be protected by a CSpace mutex.

Synchronous methods are not supported

Many middleware frameworks support synchronous invocations over the wire (i.e. where the calling thread blocks, waiting for a response).

However messaging like this is an antipattern. See http://www.cedanet.com.au/antipatterns/location-transparency.php

This is mainly because:

network latency can easily make a synchronous function call take millions (perhaps billions) of times longer than an in-process call

networks are fallible, so synchronous calls are fallible. The need to deal with errors on every function call greatly increases complexity for the application programmer

Supporting synchronous messages significantly complicates the middleware layer:

Threads must block until responses arrive
Out parameters on IPC methods must be supported.
Responses need to be queued so that can be passed to the right invocation

There is an older CEDA core library named cxRmi which supports synchronous messages, but is accordingly much more complex than cxRpc. It also has much poorer performance - even for asynchronous messages.

Single stub

At the moment only a single stub is supported. This eliminates the need to send an object and interface identifier as part of the serialised state. Instead each invocation only requires a method index and the in-parameters.

Multiple stubs

[todo]

All the stubs write to the same Archive. The Archive is used to serialise all the method calls into memory with infinite buffering. We need memory pooling of pages to avoid heap allocations

There is a generic mechanism for writing the id of the object/interface on which the subsequent method calls are being made. This only needs to be done when the id changes.

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.

Use of TcpMsg framework

Sending the memory buffer down a socket is a separate concern.

A set of recorded procedure calls can be sent as a single message using a TcpMsgSession.

TcpMsgSession is implemented in cxMessage2.dll (see Ceda/cxMessage2/TcpMsg.h) and represents an implementation of message queues on top of TCP. The message queue supports ordered delivery of messages. i.e. messages are received in the same order that messages are sent.

The message queue is transient. i.e. there is no sense in which the message queue itself persists. However within a transient session the framework ensures ordered delivery without missing messages or duplicates. This arises from the use of TCP/IP.

Interface to define a message protocol

Consider the following interface used to define a message protocol.


@import "Ceda/cxRpc/Rpc.h"

$interface+ rmi Ix
{
    void PrintInt([in]int32 x);  // Print the value of x
};

Note the keyword 'rmi' which causes xcpp to generate stub/skeleton code for the interface. The interface should not subtype IObject. Each parameter must support serialisation to an Archive or from an InputArchive.

[in] designates an in-parameter.

PrintInt() has no out-parameters and therefore is referred to as an asynchronous method. This means there will be no response message (not even an acknowledge) paired with the request message.

Supporting object identifiers in the protocol

The protocol allows for sending some kind of object identifier to identify the recipient of the procedure call.

Overview

On the stub side the following steps are used

Create a MultiplexedMsgConnection
Call CreateRpcCaller() to create and bind a stub to the MultiplexedMsgConnection for a given interface (identified by its ReflectedInterface).
Make calls through the stub as required.

On the skeleton side the following steps are used

Create a MultiplexedMsgConnection
Create an object that implements the rmi interface, that will receive incoming messages
Call CreateRpcCallee() to create and bind a skeleton to the MultiplexedMsgConnection for the given interface (identified by its qualified name), and which will translating incoming messages to calls on the receiver object
Make calls through the stub as required.

Note that applications can use a given MultiplexedMsgConnection for multiple and distinct purposes. For example, it is possible to bind 3 stubs and 2 skeletons to the one MultiplexedMsgConnection.

Thread safety

The stub is not threadsafe. Clients must ensure that all calls through the stub are serialised, as well as calls to GetBuffer() on the RpcCaller.

Infinite Write Buffering

The stub allows for infinite write buffering.

Memory management

The implementation avoids any association with CSpaces and visiting of IObjects. The stub is automatically deleted when the RpcCaller is closed. The skeleton is automatically deleted when the RpcCallee is closed. The client must ensure the receiver object outlives the RpcCallee.

Avoiding memory allocations

If we use a PagedBuffer, then we might need to make it keep the pages it has allocated previously when it is cleared. That's rather like a memory pool - it only needs to allocate memory when memory usage increases.

Example usage

Caller side


typedef xvector<octet_t> ObjectInterfaceId;

RpcCaller* c = CreateRpcCaller();

ObjectInterfaceId id1 = GetObjectInterfaceId1();
ObjectInterfaceId id2 = GetObjectInterfaceId2();
ObjectInterfaceId id3 = GetObjectInterfaceId3();

// These stubs exist for the life of the RpcCaller
ptr<Ix> stub1 = GetStub<Ix>(c, id1);
ptr<Ix> stub2 = GetStub<Ix>(c, id2);
ptr<Iy> stub3 = GetStub<Iy>(c, id3);

while(!finished)
{
    // Make calls on stubs.  These invocations will be recorded in pb
    // The recording is infallible except for running out of memory.  There is no I/O
    stub1->f(1,2,"x");
    stub1->g();
    stub2->f(3,4,"y");
    for (int i=0 ; i < 10 ; ++i)
        stub3->h(i,20);

    // Flush the archive, get the result (perhaps an alias to memory managed by the RpcCaller to avoid memcpy)
    // and clear
    // Note how this breaks up the recorded invocations into a stream of coarser level messages
    auto buffer = GetBuffer(c);

    socket.send(buffer);
}

Close(c);

Callee side


ObjectInterfaceId id1 = ...;
ObjectInterfaceId id2 = ...;
ObjectInterfaceId id3 = ...;

// We have some objects that will receive the messages
ptr<Ix> obj1 = GetObject1();
ptr<Ix> obj2 = GetObject2();
ptr<Iy> obj3 = GetObject3();

RpcCallee* c = CreateRpcCallee(
    [](const ObjectInterfaceId& id) -> AnyInterface
    {
        switch(id)
        {
        case id1: return obj1;
        case id2: return obj2;
        case id3: return obj3;
        default : return nullptr;       // not found
        }
    });

while(!finished)
{
    auto buffer = socket.recv();
    auto errorCode = PlayMessages(c, buffer);
    if (errorCode) < handle error >
}

Close(c);

Alternative which avoids assumptions about an ObjectInterfaceId type

Caller side


RpcCaller* c = CreateRpcCaller();

auto id1 = GetObjectInterfaceId1();
auto id2 = GetObjectInterfaceId2();
auto id3 = GetObjectInterfaceId3();

// These stubs exist for the life of the RpcCaller
ptr<Ix> stub1 = MakeStub<Ix>(c, [](Archive& ar) { ar << id1; } );
ptr<Ix> stub2 = MakeStub<Ix>(c, [](Archive& ar) { ar << id2; } );
ptr<Iy> stub3 = MakeStub<Iy>(c, [](Archive& ar) { ar << id3; });

while(!finished)
{
    // Make calls on stubs.  These invocations will be recorded in pb
    // The recording is infallible except for running out of memory.  There is no I/O
    stub1->f(1,2,"x");
    stub1->g();
    stub2->f(3,4,"y");
    for (int i=0 ; i < 10 ; ++i)
        stub3->h(i,20);

    // Flush the archive, get the result (perhaps an alias to memory managed by the RpcCaller to avoid memcpy)
    // and clear
    // Note how this breaks up the recorded invocations into a stream of coarser level messages
    auto buffer = GetBuffer(c);

    socket.send(buffer);
}

Close(c);

Callee side


ObjectInterfaceId id1 = ...;
ObjectInterfaceId id2 = ...;
ObjectInterfaceId id3 = ...;

// We have some objects that will receive the messages
ptr<Ix> obj1 = GetObject1();
ptr<Ix> obj2 = GetObject2();
ptr<Iy> obj3 = GetObject3();

RpcCallee* c = CreateRpcCallee(
    [](InputArchive& ar) -> AnyInterface
    {
        ObjectInterfaceId id;
        ar >> id;
        switch(id)
        {
        case id1: return obj1;
        case id2: return obj2;
        case id3: return obj3;
        default : return nullptr;       // not found
        }
    });

while(!finished)
{
    auto buffer = socket.recv();
    auto errorCode = PlayMessages(c, buffer);
    if (errorCode) < handle error >
}

Close(c);

For now we assume there's only one stub

Caller side


$interface rpc Ix
{
    void f(int32 a, uint16 b, const xstring& c);
    void g();
    void h(int64 a, float64 b);
};

RpcCaller* c = CreateRpcCaller();
ptr<Ix> stub = GetStub<Ix>(c);
stub->f(1,2,"x");
stub->g();
stub->f(3,4,"y");
for (int i=0 ; i < 10 ; ++i)
    stub->h(i,20);
auto buffer = GetBuffer(c);
Close(c);


RpcCaller* c = CreateRpcCaller();

// These stubs exist for the life of the RpcCaller
ptr<Ix> stub = GetStub<Ix>(c);

while(!finished)
{
    // Make calls on the stub.  These invocations will be recorded in a PagedBuffer in the RpcCaller
    // The recording is infallible except for running out of memory.  There is no I/O
    stub->f(1,2,"x");
    stub->g();
    stub->f(3,4,"y");
    for (int i=0 ; i < 10 ; ++i)
        stub->h(i,20);

    // Flush the archive, get the result (perhaps an alias to memory managed by the RpcCaller to avoid memcpy)
    // and clear
    // Note how this breaks up the recorded invocations into a stream of coarser level messages
    auto buffer = GetBuffer(c);

    socket.send(buffer);
}

Close(c);

Callee side


auto buffer = < some recorded buffer >
ptr<Ix> obj = < some object >
RpcCallee* c = CreateRpcCallee(obj);
PlayMessages(c, buffer);
Close(c);


// The object that will receive the messages
ptr<Ix> obj = GetObject();

RpcCallee* c = CreateRpcCallee(obj);

while(!finished)
{
    auto buffer = socket.recv();
    auto errorCode = PlayMessages(c, buffer);
    if (errorCode) < handle error >
}

Close(c);

TODO

DONE

cxRpc Source code

A snapshot of the files referenced by these implementation chapters is retained under implementation/messaging/supporting/cxrpc.

PreviousPart V: Recording and replaying calls with cxRpc Next62 Rpc

61 cxRpc

Recording and playing invocations of interface methods

Performance

Stub/skeleton

Thread safety

Synchronous methods are not supported

Single stub

Multiple stubs

Use of TcpMsg framework

Interface to define a message protocol

Supporting object identifiers in the protocol

Overview

Thread safety

Infinite Write Buffering

Memory management

Avoiding memory allocations

Example usage

Caller side

Callee side

Alternative which avoids assumptions about an ObjectInterfaceId type

Caller side

Callee side

For now we assume there's only one stub

Caller side

Callee side

Links

TODO

DONE

cxRpc Source code