Syncing a time-sensitive API application with Unet's DiscreteEventSimulator

[tuff-krister]

Scenario

I have written an external C++ application that uses the UnetSocket API to send packets at random. The application relies on a framework written in C++, so I cannot easily rewrite it in Groovy.

My application keeps track of time on the Unet node via incoming messages, which are sent by a simple UnetAgent. The packets are sent at random time instants, that I do not know beforehand. Had I known when to send the packets, I would have requested to schedule the transmission.

My implementation

My application polls the connected Unet node for messages on a known topic, which I call mweTic. Every time it receives a message on topic mweTic, its local time is updated to match sentAt of this message.
If the new time is past the scheduled transmission time, it asks the connected Unet node to broadcast a packet and schedules the next packet internally.
Whether or not a packet is sent, the application replies with a message on topic ntfToc, and the next outgoing packet is delayed by a random amount of time.

extern "C" {
#include "unet.h"
}

#include <format>
#include <iostream>
#include <string>
#include <random>

int main(int argc, char** argv) {
    // initalisation phase
    std::string ip = argc > 1 ? argv[1] : "localhost";
    int port = argc > 2 ? std::stoi(argv[2]) : 1100;
    long local_time = 0L;
    long next_tx = 5000L;
    const long min_wait = 5000L;
    const long ticker_ms = 50L;  // at most agent's TickerBehavior period

    // random number generation
    std::seed_seq sseq{2, 4, 6, 8, port};
    std::mt19937 rng;
    rng.seed(sseq);
    std::uniform_int_distribution<long> dist(1, ticker_ms);

    // startup-like phase
    unetsocket_t sock = unetsocket_open(ip.c_str(), port);
    if (sock == NULL) {
        std::clog << std::format("Could not open socket to {}:{}",
                ip, port) << std::endl;
        return 1;
    }
    // we want to send some Fjage messages directly
    fjage_gw_t gw = unetsocket_get_gateway(sock);
    const fjage_aid_t self = fjage_aid_topic("mweTic");  // inbound traffic
    const fjage_aid_t node = fjage_aid_topic("mweToc");  // outbound traffic
    fjage_subscribe(gw, self);
    fjage_msg_t ntf_in;
    fjage_msg_t req_out;
    fjage_msg_t ntf_out;

    // agent loop
    while (true) {
        // receive a message and get when it was sent
        ntf_in = fjage_receive(gw, "org.arl.fjage.Message", NULL, 1000L);
        if (ntf_in == NULL) {
            continue;
        }
        local_time = fjage_msg_get_long(ntf_in, "sentAt", local_time);
        fjage_msg_destroy(ntf_in);
        // check if inbound message told us it's time to send, and send if so
        if (local_time >= next_tx) {
            char buf[5] = "MWE!";
            unetsocket_send(sock, reinterpret_cast<uint8_t*>(&buf[0]),
                    5, 0, 0);
            next_tx += min_wait;
        }
        // publish "we are done" on outbound topic to end block on node
        ntf_out = fjage_msg_create("org.arl.fjage.Message", FJAGE_INFORM);
        fjage_msg_set_recipient(ntf_out, node);
        fjage_send(gw, ntf_out);
        next_tx += dist(rng);  // delay TX by random amount
    }
}

I also have a UnetAgent on the connected Unet node, that subscribes to the topic mweToc and publishes messages on topic mweTic at a fixed interval. After each mweTic is sent, the agent blocks until it receives a message on the topic mweToc or until timeout.

package com.example.mwe

import org.arl.fjage.AgentID
import org.arl.fjage.Performative
import org.arl.fjage.TickerBehavior
import org.arl.fjage.Message
import org.arl.unet.UnetAgent

class TimeBrokerAgent extends UnetAgent {
    final String name = "Timing Broker"
    final String description = ("Agent that should help time-dependent"
        + " external applications to use the discrete-event Unet simulator.")

    static final AgentID EXT_TOPIC = new AgentID("mweTic", true)
    static final AgentID INT_TOPIC = new AgentID("mweToc", true)

    int interval = 50  // at least application's maximum random delay
    
    @Override
    void startup() {
        // announce that we are interested in "tocs"
        subscribe(INT_TOPIC)
        add new TickerBehavior(interval, {
            // send "tic" to the external application
            // its "sentAt" property is used by the external application
            send new Message(recipient: EXT_TOPIC,
                performative: Performative.INFORM)
            block(10000) // blocking forever might not be a good idea
        })
    }

    @Override
    void processMessage(Message ntf) {
        // should only trigger when the external application sends a "toc"
        wake()
    }
}

Problem

My implementation introduces an unwanted trade-off between artificial lag and simulation speed, when I run the application to connect to a Unet running in a DiscreteEventSimulator. A shorter interval leads to a slower simulation, and a longer interval results in more artificial lag in my application. Hence, my implementation is not ideal for timing-sensitive applications.

Question

Is there a better method to synchronise my external application with the DiscreteEventSimulator clock, using only the APIs? By "better", I mean a method that is more efficient and creates less artificial lag in my application.

[mchitre]

The essential problem here is that your application isn't running in discrete time. By having the time sync interval, you are forcing discrete events at those intervals, and thus slowing down the simulations. If you can pass your next_tx to UnetStack, you can get the TimeBrokerAgent to only schedule the next time sync at the next time you need to transmit and avoid the excessive time sync points?

[tuff-krister]

Thanks for the tip! I experimented with this over the holidays and learned a bit along the way. Letting TimeBrokerAgent know when next_tx is via a subclass of Message with a nextTx property, and using the WakerBehavior properly, does help with the issue. Perhaps I should consider having the TimeBrokerAgent use Agent.request(Message, long) with an opportunely chosen timeout, and have the application respond to these requests?

[mchitre]

I am not sure I understand what you are suggesting to do. Your current solution of using a message subclass seems reasonable to me.

[tuff-krister]

I thought I would think loudly to help future Unet-application devs. What I was suggesting to do was something along these lines:

1. TimeBrokerAgent sends a request to the application, e.g. GetNextEventReq, asking for when the next event is expected. It uses the Message request(Message[, long]) method.
2. The application receives the request and resolves all events that occurred from the time the last request was received up to the new time. Any transmissions are requested here.
3. The application sends a reply to the GetNextEventReq. The message, e.g. NextEventRsp, has the nextTx property.
4. TimeBrokerAgent receives the reply from the application and schedules the next time sync point with a WakerBehavior.
5. When the WakerBehavior triggers, repeat from step 1.

Hopefully this post clears things up! Thanks again @mchitre for your guidance!