multiprocessing.Queue unexpected behaviour when using multiple queues and multiple producers

[CamLovell]

Bug report

Bug description:

When using two multiprocessing queues and multiple producers to put into these queues the queues will, at some point, get out of sync. Given two multiprocessing queues which are meant to be "synced" acquiring a shared lock before putting into both of the queues should ensure that there is no scenario where another producer can jump in and put elements in the queues. Acquiring the shared lock before getting from both the queues should also prevent any concurrency abnormalities on the consumer side. Despite this, under these conditions when multiple producers are used the consumer will at some point read elements from the queues which were not put into the queue by the same producer. For clarity see the reproduction below.

import multiprocessing as mp
import time
from queue import Empty


def producer(q0, q1, lock, num_puts):
    for i in range(num_puts):
        with lock:
            q0.put(i)
            q1.put(i)


def main():
    num_puts = 10000
    num_producers = 2

    q0 = mp.Queue()
    q1 = mp.Queue()
    lock = mp.Lock()

    producers = [mp.Process(target=producer, args=(q0, q1, lock, num_puts)) for _ in range(num_producers)]
    [producer.start() for producer in producers]

    while True:
        if lock.acquire(timeout=1):
            try:
                r0 = q0.get(timeout=1)
                r1 = q1.get(timeout=1)
                lock.release()
            except Empty:
                print("Empty queue")
                lock.release()
                time.sleep(0.1)
                continue
        else:
            print("Lock acquisition failed")
            continue

        print(f"res = {(r0, r1)}")

        # This assert will fail after a seemingly random number of gets
        assert r0 == r1


if __name__ == "__main__":
    main()

Using a different Queue implementation such as faster-fifo while maintaining all other aspects removes the issue and everything behaves as expected.

Tested on these versions (provided by uv):
3.10, 3.11, 3.12, 3.13, 3.14.0a5

CPython versions tested on:

3.10

Operating systems tested on:

Linux

[YvesDup]

IMO, using a Lock does not guarantee that each process will add one item alternately.
Releasing the Lock does not allow you to switch to the other process. So it is possible to have several successive additions to a given process, which generates the offset when reading.

[CamLovell]

I may not have explained properly here or maybe I am misunderstanding you. The issue is not that sequential additions happen from the same process, given that concurrency is non-deterministic this is expected. My expectation, which may be wrong, is that given the way the locks are set up the each queue should have the same elements in it.

For example:
After process one has put once and process two has put 4 times we would expect something like

q0 == [0, 0, 1, 2, 3]
q1 == [0, 0, 1, 2, 3]

or any other combination like

q0 == [0, 1, 0, 2, 3]
q1 == [0, 1, 0, 2, 3]

Meaning that for any get the element retrieved from each queue should be the same. The issue here is that for awhile this is the case however at some point the get returns different elements.

For example, say process one has put 5 times and process two has put 50 times we may get a response like below (assuming the consumer is slightly behind).

with lock:
    result0 = q0.get()
    resultq = q1.get()

    # result0 == 2
    # result1 == 47

Prior to this particular iteration result0 and result1 would have returned the same number on each get call. The number being enqueued is simply for ease of debugging, I first noticed the issue due to other data as part of a work project.

More than happy to be wrong about this one, just not the behaviour I would have expected but I may not have a deep enough understanding!

[ZeroIntensity]

It's because you have timeouts on your queues.

• get(timeout=...) raises Empty if the timeout is hit.
• Multiprocessing is slow, so reading q0/q1 can take longer than a second.
• q0/q1 has an item, but the read took too long, so the item remains in the queue and the lock is released.
• If q0 worked but q1 subsequently didn't, then there will be a mismatch between items.

I would just remove the timeout.

[CamLovell]

Apologies I should have removed the timeouts, they are simply there to ensure that the lock us able to be released to allow the producers to put into the queues. The lock itself I put there during debugging to try and remove as many variables as possible.

If you remove the lock in the consumer altogether as well as the timeouts so the get calls hang waiting for something in the queue the issue still persists. See the updated example below.

import multiprocessing as mp
import time
from queue import Empty


def producer(q0, q1, lock, num_puts):
    for i in range(num_puts):
        with lock:
            q0.put(i)
            q1.put(i)


def main():
    num_puts = 10000
    num_producers = 2

    q0 = mp.Queue()
    q1 = mp.Queue()
    lock = mp.Lock()

    producers = [
        mp.Process(target=producer, args=(q0, q1, lock, num_puts))
        for _ in range(num_producers)
    ]
    [producer.start() for producer in producers]

    while True:
        r0 = q0.get()
        r1 = q1.get()

        print(f"res = {(r0, r1)}")
        assert r0 == r1


if __name__ == "__main__":
    main()

[ZeroIntensity]

Yeah, that example makes plenty of sense. There's no synchronization between the main process and the other processes at all, so they're free to add to the queues in any order they want.

In this example:

• q0/q1 writes are synchronized, but reads aren't. (Technically, the reads are writes too, but it doesn't matter much here.)
• A process grabs the lock and inserts the numbers into the queues.
• The main process goes to read, but without the lock. It's allowed to suspend in any way it likes, e.g., a garbage collection.
• So, q0 and q1 have an item.
• The main process goes to read from q0. Success! The main process decides to go for a garbage collection.
• Another process grabs the lock and fills q0 again. Now, in the next iteration, q0 and q1 will be out of sync.

[CamLovell]

Ok thanks for that, just trying to get me head around this. In the example you described there is a possibility that the main process loop only executes part of the code inside it?

1. Successful puts leading to q0 == [0,1,0,2,3], q1 == [0,1,0,2,3]
2. successful read from q0
3. garbage collection in main process
4. producer grabs lock and puts in q0
5. execution suspended on producer due to gc, leading to q0 == [1,0,2,3,1], q1 == [0,1,0,2,3]
6. main process skips remaining code, returning to top of loop
7. gets are now out of sync

I may have misunderstood your explanation however I did not think this would be expected behavior. Whether or not the main process triggers garbage collection (or any suspension) I would expect it to continue on afterwards and attempt to read from q1 rather than return to the top of the loop. As the gets now hang until a result is returned this should yield 0 from q1 before continuing onwards, leading to no sync issues.

[ZeroIntensity]

No, the GC doesn't skip code, but it's basically an invisible pause in your program; that pause gives plenty of chance for another process to mess with the queues while the main process is in the middle of a garbage collection.

[CamLovell]

I understand that however I do not understand how the next element in q1 could change during this time as there are no other consumers to retrieve this element and the producers simply add elements to the back of the queue?

I am sure there is something I am missing here and I really apologise for that!

[YvesDup]

I believe the main problem is when you fill the queues.
Please look at the modified script below, where we wait for the two queues been full :

# Just main changed
def main():
    num_puts = 20
    num_producers = 2

    q0 = mp.Queue()
    q1 = mp.Queue()
    lock = mp.Lock()

    producers = [mp.Process(target=producer, args=(q0, q1, lock, num_puts)) for _ in range(num_producers)]
    [producer.start() for producer in producers]
    
    # Wait for all items in queues
    [producer.join() for producer in producers]

    # Just shows content of queues
    while not q0.empty() and not q1.empty():
        print(f"\tres = {q0.get()}, {q1.get()}")

If you run that script, sometimes there is an offset between content of the two queues, see around 9 value . Here the output was:

        res = 0, 0
        res = 0, 0
        res = 1, 1
        res = 1, 1
        res = 2, 2
        res = 2, 2
        res = 3, 3
        res = 3, 3
        res = 4, 4
        res = 4, 4
        res = 5, 5
        res = 5, 5
        res = 6, 6
        res = 6, 6
        res = 7, 7
        res = 7, 7
        res = 8, 8
        res = 9, 9
        res = 10, 8
        res = 8, 9
        res = 9, 10
        res = 10, 10
        res = 11, 11
        res = 11, 11
        res = 12, 12
        res = 12, 12
        res = 13, 13
        res = 13, 13
        res = 14, 14
        res = 14, 14
        res = 15, 15
        res = 16, 16
        res = 15, 15
        res = 16, 17
        res = 17, 16
        res = 17, 17
        res = 18, 18
        res = 19, 18
        res = 18, 19
        res = 19, 19

Do you have any explanation ?

[ZeroIntensity]

Oh wait, this is FIFO queue, not a LIFO queue. Ignore what I said about garbage collection, it doesn't apply here.

Anyways, queues don't (and can't) really guarantee precise ordering when the consumer isn't synchronized. The calls to put might be synchronized in the calling process, but there's no way to ensure that they make it to the main process in that order, because we're operating over sockets or pipes that are dependent on I/O devices.

---

full() and empty() aren't precise. See the docs:

Because of multithreading/multiprocessing semantics, this is not reliable.

[CamLovell]

Great, thanks for that, will be sure to use a different queue implementation when I care about ordering ahaha

[ZeroIntensity]

Or just synchronize your readers with a lock :)

I don't think any other implementation could effectively bypass this issue.