{"action":"created","issue":{"url":"https://api.github.com/repos/zeromq/pyzmq/issues/755","labels_url":"https://api.github.com/repos/zeromq/pyzmq/issues/755/labels{/name}","comments_url":"https://api.github.com/repos/zeromq/pyzmq/issues/755/comments","events_url":"https://api.github.com/repos/zeromq/pyzmq/issues/755/events","html_url":"https://github.com/zeromq/pyzmq/issues/755","id":118397173,"number":755,"title":"Terminating multiprocessing.Process's with pyzmq in the mix","user":{"login":"nkoep","id":1410485,"avatar_url":"https://avatars.githubusercontent.com/u/1410485?v=3","gravatar_id":"","url":"https://api.github.com/users/nkoep","html_url":"https://github.com/nkoep","followers_url":"https://api.github.com/users/nkoep/followers","following_url":"https://api.github.com/users/nkoep/following{/other_user}","gists_url":"https://api.github.com/users/nkoep/gists{/gist_id}","starred_url":"https://api.github.com/users/nkoep/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/nkoep/subscriptions","organizations_url":"https://api.github.com/users/nkoep/orgs","repos_url":"https://api.github.com/users/nkoep/repos","events_url":"https://api.github.com/users/nkoep/events{/privacy}","received_events_url":"https://api.github.com/users/nkoep/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":1,"created_at":"2015-11-23T14:50:01Z","updated_at":"2015-11-24T10:34:02Z","closed_at":null,"body":"Hello,\r\n\r\nI'm currently using pyzmq to perform some basic distributed computing. To that end, I'm running code that looks something like this:\r\n\r\n```python\r\n#!/usr/bin/env python\r\n\r\nimport sys\r\nimport time\r\nimport enum\r\nimport collections\r\nimport socket\r\nimport multiprocessing\r\nimport zmq\r\n\r\nHOSTNAME = socket.gethostname()\r\nWORK = range(10)\r\nTOTAL = len(WORK)\r\n\r\n\r\ndef producer():\r\n    print(\"Running as producer...\")\r\n    context = zmq.Context()\r\n    socket = context.socket(zmq.REP)\r\n    socket.bind(\"tcp://*:{}\".format(5555))\r\n    for idx, i in enumerate(WORK):\r\n        # Wait for a consumer to ask for work.\r\n        socket.recv_pyobj()\r\n        socket.send_pyobj(i)\r\n        sys.stdout.write(\"\\r{} of {} jobs distributed\".format(idx + 1, TOTAL))\r\n        sys.stdout.flush()\r\n\r\ndef consumer():\r\n    def inner():\r\n        print(\"Running as consumer for `{}`...\".format(HOSTNAME))\r\n        context = zmq.Context()\r\n        # Create a socket to receive work items.\r\n        socket_task = context.socket(zmq.REQ)\r\n        socket_task.connect(\r\n            \"tcp://{host}:{port}\".format(host=HOSTNAME, port=5555))\r\n        # Create a socket to push results downstream.\r\n        socket_result = context.socket(zmq.PUSH)\r\n        socket_result.connect(\r\n            \"tcp://{host}:{port}\".format(host=HOSTNAME, port=5556))\r\n\r\n        # Subscribe to the command socket.\r\n        socket_command = context.socket(zmq.SUB)\r\n        socket_command.connect(\r\n            \"tcp://{host}:{port}\".format(host=HOSTNAME, port=5557))\r\n        socket_command.setsockopt_string(zmq.SUBSCRIBE, \"\")\r\n\r\n        # Set up a poller to receive tasks/commands.\r\n        poller = zmq.Poller()\r\n        poller.register(socket_task, zmq.POLLIN)\r\n        poller.register(socket_command, zmq.POLLIN)\r\n\r\n        while True:\r\n            # Request work.\r\n            socket_task.send_pyobj(None)\r\n            # Poll for requests.\r\n            sockets = dict(poller.poll())\r\n            # Handle a task.\r\n            if sockets.get(socket_task) == zmq.POLLIN:\r\n                # Retrieve work and run the computation.\r\n                params = socket_task.recv_pyobj()\r\n                time.sleep(1)\r\n                result = params\r\n                # Send the result downstream.\r\n                socket_result.send_pyobj(result)\r\n            # Handle a command.\r\n            if sockets.get(socket_command) == zmq.POLLIN:\r\n                cmd = socket_command.recv_pyobj()\r\n                if cmd == \"stop\":\r\n                    print(\"Stop command received\")\r\n                    # FIXME: Child processes don't seem to terminate here.\r\n                    break\r\n\r\n    def Process():\r\n        proc = multiprocessing.Process(target=inner)\r\n        proc.daemon = True\r\n        proc.start()\r\n        return proc\r\n    pool = [Process() for i in range(8)]\r\n    # Zombies are bad for your health.\r\n    for proc in pool:\r\n        proc.join()\r\n\r\ndef collector():\r\n    print(\"Running as collector...\")\r\n\r\n    context = zmq.Context()\r\n    socket_result = context.socket(zmq.PULL)\r\n    socket_result.bind(\"tcp://*:{}\".format(5556))\r\n\r\n    socket_command = context.socket(zmq.PUB)\r\n    socket_command.bind(\"tcp://*:{}\".format(5557))\r\n\r\n    n_results = 0\r\n    while True:\r\n        result = socket_result.recv_pyobj()\r\n        n_results += 1\r\n        sys.stdout.write(\r\n            \"\\r{} of {} results collected\".format(n_results, TOTAL))\r\n        sys.stdout.flush()\r\n        if n_results == TOTAL:\r\n            break\r\n\r\n    # Broadcast a stop instruction to shut down worker processes.\r\n    socket_command.send_pyobj(\"stop\")\r\n\r\ndef main():\r\n    role = sys.argv[1]\r\n    if role == \"producer\":\r\n        producer()\r\n    elif role == \"collector\":\r\n        collector()\r\n    elif role == \"consumer\":\r\n        consumer()\r\n    else:\r\n        raise SystemExit(\"Invalid role\")\r\n\r\nif __name__ == \"__main__\":\r\n    main()\r\n```\r\n\r\nIf I run this on my local machine using\r\n```\r\n$ python sim.py producer\r\n$ python sim.py collector\r\n$ python sim.py consumer\r\n```\r\n(in three different terminals of course), everything works fine for the most part. Each process is receiving its work items, and each process successfully transmits its results to the collector. The problem occurs when shutting down the worker processes, which never get joined for some reason. Interrupting the consumer process shows that it is hanging at `pid, sts = os.waitpid(self.pid, flag)` in `multiprocessing/popen_fork.py`. The issue seems to be that the producer and collector terminate before the consumer(s) get a chance to. If I add a simple timeout at the end of the producer and consumer functions, every process terminates properly.\r\n\r\nCould somebody explain to me why that is, and possibly suggest a more reliable solution than adding a random timeout?"},"comment":{"url":"https://api.github.com/repos/zeromq/pyzmq/issues/comments/159224275","html_url":"https://github.com/zeromq/pyzmq/issues/755#issuecomment-159224275","issue_url":"https://api.github.com/repos/zeromq/pyzmq/issues/755","id":159224275,"user":{"login":"minrk","id":151929,"avatar_url":"https://avatars.githubusercontent.com/u/151929?v=3","gravatar_id":"","url":"https://api.github.com/users/minrk","html_url":"https://github.com/minrk","followers_url":"https://api.github.com/users/minrk/followers","following_url":"https://api.github.com/users/minrk/following{/other_user}","gists_url":"https://api.github.com/users/minrk/gists{/gist_id}","starred_url":"https://api.github.com/users/minrk/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/minrk/subscriptions","organizations_url":"https://api.github.com/users/minrk/orgs","repos_url":"https://api.github.com/users/minrk/repos","events_url":"https://api.github.com/users/minrk/events{/privacy}","received_events_url":"https://api.github.com/users/minrk/received_events","type":"User","site_admin":false},"created_at":"2015-11-24T10:34:02Z","updated_at":"2015-11-24T10:34:02Z","body":"The hang is caused by LINGER behavior on `socket_task`. At the beginning of every loop, `socket_task` is sending a ping to request new tasks. The last request from one or more workers is going out after the producer has closed its socket. This means that the request for a new task is waiting in a queue on the workers. The default LINGER behavior is to block Context.term until undelivered messages can be sent. If you set `socket_task.linger = 0`, it will immediately discard unsent messages.\r\n\r\nIf you change\r\n\r\n```python\r\nsocket_task = context.socket(zmq.REQ)\r\n```\r\n\r\nto\r\n\r\n```python\r\nsocket_task = context.socket(zmq.REQ)\r\nsocket_task.linger = 0\r\n```\r\n\r\nyou should be set."}}