{"action":"created","issue":{"url":"https://api.github.com/repos/pika/pika/issues/963","repository_url":"https://api.github.com/repos/pika/pika","labels_url":"https://api.github.com/repos/pika/pika/issues/963/labels{/name}","comments_url":"https://api.github.com/repos/pika/pika/issues/963/comments","events_url":"https://api.github.com/repos/pika/pika/issues/963/events","html_url":"https://github.com/pika/pika/issues/963","id":297985874,"number":963,"title":"Acknowledging a single message from a different thread may fail or get stuck","user":{"login":"adah1972","id":2137834,"avatar_url":"https://avatars1.githubusercontent.com/u/2137834?v=4","gravatar_id":"","url":"https://api.github.com/users/adah1972","html_url":"https://github.com/adah1972","followers_url":"https://api.github.com/users/adah1972/followers","following_url":"https://api.github.com/users/adah1972/following{/other_user}","gists_url":"https://api.github.com/users/adah1972/gists{/gist_id}","starred_url":"https://api.github.com/users/adah1972/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/adah1972/subscriptions","organizations_url":"https://api.github.com/users/adah1972/orgs","repos_url":"https://api.github.com/users/adah1972/repos","events_url":"https://api.github.com/users/adah1972/events{/privacy}","received_events_url":"https://api.github.com/users/adah1972/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":{"login":"lukebakken","id":514926,"avatar_url":"https://avatars2.githubusercontent.com/u/514926?v=4","gravatar_id":"","url":"https://api.github.com/users/lukebakken","html_url":"https://github.com/lukebakken","followers_url":"https://api.github.com/users/lukebakken/followers","following_url":"https://api.github.com/users/lukebakken/following{/other_user}","gists_url":"https://api.github.com/users/lukebakken/gists{/gist_id}","starred_url":"https://api.github.com/users/lukebakken/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/lukebakken/subscriptions","organizations_url":"https://api.github.com/users/lukebakken/orgs","repos_url":"https://api.github.com/users/lukebakken/repos","events_url":"https://api.github.com/users/lukebakken/events{/privacy}","received_events_url":"https://api.github.com/users/lukebakken/received_events","type":"User","site_admin":false},"assignees":[{"login":"lukebakken","id":514926,"avatar_url":"https://avatars2.githubusercontent.com/u/514926?v=4","gravatar_id":"","url":"https://api.github.com/users/lukebakken","html_url":"https://github.com/lukebakken","followers_url":"https://api.github.com/users/lukebakken/followers","following_url":"https://api.github.com/users/lukebakken/following{/other_user}","gists_url":"https://api.github.com/users/lukebakken/gists{/gist_id}","starred_url":"https://api.github.com/users/lukebakken/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/lukebakken/subscriptions","organizations_url":"https://api.github.com/users/lukebakken/orgs","repos_url":"https://api.github.com/users/lukebakken/repos","events_url":"https://api.github.com/users/lukebakken/events{/privacy}","received_events_url":"https://api.github.com/users/lukebakken/received_events","type":"User","site_admin":false}],"milestone":{"url":"https://api.github.com/repos/pika/pika/milestones/8","html_url":"https://github.com/pika/pika/milestone/8","labels_url":"https://api.github.com/repos/pika/pika/milestones/8/labels","id":2664000,"number":8,"title":"1.0.0","description":"","creator":{"login":"lukebakken","id":514926,"avatar_url":"https://avatars2.githubusercontent.com/u/514926?v=4","gravatar_id":"","url":"https://api.github.com/users/lukebakken","html_url":"https://github.com/lukebakken","followers_url":"https://api.github.com/users/lukebakken/followers","following_url":"https://api.github.com/users/lukebakken/following{/other_user}","gists_url":"https://api.github.com/users/lukebakken/gists{/gist_id}","starred_url":"https://api.github.com/users/lukebakken/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/lukebakken/subscriptions","organizations_url":"https://api.github.com/users/lukebakken/orgs","repos_url":"https://api.github.com/users/lukebakken/repos","events_url":"https://api.github.com/users/lukebakken/events{/privacy}","received_events_url":"https://api.github.com/users/lukebakken/received_events","type":"User","site_admin":false},"open_issues":9,"closed_issues":45,"state":"open","created_at":"2017-07-26T21:40:33Z","updated_at":"2018-02-23T02:37:54Z","due_on":null,"closed_at":null},"comments":15,"created_at":"2018-02-17T07:14:35Z","updated_at":"2018-02-25T15:51:16Z","closed_at":null,"author_association":"NONE","body":"According to the documents I've seen, ack'ing a single message from a different thread should be OK. However, my code generates errors, which seems to indicate there is a bug somewhere.\r\n\r\nThe minimal code that can reproduce the error is like the following (simulating my real logic but removed irrelevant code:\r\n\r\n```python\r\n#!/usr/bin/env python3\r\n\r\nimport json\r\nimport pika\r\nimport queue\r\nimport signal\r\nimport threading\r\n\r\nsend_queue = None\r\nsend_channel = None\r\nconsumer_tag = None\r\naccess_tokens = {}\r\nexit_event = threading.Event()\r\ntotal_count = 10000\r\n\r\nrabbitmq_credentials = pika.PlainCredentials('guest', 'guest')\r\nrabbitmq_host = 'localhost'\r\nrabbitmq_port = 5672\r\nrabbitmq_queue = 'test_queue'\r\n\r\n\r\ndef stop_consume():\r\n    global consumer_tag\r\n    if consumer_tag is not None:\r\n        send_channel.basic_cancel(consumer_tag)\r\n        consumer_tag = None\r\n\r\n\r\ndef send_from_queue():\r\n    is_ending = False\r\n    while not exit_event.is_set() and not is_ending:\r\n        count = 0\r\n        ack_list = []\r\n        while count < 100 and not send_queue.empty():\r\n            message = send_queue.get(block=False)\r\n            ack_list.append(message['delivery_tag'])\r\n            count += 1\r\n            global total_count\r\n            total_count -= 1\r\n\r\n        if ack_list:\r\n            print('Acking messages')\r\n            for tag in ack_list:\r\n                print('Acking', tag)\r\n                send_channel.basic_ack(delivery_tag=tag)\r\n            print('Acking done')\r\n\r\n        if send_queue.empty():\r\n            if total_count <= 0:\r\n                exit_event.set()\r\n                stop_consume()\r\n            exit_event.wait(0.25)\r\n\r\n\r\ndef rabbitmq_callback(ch, method, properties, body):\r\n    try:\r\n        message = json.loads(body.decode())\r\n        message['delivery_tag'] = method.delivery_tag\r\n        send_queue.put(message)\r\n    except ValueError:\r\n        print('Non-JSON message encountered:', body)\r\n        ch.basic_ack(delivery_tag=method.delivery_tag)\r\n\r\n\r\ndef handle_signal(signum, stack):\r\n    exit_event.set()\r\n\r\n    global consumer_tag\r\n    if consumer_tag is not None:\r\n        print('Signal', signum,\r\n              'is received.  Stopped receiving requests...')\r\n        stop_consume()\r\n\r\n\r\ndef repeat_send(channel, tries):\r\n    message = {\r\n        \"id\": 0,\r\n        \"content\": \"\"\r\n    }\r\n    for i in range(tries):\r\n        message['id'] = i\r\n        message['content'] = 'Hello World' + str(i)\r\n        channel.basic_publish(exchange='',\r\n                              routing_key=rabbitmq_queue,\r\n                              body=json.dumps(message),\r\n                              properties=pika.BasicProperties(\r\n                                  delivery_mode=2,  # make message persistent\r\n                              ))\r\n\r\n\r\ndef main():\r\n    signal.signal(signal.SIGTERM, handle_signal)\r\n\r\n    global send_queue\r\n    global send_channel\r\n    global consumer_tag\r\n\r\n    connection = pika.BlockingConnection(pika.ConnectionParameters(\r\n        rabbitmq_host, rabbitmq_port, '/', rabbitmq_credentials))\r\n    send_channel = connection.channel()\r\n    send_channel.queue_declare(queue=rabbitmq_queue,\r\n                               durable=True)\r\n\r\n    repeat_send(send_channel, total_count)\r\n\r\n    send_queue = queue.Queue(10000)\r\n    send_thread = threading.Thread(target=send_from_queue)\r\n    send_thread.start()\r\n\r\n    send_channel.basic_qos(prefetch_count=10)\r\n    consumer_tag = send_channel.basic_consume(\r\n        rabbitmq_callback, queue=rabbitmq_queue)\r\n    send_channel.start_consuming()\r\n\r\n\r\nsignal.signal(signal.SIGINT, handle_signal)\r\n\r\nif __name__ == \"__main__\":\r\n    main()\r\n```\r\n\r\nA sample error generated on macOS Sierra 10.12.6 + Anaconda Python 3.6.2 + pika 0.11.2 + rabbitmq 3.7.3:\r\n\r\n```\r\n…\r\nAcking messages\r\nAcking 9354\r\nAcking 9355\r\nException in thread Thread-1:\r\nTraceback (most recent call last):\r\n  File \"/Users/yongwei/anaconda/lib/python3.6/threading.py\", line 916, in _bootstrap_inner\r\n    self.run()\r\n  File \"/Users/yongwei/anaconda/lib/python3.6/threading.py\", line 864, in run\r\n    self._target(*self._args, **self._kwargs)\r\n  File \"./test_pika.py\", line 53, in send_from_queue\r\n    send_channel.basic_ack(delivery_tag=tag)\r\n  File \"/Users/yongwei/anaconda/lib/python3.6/site-packages/pika/adapters/blocking_connection.py\", line 1988, in basic_ack\r\n    self._flush_output()\r\n  File \"/Users/yongwei/anaconda/lib/python3.6/site-packages/pika/adapters/blocking_connection.py\", line 1250, in _flush_output\r\n    *waiters)\r\n  File \"/Users/yongwei/anaconda/lib/python3.6/site-packages/pika/adapters/blocking_connection.py\", line 455, in _flush_output\r\n    self._impl.ioloop.poll()\r\n  File \"/Users/yongwei/anaconda/lib/python3.6/site-packages/pika/adapters/select_connection.py\", line 245, in poll\r\n    self._poller.poll()\r\n  File \"/Users/yongwei/anaconda/lib/python3.6/site-packages/pika/adapters/select_connection.py\", line 816, in poll\r\n    self._dispatch_fd_events(fd_event_map)\r\n  File \"/Users/yongwei/anaconda/lib/python3.6/site-packages/pika/adapters/select_connection.py\", line 625, in _dispatch_fd_events\r\n    handler(fileno, events)\r\n  File \"/Users/yongwei/anaconda/lib/python3.6/site-packages/pika/adapters/base_connection.py\", line 392, in _handle_events\r\n    self._manage_event_state()\r\n  File \"/Users/yongwei/anaconda/lib/python3.6/site-packages/pika/adapters/base_connection.py\", line 514, in _manage_event_state\r\n    self.ioloop.update_handler(self.socket.fileno(), self.event_state)\r\n  File \"/Users/yongwei/anaconda/lib/python3.6/site-packages/pika/adapters/select_connection.py\", line 194, in update_handler\r\n    self._poller.update_handler(fileno, events)\r\n  File \"/Users/yongwei/anaconda/lib/python3.6/site-packages/pika/adapters/select_connection.py\", line 401, in update_handler\r\n    events_to_set=events_set)\r\n  File \"/Users/yongwei/anaconda/lib/python3.6/site-packages/pika/adapters/select_connection.py\", line 874, in _modify_fd_events\r\n    self._kqueue.control(kevents, 0)\r\nBlockingIOError: [Errno 36] Operation now in progress\r\n```\r\n\r\nThe number of successful acknowledgements before the error is not deterministic. I've seen something like 3321, but it may go on to 10000 without any problems.\r\n\r\nActually I first encountered a similar problem ('NoneType' object has no attribute 'body_size') in a CentOS 7 Docker container + Python 3.6.4 + pika 0.11.2 (running on an old CentOS 6.9 box) + rabbitmq-server 3.1.5. I have not been able to reproduce the error again, but I am seeing apparent pauses while acknowledging messages, sometimes like being stuck forever.\r\n\r\nI have not been able to reproduce such problem on a vanilla CentOS 7 system."},"comment":{"url":"https://api.github.com/repos/pika/pika/issues/comments/368320003","html_url":"https://github.com/pika/pika/issues/963#issuecomment-368320003","issue_url":"https://api.github.com/repos/pika/pika/issues/963","id":368320003,"user":{"login":"adah1972","id":2137834,"avatar_url":"https://avatars1.githubusercontent.com/u/2137834?v=4","gravatar_id":"","url":"https://api.github.com/users/adah1972","html_url":"https://github.com/adah1972","followers_url":"https://api.github.com/users/adah1972/followers","following_url":"https://api.github.com/users/adah1972/following{/other_user}","gists_url":"https://api.github.com/users/adah1972/gists{/gist_id}","starred_url":"https://api.github.com/users/adah1972/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/adah1972/subscriptions","organizations_url":"https://api.github.com/users/adah1972/orgs","repos_url":"https://api.github.com/users/adah1972/repos","events_url":"https://api.github.com/users/adah1972/events{/privacy}","received_events_url":"https://api.github.com/users/adah1972/received_events","type":"User","site_admin":false},"created_at":"2018-02-25T15:51:16Z","updated_at":"2018-02-25T15:51:16Z","author_association":"NONE","body":"@lukebakken Let me be very specific about the API change. Hopefully we can argue over it for once and all.\r\n\r\n**0: Any API change is bad** (or, *do not punish your users*).\r\n\r\nI suppose this is self-evident. API change breaks existing code, and invalidates documentation. Your users will need to re-read the documentation (if it is up to date), and change the current code.\r\n\r\nGranted, some API might not be originally designed very well. We are human, and we all err. No code is perfect, and no code can be (anyway, non-trivial code can't be). Sometimes we need to live with imperfections. Over-pursuit of perfection may even be harmful.\r\n\r\n**1: When change is necessary, the benefits should be big enough to overcome the harm.**\r\n\r\nSometimes we just have to change. If software does not change, in most cases it is dead. It does not conflict with point 0, but amends it. Just be sure the trade-off is good overall.\r\n\r\nThe change from `basic_consume(self, consumer_callback, queue, …)` to `basic_consume(self, queue, on_message_callback, …)` looks a bit arbitrary, especially the order change. It does not look to me beneficial enough to justify the API change.\r\n\r\n**2: When change is necessary, provide a smooth path for the upgrade.**\r\n\r\nIt is important not to make API users err easily. The common practice in kernel develop is always to use a new syscall number for new/changed APIs, and obsolete syscall numbers are never reused. Libtool provides shared library versioning to make sure when a library is upgraded, existing binaries can either link to an old version or abort on running. Compiled languages usually ensure that changed API will cause build-time failures (by using different types of arguments, or even changing the function name).\r\n\r\nRegretfully, nothing like this happens for the `basic_consume` API change. Programs just break at runtime. The change is neither backward-compatible nor forward-compatible. A breakdown may happen when someone accidentally updates a package. *This is very dangerous and very bad.*\r\n\r\nWhat is worse, there is no way to write code that can run on both 0.11.2 and 1.0.0b, excepting the ugliness as follows:\r\n\r\n```python\r\nfrom packaging import version\r\n…\r\nif version.parse(pika.__version__) < version.parse('0.99'):\r\n    channel.basic_consume(rabbitmq_callback, rabbitmq_queue)\r\nelse:\r\n    channel.basic_consume(rabbitmq_queue, rabbitmq_callback)\r\n```\r\n\r\nIn summary, I do not think this API change is worthwhile. I would suggest keeping the new name but revert the order (`basic_consume(self, on_message_callback, queue, …)` is acceptable to me)."}}
{"actor":{"display_login":"adah1972"}}