Consistency of diamonds in a concurrent system

Question

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.

Answer 1

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.

Answer 2

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.

Answer 3

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.