Skip to content
v1.11dev
GitHub

Examples

1-n-1: sampler, processors and sink

The 1-n-1 example puts processors between a sampler and a sink. The sampler distributes its messages over the processors, each processor changes the text, and the sink collects the results. A JSON file configures the channels of all devices, and the two processors share one entry of this file.

The files of the example are in examples/1-n-1 on GitHub.

Topology

Section titled “Topology”

The topology comes from the file ex-1-n-1.json:

Device Channel Type Method Address
sampler1 data1 push bind tcp://*:5555
processor1, processor2 data1 pull connect tcp://localhost:5555
processor1, processor2 data2 push connect tcp://localhost:5556
sink1 data2 pull bind tcp://*:5556

All sockets have sndBufSize 1000, rcvBufSize 1000 and rateLogging 0. The sampler and the sink bind, and the processors connect. Thus all processors have the same configuration, and you can add processors at any time. The push socket of the sampler sends each message to one connected processor, in turn (round-robin). The pull socket of the sink receives from all processors.

The JSON file

Section titled “The JSON file”

The file has one entry for each device in fairMQOptions.devices. This is the entry of the processors:

examples/1-n-1/ex-1-n-1.json
"key": "processor",
"channels": [
    {
        "name": "data1",
        "sockets": [
            {
                "type": "pull",
                "method": "connect",
                "address": "tcp://localhost:5555",
                "sndBufSize": 1000,
                "rcvBufSize": 1000,
                "rateLogging": 0
            }
        ]
    },
    {
        "name": "data2",
        "sockets": [
            {
                "type": "push",
                "method": "connect",
                "address": "tcp://localhost:5556",
                "sndBufSize": 1000,
                "rcvBufSize": 1000,
                "rateLogging": 0
            }
        ]
    }
]
examples/1-n-1/ex-1-n-1.json:23–51

A device uses the entry whose id is equal to its --id. With --config-key processor, the device uses the entry with "key": "processor" instead (Config.cxx, JSONParser.cxx). Thus processor1 and processor2 have different IDs, but the same channels. The JSON configuration page describes the format.

Build the example

Section titled “Build the example”

The examples build together with FairMQ when the CMake option BUILD_EXAMPLES is ON (the default). The build copies the JSON file to the build directory of the example:

  • Directorybuild/examples/1-n-1/
    • fairmq-ex-1-n-1-sampler
    • fairmq-ex-1-n-1-processor
    • fairmq-ex-1-n-1-sink
    • ex-1-n-1.json
    • fairmq-start-ex-1-n-1.sh
    • test-ex-1-n-1.sh

The install step copies the JSON file to <prefix>/share/fairmq/. The installed start script uses this copy.

Run the example

Section titled “Run the example”

  1. Go to the directory of the example in the build tree:

    Terminal window
    cd build/examples/1-n-1

  2. Run the start script. It opens four xterm windows, one for each device:

    Terminal window
    ./fairmq-start-ex-1-n-1.sh

To run the example without xterm, run each command in its own terminal in build/examples/1-n-1:

Terminal window
./fairmq-ex-1-n-1-sampler --id sampler1 --mq-config ex-1-n-1.json
./fairmq-ex-1-n-1-processor --id processor1 --mq-config ex-1-n-1.json --config-key processor
./fairmq-ex-1-n-1-processor --id processor2 --mq-config ex-1-n-1.json --config-key processor
./fairmq-ex-1-n-1-sink --id sink1 --mq-config ex-1-n-1.json

Push q in a terminal to stop the device in it.

The code

Section titled “The code”
examples/1-n-1/sampler.cxx
bool ConditionalRun() override
{
    // Initializing message with NewStaticMessage will avoid copy
    // but won't delete the data after the sending is completed.
    fair::mq::MessagePtr msg(NewStaticMessage(fText));


    LOG(info) << "Sending \"" << fText << "\"";


    // in case of error or transfer interruption, return false to go to IDLE state
    // successfull transfer will return number of bytes transfered (can be 0 if sending an empty message).
    if (Send(msg, "data1") < 0) {
        return false;
    } else if (fMaxIterations > 0 && ++fNumIterations >= fMaxIterations) {
        LOG(info) << "Configured maximum number of iterations reached. Leaving RUNNING state.";
        return false;
    }


    std::this_thread::sleep_for(std::chrono::seconds(1));


    return true;
}
examples/1-n-1/sampler.cxx:27–47

NewStaticMessage() creates a message for the data of fText and does not delete the data after the transfer. The zeromq transport does not copy the data, so the data must stay valid until the transfer is complete. fText is a member of the device, so it stays valid. The shmem transport copies the data into shared memory when it creates the message (shmem/Message.h). After each message, the sampler waits 1 s.

examples/1-n-1/processor.cxx
Processor()
{
    OnData("data1", &Processor::HandleData);
}
examples/1-n-1/processor.cxx:18–21
examples/1-n-1/processor.cxx
bool HandleData(fair::mq::MessagePtr& msg, int)
{
    LOG(info) << "Received data, processing...";


    // Modify the received string
    std::string* text = new std::string(static_cast<char*>(msg->GetData()), msg->GetSize());
    *text += " (modified by " + fId + ")";


    // create message object with a pointer to the data buffer,
    // its size,
    // custom deletion function (called when transfer is done),
    // and pointer to the object managing the data buffer
    fair::mq::MessagePtr msg2(NewMessage(const_cast<char*>(text->c_str()),
                                    text->length(),
                                    [](void* /*data*/, void* object) { delete static_cast<std::string*>(object); },
                                    text));


    // Send out the output message
    if (Send(msg2, "data2") < 0) {
        return false;
    }


    return true;
}
examples/1-n-1/processor.cxx:23–46

The processor registers HandleData() for the channel data1 with OnData(). For each message, it copies the text into a new string and adds its device ID fId. Then it sends the new message on data2. The processor has no custom options, so it runs until you stop it.

The sink is the sink of the 1-1 example, with the channel data2.

Output

Section titled “Output”

The listings show the severity and the message of each line, without the time.

The sampler logs Sending "Hello" once per second. Each processor logs one line for each message that it gets:

[INFO] Received data, processing...

The sink gets the messages of both processors:

[INFO] Received: "Hello (modified by processor1)"
[INFO] Received: "Hello (modified by processor2)"
[INFO] Received: "Hello (modified by processor1)"

Test the example

Section titled “Test the example”

The test script test-ex-1-n-1.sh configures the channels with --channel-config instead of the JSON file. It uses ipc:// addresses in /tmp, a new session, and --control static. The sampler and the sink stop after two messages (--max-iterations 2). The processors have no such option, so the script stops them with SIGINT after the sampler and the sink end.

CTest runs the script once for each transport: Example.1-n-1.zeromq and Example.1-n-1.shmem. The tests exist if you configure with -DBUILD_TESTING=ON.

Terminal window
cd build
ctest -R '^Example\.1-n-1\.' --output-on-failure

Things to try

Section titled “Things to try”

  • Add a processor while the other devices run. The sampler then distributes the messages over three processors:

    Terminal window
    ./fairmq-ex-1-n-1-processor --id processor3 --mq-config ex-1-n-1.json --config-key processor

  • Configure a processor without the JSON file, as the test script does. One --channel-config option takes one value for each channel:

    Terminal window
    ./fairmq-ex-1-n-1-processor --id processor3 \
      --channel-config "name=data1,type=pull,method=connect,address=tcp://localhost:5555,rateLogging=0" \
                       "name=data2,type=push,method=connect,address=tcp://localhost:5556,rateLogging=0"

    If you give --mq-config and --channel-config together, the device uses only --mq-config (Config.cxx).

  • Send another text: start the sampler with --text "Hi".

  • Use shared memory: add --transport shmem to all four commands.

  • Stop after four messages: add --max-iterations 4 to the commands of the sampler and the sink.

Section titled “Related pages”