This is a simple nanomsg-based message buffer.
Connectivity looks like this:
flowchart LR
subgraph caches
direction TB
C1[(Cache1)]
C2[(Cache2)]
end
Control ---|Bus| caches
S1[/Source1/] -->|Push/Pull| C1
S2[/Source2/] -->|Push/Pull| C2
S3[/Source3/] -->|Push/Pull| C2
C1 -->|Push/Pull| Client1
C1 --> Client2
C2 -->|Push/Pull| Client2
C2 --> Client4
Every source pushes its data to a unique cache server. The cache servers push their data to connected compute nodes. Compute nodes typically pull from just one cache.
Normal termination occurs when all producers have disconnected, and the cache sends the last available message. To prevent an early disconnection from shutting down the cache, the "Active" state tracks whether all expected producers have connected at least once. These expected producers must be communicated in the Control's Start message.
Caches push data to clients in FIFO order.
stateDiagram
Init --> Active: C/Start
Active --> Drain: C/Stop or LastSend
Drain --> Init: LastSend
%% Alternatively, we could have the Active state
%% cycle between pulling data from Producers and other Caches
%% However, this requires more tuning to get right.
(using the rc shell)
./cache tcp://127.0.0.1:5551 tcp://127.0.0.1:5561 & C1=$apid
./cache tcp://127.0.0.1:5552 tcp://127.0.0.1:5562 & C2=$apid
./producer tcp://127.0.0.1:5551 2048 & P1=$apid
./producer tcp://127.0.0.1:5552 2048 & P2=$apid
./producer tcp://127.0.0.1:5552 2048 & P3=$apid
python3 client.py tcp://127.0.0.1:5561 & S1=$apid
python3 client.py tcp://127.0.0.1:5561 & S2=$apid
python3 client.py tcp://127.0.0.1:5562 & S3=$apid
python3 client.py tcp://127.0.0.1:5562 & S4=$apid
for(C in $S1 $S2 $S3 $S4) { wait $C }
kill $C1 $C2
Output for 1 producer, 1 cache, 1 consumer:
Connecting to tcp://127.0.0.1:5561
Sending request 1 ...
633172 Client pipe event: 0
Received reply 1 len = 131072
Sending request 2 ...
Received reply 2 len = 131072
Sending request 3 ...
Received reply 3 len = 131072
Sending request 4 ...
Received reply 4 len = 131072
Sending request 5 ...
Received reply 5 len = 131072
Sending request 6 ...
Received reply 6 len = 131072
Sending request 7 ...
Received reply 7 len = 131072
Sending request 8 ...
Received reply 8 len = 131072
Sending request 9 ...
Received reply 9 len = 131072
633227 Sent 10000 messages in 963 milliseconds: 1298.026999 MB/s
633172 Client pipe event: 2
633172 PUSH thread exiting.
633172 PULL thread exiting.
Received 10001 messages in 1.2032361030578613 seconds: 1038.8651045487204 MB/sec
633172 Closing push socket.
Output for 3 producer, 2 cache, 4 consumer:
nng_stream% rc ./run.rc
Connecting to tcp://127.0.0.1:5561
Sending request 1 ...
Connecting to tcp://127.0.0.1:5562
Sending request 1 ...
Connecting to tcp://127.0.0.1:5561
Connecting to tcp://127.0.0.1:5562
Sending request 1 ...
Sending request 1 ...
633504 Client pipe event: 0
633503 Client pipe event: 0
633504 Client pipe event: 0
Received reply 1 len = 131072
Sending request 2 ...
Received reply 1 len = 131072
Sending request 2 ...
Received reply 2 len = 131072
Sending request 3 ...
Received reply 2 len = 131072
Sending request 3 ...
Received reply 3 len = 131072
Sending request 4 ...
Received reply 3 len = 131072
Sending request 4 ...
Received reply 4 len = 131072
Sending request 5 ...
Received reply 4 len = 131072
Sending request 5 ...
Received reply 5 len = 131072
Sending request 6 ...
Received reply 5 len = 131072
Sending request 6 ...
Received reply 6 len = 131072
Sending request 7 ...
Received reply 6 len = 131072
Sending request 7 ...
Received reply 7 len = 131072
Sending request 8 ...
Received reply 7 len = 131072
Sending request 8 ...
Received reply 8 len = 131072
Sending request 9 ...
Received reply 8 len = 131072
Sending request 9 ...
Received reply 9 len = 131072
Received reply 9 len = 131072
Received reply 1 len = 131072
Sending request 2 ...
Received reply 1 len = 131072
Sending request 2 ...
Received reply 2 len = 131072
Sending request 3 ...
Received reply 2 len = 131072
Sending request 3 ...
Received reply 3 len = 131072
Sending request 4 ...
Received reply 3 len = 131072
Sending request 4 ...
Received reply 4 len = 131072
Sending request 5 ...
Received reply 4 len = 131072
Sending request 5 ...
Received reply 5 len = 131072
Sending request 6 ...
Received reply 5 len = 131072
Sending request 6 ...
Received reply 6 len = 131072
Sending request 7 ...
Received reply 6 len = 131072
Received reply 7 len = 131072
Sending request 7 ...
Sending request 8 ...
Received reply 8 len = 131072
Sending request 9 ...
Received reply 7 len = 131072
Sending request 8 ...
Received reply 9 len = 131072
Received reply 8 len = 131072
Sending request 9 ...
Received reply 9 len = 131072
633611 Sent 3000 messages in 292 milliseconds: 1284.246575 MB/s
633503 Client pipe event: 2
633503 PUSH thread exiting.
633503 PULL thread exiting.
633503 Closing push socket.
Received 1498 messages in 0.4615328311920166 seconds: 405.44244602644164 MB/sec
633613 Sent 3000 messages in 772 milliseconds: 485.751295 MB/s
633612 Sent 3000 messages in 773 milliseconds: 485.122898 MB/s
633504 Client pipe event: 2
633504 Client pipe event: 2
633504 PUSH thread exiting.
633504 PULL thread exiting.
Received 3013 messages in 1.0548529624938965 seconds: 356.92178283300643 MB/sec
633504 Closing push socket.
Python producer/consumer example:
% python3 serve.py tcp://127.0.0.1:4000 7 &
% python3 client.py tcp://127.0.0.1:4000
Connecting to tcp://127.0.0.1:4000
Sending request 1 ...
Received reply 1 len = 2
Sending request 2 ...
Received reply 2 len = 2
Sending request 3 ...
Received reply 3 len = 2
Sending request 4 ...
Received reply 4 len = 2
Sending request 5 ...
Received reply 5 len = 2
Sending request 6 ...
Received reply 6 len = 2
Sending request 7 ...
Received reply 7 len = 2
Sending request 8 ...
Received reply 8 len = 0
The NNG library runs callbacks on separate threads,
so it is possible for multiple callbacks
to be executing in parallel.
This was determined by experimental testing, toggling
a global cb_running state on entry/exit of each
callback.
x. Add outer loop (start/stop client/server listeners).
x. Add cache transition to Drain state.
x. Add pipe notification to client connections
maintain list of client connections
- prevent new client connections when in DRAIN state
- close consumer pipes on exit ~> triggers notifications
-
Trigger outer loop transitions from Control connection
-
Test outer [Init -> Active -> Drain] loop for cache.
-
Have cache Dial new Pull connections on Active -> Drain trs.