33  Developing agents

By now you are very familiar with the idea of agents — you have interacted with them through commands and messages throughout this handbook. In this chapter we cross over from using agents to writing them. While the prospect of writing your own agent might sound daunting, you’ll soon see that it is actually quite easy.

33.1 Unet agents

Agents are the basic building blocks of UnetStack. They exchange messages, provide services and implement protocols. Most of the demanding work expected of a well-behaved agent is already taken care of by the UnetAgent base class. All you need to do is extend it and add a little code to teach the agent what you want it to do.

Tip

You can write agents in Java or Groovy (or any other JVM language), but we recommend Groovy: Groovy agents need less boilerplate, are more readable, and can be loaded dynamically from source without precompilation. If you are an expert Java programmer and prefer Java, you are welcome to use it.

The skeleton of a Groovy agent looks like this:

import org.arl.fjage.*
import org.arl.unet.*
import groovy.transform.CompileStatic

@CompileStatic
class MyAgent extends UnetAgent {

  @Override
  void setup() {
    // called when the stack is initialized
    // register services and capabilities that you provide here
  }

  @Override
  void startup() {
    // called just after the stack is running
    // look up other agents and services here, as needed
    // subscribe to topics of interest to get notifications
  }

  @Override
  Message processRequest(Message msg) {
    // process requests supported by the agent, and return responses
    // if request is not processed, return null
    return null
  }

  @Override
  void processMessage(Message msg) {
    // process other messages, such as notifications, here
    // messages that aren't interesting can simply be ignored
  }

  @Override
  void shutdown() {
    // called when the agent is stopped or the stack is shut down
    // release any resources (sockets, files, devices) acquired here
  }

}

The @CompileStatic and @Override annotations are not strictly necessary, but are good practice: they let the compiler catch type errors and typos in method names. You only need to define the methods you actually use — the base class provides sensible defaults for the rest. The lifecycle hooks are called in order: setup() once, then startup() once the stack is running, then processRequest()/processMessage() repeatedly as messages arrive, and finally shutdown() when the agent is killed or the stack stops.

NoteThe fjåge lifecycle behind the hooks

UnetStack is built on the fjåge agent framework, and the UnetAgent hooks map onto the standard fjåge agent lifecycle. A fjåge agent is initialized (init()), runs cooperatively while it has active behaviors, sits idle when it does not, and is cleaned up (shutdown()) before termination. UnetAgent wires the hooks into this lifecycle for you:

• setup() is called from the agent’s init(), before the stack is fully up.
• startup() is called from a OneShotBehavior scheduled during initialization, so it runs once the stack is running and all agents have had a chance to register their services.
• processRequest()/processMessage() are called from a MessageBehavior added during initialization.
• shutdown() is called from fjåge’s shutdown() when the agent is stopped.

See the fjåge documentation on Agents & Behaviors for the underlying model.

33.2 A first agent: the echo daemon

The best way to learn is by example. Let’s develop an echo daemon that responds to each incoming echo request datagram with an echo response datagram carrying the same data.

We need a way to tell an echo request apart from any other datagram, so that we don’t echo datagrams intended for other agents. We define an echo request as any datagram with protocol USER (recall that protocol numbers from USER onwards are available for your own applications). We must not echo the response with the same protocol, or the daemon on the source node would echo it again, ad infinitum — so the response uses protocol DATA, which is intended for generic application data.

Here’s the daemon:

import org.arl.fjage.*
import org.arl.unet.*
import groovy.transform.CompileStatic

@CompileStatic
class EchoDaemon extends UnetAgent {

  @Override
  void startup() {
    // subscribe to the topic where datagrams delivered to this node are published
    subscribe topic(Topics.DATAGRAM)
  }

  @Override
  void processMessage(Message msg) {
    if (msg instanceof DatagramNtf && msg.protocol == Protocol.USER) {
      // respond to protocol USER datagram with protocol DATA datagram
      send new DatagramReq(
        recipient: msg.sender,
        to: msg.from,
        protocol: Protocol.DATA,
        data: msg.data
      )
    }
  }

}

Walking through the code:

1. Our agent provides no formal services or capabilities, so we skip setup() and processRequest().
2. startup() subscribes to the DATAGRAM topic. Whenever a datagram addressed to this node arrives, the agent that received it (a physical layer, link or router agent providing the DATAGRAM service, see Chapter 16) publishes a DatagramNtf on this topic, and processMessage() is called.
3. In processMessage(), we check for datagram notifications with protocol USER, and respond by asking the sender to send a protocol DATA datagram back to the originating node, copying the data across.

The @CompileStatic annotation deserves special mention. It is not required — the agent works identically without it — but we recommend it for all production agents. By default, Groovy compiles code with dynamic dispatch, resolving methods and properties at runtime; with @CompileStatic, the class is type-checked and compiled statically, like Java. Typos and type errors are then caught when the agent is compiled, rather than surfacing as runtime failures in the middle of a deployment, and the generated bytecode runs faster and allocates less garbage — a difference that matters on resource-constrained modem hardware. If a method genuinely needs Groovy’s dynamic features, annotate just that method with @CompileDynamic and keep the rest of the class static.

Tip

Do not confuse sender/recipient with from/to. The sender and recipient always refer to the agents that generate and consume a message, within a single node. The from and to are node addresses identifying which node transmits a datagram and which is its intended destination.

To test the daemon, create a file called EchoDaemon.groovy in the classes folder and paste in the code above. Start the 2-node network simulation, and on node B load the agent:

> container.add 'echo', new EchoDaemon();

Our daemon is up and running! Now test it from node A:

> subscribe phy
> phy << new DatagramReq(to: host('B'), protocol: Protocol.USER, data: [42])
AGREE
phy >> TxFrameNtf:INFORM[type:DATA txTime:2809812247]
phy >> RxFrameStartNtf:INFORM[type:DATA rxTime:2811767943]
phy >> RxFrameNtf:INFORM[type:DATA from:31 to:232 rxTime:2811767943 (1 byte)]
> ntf.data
[42]

We subscribed to phy to see the incoming echo response, transmitted an echo request with some data, and saw the same data echoed back. We have written our first agent!

NoteWriting agents in Java

If you prefer Java, you can write the same agent in pure Java, at the cost of some verbosity and an extra compilation step:

import org.arl.fjage.Message;
import org.arl.unet.*;

public class EchoDaemon extends UnetAgent {

  @Override
  protected void startup() {
    // subscribe to the topic where datagrams delivered to this node are published
    subscribe(topic(Topics.DATAGRAM));
  }

  @Override
  protected void processMessage(Message msg) {
    if (msg instanceof DatagramNtf) {
      DatagramNtf ntf = (DatagramNtf)msg;
      if (ntf.getProtocol() == Protocol.USER) {
        // respond to protocol USER datagram with protocol DATA datagram
        DatagramReq req = new DatagramReq(ntf.getSender());
        req.setTo(ntf.getFrom());
        req.setProtocol(Protocol.DATA);
        req.setData(ntf.getData());
        send(req);
      }
    }
  }

}

The differences are the usual Java idioms: explicit casts on Message, getter/setter accessors (getProtocol(), setTo()) in place of Groovy’s property syntax, and explicit parentheses on method calls. One difference that matters later: the Groovy closures used to construct behaviors (e.g. add new WakerBehavior(1000, { ... })) are Groovy syntactic sugar — in Java, you instead create an anonymous subclass and override the appropriate method (onWake(), onTick(), etc.).

Create EchoDaemon.java, then compile it with the fjåge and unet-framework jars on the classpath:

$ javac -cp "lib/*" EchoDaemon.java

Copy the resulting EchoDaemon.class into the classes folder (delete any EchoDaemon.groovy first, to avoid duplicate classes), and test it just as before.

Tip

Unet modems also have a classes folder that accepts Groovy source files or compiled Java/Groovy class files. You can upload files through the modem’s web interface, or create and edit them right there in the web shell’s script editor — a convenient workflow when iterating on an agent that runs on the modem. If your code has many classes, package them into a jar and place it in the jars folder.

33.3 Behaviors

Agents implement most of their functionality through behaviors. UnetStack is built on the fjåge agent framework, which provides a set of standard behaviors for agents to use.

We have already used two behaviors implicitly: the startup() method is run by the base class using a OneShotBehavior, and processMessage() is called from a MessageBehavior. The base class adds these for us, because almost every agent needs them.

Suppose we wanted our echo daemon to respond after a delay of 7 seconds. We could naïvely add a delay(7000) in processMessage(), but that would be a bad idea — the agent would sleep for 7 seconds and be unable to process requests from any other node in the meantime. Instead, we want a behavior that fires 7 seconds later without blocking — a WakerBehavior:

  @Override
  void processMessage(Message msg) {
    if (msg instanceof DatagramNtf && msg.protocol == Protocol.USER) {
      // respond after 7 seconds
      add new WakerBehavior(7000, {
        send new DatagramReq(
          recipient: msg.sender,
          to: msg.from,
          protocol: Protocol.DATA,
          data: msg.data
        )
      })
    }
  }

The agent remains responsive while waiting, and can process several requests concurrently — send two echo requests in quick succession, and you’ll get two responses 7 seconds later, also in quick succession.

To reload the agent on node B after editing:

> container.kill echo
true
> container.add 'echo', new EchoDaemon();

Notefjåge behaviors

fjåge provides several behaviors commonly used in Unet agents:

• One-shot behavior — runs once at the earliest opportunity.
• Cyclic behavior — runs repeatedly while active; may be blocked and restarted as needed.
• Waker behavior — runs once after a specified delay (ms).
• Ticker behavior — runs repeatedly with a fixed delay between invocations.
• Backoff behavior — like a waker behavior, but the wakeup time can be extended dynamically; useful for backoff/retry timeouts.
• Poisson behavior — like a ticker behavior, but with exponentially distributed intervals, modelling a Poisson arrival process (commonly used for network data sources).
• Finite state machine behavior — implements protocols as a set of states with event-triggered transitions.

See the fjåge documentation on Agents & Behaviors to learn more.

33.4 Parameters

Many agents expose parameters that you can get and set. Let’s make our echo daemon’s delay configurable. By default the daemon has no parameters:

> echo
<<< EchoDaemon >>>

Adding a title, description and a delay parameter takes three steps — declare a parameter list, declare the parameter, and advertise the list:

import org.arl.fjage.*
import org.arl.fjage.param.Parameter
import org.arl.unet.*
import groovy.transform.CompileStatic

@CompileStatic
class EchoDaemon extends UnetAgent {

1  enum Params implements Parameter {
    delay
  }

2  final String title = 'Echo Daemon'
  final String description = 'Echoes any USER datagrams back as DATA'

3  int delay = 7000

  @Override
  void startup() {
    subscribe topic(Topics.DATAGRAM)
  }

  @Override
  void processMessage(Message msg) {
    if (msg instanceof DatagramNtf && msg.protocol == Protocol.USER) {
4      add new WakerBehavior(delay, {
        send new DatagramReq(
          recipient: msg.sender,
          to: msg.from,
          protocol: Protocol.DATA,
          data: msg.data
        )
      })
    }
  }

  @Override
5  List<Parameter> getParameterList() {
    allOf(Params)
  }

}

1

Declare the list of parameters the agent advertises (here, as an inner enum).

2

Provide a descriptive title and description for the agent.

3

Declare the parameter, with its default value.

4

Use the parameter in the behavior.

5

Advertise the parameter list.

Reload the agent and inspect it:

> echo
<<< Echo Daemon >>>

Echoes any USER datagrams back as DATA

[EchoDaemon.Params]
  delay = 7000

> echo.delay = 5000
5000

Tip

Parameters are much more than class attributes — they can be read and set remotely, even from a different JVM, a different computer, or through a UnetSocket gateway. In Java you’d implement getters and setters explicitly; in Groovy these are generated for you. To compute or validate a parameter on demand, implement a getter/setter and it will be called. For a read-only parameter, declare the attribute private and implement only a getter. Descriptions can be dynamic too — replace the description attribute with a getDescription() method to surface live status.

NoteDocumenting your agent

The title and description attributes give a one-line summary. For richer help — usage notes, parameter and command descriptions that show up under the shell help command — define a static __doc__ string. By convention @@ is a placeholder for the agent’s name, and ## Parameters: / ## Commands: sections (with ### entries) document individual parameters and commands:

public final static String __doc__ = '''\
# @@ - echoes USER datagrams back as DATA

EchoDaemon listens for datagrams sent with protocol USER and echoes their
payload back to the originating node as a protocol DATA datagram.

## Parameters:

### @@.delay - delay (ms) before echoing a datagram
'''.stripIndent()

Many of the built-in agents documented in this handbook expose their parameters and commands exactly this way.

33.5 Implementing network protocols

Real-world protocols demand more: advertising services, looking up other agents, computing parameters on demand, encoding/decoding PDUs, generating random variates, and describing behavior as a finite state machine (FSM). We illustrate all of these by developing three simple MAC agents (Chapter 19), kept intentionally unoptimized so the mechanics stay clear.

33.5.1 A simple MAC without handshake

Recall the message flow of the MAC service (Chapter 19): a client asks to reserve the channel with a ReservationReq, and the MAC agent answers with a ReservationRsp. The MAC then tells the client when its reservation starts and ends with ReservationStatusNtf messages — the client transmits its data between the START and the END notifications (Figure 33.1).

sequenceDiagram
    participant C as Client agent
    participant M as MAC agent
    C->>M: ReservationReq(to, duration)
    M-->>C: ReservationRsp (AGREE)
    M-->>C: ReservationStatusNtf (START)
    note over C: client transmits<br/>for the reservation duration
    M-->>C: ReservationStatusNtf (END)

Figure 33.1: Message exchange between a client agent and a MAC agent for a channel reservation.

Our first MAC grants every reservation request immediately. To comply with the MAC service specification, it must also respond to ReservationCancelReq, ReservationAcceptReq and TxAckReq (which we refuse), and expose the MAC service parameters:

import org.arl.fjage.*
import org.arl.fjage.param.Parameter
import org.arl.unet.*
import org.arl.unet.mac.*
import groovy.transform.CompileStatic

@CompileStatic
class MySimplestMac extends UnetAgent {

  @Override
  void setup() {
    register Services.MAC                  // advertise the MAC service
  }

  @Override
  Message processRequest(Message msg) {
    switch (msg) {
      case ReservationReq:
        ReservationReq req = (ReservationReq)msg
        if (req.duration <= 0) return new RefuseRsp(req, 'Bad reservation duration')
        ReservationStatusNtf ntf1 = new ReservationStatusNtf(
          recipient: req.sender, inReplyTo: req.messageID, to: req.to,
          status: ReservationStatus.START)
        ReservationStatusNtf ntf2 = new ReservationStatusNtf(
          recipient: req.sender, inReplyTo: req.messageID, to: req.to,
          status: ReservationStatus.END)
        add new OneShotBehavior({ send ntf1 })                       // START immediately
        add new WakerBehavior(Math.round(1000*req.duration), {       // END after duration
          send ntf2
        })
        return new ReservationRsp(req)     // defaults to an AGREE performative
      case ReservationCancelReq:
      case ReservationAcceptReq:
      case TxAckReq:
        return new RefuseRsp(msg, 'Not supported')
    }
    return null                            // NOT_UNDERSTOOD for anything else
  }

  @Override
  List<Parameter> getParameterList() {
    return allOf(MacParam)                 // advertise MAC service parameters
  }

  final boolean channelBusy = false        // 'final' makes parameters read-only
  final int reservationPayloadSize = 0
  final int ackPayloadSize = 0
  final float maxReservationDuration = Float.POSITIVE_INFINITY
  final Float recommendedReservationDuration = null

}

Note how we prepare the AGREE response and the START/END notifications together. The OneShotBehavior ensures the START notification is sent after the AGREE response, and the WakerBehavior schedules the END notification at the right time. Returning null for an unrecognized request lets the base class respond with NOT_UNDERSTOOD.

Two small idioms are worth noting, both consequences of static compilation. A case ReservationReq: label does not narrow the type of msg the way an instanceof check does, so we introduce a typed local variable (req) at the top of the case — without it, accesses like msg.duration will not compile. And we read the message ID through the public messageID property; the msgID shorthand seen in older (dynamically compiled) examples reaches into a protected field, which the static compiler rightly rejects.

To try it on the 2-node network, kill the default CSMA mac and load ours:

> container.kill mac
true
> container.add 'mac', new MySimplestMac()
mac
> mac << new ReservationReq(to: 31, duration: 3.seconds)
ReservationRsp:AGREE
mac >> ReservationStatusNtf:INFORM[to:31 status:START]
mac >> ReservationStatusNtf:INFORM[to:31 status:END]

The START notification arrives right after the AGREE, and the END about 3 seconds later.

Typing the agent’s name shows that our getParameterList() declaration did its job — the standard MacParam service parameters surface in the shell automatically, with read-only markers on the ones we exposed as plain properties:

> mac
« MAC »

[org.arl.unet.mac.MacParam]
  channelBusy ⤇ false
  maxReservationDuration ⤇ Infinity
  recommendedReservationDuration = null
  reservationPayloadSize ⤇ 0

NoteLogging and debugging

Every agent has a Java logger (log) writing to logs/log-0.txt, supporting levels severe, warning, info, fine, finer, finest:

log.fine 'Some debugging information'

Control the log level per class or package with the logLevel command (logLevel FINE), and view the last few log lines from the shell with tail. Logs are also available in the web interface “Logs” tab.

When an agent will not even load, the same log file is where to look: if a class in the classes folder fails to compile, the container.add reports an error on the shell, and the compiler’s diagnostics land in logs/log-0.txt. One gotcha to recognize: if the class file is malformed in a way that leaves the shell parser wanting more input (for example, an unbalanced brace), the shell may show a - continuation prompt instead of an error — press Ctrl-C (or enter a lone }) to get back to the normal prompt and check the log. When in doubt, compile the class manually with groovyc -cp "lib/*" MyAgent.groovy and read the compiler output directly.

TipEvent tracing

Logs tell you what one agent did; the event tracing framework tells you why — it records each stimulus (a message received) together with the response it provoked (a message sent), threading causally related events together across agents and even across nodes. To make your agent’s events traceable, wrap the messages it generates in trace(stimulus, response) — a UnetAgent convenience method that is a no-op unless tracing is active:

send trace(req, new DatagramDeliveryNtf(req))
request trace(ntf, req2), timeout

All the built-in agents do this. Tracing is enabled automatically in simulations (producing logs/trace.json) and can be turned on on real nodes with EventTracer.enable(filename); the resulting trace can be rendered as a sequence diagram of your protocol in action — see Section 34.6.1.

33.5.2 A simple MAC with throttling

The simplest MAC performs poorly under heavy load — every node grabs the channel at once and collisions destroy throughput. We can fix this by adding an exponentially distributed random backoff (a Poisson arrival process), chosen to target a normalized load of about 0.5, which maximizes ALOHA throughput. Here is the full agent:

import org.arl.fjage.*
import org.arl.fjage.param.Parameter
import org.arl.unet.*
import org.arl.unet.mac.*
import org.arl.unet.phy.*
import groovy.transform.CompileStatic

@CompileStatic
class MySimpleThrottledMac extends UnetAgent {

  private final static double TARGET_LOAD = 0.5
  private final static int    MAX_QUEUE_LEN = 16

  private AgentID phy
  boolean busy = false
  Long t0 = null
  Long t1 = null
  int waiting = 0

  @Override void setup()   { register Services.MAC }
  @Override void startup() { phy = agentForService(Services.PHYSICAL) }   // lookup after services are advertised

  @Override
  Message processRequest(Message msg) {
    switch (msg) {
      case ReservationReq:
        ReservationReq req = (ReservationReq)msg
        if (req.duration <= 0) return new RefuseRsp(req, 'Bad reservation duration')
        if (waiting >= MAX_QUEUE_LEN) return new RefuseRsp(req, 'Queue full')
        ReservationStatusNtf ntf1 = new ReservationStatusNtf(
          recipient: req.sender, inReplyTo: req.messageID, to: req.to,
          status: ReservationStatus.START)
        ReservationStatusNtf ntf2 = new ReservationStatusNtf(
          recipient: req.sender, inReplyTo: req.messageID, to: req.to,
          status: ReservationStatus.END)
        AgentLocalRandom rnd = AgentLocalRandom.current()                 // repeatable in discrete-event simulation
        double backoff = rnd.nextExp(TARGET_LOAD/req.duration/nodes)      // rate chosen to hit the target load
        long t = currentTimeMillis()
        if (t0 == null || t0 < t) t0 = t
        t0 += Math.round(1000*backoff)
        if (t0 < t1) t0 = t1
        long duration = Math.round(1000*req.duration)
        t1 = t0 + duration
        waiting++
        add new WakerBehavior(t0-t, {                                     // START after backoff, END after duration
          send ntf1
          busy = true
          waiting--
          add new WakerBehavior(duration, { send ntf2; busy = false })
        })
        return new ReservationRsp(req)
      case ReservationCancelReq:
      case ReservationAcceptReq:
      case TxAckReq:
        return new RefuseRsp(msg, 'Not supported')
    }
    return null
  }

  @Override
  List<Parameter> getParameterList() {
    return allOf(MacParam, Param)
  }

  enum Param implements Parameter { nodes }                              // one user-configurable parameter

  int nodes = 6                            // number of nodes, set by the user
  final int reservationPayloadSize = 0
  final int ackPayloadSize = 0
  final float maxReservationDuration = Float.POSITIVE_INFINITY

  boolean getChannelBusy() { return busy }                               // reflects the real reservation state

  float getRecommendedReservationDuration() {                            // assume one frame per reservation
    return (float)get(phy, Physical.DATA, PhysicalChannelParam.frameDuration)
  }

}

A few points to note. Other agents are looked up in startup(), after they’ve had a chance to advertise their services during setup. Random numbers are drawn from AgentLocalRandom, which gives repeatable results during discrete-event simulation — the preferred way to generate random variates in an agent — and nextExp() returns an exponentially distributed backoff, with a rate chosen to hit the target load. The START notification is no longer immediate: it is scheduled after the backoff, and the END after the reservation duration. We advertise one user-configurable parameter, nodes, and channelBusy now reflects the real reservation state. Finally, recommendedReservationDuration is derived from the PHYSICAL service’s frame duration (Chapter 17), assuming most reservations carry one frame — the generic get() helper returns an Object, so under static compilation we cast it to the declared type.

33.5.3 A simple MAC with handshake

Many MAC protocols (MACA, FAMA, …) use an RTS/CTS handshake. The initiator’s MAC sends a short request-to-send (RTS) PDU to the responder, which answers with a clear-to-send (CTS); on receiving the CTS, the initiator tells its client to go ahead with the transmission. Any neighbor that overhears an RTS or CTS knows the channel is about to be busy and backs off silently — this is what protects the exchange from being trampled by nodes that can hear only one of the two parties. Figure 33.2 shows a successful handshake, together with the messages exchanged between the agents that implement it.

sequenceDiagram
    participant C as Client agent<br/>(node A)
    participant A as MAC<br/>(node A)
    participant B as MAC<br/>(node B)
    participant N as MAC<br/>(node C, neighbor)
    C->>A: ReservationReq(to: B, duration)
    A-->>C: ReservationRsp (AGREE)
    note over A: IDLE → RTS<br/>(after a random backoff)
    A->>B: RTS PDU
    note over B: IDLE → RX
    note over N: overhears RTS<br/>IDLE → BACKOFF
    B->>A: CTS PDU
    note over A: RTS → TX
    note over N: overhears CTS<br/>backoff extended
    A-->>C: ReservationStatusNtf (START)
    note over C,B: client transmits data for the reservation duration
    A-->>C: ReservationStatusNtf (END)
    note over A: TX → IDLE

Figure 33.2: A successful RTS/CTS handshake between nodes A and B, overheard by a neighbor node C. Notes show the state transitions of each MAC agent.

Such protocols are naturally described as a finite state machine (FSM). When idle, the agent waits; on a ReservationReq it sends an RTS and moves to the RTS state; a returning CTS moves it to TX (where it tells the client to transmit); after the reservation it returns to idle. If no CTS arrives in time, it retries or fails. An incoming RTS moves it to RX, where it replies with a CTS. Overheard (snooped) RTS/CTS PDUs move it to a BACKOFF state to avoid interfering with other nodes’ exchanges.

We develop the agent as class MySimpleHandshakeMac, statically compiled like the others, and walk through it fragment by fragment. We tag our RTS/CTS PDUs with the MAC protocol number, and define some timeouts:

import org.arl.fjage.*
import org.arl.fjage.param.Parameter
import org.arl.unet.*
import org.arl.unet.mac.*
import org.arl.unet.phy.*
import org.arl.unet.nodeinfo.NodeInfoParam
import groovy.transform.*

@CompileStatic
class MySimpleHandshakeMac extends UnetAgent {

  int PROTOCOL = Protocol.MAC

  float RTS_BACKOFF    = 2     // seconds
  float CTS_TIMEOUT    = 5     // seconds
  float BACKOFF_RANDOM = 5     // seconds
  float MAX_PROP_DELAY = 2     // seconds

Note that we write the timeouts as plain numbers: the 2.seconds-style unit sugar we enjoy in shell and simulation scripts is a runtime Groovy extension, which a statically compiled class cannot use.

UnetStack provides a PDU class to encode and decode protocol data units declaratively. Our PDU has a type field and a duration field:

int RTS_PDU = 0x01
int CTS_PDU = 0x02

PDU pdu = PDU.withFormat {
  uint8('type')         // RTS_PDU / CTS_PDU
  uint16('duration')    // ms
}

NoteEncoding and decoding PDUs

The PDU class describes a format declaratively; encoding and decoding is then just encode() and decode():

> import java.nio.ByteOrder
> pdu = PDU.withFormat {
-   length(16)                     // 16 byte PDU
-   order(ByteOrder.BIG_ENDIAN)    // big-endian byte ordering
-   uint8('type')                  // 1 byte field 'type'
-   uint8(0x01)                    // literal byte 0x01
-   filler(2)                      // 2 filler bytes
-   uint16('data')                 // 2 byte field 'data'
-   padding(0xff)                  // pad with 0xff to 16 bytes
- };
> bytes = pdu.encode([type: 7, data: 42])
[7, 1, 0, 0, 0, 42, -1, -1, -1, -1, -1, -1, -1, -1, -1, -1]
> pdu.decode(bytes)
[data:42, type:7]

Supported fields include uint8, int8, uint16, int16, uint32, int32, int64 and chars (string), with length, order, filler and padding declarations.

The heart of the protocol is the FSM, drawn in Figure 33.3. Transitions labelled with events (CTS received, overheard RTS/CTS, …) are triggered by incoming PDUs; the remaining transitions are driven by timers started when a state is entered. An action after a slash is performed on entering the target state — for instance, request queued / RTS means that when a reservation request is queued, the agent moves to the RTS state and transmits an RTS PDU.

stateDiagram-v2
    direction LR
    [*] --> IDLE
    IDLE --> RTS: request queued / RTS
    RTS --> TX: CTS received
    RTS --> IDLE: CTS timeout
    TX --> IDLE: reservation over
    IDLE --> RX: RTS received / CTS
    RX --> IDLE: peer's reservation over
    IDLE --> BACKOFF: overheard RTS/CTS
    BACKOFF --> BACKOFF: overheard RTS/CTS
    BACKOFF --> IDLE: backoff expired

Figure 33.3: The handshake MAC finite state machine. RTS and TX handle our own reservations, RX other nodes’ requests to us, and BACKOFF overheard exchanges between other nodes.

In code, we define the states and events as enums, then build an FSMBehavior using the FSMBuilder utility — each state(...) block corresponds to a state in the diagram, onEvent closures to the event-triggered transitions, and after(...) timers to the timeout transitions:

enum State { IDLE, RTS, TX, RX, BACKOFF }
enum Event { RX_RTS, RX_CTS, SNOOP_RTS, SNOOP_CTS }

int MAX_RETRY = 3
int MAX_QUEUE_LEN = 16

Queue<ReservationReq> queue = new ArrayDeque<ReservationReq>(MAX_QUEUE_LEN)

FSMBehavior fsm = buildFSM()

private FSMBehavior buildFSM() {
  return FSMBuilder.build {

    int retryCount = 0
    double backoff = 0
    Map rxInfo
    def rnd = AgentLocalRandom.current()

    state(State.IDLE) {
      action {
        if (!queue.isEmpty()) {
          after(rnd.nextDouble(0, BACKOFF_RANDOM)) { setNextState(State.RTS) }
        }
        block()
      }
      onEvent(Event.RX_RTS)    { Map info -> rxInfo = info; setNextState(State.RX) }
      onEvent(Event.SNOOP_RTS) { backoff = RTS_BACKOFF; setNextState(State.BACKOFF) }
      onEvent(Event.SNOOP_CTS) { Map info ->
        backoff = (double)info.duration + 2*MAX_PROP_DELAY
        setNextState(State.BACKOFF)
      }
    }

    state(State.RTS) {
      onEnter {
        ReservationReq msg = queue.peek()
        def bytes = pdu.encode(type: RTS_PDU, duration: Math.ceil(msg.duration*1000))
        phy << new TxFrameReq(to: msg.to, type: Physical.CONTROL, protocol: PROTOCOL, data: bytes)
        after(CTS_TIMEOUT) {
          if (++retryCount >= MAX_RETRY) {
            sendReservationStatusNtf(queue.poll(), ReservationStatus.FAILURE)
            retryCount = 0
          }
          setNextState(State.IDLE)
        }
      }
      onEvent(Event.RX_CTS) { setNextState(State.TX) }
    }

    state(State.TX) {
      onEnter {
        ReservationReq msg = queue.poll()
        retryCount = 0
        sendReservationStatusNtf(msg, ReservationStatus.START)
        after(msg.duration) {
          sendReservationStatusNtf(msg, ReservationStatus.END)
          setNextState(State.IDLE)
        }
      }
    }

    state(State.RX) {
      onEnter {
        def bytes = pdu.encode(type: CTS_PDU, duration: Math.round((double)rxInfo.duration*1000))
        phy << new TxFrameReq(to: (int)rxInfo.from, type: Physical.CONTROL, protocol: PROTOCOL, data: bytes)
        after((double)rxInfo.duration + 2*MAX_PROP_DELAY) { setNextState(State.IDLE) }
        rxInfo = null
      }
    }

    state(State.BACKOFF) {
      onEnter { after(backoff) { setNextState(State.IDLE) } }
      onEvent(Event.SNOOP_RTS) { backoff = RTS_BACKOFF; reenterState() }
      onEvent(Event.SNOOP_CTS) { Map info ->
        backoff = (double)info.duration + 2*MAX_PROP_DELAY
        reenterState()
      }
    }

  }
}

Notice that the whole agent — FSM included — compiles statically. FSMBuilder implements a little domain-specific language, in which calls like state(...), onEnter, onEvent, after and setNextState are not methods of our agent: they are resolved against the builder and FSM objects through closure delegates, which fjåge declares with @DelegatesTo annotations so the static compiler can verify them. The only concession to static typing is at the loosely-typed edge of the FSM: the info delivered with each event is declared as a Map closure parameter, and its entries are cast where they are used.

Three design details are easy to miss in the code, because they don’t appear in the FSM diagram. First, the IDLE state does not fire off an RTS the instant a request is queued — it waits a random delay (rnd.nextDouble(0, BACKOFF_RANDOM)) first, so that on a loaded network two nodes with queued requests don’t keep colliding their RTS PDUs. Second, the retryCount/MAX_RETRY logic in the RTS state gives up on a reservation after a few failed handshake attempts, reporting FAILURE to the requester rather than retrying forever. Third, backoff is a variable, deliberately set before each transition into BACKOFF: an overheard RTS warrants only a short wait (RTS_BACKOFF — the handshake may yet fail), while an overheard CTS means a reservation is definitely starting, so we back off for its whole duration plus propagation allowance.

NoteFinite state machines

The FSMBuilder makes setting up an FSM behavior easy:

1. Define states and events as enum declarations.
2. Build the FSMBehavior with FSMBuilder.build, with a state(...) for each state.
3. In each state, use action (run continuously, like a cyclic behavior — call block()/restart() to avoid busy loops), onEnter/onExit (run on entry/exit), onEvent (triggered via trigger()), and after (timers fired a set time after entry).
4. Use setNextState() and reenterState() to effect transitions.
5. For short-lived FSMs, call terminate() when done.

A small helper sends the reservation status notifications:

void sendReservationStatusNtf(ReservationReq msg, ReservationStatus status) {
  send new ReservationStatusNtf(
    recipient: msg.sender, inReplyTo: msg.messageID,
    to: msg.to, from: addr, status: status)
}

We wire everything together in setup() and startup() — registering the MAC service, looking up the PHYSICAL service, subscribing to its topic (to overhear other nodes’ PDUs) and to the DATAGRAM topic (for PDUs addressed to us), looking up our own address, and starting the FSM:

AgentID phy
int addr

void setup() {
  register Services.MAC
}

void startup() {
  phy = agentForService(Services.PHYSICAL)
  subscribe topic(phy)                  // overheard (snooped) frames
  subscribe topic(Topics.DATAGRAM)      // frames addressed to this node
  add new OneShotBehavior({
    def nodeInfo = agentForService(Services.NODE_INFO)
    addr = (int)get(nodeInfo, NodeInfoParam.address)
  })
  add(fsm)
}

Why does the address lookup sit inside a OneShotBehavior rather than running directly in startup()? The get() call blocks while it queries the node-info agent, which may itself still be starting up. Wrapping it in a behavior lets startup() return immediately — the lookup then runs once the agent (and the rest of the stack) is up, without holding anything else back.

Tip

Here we looked up one PHYSICAL provider and subscribed to its topic. When an agent instead wants notifications from every provider of a service — including providers that are loaded later — it can call subscribeForService(service), which tracks the providers of the service and keeps the subscriptions up to date.

Incoming PDUs arrive as RxFrameNtf messages; we decode them and trigger the matching FSM events — RX_* for PDUs destined to us, SNOOP_* for those we overhear. PDU.decode() returns a map of field values, so we cast the duration to a number before converting it to seconds:

void processMessage(Message msg) {
  if (msg instanceof RxFrameNtf && msg.protocol == PROTOCOL) {
    def rx = pdu.decode(msg.data)
    def info = [from: msg.from, to: msg.to, duration: ((int)rx.duration)/1000.0]
    if (rx.type == RTS_PDU)
      fsm.trigger(info.to == addr ? Event.RX_RTS : Event.SNOOP_RTS, info)
    else if (rx.type == CTS_PDU)
      fsm.trigger(info.to == addr ? Event.RX_CTS : Event.SNOOP_CTS, info)
  }
}

As before, we expose the MAC service parameters, with channelBusy now reflecting whether the FSM is in a non-idle state. Unlike the earlier MACs, maxReservationDuration is now finite — our PDU carries the reservation duration in a uint16 millisecond field, which caps out at 65535 ms, so the parameter advertises exactly that limit (and the ReservationReq validation below enforces it):

@Override
List<Parameter> getParameterList() {
  return allOf(MacParam)
}

final int reservationPayloadSize = 0
final int ackPayloadSize = 0
final float maxReservationDuration = 65.535    // uint16 ms field in the PDU
final Float recommendedReservationDuration = null

boolean getChannelBusy() {
  return fsm.currentState != State.IDLE
}

Finally, a ReservationReq is validated, queued, and the FSM restarted:

@Override
Message processRequest(Message msg) {
  switch (msg) {
    case ReservationReq:
      ReservationReq req = (ReservationReq)msg
      if (req.to == Address.BROADCAST || req.to == addr)
        return new RefuseRsp(req, 'Reservation must have a destination node')
      if (req.duration <= 0 || req.duration > maxReservationDuration)
        return new RefuseRsp(req, 'Bad reservation duration')
      if (queue.size() >= MAX_QUEUE_LEN)
        return new RefuseRsp(req, 'Queue full')
      queue.add(req)
      fsm.restart()    // wake the fsm, which blocks when the queue is empty
      return new ReservationRsp(req)
    case ReservationCancelReq:
    case ReservationAcceptReq:
    case TxAckReq:
      return new RefuseRsp(msg, 'Not supported')
  }
  return null
}

To test the handshake MAC, the protocol must run on all nodes, so load MySimpleHandshakeMac on both node A and node B (after killing the default mac). Subscribe to phy on node A and make a reservation, and you’ll see the RTS/CTS PDUs exchanged before the reservation starts:

> subscribe phy
> mac << new ReservationReq(to: 31, duration: 3.seconds)
ReservationRsp:AGREE
phy >> TxFrameStartNtf:INFORM[type:CONTROL txTime:3631928985 txDuration:950]
phy >> RxFrameStartNtf:INFORM[type:CONTROL rxTime:3634151681]
phy >> RxFrameNtf:INFORM[type:CONTROL from:31 to:232 protocol:4 rxTime:3634151681 (3 bytes)]
mac >> ReservationStatusNtf:INFORM[to:31 from:232 status:START]
mac >> ReservationStatusNtf:INFORM[to:31 from:232 status:END]

The reservation begins once the handshake completes, and ends 3 seconds later — exactly as designed.

The walkthrough above showed the agent in fragments; the complete, tested source of MySimpleHandshakeMac is listed in Chapter 37. All three MAC agents from this chapter (MySimplestMac, MySimpleThrottledMac and MySimpleHandshakeMac) also ship in the samples folder of the Unet distribution, ready to load and experiment with.

Tip

Advertising services and optional capabilities, honouring requests, and publishing notifications are all straightforward to implement. For a worked example of a PHYSICAL service agent (a modem driver), see this blog article.

33.6 Fragmentation and reassembly

An agent that delivers datagrams larger than the frames of the layer below — a link agent carrying multi-frame datagrams over the physical layer, or a transport agent carrying large transfers over a link — has to split the data into fragments on transmission and stitch them back together on reception. Rather than have every such agent reinvent this, UnetStack provides a fragmentation framework in org.arl.unet.utils: a Fragmenter interface that turns a byte array into fragments, and a matching Reassembler that reconstructs the original data from them.

The contract is deliberately simple. A fragmenter is set up with the data and a fragment size, and then iterated:

import org.arl.unet.utils.*

Fragmenter frag = new SimpleFragmenter().setup(data, fragLen)
while (frag.hasMoreFragments()) {
  byte[] fragment = frag.nextFragment()
  // transmit the fragment, e.g. as one frame
}

A reassembler is set up with the expected data length and the same fragment size, and fed fragments as they arrive (in any order) until it reports completion:

Reassembler reasm = new SimpleReassembler().setup(dataLen, fragLen)
// as each fragment arrives:
reasm.addFragment(fragment)
if (reasm.hasFinishedReassembly()) {
  byte[] data = reasm.getData()
  // deliver the reassembled datagram
}

Both sides also expose introspection methods (getMinFragmentCount(), getMaxFragmentCount(), getFragmentCount(), etc.) for sizing headers and tracking progress. The framework does not dictate how fragments travel — your protocol still decides what per-fragment header (source, sequence number, data length, …) it needs so that the receiver can direct each fragment to the right reassembler.

Two implementations ship with UnetStack:

• SimpleFragmenter / SimpleReassembler — each fragment carries a small offset header (2 bytes, or 4 bytes for data of 64 kB or more) within the fragment size you specify. Every fragment is required for reassembly, so pair it with a reliable delivery mechanism (or accept that a lost fragment costs the whole datagram). This is what UdpLink uses over IP networks.
• ECFragmenter / ECReassembler — an erasure-coded engine: the fragmenter can generate more fragments than the minimum needed, and the reassembler reconstructs the data from any sufficiently large subset, no matter which fragments were lost. This suits half-duplex, high-latency channels where per-fragment acknowledgements are expensive — it is the engine inside the default underwater link, ECLink (Section 18.6.1).

33.7 Writing shell extensions

All the service commands used throughout this handbook — routes, addroute, bbrec, pclr, and the rest — are not built into the shell. Each is contributed by a shell extension: a class that the shell loads to gain a set of commands, variables and help topics (Chapter 30). When your agent would benefit from convenient shell commands — and any agent an operator interacts with does — you write one the same way.

A shell extension is a class implementing the (empty) marker interface org.arl.fjage.shell.ShellExtension, on which the shell recognizes a few conventions:

• A static void __init__(ScriptEngine engine) method, if present, is called when the extension is loaded into a shell. Use it to import classes into the shell’s namespace and to keep the engine for later use — through it, your commands can read shell variables, most usefully __agent__, the shell agent itself, which lets a command send and receive messages on the user’s behalf.
• Public static methods become shell commands (routes 2 calls routes(2)), and static getters become shell variables (getRoutes() makes the bare command routes work).
• A public final static String __doc__ markdown string provides the help entries, in the same format used to document agents (Chapter 31).
• Errors are reported by throwing ShellCommandFailed with a user-facing message.

The router’s own extension shows all of these conventions in a few lines (abridged from org.arl.unet.net.RouterShellExt):

import org.arl.fjage.AgentID
import org.arl.fjage.Performative
import org.arl.fjage.shell.*
import org.arl.unet.*

1class RouterShellExt implements ShellExtension {

  public final static String __doc__ = '''\
# router - routing service

## Commands:

### routes - print routing table

Examples:
 routes              // display routing table
 routes 2            // display routes to node 2
2'''.stripIndent()

  static private final ThreadLocal<ScriptEngine> engine = new ThreadLocal<ScriptEngine>()

3  static void __init__(ScriptEngine engine) {
    this.engine.set(engine)
    engine.importClasses('org.arl.unet.net.*')
  }

4  static AgentID getRouter() {
    return ShellExtUtils.selectAgentID(engine.get(), Services.ROUTING, 'router')
  }

5  static String routes(int to = -1) {
    def rr = router
6    if (rr == null) throw new ShellCommandFailed('Router not found')
    def req = new GetRouteReq(to: to, all: true)
    rr.send(req)
7    def agent = engine.get().getVariable('__agent__')
    def s = ''
    def rsp = agent.receive(req, 1000)
    while (rsp) {
      if (rsp instanceof RouteInfo) s += "$rsp\n"          // format table row
      if (rsp.performative == Performative.AGREE) break
      rsp = agent.receive(req, 1000)
    }
    return s ?: 'No routes'
  }

}

1

The marker interface that tells the shell this class contributes commands.

2

The __doc__ string feeds help router and help routes — and, via the DOCUMENTATION service, this handbook’s own command tables.

3

Called on load; the engine is kept in a ThreadLocal (one per shell), and useful classes are imported so users can type GetRouteReq unqualified.

4

A static getter — in the shell, typing router now resolves to the routing agent’s AgentID. The ShellExtUtils.selectAgentID helper finds the agent providing a service, preferring one whose name matches (and letting the user disambiguate by setting a shell variable when several match).

5

A static method — in the shell, routes or routes 2 invokes it, and its return value is displayed.

6

Failures are signalled with ShellCommandFailed, which the shell reports cleanly.

7

Commands run in the shell agent’s context: __agent__ lets the command use the shell agent to send requests and collect responses.

Loading an extension:

Put your extension class in the classes folder (or a jar in jars), like any custom agent. Extensions are then loaded into a shell with the shell’s run command using a cls:// URL — UnetStack’s own extensions are loaded exactly this way in etc/fshrc.groovy. To load your own automatically in every Groovy shell, add a line to a fshrc.groovy script in the scripts folder (the same startup script that holds convenience closures, Chapter 3):

run 'cls://my.pkg.MyShellExt'

On AT shells (Chapter 12), extensions are loaded at runtime with AT~EXT=my.pkg.MyShellExt instead.

33.8 Do’s and don’ts

A few practices keep agents fast, robust and consistent with the rest of the stack.

Do

• Compile Groovy agents statically. Annotate your agent class with @CompileStatic (from groovy.transform). Static compilation catches type errors at build time and runs much faster than dynamic dispatch — important on resource-constrained modems. Where you genuinely need dynamic features in a method, exclude just that method with @CompileDynamic. One caveat: do not build an FSM inside a non-static inner class — the Groovy compiler mis-resolves the DSL closures there (the code compiles, then fails at runtime). Make the state-holder a static nested class carrying an explicit reference to its agent instead; any agent member you forget to qualify through that reference then fails at compile time rather than at sea.
• Define parameters with an inner enum implementing Parameter, and provide a __doc__ string (as shown in Chapter 33 above). This is what makes your parameters and commands discoverable from the shell and in this handbook.
• Ship a shell extension for operator-facing agents (Section 33.7) — commands documented under help and loadable into any shell beat ad-hoc convenience closures in fshrc.groovy.
• Register services and capabilities in setup(), but defer looking up other agents’ services to startup() — at setup() time other agents may not have registered yet.
• Do all timing and waiting with behaviors — WakerBehavior, TickerBehavior, BackoffBehavior, an FSMBehavior, etc. — so the agent stays responsive to other messages.
• Reply to requests by returning a response from processRequest() (an AGREE/REFUSE/FAILURE or a typed response), and report asynchronous outcomes by publishing notifications on your topic.
• Use protocol numbers Protocol.USER and above for your own application protocols; the lower numbers are reserved (see Chapter 16).

Don’t

• Don’t block the agent’s thread. Never call delay()/Thread.sleep() or do blocking I/O inside processMessage()/processRequest() — the agent cannot handle anything else while blocked. Schedule the work in a behavior instead.
• Don’t do heavy work — or service lookups — in the constructor. The container and other agents may not be ready. Use setup()/startup().
• Don’t bypass the message API. Interact with other agents by sending messages (request/response, notifications), not by calling their methods directly; this keeps agents decoupled and lets them live on different nodes (see Chapter 11).
• Don’t create feedback loops. When echoing or forwarding datagrams, change the protocol number (or otherwise mark them) so a response is not itself re-processed and forwarded forever.