First impressions good, but fails under load

[mark-99]

So first the good: I've ported the basic functionalty of our job system from RQ to MRQ and generally it went well, at least so far. Dashboard is certainly a huge improvement. The burst mode you added also works well (note as such I'm using latest code from github, rather than the pip install version).

Some observations/suggestions:

• I couldn't actually find any example of queueing jobs from Python (vs command line). Eventually I reverse-engineered mrq-run.py and figured out I had to do this to get jop.queue_jobs() to work:
  mrq.context.set_current_config(mrq.config.get_config())
• Python being untyped, it would be good to document params better. E.g. in RQ enqueue() takes a callable, but MRQ job.queue_job(s) takes a string. This isn't immediately obvious.
• Dashboard frequently reads -ve jobs/second.

However big issue is that the system starts to fail above a certain number of workers, somewhere in the region of 1000. It seems like database updates fail or time out and aren't retried. In some cases the job fails with an exception, but mostly just the status is wrong, ie it gets permanently stuck in either "queued" or "started" state in the DB. Note that workers are still running (variously in "wait" or "full" status) but the queues are not consumed, ie there's an inconsistency between the worker process and/or mongo and/or redis which never resolves (I've just been wiping mongo+redis between tests).

'Failed' is perhaps fixable with the retry operations, but the stuck jobs where presumably some other part of the socket communications failed/timed out are more of an issue.

Here's a code snippet:

    for job_id in job_ids:
        job_res = mrq.job.get_job_result(job_id)
        if job_res["status"] == "queued": num_queued += 1
        elif job_res["status"] == "started": num_started += 1
        elif job_res["status"] == "success": num_finished += 1
        else: num_failed += 1

    log("num_queued: %d, num_started: %d, num_finished:%d, num_failed: %d" % 
            (num_queued, num_started, num_finished, num_failed))

Here's the final output for 50k jobs (number of jobs is less important than number of workers):
09:19:26.906279: num_queued: 32, num_started: 8, num_finished:49957, num_failed: 3

Different runs give different numbers of stuck or failed (sometimes it will complete ok). Fewer workers (<500) and it's always ok, it goes to zero queued+started, and my script exits. Note the stats also agree on the dashboard "Statuses" tab.

For the failed jobs, here's the backtrace:

Traceback (most recent call last): 
  File "/apps/infrafs1/matkinson/venv-infra/lib/python2.7/site-packages/mrq/worker.py", line 540, in perform_job 
    job.perform() 
  File "/apps/infrafs1/matkinson/venv-infra/lib/python2.7/site-packages/mrq/job.py", line 279, in perform 
    self.save_success(result) 
  File "/apps/infrafs1/matkinson/venv-infra/lib/python2.7/site-packages/mrq/job.py", line 364, in save_success 
    self._save_status("success", updates) 
  File "/apps/infrafs1/matkinson/venv-infra/lib/python2.7/site-packages/mrq/job.py", line 437, in _save_status 
    }, {"$set": db_updates}, w=w, j=j, manipulate=False) 
  File "/apps/infrafs1/matkinson/venv-infra/lib/python2.7/site-packages/pymongo/collection.py", line 1956, in update 
    with self._socket_for_writes() as sock_info: 
  File "/apps/infrafs1/matkinson/venv-infra/lib/python2.7/contextlib.py", line 17, in __enter__ 
    return self.gen.next() 
  File "/apps/infrafs1/matkinson/venv-infra/lib/python2.7/site-packages/pymongo/mongo_client.py", line 665, in _get_socket 
    with server.get_socket(self.__all_credentials) as sock_info: 
  File "/apps/infrafs1/matkinson/venv-infra/lib/python2.7/contextlib.py", line 17, in __enter__ 
    return self.gen.next() 
  File "/apps/infrafs1/matkinson/venv-infra/lib/python2.7/site-packages/pymongo/server.py", line 102, in get_socket 
    with self.pool.get_socket(all_credentials, checkout) as sock_info: 
  File "/apps/infrafs1/matkinson/venv-infra/lib/python2.7/contextlib.py", line 17, in __enter__ 
    return self.gen.next() 
  File "/apps/infrafs1/matkinson/venv-infra/lib/python2.7/site-packages/pymongo/pool.py", line 509, in get_socket 
    sock_info = self._get_socket_no_auth() 
  File "/apps/infrafs1/matkinson/venv-infra/lib/python2.7/site-packages/pymongo/pool.py", line 543, in _get_socket_no_auth 
    sock_info, from_pool = self.connect(), False 
  File "/apps/infrafs1/matkinson/venv-infra/lib/python2.7/site-packages/pymongo/pool.py", line 475, in connect 
    DEFAULT_CODEC_OPTIONS)) 
  File "/apps/infrafs1/matkinson/venv-infra/lib/python2.7/site-packages/pymongo/network.py", line 48, in command 
    response = receive_message(sock, 1, request_id) 
  File "/apps/infrafs1/matkinson/venv-infra/lib/python2.7/site-packages/pymongo/network.py", line 60, in receive_message 
    header = _receive_data_on_socket(sock, 16) 
  File "/apps/infrafs1/matkinson/venv-infra/lib/python2.7/site-packages/pymongo/network.py", line 84, in _receive_data_on_socket 
    raise AutoReconnect("connection closed") 
AutoReconnect: connection closed 

Any ideas? Thanks.

[mark-99]

Ok so I seem to have fixed this by increasing the ulimit for fds and threads on the mongod process - looks like it needs 4 connections per worker. I also had systems increase the socket timeout on the Condor nodes, so that might have helped also.

So far it's holding up at 100k jobs on 5k workers...

BTW I realised -ve jobs/sec is normal, just means decreasing...

[sylvinus]

Hi mark! Thanks for your feedbacks! Can you give a few details about the configuration of your workers? (number or processes & greenlets)

There are definitely some areas to improve in the docs, I'll open a separate issue with your feedback.

By the way, have you added the Job maintenance tasks? They are useful when the network or the worker environment is unreliable:
http://mrq.readthedocs.org/en/latest/jobs-maintenance/

[sylvinus]

With lots of workers, you might also want to set --report_interval to something higher than the default 10 seconds.

[mark-99]

The params to the workers are just the redis & mongo URIs, and --dequeue_strategy burst. I've added --report_interval 30 also now.

The Condor nodes are each single CPU cores, so typically we run 1 process/thread. However the jobs to tend to be I/O bound so there might be some benefit perhaps. Not sure about greenlets (I'm new to Python) but looking at the docs that seems to be a cooperative multi-tasking scheme (like coroutines or fibers), so I'm not sure if that would be helpful.

The system is intended to run arbitrary jobs from user-supplied Python modules (which may in turn call C++ libraries or even spawn other processes). We've found 'raw' Condor has a lot of per-job overhead, as well as some reliability/responsiveness issues when fed large numbers of jobs all at once, hence I'm trying out this idea of layering a queueing system on top.

BTW is there any better way to query for when (queued+started==0) as per the code I posted above (ie the exit condition)? Iterating large numbers of jobs and doing string tests gets quite slow.

[sylvinus]

Hi mark,

If your jobs are I/O bound, then taking advantage of gevent with --greenlets is definitely a good way to reduce overhead. We do web crawling and run most of our workers with --greenlets 30 which really boosts performance.

To query the jobs, you could do something like this?

from mrq.context import connections, setup_context
setup_context()
empty = (connections.mongodb_jobs.mrq_jobs.count({"status": {"$in": ["started", "queued"]}, "queue": "myqueue"}) == 0)

[mark-99]

That works great, and much faster. Might I suggest that becomes a 1st class API? e.g.:

def get_queue_count(queue_name, status_list):
    return mrq.context.connections.mongodb_jobs.mrq_jobs.count({"status": {"$in": status_list}, "queue": queue_name})

[sylvinus]

I'll add a separate ticket on that!