Bi-directional sub-process communication

[Ian Kelly]

On Mon, Nov 23, 2015 at 12:55 PM, Ian Kelly <ian.g.kelly at gmail.com> wrote:
> On Mon, Nov 23, 2015 at 10:54 AM, Israel Brewster <israel at ravnalaska.net> wrote:
>> Concern: Since the master process is multi-threaded, it seems likely enough that multiple threads on the master side would make requests at the same time. I understand that the Queue class has locks that make this fine (one thread will complete posting the message before the next is allowed to start), and since the child process only has a single thread processing messages from the queue, it should process them in order and post the responses (if any) to the master_queue in order. But now I have multiple master processes all trying to read master_queue at the same time. Again, the locks will take care of this and prevent any overlapping reads, but am I guaranteed that the threads will obtain the lock and therefore read the responses in the right order? Or is there a possibility that, say, thread three will get the response that should have been for thread one? Is this something I need to take into consideration, and if so, how?
>
> Yes, if multiple master threads are waiting on the queue, it's
> possible that a master thread could get a response that was not
> intended for it. As far as I know there's no guarantee that the
> waiting threads will be woken up in the order that they called get(),
> but even if there are, consider this case:
>
> Thread A enqueues a request.
> Thread B preempts A and enqueues a request.
> Thread B calls get on the response queue.
> Thread A calls get on the response queue.
> The response from A's request arrives and is given to B.
>
> Instead of having the master threads pull objects off the response
> queue directly, you might create another thread whose sole purpose is
> to handle the response queue. That could look like this:
>
>
> request_condition = threading.Condition()
> response_global = None
>
> def master_thread():
>     global response_global
>     with request_condition:
>         request_queue.put(request)
>         request_condition.wait()
>         # Note: the Condition should remain acquired until
> response_global is reset.
>         response = response_global
>         response_global = None
>     if wrong_response(response):
>         raise RuntimeError("got a response for the wrong request")
>     handle_response(response)
>
> def response_thread():
>     global response_global
>     while True:
>         response = response_queue.get()
>         with request_condition:
>             response_global = response
>             request_condition.notify()

Actually I realized that this fails because if two threads get
notified at about the same time, they could reacquire the Condition in
the wrong order and so get the wrong responses.

Concurrency, ugh.

It's probably better just to have a Condition/Event per thread and
have the response thread identify the correct one to notify, rather
than just notify a single shared Condition and hope the threads wake
up in the right order.