Skip to content

Instantly share code, notes, and snippets.

@pitrou
Created Apr 28, 2020
Embed
What would you like to do?
import faulthandler
from queue import Queue
import multiprocessing
import os
import sys
import threading
import time
if __name__ == "__main__":
queue = Queue()
def consume():
while True:
print("Consumed:", queue.get())
thread = threading.Thread(target=consume)
thread.start()
def proc():
faulthandler.dump_traceback_later(1.0, exit=True)
for i in range(10000):
queue.put(i)
p = multiprocessing.Process(target=proc)
p.start()
p.join()
if p.exitcode != 0:
print("Child process failed!")
os._exit(0)
print("Not hanging yet", i)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment