Delta V Software

Client stubs

Documentation

PrevUpHomeNext
Timeouts
Pinging
Oneway calls
Batched oneway calls
Client progress callbacks
Per-request user data
Client stub copying

Every RcfClient<> object has a member function called getClientStub(), which allows access to an underlying RCF::ClientStub object. Almost all client-side configuration in RCF is done via RCF::ClientStub.

We previously saw how to set the timeouts for a remote call: 

// 5 second connect timeout
client.getClientStub().setConnectTimeoutMs(5*1000);


// 60 second remote call timeout
client.getClientStub().setRemoteCallTimeoutMs(60*1000);

Pings
Pingbacks

Client stubs have a built in ping() method, which can be used to determine if the connection to the server is still functional. A ping behaves exactly as a remote call with no in or out parameters, and is subject to the same timeouts. 

RcfClient<I_X> client(( RCF::TcpEndpoint(port)));        

// ...

// Set a short timeout for the ping.
int previousTimeoutMs = client.getClientStub().getRemoteCallTimeoutMs();
client.getClientStub().setRemoteCallTimeoutMs(1000);

// Ping.
client.getClientStub().ping();

// Set the timeout back to what it was.
client.getClientStub().setRemoteCallTimeoutMs(previousTimeoutMs);

Pings are a basic tool of distributed applications. It is often necessary to verify the connectedness of a connection, even if no network level disconnects have been detected. For example, it is common for routers on IP networks to drop TCP connections that have been idle for some time. Unfortunately, many routers will do so without sending any disconnect notifications to the two endpoints of the connection. In this situation, pinging the connection is the only way to discover that it is down.

Of course, by regularly pinging an otherwise idle connection, we can prevent network hardware from harvesting the connection in the first place.

When making remote calls that take some time to execute on the server, we run the risk of encountering network failures, while the call is in progress. Ideally such a failure would be detected immediately, and an exception thrown on the client. Unfortunately the nature of network failures (and a number of other failures) is such that there is no way of detecting them, short of testing the connection with a ping.

In this situation, however, the client has already initiated a call, and is waiting for a response, and is not in a position where it can ping the server. What RCF provides for these situations is a PingBackService on the server, which will regularly send pings back down the connection from the server, while the call is in progress. If there is an interruption in the stream of pingbacks, the client will draw the conclusion that the network connection has been compromised, and will throw an exception immediately, instead of waiting for its remote call timeout to expire.

Pingbacks require a multithreaded server, with a PingBackService installed. The client needs to explicitly request pingbacks at a certain interval: 

RCF_BEGIN(I_X, "I_X")
    RCF_METHOD_V1(void, wait, int)
RCF_END(I_X)

class X
{
public:
    void wait(int waitMs)
    {
        Sleep(waitMs);
    }
};


RCF::RcfServer server( RCF::TcpEndpoint(0));

// Pingbacks require at least two server threads.
server.setThreadPool( RCF::ThreadPoolPtr( new RCF::ThreadPool(2) ) );

X x;
server.bind<I_X>(x);

RCF::PingBackServicePtr pbsPtr( new RCF::PingBackService() );
server.addService(pbsPtr);

server.start();

int port = server.getIpServerTransport().getPort();
RcfClient<I_X> client(( RCF::TcpEndpoint(port)));

// No ping backs configured.
client.wait(5*1000);

// Ping backs configured, every 1000 ms.
client.getClientStub().setPingBackIntervalMs(1000);
client.wait(5*1000);

Both calls to wait() will wait for 5 seconds on the server. The second call has pingbacks configured - if one were to pull the network cable out of the server right after the call started, 2 seconds later, the client would throw an exception. Without pingbacks, the client would wait until its remote call timeout expires, before throwing an exception.

Via the client stub, we can also set the calling semantics. RCF::Twoway is the default calling semantic. It implies that the client will, after sending a request, wait for the corresponding response from the server. Alternatively, with the RCF::Oneway semantic, the client will, after sending a request to the server, immediately return control to the application. That means that any return values or out parameters will be in the same state they were in before the call. On the server-side, a oneway call is executed in the same way as a twoway call, with the exception that return values and out parameters are not sent back to the client.

As an illustration of RCF::Oneway, the following program demonstrates two servers bouncing oneway messages off each other: 

#include <iostream>

#include <RCF/Idl.hpp>
#include <RCF/RcfServer.hpp>
#include <RCF/TcpEndpoint.hpp>

RCF_BEGIN(I_Echo, "I_Echo")
    RCF_METHOD_R1(std::string, echo, std::string)
RCF_END(I_Echo)

class Echo
{
public:
    Echo(int port) : mClient(RCF::TcpEndpoint(port)), mCount()
    {}

    std::string echo(const std::string &s)
    {
        ++mCount;
        mClient.echo(RCF::Oneway, s);
        return s;
    }

    RcfClient<I_Echo> mClient;
    int mCount;
};

int main()
{
    int portA = 50001;
    int portB = 50002;

    Echo echoA(portB);
    Echo echoB(portA);

    RCF::RcfServer serverA(( RCF::TcpEndpoint(portA)));
    serverA.bind<I_Echo>(echoA);
    serverA.start();

    RCF::RcfServer serverB(( RCF::TcpEndpoint(portB)));
    serverB.bind<I_Echo>(echoB);
    serverB.start();

    // Get things rolling.
    RcfClient<I_Echo>( RCF::TcpEndpoint(portA)).echo(RCF::Oneway, "dummy");

    // Let them play for a while.
    Sleep(5*1000);

    serverA.stop();
    serverB.stop();

    std::cout << "echoA sent " << echoA.mCount << " messages." << std::endl;
    std::cout << "echoB sent " << echoB.mCount << " messages." << std::endl;

    return 0;
}


echoA sent 19796 messages.
echoB sent 19796 messages.

Sending many frequent oneway messages has the potential to become inefficient, as each send operation involves non-trivial operating system overhead. Rather than sending each message separately, an RcfClient<> can be configured to accumulate oneway messages in an internal buffer, to be sent later, in a single send operation.

Oneway batching is controlled through the ClientStub::enableBatching(), ClientStub::disableBatching() and ClientStub::flushBatch() functions. For example: 

#include <RCF/Idl.hpp>
#include <RCF/InitDeinit.hpp>
#include <RCF/IpServerTransport.hpp>
#include <RCF/RcfServer.hpp>
#include <RCF/TcpEndpoint.hpp>

RCF_BEGIN(I_X, "I_X")
RCF_METHOD_V1(void, add, int)
RCF_END(I_X)

class X
{
public:
    X() : mTotal(0) {}
    void add(int x) {mTotal += x; }
    int mTotal;
};

int main()
{
    RCF::RcfInitDeinit rcfInitDeinit;

    X x;
    RCF::RcfServer server( RCF::TcpEndpoint(0));
    server.bind<I_X>(x);
    server.start();

    int port = server.getIpServerTransport().getPort();

    RcfClient<I_X> client(( RCF::TcpEndpoint(port) ));

    // Turn on batching mode.
    client.getClientStub().enableBatching();

    for (std::size_t i=0; i<99; ++i)
    {
        client.add(RCF::Oneway, 1);
    }

    std::cout << "Total so far: " << x.mTotal << std::endl;

    // Send the 99 messages accumulated so far.
    client.getClientStub().flushBatch();

    // Let the server catch up.
    Sleep(1000);

    std::cout << "Total so far: " << x.mTotal << std::endl;

    // Turn off batching mode.
    client.getClientStub().disableBatching();

    // Regular twoway call.
    client.add(1);

    std::cout << "Total so far: " << x.mTotal << std::endl;

    return 0;
}


Total so far: 0
Total so far: 99
Total so far: 100

Some points to keep in mind when using oneway batching:

  • If an error occurs while sending, or the connection is disconnected explicitly, the internal batching buffer is cleared, thus discarding any oneway messages that have not yet been sent.
  • The messages in a batch will be processed by the server in the order they were sent (but possibly by different threads).
  • Message and transport filters can be applied as usual.
  • There is an upper limit on the size of the internal batching buffer. RCF will automatically send a batch when this limit is reached. By default this limit is 1 MB. To change the limit, call the ClientStub::setMaxBatchMessageLength(). To disable automatic flushing, set the limit to zero.
  • Batching can be used on connection-less transports (e.g. UDP), but the batch size then needs to be small enough that the transport layer doesn't fragment the batch.

Another important feature of client stubs are progress callbacks. It is usually not desirable to have your application freeze while making a remote call, especially if the call might take a long time to complete. By using a progress callback, we can direct RCF to return control to us periodically, even though the call is still in progress.

Client progress callbacks can be used to repaint the user interface of the application, to show a progress bar, or even to cancel the remote call itself. The following example demonstrates timer based progress callbacks. 

#include <iostream>

#include <RCF/ClientProgress.hpp>
#include <RCF/Idl.hpp>
#include <RCF/IpServerTransport.hpp>
#include <RCF/RcfServer.hpp>
#include <RCF/TcpEndpoint.hpp>

RCF_BEGIN(I_Echo, "I_Echo")
    RCF_METHOD_R1(std::string, echo, std::string)
RCF_END(I_Echo)

class Echo
{
public:
    std::string echo(const std::string &s)
    {
        Sleep(5*1000);
        return s;
    }
};

void onTimer(RCF::ClientProgress::Action &action)
{
    std::cout << "Timer event" << std::endl;

    // To cancel the call...
    action = RCF::ClientProgress::Cancel;

    // Or to continue the call.
    action = RCF::ClientProgress::Continue;
}

int main()
{
    Echo echo;
    RCF::RcfServer server( RCF::TcpEndpoint(0));
    server.bind<I_Echo>(echo);
    server.start();

    // Set up a timer callback every 1000 ms, to onTimer().
    RCF::ClientProgressPtr clientProgressPtr( new RCF::ClientProgress() );
    clientProgressPtr->mTriggerMask = RCF::ClientProgress::Timer;
    clientProgressPtr->mTimerIntervalMs = 1000;
    clientProgressPtr->mProgressCallback = boost::bind(onTimer, _5);

    int port = server.getIpServerTransport().getPort();
    RcfClient<I_Echo> client(( RCF::TcpEndpoint(port)));
    client.getClientStub().setClientProgressPtr(clientProgressPtr);

    std::cout << "Call initiated" << std::endl;
    client.echo("");
    std::cout << "Call completed" << std::endl;

    return 0;
}


Call initiated
Timer event
Timer event
Timer event
Timer event
Call completed

Progress callbacks can also be triggered by events in the lifecycle of a remote call (connecting/sending/receiving), and, on Windows, by the reception of specific Windows messages (such as WM_PAINT).

A RCF remote call involves marshaling a number of values between the client and the server. The type of these values is determined by the RCF interface declaration. In some situations, however, there may also be other data that needs to be transmitted, in either direction, as part of the remote call.

For this purpose, RCF provides user data fields in the request and response headers. The user data fields are set by calling ClientStub::setRequestUserData() and RcfSession::setResponseUserData(). Here is an example: 

// Client-side.
RcfClient<I_Echo> client(( RCF::TcpEndpoint(port) ));
client.getClientStub().setRequestUserData("ABC");
client.echo();
std::string responseUserData = client.getClientStub().getResponseUserData();


// Server-side.
class Echo
{
public:
    void echo()
    {
        // Retrieve request user data, and copy it to response user data.
        RCF::RcfSession & session = RCF::getCurrentRcfSession();
        std::string requestUserData = session.getRequestUserData();
        session.setResponseUserData(requestUserData);
    }
};

User data slots can be used to implement custom authentication schemes, where a clients first call results in a token, that is then passed along on subsequent calls.

When a copy is made of a RcfClient<> object, it's important to realize that only the client stub attributes are copied, and not the physical connection itself. For example: 

// Create some clients.
RcfClient<I_Echo> client1(( RCF::TcpEndpoint(port)));
RcfClient<I_Echo> client2( client1);
RcfClient<I_Echo> client3 = client1;
RcfClient<I_Echo> client4;
client4 = client1;

// Copy them into a vector.
std::vector< RcfClient<I_Echo> > clients;
clients.push_back(client1);
clients.push_back(client2);
clients.push_back(client3);
clients.push_back(client4);

The copied RcfClient<> objects will have the same client stub attributes as client1 , but when calls are made on them, they will create new connections of their own, distinct from the connection of client1.

If we actually want a copied RcfClient<> to use the same connection as the RcfClient<> it is copied from, we have to explicitly move the connection: 

std::cout << clients[1].echo("asdf\n");

// Move the connection from client 1 to client 0.
clients[0].getClientStub().setTransport(
    clients[1].getClientStub().releaseTransport());

std::cout << clients[0].echo("asdf\n");

After the call to releaseTransport(), clients[1] no longer has a connection. If we make a call on it, it will use the information in its cached TcpEndpoint object, to create a brand new connection. If we didn't want that to happen, and instead wanted the original connection reused, we would need to move the connection back again: 

// Move the connection from client 0 to client 1.
clients[1].getClientStub().setTransport(
    clients[0].getClientStub().releaseTransport());

std::cout << clients[1].echo("asdf\n");

// This call will create a new connection.
std::cout << clients[0].echo("asdf\n");

In C++ terms, RcfClient<> objects have value semantics. They can be used as parameters or return values, stored in containers, etc. Just be aware that each copy will create its own connection to the server, if and when a remote method is called on it.

PrevUpHomeNext