Streaming messages

The fundamental class for communication in yarp is yarp::os::Port. Ports have names and can connect to any number of other Ports. Data flows along these connections. YARP has a strong bias towards "streaming" communication, where the sender of data is maximally decoupled from the receiver of data, but also supports "send/reply" communication where sender and receiver are tightly coupled.

Here is an example of a receiver for streamed messages:

Port p;            // Create a port.
p.open("/in");     // Give it a name on the network.
Bottle b;          // Make a place to store things.
while (true) {
  p.read(b);         // Read from the port.  Waits until data arrives.
  ...                // Do something with data.
}

Here is an example of a sender of streamed messages:

Port p;            // Create a port.
p.open("/out");    // Give it a name on the network.
while (true) {
  Bottle b;        // Make a place to store things.
  ...              // Generate data.
  p.write(b);      // Send the data.
}

Here's an example of some code that connects the sender to the receiver:

Network::connect("/out","/in"); // connect two ports.

Characteristic of streaming operation, the sender and the receiver don't know much about each other. The sender or receiver could be restarted without needing to restart the other. There could be multiple receivers or multiple senders.

Our sender and receiver do have some dependencies on each other. In particular, once a receiver starts receiving an object, the sender must wait for that process to finish before continuing. The sender can change this behavior by calling: yarp::os::Port::enableBackgroundWrite() before the port is opened. However, this is only appropriate if we can promise the port that any object (b in this case) we ask it to write will stay in existence until it is communicated. The port won't take a copy of our object (that is inefficient in general). We can tell when writing is finished in two ways. One is by calling yarp::os::Port::isWriting(). Another is by overriding the yarp::os::PortWriter::onCompletion() method on the object. This will be called when the port is finished with the object.

Decoupling sender/receiver timing

Keeping objects around until they have been transmitted is actually quite complicated to do, especially when there are multiple receivers operating at different rates. YARP supplies the yarp::os::PortWriterBuffer class to manage these details. There is a similar class for reducing timing dependencies during reading, called yarp::os::PortReaderBuffer. These two kinds of buffering are bundled with the yarp::os::Port class in yarp::os::BufferedPort. For streaming communication, this class offers many advantages. We can rewrite our previous example like this – the receiver:

BufferedPort<Bottle> p; // Create a port.
p.open("/in");          // Give it a name on the network.
while (true) {
  Bottle *b = p.read(); // Read from the port.  Waits until data arrives.
  ...                   // Do something with data in *b.
}

The sender:

BufferedPort<Bottle> p; // Create a port.
p.open("/out");         // Give it a name on the network.
while (true) {
  Bottle& b = p.prepare(); // Get a place to store things.
  ...                   // Generate data.
  p.write();            // Send the data.
}

The code is very similar. The big difference is that the yarp::os::BufferedPort is now responsible for the lifetime of the objects being communicated. It will make a pool of them, growing upon need, in order to give smooth performance with minimal timing dependencies between sender and receiver.

By default, a BufferedPort will just keep the newest message received if several have come in between calls to read(), dropping old messages. If you'd prefer to have all messages received, call yarp::os::BufferedPort::setStrict().

By default, if asked to send a message while some connections are still writing a previous message, BufferedPort will simply skip sending that message on those busy connections. If you'd rather all messages were send, call yarp::os::BufferedPort::writeStrict() when writing.

Polling ports

If you want to know if data has arrived, but don't want to wait if it has not, you can do:

BufferedPort<Bottle> p;
...
Bottle *b = p.read(false);
if (b!=NULL) {
  // data received in *b
}

The read method here will return immediately if there is no data. Without the "false", it would wait until data is present.

It is not possible to do something like this with yarp::os::Port, since this functionality requires buffering. You can, however, get a callback.

Getting a callback

Often when reading from a port, it is preferable to ask the port to call a method when data arrives rather than to sit around calling read(). With the Port class, you can do that with the yarp::os::Port::setReader() method:

class DataProcessor : public PortReader
{
    bool read(ConnectionReader& connection) override
    {
        Bottle b;
        bool ok = b.read(connection);
        if (!ok) {
            return false;
        }
        // process data in b
        return true;
    }
};
DataProcessor processor;
...
p.setReader(processor);  // no need to call p.read() on port any more.

With BufferedPorts, things are a bit easier since these ports know what type of data they are dealing with. You can override the port's onRead method:

class DataPort : public BufferedPort<Bottle>
{
    using BufferedPort<Bottle>::onRead;
    void onRead(Bottle& b) override
    {
        // process data in b
    }
};
...
DataPort p;
p.useCallback();  // input should go to onRead() callback
p.open("/in");

Alternatively, you can leave the port unchanged and use an external callback as follows:

class DataProcessor : public TypedReaderCallback<Bottle>
{
    using TypedReaderCallback<Bottle>::onRead;
    void onRead(Bottle& b) override
    {
        // process data in b
    }
};
...
DataProcessor processor;
BufferedPort<Bottle> p;
p.useCallback(processor);  // input should go to processor.onRead()

Getting replies

Sometimes we want to send messages and wait for replies to them. If you're considering doing this, beware – the timing of your processes is going to become tightly coupled, and a network of processes written this way is much less robust and malleable than with streaming communication. However, sometimes you really want to be sure a particular command gets through, and what the response to it is. In this case, we need something new. Messages with replies are possible with normal ports. However, this can be dangerous. For this reason YARP provides specialized classes (yarp::os::RpcClient and yarp::os::RpcServer) which allows to catch errors and provide informative messages (see Specialized RPC ports for details).

Here's how we can do it, on the sender side:

RpcClient p;       // Create a port.
p.open("/out");    // Give it a name on the network.
while (true) {
  Bottle in,out;   // Make places to store things.
  ...              // prepare command "out".
  p.write(out,in); // send command, wait for reply.
  ...              // process response "in".
}

On the receiver side:

RpcServer p;       // Create a port.
p.open("/in");     // Give it a name on the network.
Bottle in, out;    // Make places to store things.
while (true) {
  p.read(in,true); // Read from the port, and warn that we'll be replying.
  ...              // Do something with data, prepare reply
  p.reply(out);    // send reply.
 
}

If you expect replies, it is better to use one-to-one port connections rather than many-to-many – otherwise it will get confusing.

If we are using callbacks, our receiver should do something like this:

class DataProcessor : public PortReader
{
    bool read(ConnectionReader& connection) override
    {
        Bottle in, out;
        bool ok = in.read(connection);
        if (!ok) {
            return false;
        }
        ...      // process data "in", prepare "out"
        ConnectionWriter *returnToSender = connection.getWriter();
        if (returnToSender!=NULL) {
            out.write(*returnToSender);
        }
        return true;
    }
};
DataProcessor processor;
...
p.setReader(processor);  // no need to call p.read() on port any more.

(Technical note: beware that the requested reply does not happen until the end of the DataProcessor read() method, since only then is the full size of the reply known and some lower-level network protocols require that information to be send in a header.)

Expecting replies is incompatible with buffering, and so can't be done with the yarp::os::BufferedPort class. There's simply no way to isolate the timing of the sender and receiver in this case, since the one is explicitly waiting for the other to do some processing and respond.

There is a yarp::os::BufferedPort::setReplier() method which can be used to establish a "replier" for buffered ports. If someone connects to these ports and requests them to reply, this callback will be used (bypassing buffering).

Datatypes

What kind of data can be sent on ports? Anything you like, YARP doesn't care. If you communicate between machines with different OSes and compilers, you may need to be careful if you send your own custom data-typle, but apart from that there's no limit.

Suppose you have the following data-structure:

class Target {
public:
   int x;
   int y;
};

Then you can create a buffered port for it this way:

BufferedPort< BinPortable<Target> > port;

The yarp::os::BinPortable class tells YARP that you want to send the Target type across the network by encoding it exactly as it is represented in memory. The method yarp::os::BinPortable::content() will give you access to the actual Target object.

If you try this between, say, an intel machine and a non-intel Mac, you'll run into trouble since integers are represented in different ways on these machines ("big-endian" versus "little-endian"). If you try it between machines with different compilers, you may have trouble too – compilers may add different amounts of "padding" into your structure

If you define your data-structure as follows:

#include <yarp/conf/system.h>
YARP_BEGIN_PACK
class Target {
public:
  NetInt32 x;
  NetInt32 y;
};
YARP_END_PACK

Then you'll be in better shape. The integers now have a well defined representation, and the compiler is requested not to introduce padding. So this is a portable representation.

Even better, for YARP, is to add in a small "header" that describes your data-type. If you defined Target as follows:

#include <yarp/conf/system.h>
#include <yarp/os/Bottle.h>
YARP_BEGIN_PACK
class Target {
public:
    NetInt32 tag;
    NetInt32 len;
    NetInt32 x;
    NetInt32 y;
    Target() {
        tag = BOTTLE_TAG_LIST + BOTTLE_TAG_INT32;
        len = 2;
    }
};
YARP_END_PACK

Then suddenly ports carrying this data become a lot easy to interoperate with. You can read from them using "yarp read", write to them using "yarp write", send/reply to them using "yarp rpc". You can even connect your own sockets to them without using YARP to write their values in a standard, document text/binary format. You may not see why you'd want to do any of that today, but it might be worth bearing in mind for the future.

More usually, rather than sending memory images across the network (a procedure fraught with gotchas), it is better to provide explicit serialization methods for your class by implementing the yarp::os::Portable interface. For example:

class Target : public Portable {
public:
  int x;
  int y;
  bool write(ConnectionWriter& connection) override
  {
    connection.appendInt32(x);
    connection.appendInt32(y);
    return true;
  }
  bool read(ConnectionReader& connection) override
  {
    x = connection.expectInt32();
    y = connection.expectInt32();
    return !connection.isError();
  }
};

See the yarp::os::ConnectionWriter and yarp::os::ConnectionReader for the serialization methods available. These take care of using neutral formats for your data types. Now you no longer need the BinPortable wrapper:

BufferedPort<Target> port;

This is how classes like yarp::os::Bottle work.

There is one other thing you might like to consider when deciding how to express your data on the network. If you embed a few "tags" in your data, then it can be made compatible with the network format used by Bottles. Then ports using the format can be read from and written to the command line (with "yarp read", "yarp write", and "yarp rpc"), or a web-browser, etc. For example, we could make Target's write method be:

bool write(ConnectionWriter& connection) override
{
  connection.appendInt32(BOTTLE_TAG_LIST+BOTTLE_TAG_INT32);
  connection.appendInt32(2); // two elements
  connection.appendInt32(x);
  connection.appendInt32(y);
  connection.convertTextMode(); // if connection is text-mode, convert!
  return true;
}

The first integer added says that the data is a list of integers. The second integer says there are 2 integers. The call to yarp::os::ConnectionWriter::convertTextMode is a friendly thing to do. It means if someone connects to a port outputting this data, and they are in text-mode, they will see our two integers in text form. The conversion is done by reading the data we've written as a Bottle, and then expressing that Bottle in text form.

The corresponding read method would be:

bool read(ConnectionReader& connection) override
{
  connection.convertTextMode(); // if connection is text-mode, convert!
  int tag = connection.expectInt32();
  if (tag!=BOTTLE_TAG_LIST+BOTTLE_TAG_INT32) return false;
  int ct = connection.expectInt32();
  if (ct!=2) return false;
  x = connection.expectInt32();
  y = connection.expectInt32();
  return !connection.isError();
}

The advantages of doing this are for testing - no need to write special purpose test programs to send and read particular messages, the yarp tools already work. The disadvange is you have to do a bit more work. It is entirely up to you, this issue is independent of the rest of YARP.

NOTE: For automatic text conversion to work, your message will need to be represented as a list at the top level. If you try to send a single integer, for example, translation will not work. This is because we chose to omit parentheses from lists at the top level of messages in text-mode, so there would be no way to differentiate a list containing an integer from an integer on its own.

Carriers

We've said a lot about communication. But what exactly happens when you read or write? What bytes get sent on the network? How reliable is transmission? The previous sections on datatypes and replies have started to touch on such issues, and now we go into more detail.

The first thing to realize is that we are now looking closely at the connections between ports, rather than the ports themselves. The properties of connections are quite loosely coupled with ports. We are free to implement connections in all sorts of weird and wonderful ways, and ports won't care at all. The abstraction in yarp for what a connection really does is called a "Carrier". There are currently carriers for tcp, udp, multi-cast, and shared memory communication. There's nothing to stop you adding more if you have some novel kind of network, or want to send messages via audio, etc. Ports will still work.

Carriers can be connection based (tcp, shmem) or connectionless (udp, mcast). Be aware that on connectionless carriers data flows in one direction only. As a consequence of this, replies are not possible. Data transmission may also be unreliable for such carriers. YARP guarantees that if a message is sent and successfully received, it has not been corrupted. For connectionless carriers it does not guarantee that the message will in fact be received.

Ideally, for all carriers, YARP will send messages of arbitrary length. At the time of writing, there is a limit on shmem message sizes.

There are some extra carriers designed to be as simple as possible for programs not written in YARP to use. One of these is the "text" Carrier. This allows messages in bottle-compatible format to be sent as plain text.

There is experimental support for http, as a way to browse port information from a browser.

Table of Contents