{"action":"created","issue":{"url":"https://api.github.com/repos/nvie/rq/issues/526","labels_url":"https://api.github.com/repos/nvie/rq/issues/526/labels{/name}","comments_url":"https://api.github.com/repos/nvie/rq/issues/526/comments","events_url":"https://api.github.com/repos/nvie/rq/issues/526/events","html_url":"https://github.com/nvie/rq/issues/526","id":72525788,"number":526,"title":"After suspend, each worker accepts one more job before becoming \"suspended\"?","user":{"login":"smaccona","id":394952,"avatar_url":"https://avatars.githubusercontent.com/u/394952?v=3","gravatar_id":"","url":"https://api.github.com/users/smaccona","html_url":"https://github.com/smaccona","followers_url":"https://api.github.com/users/smaccona/followers","following_url":"https://api.github.com/users/smaccona/following{/other_user}","gists_url":"https://api.github.com/users/smaccona/gists{/gist_id}","starred_url":"https://api.github.com/users/smaccona/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/smaccona/subscriptions","organizations_url":"https://api.github.com/users/smaccona/orgs","repos_url":"https://api.github.com/users/smaccona/repos","events_url":"https://api.github.com/users/smaccona/events{/privacy}","received_events_url":"https://api.github.com/users/smaccona/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":2,"created_at":"2015-05-01T21:19:23Z","updated_at":"2015-05-06T21:37:58Z","closed_at":null,"body":"I am running RQ 1.5.2, and perhaps I don't fully understand how suspend/resume is supposed to work, but here's what I observe:\r\n\r\n- I have one (or more) workers running\r\n- I suspend RQ using `rq suspend`\r\n- I manually submit a job (see code sample below)\r\n- the job gets accepted by the worker and executes (this is not what I expected to happen)\r\n- after completion, the worker sets its status to `suspended` and prints a message to that effect to `stdout`\r\n- after that, the worker will accept no more jobs until I run `rq resume`\r\n\r\nIs this expected behavior? I would have expected the worker would accept no new jobs after RQ has been suspended, instead of accepting one more and *then* suspending.\r\n\r\nTo reproduce, first create two simple python files:\r\n\r\n**jobs.py**\r\n```python\r\ndef simple():\r\n  return 123\r\n```\r\n**test.py**\r\n```python\r\nfrom rq import Queue\r\nfrom redis import Redis\r\nfrom datetime import datetime\r\nfrom jobs import simple\r\n\r\nredis_conn = Redis()\r\nq = Queue(connection=redis_conn)\r\n\r\nprint datetime.now()\r\njob = q.enqueue(simple)\r\n```\r\nNow run a worker from the directory containing those files and place it in the background so we can easily observe the timeline\r\n```bash\r\n~/test$ rq worker &\r\n[1] 21116\r\n~/test$ 20:58:23 RQ worker started, version 0.5.2\r\n20:58:23 \r\n20:58:23 *** Listening on default...\r\n\r\n~/test$ python test.py\r\n2015-05-01 20:58:27.963802\r\nNone\r\n20:58:27 default: jobs.simple() (7d4c0bd7-a2b7-4b33-8565-2871bbf622c0)\r\n~/test$ 20:58:28 Job OK, result = 123\r\n20:58:28 Result is kept for 500 seconds.\r\n20:58:28 \r\n20:58:28 *** Listening on default...\r\n```\r\nAs expected the job runs fine. Now suspend RQ and submit another job. Note that the job executes, and the background worker does not suspend itself until AFTER the execution of the job.\r\n```bash\r\n~/test$ rq suspend\r\nSuspending workers.  No new jobs will be started.  But current jobs will be completed\r\n~/test$ python test.py\r\n2015-05-01 20:58:54.011261\r\nNone\r\n20:58:54 default: jobs.simple() (55b03430-17de-459c-b14f-8849e0cc9441)\r\n20:58:54 Job OK, result = 123\r\n~/test$ 20:58:54 Result is kept for 500 seconds.\r\n20:58:54 Worker suspended, use \"rq resume\" command to resume\r\n```\r\nUsing `rq info` to monitor the workers mirrors this - the status of each worker is still `idle` after I issue the `rq suspend` command, until the worker receives a new job and completes it, whereupon I see in `rq info` that the status switches to `suspended`.\r\n\r\nThe issue here is that I am trying to wait until all workers are suspended before cycling code (the original reason for the PR containing the suspend/resume functionality, I think), and simply counting the jobs in my various queues doesn't cut it, since a new job will execute on each worker before that worker actually suspends. \r\n\r\nInstead, I have to monitor the status of each worker instead of counting the jobs on the queues, and wait for the workers to become `suspended`. I have about 30 workers at present to handle peak load, but during low load periods I end up having to wait quite a while after suspending: 30 new jobs have to come on the queues, then they are accepted by the workers, which execute the jobs and then suspend themselves afterwards. Only then is it safe to cycle the code.\r\n\r\nI may be missing something here - comments welcome! "},"comment":{"url":"https://api.github.com/repos/nvie/rq/issues/comments/99617484","html_url":"https://github.com/nvie/rq/issues/526#issuecomment-99617484","issue_url":"https://api.github.com/repos/nvie/rq/issues/526","id":99617484,"user":{"login":"smaccona","id":394952,"avatar_url":"https://avatars.githubusercontent.com/u/394952?v=3","gravatar_id":"","url":"https://api.github.com/users/smaccona","html_url":"https://github.com/smaccona","followers_url":"https://api.github.com/users/smaccona/followers","following_url":"https://api.github.com/users/smaccona/following{/other_user}","gists_url":"https://api.github.com/users/smaccona/gists{/gist_id}","starred_url":"https://api.github.com/users/smaccona/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/smaccona/subscriptions","organizations_url":"https://api.github.com/users/smaccona/orgs","repos_url":"https://api.github.com/users/smaccona/repos","events_url":"https://api.github.com/users/smaccona/events{/privacy}","received_events_url":"https://api.github.com/users/smaccona/received_events","type":"User","site_admin":false},"created_at":"2015-05-06T21:37:58Z","updated_at":"2015-05-06T21:37:58Z","body":"I looked in the code and here is the flow:\r\n- `Worker.check_for_suspension()` checks to see whether the underlying connection is suspended, and if it is it sets the worker's `state` appropriately to `suspended` and then blocks until it's no longer suspended.\r\n- `Worker.work()` has a loop which calls `check_for_suspension()` and then proceeds to (among other things) call `dequeue_job_and_maintain_ttl()` to get a job and then executes it if it's a valid job.\r\n- `dequeue_job_and_maintain_ttl()` in turn has a loop which calls `Queue.dequeue_any()`, which in turn calls `Queue.lpop()` which in turn in most cases will translate to a Redis blocking `BLPOP` call. It only ultimately returns to `Worker.work()` if there's a valid job to return.\r\n\r\nSo here's how this issue arises:\r\n- When the queue is idle, the worker is effectively blocking on a Redis call (via `Worker.dequeue_job_and_maintain_ttl()` and `Queue.dequeue_any()` and finally `Queue.lpop()`) for at least `timeout` seconds.\r\n- If RQ is set to `suspended` while the worker is thus blocked, then nothing happens in the worker until it's unblocked, which happens either when a job comes on the queue or the timeout expires. If it's a job, then it eventually makes its way back up to `Worker.work()` and gets executed, and then the worker gets `suspended`. It the timeout expires, then `dequeue_job_and_maintain_ttl()` just re-issues the call to `Queue.dequeue_any()`, so the worker doesn't get to know it has been suspended. So it either takes a while (however long it takes for a job to come on the queue) or never happens (no jobs come on the queue). In the former case, a job gets executed that shouldn't have been executed.\r\n\r\nOne solution that comes to mind is this:\r\n\r\n1. Add the `suspension.WORKERS_SUSPENDED` key to the keys we are requesting in the call to `BLPOP`, so we can immediately return from this blocking Redis call if RQ has been suspended.\r\n2. Add an additional call to `suspension.is_suspended()` in `dequeue_job_and_maintain_ttl()` after the return from `Queue.dequeue_any()`. If the queue is suspended, hand control back to `Worker.work()` which will then get to call `check_for_suspension()` and block as it's supposed to.\r\n\r\nI can make a PR for this if everyone's agreed on the approach. Pulling @jtushman and @selwin into the discussion since they created and merged the original suspend/resume PR respectively, and also @nvie may want to have a say."}}