From Chapter 10 we know robots must respect real-time constraints and handle multiple sensor/actuator interfaces. The software architecture introduced in Lessons 1–2 layers drivers, middleware, and applications. Data flow is the operational glue: it defines what data is produced, how it is transported, and when consumers can act on it.

Examples include sensor streams (IMU, encoder, camera), state estimates, and control commands. Poorly designed data flow can introduce delays, jitter, or deadlocks, which directly degrade closed-loop control performance.

flowchart TD
  S1["Sensors + Drivers"] --> M1["Middleware Transport"]
  M1 --> P1["Perception / Estimation"]
  P1 --> D1["Decision / Planning"]
  D1 --> C1["Control / Actuation"]
  C1 --> A1["Actuators + Drivers"]
  P1 --> L1["Logging / Monitoring"]
        

We now turn this picture into a rigorous model.

A robot software pipeline can be represented by a directed graph \( G = (V,E) \) where:

• \( v \in V \) is a computational component (driver, filter, controller, logger).
• \( e=(i\rightarrow j)\in E \) is a data dependence: node \(j\) consumes messages produced by \(i\).

Each node \( i \) has: sampling or trigger rate \( f_i \) (Hz), compute time random variable \( C_i \), and output message size \( s_i \). Each edge \( e \) has transport delay random variable \( D_e \) and optional buffering.

If a message is produced at time \( t \) by node \(i\), its arrival time at consumer \(j\) is

\[ t_{arr} = t + C_i + D_{(i\rightarrow j)} . \]

For a path \( \pi = (v_1\rightarrow v_2\rightarrow\cdots\rightarrow v_m)\), the end-to-end latency is the sum of compute and transport delays:

\[ L_{\pi} = \sum_{k=1}^{m-1} \Big(C_{v_k} + D_{(v_k\rightarrow v_{k+1})}\Big). \]

Determinism vs. stochasticity. In embedded robots, \(C_i\) and \(D_e\) may have bounded support: \( 0 \le C_i \le \overline{C}_i \), \( 0 \le D_e \le \overline{D}_e \). Then

\[ 0 \le L_{\pi} \le \overline{L}_\pi = \sum_{k=1}^{m-1} \Big(\overline{C}_{v_k} + \overline{D}_{(v_k\rightarrow v_{k+1})}\Big). \]

Architectures aim to minimize both the mean latency \(E[L_\pi]\) and the worst-case bound \(\overline{L}_\pi\).

3.1 Synchronous (Blocking) Passing

In synchronous passing, sender and receiver rendezvous. The sender blocks until the receiver accepts the data. Let \(S_i(t)\) be the send time and \(R_j(t)\) be the receive-acknowledge time. A simple model is:

\[ R_j(t) = S_i(t) + C_i + D_{(i\rightarrow j)} + W_j, \]

where \( W_j \) is waiting time until node \(j\) is ready. This improves determinism of ordering but can propagate delays backward, coupling unrelated components.

3.2 Asynchronous (Non-Blocking) Passing

In asynchronous passing, messages are placed in a buffer/queue. Sender continues immediately; receiver consumes later. Modeling the buffer as a discrete-time queue of capacity \(B\), let \(q(k)\) be queue occupancy at step \(k\). If arrivals \(a(k)\) and services \(b(k)\) are counted per step:

\[ q(k+1) = \min\big(B,\; \max(0,\; q(k)+a(k)-b(k))\big). \]

Overflow occurs if \(q(k)=B\) and \(a(k) > b(k)\), implying message loss or backpressure depending on policy.

3.3 Publish–Subscribe vs. Request–Response

Two generic patterns:

• Publish–Subscribe: producers publish to a channel; any number of subscribers receive a copy. This is naturally asynchronous.
• Request–Response: clients request data or action and wait for reply; typically synchronous.

sequenceDiagram
  participant Pub as Publisher
  participant Q as Queue/Channel
  participant Sub as Subscriber
  Pub->>Q: publish(msg)
  Pub-->>Pub: continues
  Sub->>Q: take()
  Q-->>Sub: msg delivered (later)
        

Notice how the asynchronous queue decouples temporal execution of components.

Consider a single producer–consumer edge with mean arrival rate \( \lambda \) (messages/s) and mean service rate \( \mu \) (messages/s). Under standard queueing assumptions (e.g., renewal arrivals, independent service), stability requires:

\[ \rho = \frac{\lambda}{\mu} < 1. \]

Proposition (Stability Condition). If \( \rho < 1 \) then the expected queue length is finite: \(E[q] < \infty\). If \( \rho \ge 1 \), the queue diverges.

Proof (sketch via drift). Let \(q(k)\) follow the queue recursion from Section 3.2 with unbounded \(B\). Taking expectation:

\[ E[q(k+1)-q(k)] = E[a(k)] - E[b(k)] = \lambda\Delta t - \mu\Delta t = (\lambda-\mu)\Delta t. \]

If \(\lambda < \mu\), the drift is negative for large \(q(k)\), so the Markov chain is positive recurrent and \(E[q]\) is bounded. If \(\lambda \ge \mu\), drift is nonnegative and \(q(k)\) grows without bound. \(\square\)

Latency in stable queues. Little’s Law applies:

\[ E[q] = \lambda E[T], \]

where \(T\) is the waiting + service time. Hence average waiting time:

\[ E[T] = \frac{E[q]}{\lambda}. \]

For an M/M/1 approximation (Poisson arrivals, exponential service), the mean waiting time is

\[ E[T] = \frac{1}{\mu-\lambda}. \]

Even if a robot is not perfectly Poisson/exponential, this formula highlights a key design principle: approaching \(\rho \to 1\) causes latency blow-up.

4.1 End-to-End Bound with Multiple Queues

For a pipeline path \(\pi\) where each edge maintains a stable queue with arrival \(\lambda_k\) and service \(\mu_k\), the mean end-to-end latency is

\[ E[L_\pi] = \sum_{k=1}^{m-1} \left( E[C_{v_k}] + \frac{1}{\mu_k-\lambda_k} \right). \]

This gives an architectural way to allocate compute budgets: increase \(\mu_k\) (faster consumer or smaller message), reduce \(\lambda_k\) (downsample or filter), or split the path (parallelize).

Robot components often operate at different rates. Let producer \(i\) publish timestamps \(t_i[n] = n/f_i\). Consumer \(j\) runs at \(f_j\). A key problem is aligning measurements with computation time.

Suppose consumer reads the freshest message in its queue at time \(t\). The data age is \( \Delta(t) = t - t_{msg} \). With bounded compute/transport delays, if producer period is \(T_i = 1/f_i\),

\[ 0 \le \Delta(t) \le T_i + \overline{C}_i + \overline{D}_{(i\rightarrow j)}. \]

If control requires age \(\Delta(t) \le \Delta_{max}\), a sufficient feasibility condition is

\[ T_i + \overline{C}_i + \overline{D}_{(i\rightarrow j)} \le \Delta_{max}. \]

This is a software-architecture analogue to sampling constraints in control.

6.1 Python (async publish–subscribe with ZeroMQ)

# pip install pyzmq
import zmq, time, json, threading

CTX = zmq.Context()

def publisher(rate_hz=50):
    pub = CTX.socket(zmq.PUB)
    pub.bind("tcp://*:5556")
    k = 0
    T = 1.0 / rate_hz
    while True:
        msg = {"seq": k, "t_pub": time.time()}
        pub.send_string("imu " + json.dumps(msg))
        k += 1
        time.sleep(T)

def subscriber(mu_hz=40):
    sub = CTX.socket(zmq.SUB)
    sub.connect("tcp://localhost:5556")
    sub.setsockopt_string(zmq.SUBSCRIBE, "imu")
    T = 1.0 / mu_hz
    while True:
        topic, payload = sub.recv_string().split(" ", 1)
        msg = json.loads(payload)
        age = time.time() - msg["t_pub"]
        print(f"recv seq={msg['seq']} age={age:.4f}s")
        time.sleep(T)

threading.Thread(target=publisher, daemon=True).start()
subscriber()
      

Here \(\lambda \approx 50\) Hz and \(\mu \approx 40\) Hz so \(\rho > 1\): you should observe increasing age or drops depending on transport buffers.

6.2 C++ (ZeroMQ PUB/SUB)

// g++ pubsub.cpp -lzmq -std=c++17
#include <iostream>
#include <thread>
#include <chrono>
#include <zmq.hpp>
#include <nlohmann/json.hpp>

using json = nlohmann::json;

void publisher(double rate_hz){
    zmq::context_t ctx(1);
    zmq::socket_t pub(ctx, ZMQ_PUB);
    pub.bind("tcp://*:5556");
    int k=0;
    auto T = std::chrono::duration<double>(1.0/rate_hz);
    while(true){
        json msg = {
          {"seq",k},{"t_pub",(double)std::chrono::duration<double>(
            std::chrono::system_clock::now().time_since_epoch()).count()}
          };
        std::string out = "imu " + msg.dump();
        pub.send(zmq::buffer(out), zmq::send_flags::none);
        k++;
        std::this_thread::sleep_for(T);
    }
}

void subscriber(double mu_hz){
    zmq::context_t ctx(1);
    zmq::socket_t sub(ctx, ZMQ_SUB);
    sub.connect("tcp://localhost:5556");
    sub.setsockopt(ZMQ_SUBSCRIBE, "imu", 3);
    auto T = std::chrono::duration<double>(1.0/mu_hz);
    while(true){
        zmq::message_t m;
        sub.recv(m, zmq::recv_flags::none);
        std::string s((char*)m.data(), m.size());
        auto pos = s.find(" ");
        std::string payload = s.substr(pos+1);
        auto msg = json::parse(payload);
        double now = (double)std::chrono::duration<double>(
            std::chrono::system_clock::now().time_since_epoch()).count();
        std::cout << "seq=" << msg["seq"]
                  << " age=" << (now - (double)msg["t_pub"]) << "s\n";
        std::this_thread::sleep_for(T);
    }
}

int main(){
    std::thread(pub, publisher, 50.0).detach();
    subscriber(40.0);
}
      

6.3 Java (PUB/SUB via JeroMQ)

// Gradle/Maven: org.zeromq:jeromq
import org.zeromq.ZMQ;
import org.json.JSONObject;

public class PubSub {
    static void publisher(double rateHz){
        ZMQ.Context ctx = ZMQ.context(1);
        ZMQ.Socket pub = ctx.socket(ZMQ.PUB);
        pub.bind("tcp://*:5556");
        long k = 0;
        long Tms = (long)(1000.0/rateHz);
        while(true){
            JSONObject msg = new JSONObject();
            msg.put("seq", k);
            msg.put("t_pub", System.currentTimeMillis()/1000.0);
            pub.send("imu " + msg.toString());
            k++;
            try { Thread.sleep(Tms); } catch(Exception e){}
        }
    }

    static void subscriber(double muHz){
        ZMQ.Context ctx = ZMQ.context(1);
        ZMQ.Socket sub = ctx.socket(ZMQ.SUB);
        sub.connect("tcp://localhost:5556");
        sub.subscribe("imu");
        long Tms = (long)(1000.0/muHz);
        while(true){
            String s = sub.recvStr();
            String payload = s.split(" ",2)[1];
            JSONObject msg = new JSONObject(payload);
            double age = System.currentTimeMillis()/1000.0 - msg.getDouble("t_pub");
            System.out.println("seq=" + msg.getLong("seq") + " age=" + age);
            try { Thread.sleep(Tms); } catch(Exception e){}
        }
    }

    public static void main(String[] args){
        new Thread(() -> publisher(50.0)).start();
        subscriber(40.0);
    }
}
      

6.4 Matlab / Simulink (rate-transition + buffer)

In Simulink, asynchronous passing is represented using a Rate Transition block and a bounded buffer (e.g., FIFO queue). A minimal script for a discrete buffer model:

% Discrete-time FIFO buffer simulation
B = 10;          % capacity
lambda = 50;     % producer Hz
mu = 40;         % consumer Hz
dt = 0.001;      % 1 ms step
T = 5;           % total time
q = 0;           % queue occupancy
drop = 0;

for t = 0:dt:T
    a = rand < lambda*dt; % Bernoulli arrival in dt
    b = rand < mu*dt;     % Bernoulli service in dt
    q_next = min(B, max(0, q + a - b));
    if q == B && a > b
        drop = drop + 1;
    end
    q = q_next;
end

fprintf("Final q=%d, drops=%d\n", q, drop);
      

In Simulink, you would implement the same logic using: Discrete-Time Integrator (for queue state), Switch/Saturation (for min/max), and a Rate Transition between producer and consumer subsystems.

Problem 1 (End-to-End Latency): A pipeline has three nodes \(v_1 \rightarrow v_2 \rightarrow v_3\). Compute-time bounds are \( \overline{C}_{v_1}=4\text{ ms}\), \( \overline{C}_{v_2}=6\text{ ms}\). Transport bounds are \( \overline{D}_{12}=2\text{ ms}\), \( \overline{D}_{23}=3\text{ ms}\). Find the worst-case end-to-end latency \(\overline{L}_\pi\).

Solution:

\[ \overline{L}_\pi = (\overline{C}_{v_1}+\overline{D}_{12}) +(\overline{C}_{v_2}+\overline{D}_{23}) =(4+2)+(6+3)\text{ ms}=15\text{ ms}. \]

Problem 2 (Queue Stability): A sensor publishes at \(f_s=100\) Hz. A processing node consumes each message with average compute time \(E[C]=8\) ms. Assume transport delay negligible. Determine if the async queue is stable.

Solution:

Arrival rate is \(\lambda = 100\) Hz. Service rate is \(\mu = 1/E[C] = 1/0.008 = 125\) Hz. Hence

\[ \rho=\frac{\lambda}{\mu}=\frac{100}{125}=0.8 < 1, \]

so the queue is stable.

Problem 3 (Latency Blow-Up Near Saturation): Using the M/M/1 approximation, compute \(E[T]\) for \((\lambda,\mu)=(40,50)\) Hz and for \((49,50)\) Hz. Interpret the result.

Solution:

\[ E[T]=\frac{1}{\mu-\lambda}. \]

Case 1: \(\mu-\lambda=10\) Hz \(\Rightarrow E[T]=0.1\) s. Case 2: \(\mu-\lambda=1\) Hz \(\Rightarrow E[T]=1\) s. The mean waiting time increases by a factor of 10 when \(\rho\) approaches 1, illustrating why pipelines must operate with slack.

Problem 4 (Data Age Feasibility): A camera publishes at \(f_i=30\) Hz. Worst-case compute and transport are \(\overline{C}_i=5\) ms and \(\overline{D}=10\) ms. A controller needs data age \(\Delta(t)\le 60\) ms. Is the architecture feasible?

Solution:

Producer period \(T_i=1/30\approx 33.3\) ms. Check condition:

\[ T_i + \overline{C}_i+\overline{D} \approx 33.3 + 5 + 10 = 48.3\text{ ms} \le 60\text{ ms}. \]

Yes, feasibility holds with margin about \(11.7\) ms.

Problem 5 (Buffer Sizing for Bursts): A producer normally runs at 50 Hz but can burst to 80 Hz for 0.5 s. The consumer runs steadily at 60 Hz. Assume no drops if buffer is large enough. Find the minimum buffer size \(B\) to avoid overflow during a burst.

Solution:

During burst, net arrival rate is \(80-60=20\) messages/s. Over \(0.5\) s, backlog grows by

\[ \Delta q = 20 \times 0.5 = 10 \text{ messages}. \]

Thus minimum capacity is \(B \ge 10\). Adding safety margin for timing jitter is typical in real robots.

We modeled robot software as data-flow graphs and analyzed message passing mathematically. Synchronous passing improves ordering but couples delays; asynchronous passing decouples execution but requires queue stability (\(\lambda < \mu\)) and careful buffer sizing. End-to-end latency and data age bounds provide concrete architectural design rules that connect software structure to control quality.