# Consistency of diamonds in a concurrent system

I am struggling to find solutions to a problem that I encounter in a concurrent system (I am currently using the Actor Model). I basically have a diamond structure of actors:

So `D` sends messages to `C` and `B`, which consequently send messages to `A`. Each part performs some form of operation upon receiving a message.

As an example, define some arbitrary operations, e.g.:

• D: Generate random number
• C: Apply f(x) = x - 1
• B: Apply f(x) = 2x
• A: Apply f(c, b) = (c + b) / 2, where c,b are the results of `C`, `B`

After `A` applies its operation, it publishes the result. I now want `A` to only publish a result, if it has received the result from both `B` and `C` for a particular message from `D`. So if we would have an incremental index i and generate numbers x_i, I only want `A` to operate on c_k, b_j if k == j.

One property of my system that makes this problem quite a bit harder is the existence of throttling. The operations can be costly and/or the frequency with which `D` sends messages can be high, so in order to prevent queues from growing infinitely, this seems necessary. It appears to me that this would eliminate any solutions of the form "mark the message from `D` and only execute in `A` if it has received 2 messages with the same mark", as I can not guarantee to ever receive those 2 messages.

My experience in this area is fairly limited, so any resources or patterns that deal with a similar problem like this would be helpful.

• If each A result requires a message from both B and C, (i.e. there is a 1 to 1 correspondence between B and C messages) then why would any of your queues grow infinitely? Either C is generating too many messages (in which case you throw out the irrelevant ones), or you're not properly waiting for the message from B to arrive at A. Commented Mar 30, 2017 at 21:35
• Example: D is sending 1 message per second, B (or C, or both) need 2 seconds to complete their operation. Then I need to throw out some in order to "keep up". I don't need a result in A for every message in D. Just if A produces a result, it needs to have messages from B and C that are based on the same message from D Commented Mar 31, 2017 at 0:45
• If you are open to a redesign, you could submit B and C to be computed, then give the pair of resulting Futures to A. Then the respective Bs and Cs are kept together. Commented Apr 6, 2017 at 14:32

In the aerospace industry there is a file format called IRIG 106 Chapter 10. It uses a multiple-channel packet metaphor. One of the channels contains Time packets, which are used as a referent.

One of the interesting features of Chapter 10 is that (similar to TCP/IP), packets can arrive in any order. There are sequence numbers in the packets so that you can re-assemble them in the correct order, but if you're trying to lock time and a time packet doesn't arrive in the data stream when you expect it to, you have to buffer up data and wait until it arrives.

To avoid the problem of time packets that simply do not exist, the Chapter 10 Specification stipulates that the delay between the time that a time packet represents and the moment that it gets written to the Chapter 10 file cannot exceed one second. This allows data processing equipment that processes Chapter 10 files to design their systems so that they only have to buffer a maximum of 1 second of data, while still remaining compliant with the standard. It also requires that data recorders must be designed to never write a time packet later than 1 second after it is due.

It appears that you need some sort of design constraint similar to this in your system, and a way to handle its non-compliance.

• That's an interesting idea. It allows for individual packets being lost. But the problem in question is even more complicated: there most be some coordination between the packets being lost. The final consumer of the packets does not require individual packets, but pairs of packets. Commented Mar 31, 2017 at 9:05

The most natural and concurrent way to handle this kind of model is to build inbound and outbound queues on both B and D and two inbound queues on A (and possibly an outbound queue on A depending how it uses its result).

An easy way to ensure throttling is to set a maximum outbound queue length on B and C and for them to suspend if their outbound queue exceeds the configured value. A value of 1 is a trivial queue but in more complex situations the execution time of each calculation may vary or difficult to predict.

So it may make sense for (say) C to queue up a few results while B performs a long calculation in case C later receives a complex case while B is given small tasks.

So at its simplest A looks at one queue, if it finds something looks at the other else waits on that queue. When it has something looks at the other queue and waits there. When it has something in both queues it proceeds.

One way of addressing 'gross' delays (defined by time or possibly being n messages behind on one side or the other) is for A to discard entries in one of the queues and notify the other processor to skip forwards. That will involve a bit of re-syncing between the queues.

That sort of approach may be valid and is a bit similar to dropping frames in an animation.

• That works only when C generally copes with processing messages in time. When bot B and C are (sometimes) too slow, their results arriving at A may be out of sync such that A cannot set up a couple of corresponding messages for further processing. Commented Apr 7, 2017 at 7:23
• @BernhardHiller If B and C work sequentially and A requires a result from each then it may need to wait for one, the other or both but when faced with non-empty queues will always be processing a matched pair. Unless it resorts to dumping results on the 'fast' side. In which case it's wasting its time dumping them unless it sends some kind of 'dump/skip' instruction to the other. That's when re-syncing comes in. I've tried to give a flavour of what is required as a sketch. I haven't provided an algorithm which will have special cases. Commented Apr 7, 2017 at 7:31

You need some way of "synchronizing the loss" of packets. In a worst case scenario, `B` could lose every even packet, while `C` loses every odd packet, and thus `A` won't ever be able to produce a result.

Let me describe the concept of such synchronization in a simple example. When `A` produced its message, it does not yet send it out. Instead, it stores it to some field. Only after both `B` and `C` have told `D` that they are ready to receive a new message, `D` will send this stored message out to `B` and `C`. After `B` and `C` have finished their task,`A` has a couple of data corresponding to the same original message from `D`.

If `A` produces a new message before the stored message was sent out, that previously stored message gets overwritten and thus lost. Another constraint is that the message receivers `B` and `C` must register with `D`, and also inform `D` that they are ready, i.e. a tighter coupling is required.

Of course, the concept of the intermediary storage can be refactored into a class of its own, such that `D` sends the message to the intermediary immediately, which then in turn will send it to `B` and `C` when appropriate.