test_deadlock.py

import multiprocessing
import time


def generate(queue, close):
    queue.put_nowait('x' * 65523)
    close.wait()


def main():
    q = multiprocessing.Queue()
    close = multiprocessing.Event()
    process = multiprocessing.Process(target=generate, args=(q, close))
    process.start()

    time.sleep(1)

    print("Sending close event")
    close.set()
    try:
        print("Queue length: {:d}".format(q.qsize()))
    except NotImplementedError:  # Not available on OS X
        pass
    print("Joining enqueue process")
    process.join()
    print("Exited correctly")


if __name__ == "__main__":
    main()