@memoize_single_argument
def load_class_by_path(taskpath):
    """ Given a taskpath, returns the main task class. """

    return getattr(
        importlib.import_module(
            re.sub(
                r"\.[^.]+$",
                "",
                taskpath)),
        re.sub(
            r"^.*\.",
            "",
            taskpath)) class Job(object):

    timeout = None
    result_ttl = None
    abort_ttl = None
    cancel_ttl = None
    max_retries = None
    retry_delay = None

    # All values above can be overrided from the TASKS config

    progress = 0

    _memory_stop = 0
    _memory_start = 0

    _current_io = None

    # Has this job been inserted in MongoDB yet?
    stored = None

    # List of statuses that don't trigger a storage of this task in MongoDB on raw queues.
    # In task config, use ("started", "success") to avoid storing successful raw tasks at all
    statuses_no_storage = None

    def __init__(self, job_id, queue=None, start=False, fetch=False):
        self.worker = context.get_current_worker()
        self.queue = queue
        self.datestarted = None

        self.collection = context.connections.mongodb_jobs.mrq_jobs

        if job_id is None:
            self.id = None
        else:
            if isinstance(job_id, bytes):
                self.id = ObjectId(job_id.decode('utf-8'))
            else:
                self.id = ObjectId(job_id)

        self.data = None
        self.saved = True

        self.task = None
        self.greenlet_switches = 0
        self.greenlet_time = 0

        self._trace_mongodb = defaultdict(int)

        if start:
            self.fetch(start=True, full_data=False)
        elif fetch:
            self.fetch(start=False, full_data=False)

    @property
    def redis_max_concurrency_key(self):
        """ Returns the global redis key used to control job concurrency """
        return "%s:c:%s" % (context.get_current_config()["redis_prefix"], self.data["path"])

    def exists(self):
        """ Returns True if a job with the current _id exists in MongoDB. """
        return bool(self.collection.find_one({"_id": self.id}, projection={"_id": 1}))

    def fetch(self, start=False, full_data=True):
        """ Get the current job data and possibly flag it as started. """

        if self.id is None:
            return self

        if full_data is True:
            fields = None
        elif isinstance(full_data, dict):
            fields = full_data
        else:
            fields = {
                "_id": 0,
                "path": 1,
                "params": 1,
                "status": 1,
                "retry_count": 1,
            }

        if start:
            self.datestarted = datetime.datetime.utcnow()
            self.set_data(self.collection.find_and_modify(
                {
                    "_id": self.id,
                    "status": {"$nin": ["cancel", "abort", "maxretries"]}
                },
                {"$set": {
                    "status": "started",
                    "datestarted": self.datestarted,
                    "worker": self.worker.id
                },
                "$unset": {
                    "dateexpires": 1 # we don't want started jobs to expire unexpectedly
                }},
                projection=fields)
            )

            context.metric("jobs.status.started")

        else:
            self.set_data(self.collection.find_one({
                "_id": self.id
            }, projection=fields))

        if self.data is None:
            context.log.info(
                "Job %s not found in MongoDB or status was cancelled!" %
                self.id)

        self.stored = True

        return self

    def get_task_config(self):
        cfg = context.get_current_config()
        return cfg.get("tasks", {}).get(
            self.data["path"]
        ) or {}

    def set_data(self, data):
        self.data = data
        if self.data is None:
            return

        if "path" in self.data:
            cfg = context.get_current_config()
            task_def = self.get_task_config()

            self.timeout = task_def.get("timeout", cfg["default_job_timeout"])
            self.default_ttl = task_def.get("default_ttl", cfg["default_job_ttl"])
            self.result_ttl = task_def.get("result_ttl", cfg["default_job_result_ttl"]) # success ttl
            self.abort_ttl = task_def.get("abort_ttl", cfg["default_job_abort_ttl"])
            self.cancel_ttl = task_def.get("cancel_ttl", cfg["default_job_cancel_ttl"])
            self.max_retries = task_def.get("max_retries", cfg["default_job_max_retries"])
            self.retry_delay = task_def.get("retry_delay", cfg["default_job_retry_delay"])

    def set_progress(self, ratio, save=False):
        self.data["progress"] = ratio
        self.saved = False

        # If not saved, will be updated in the next worker report
        if save:
            self.save()

    def save(self):
        """ Persists the current job metadata to MongoDB. Will be called at each worker report. """

        if not self.saved and self.data and "progress" in self.data:
            # TODO should we save more fields?
            self.collection.update({"_id": self.id}, {"$set": {
                "progress": self.data["progress"]
            }})
            self.saved = True

    @classmethod
    def insert(cls, jobs_data, queue=None, statuses_no_storage=None, return_jobs=True, w=None, j=None):
        """ Insert a job into MongoDB """

        now = datetime.datetime.utcnow()
        for data in jobs_data:
            if data["status"] == "started":
                data["datestarted"] = now

        no_storage = (statuses_no_storage is not None) and ("started" in statuses_no_storage)
        if no_storage and return_jobs:
            for data in jobs_data:
                data["_id"] = ObjectId()  # Give the job a temporary ID
        else:
            inserted = context.connections.mongodb_jobs.mrq_jobs.insert(
                jobs_data,
                manipulate=True,
                w=w,
                j=j
            )

        if return_jobs:
            jobs = []
            for data in jobs_data:
                job = cls(data["_id"], queue=queue)
                job.set_data(data)
                job.statuses_no_storage = statuses_no_storage
                job.stored = (not no_storage)
                if data["status"] == "started":
                    job.datestarted = data["datestarted"]
                jobs.append(job)

            return jobs
        else:
            return inserted

    def _attach_original_exception(self, exc):
        """ Often, a retry will be raised inside an "except" block.
            This Keep track of the first exception for debugging purposes """

        original_exception = sys.exc_info()
        if original_exception[0] is not None:
            exc.original_exception = original_exception

    def retry(self, queue=None, delay=None, max_retries=None):
        """ Marks the current job as needing to be retried. Interrupts it. """

        max_retries = max_retries
        if max_retries is None:
            max_retries = self.max_retries

        if self.data.get("retry_count", 0) >= max_retries:
            raise MaxRetriesInterrupt()

        exc = RetryInterrupt()

        exc.queue = queue or self.queue or self.data.get("queue") or "default"
        exc.retry_count = self.data.get("retry_count", 0) + 1
        exc.delay = delay
        if exc.delay is None:
            exc.delay = self.retry_delay

        self._attach_original_exception(exc)

        raise exc

    def abort(self):
        """ Aborts the current task mid-excution. """
        exc = AbortInterrupt()
        self._attach_original_exception(exc)
        raise exc

    def cancel(self):
        """ Markes the current job as cancelled. Doesn't interrupt it. """
        self._save_status("cancel")

    def requeue(self, queue=None, retry_count=0):
        """ Requeues the current job. Doesn't interrupt it """

        if not queue:
            if not self.data or not self.data.get("queue"):
                self.fetch(full_data={"_id": 0, "queue": 1, "path": 1})
            queue = self.data["queue"]

        self._save_status("queued", updates={
            "queue": queue,
            "datequeued": datetime.datetime.utcnow(),
            "retry_count": retry_count
        })

    def perform(self):
        """ Loads and starts the main task for this job, the saves the result. """

        if self.data is None:
            return

        context.log.debug("Starting %s(%s)" % (self.data["path"], self.data["params"]))
        task_class = load_class_by_path(self.data["path"])

        self.task = task_class()

        self.task.is_main_task = True

        if not self.task.max_concurrency:

            result = self.task.run_wrapped(self.data["params"])

        else:

            if self.task.max_concurrency > 1:
                raise NotImplementedError()

            lock = None
            try:

                # TODO: implement a semaphore
                lock = context.connections.redis.lock(self.redis_max_concurrency_key, timeout=self.timeout + 5)
                if not lock.acquire(blocking=True, blocking_timeout=0):
                    raise MaxConcurrencyInterrupt()

                result = self.task.run_wrapped(self.data["params"])

            finally:
                try:
                    if lock:
                        lock.release()
                except LockError:
                    pass

        self.save_success(result)

        if context.get_current_config().get("trace_greenlets"):

            # TODO: this is not the exact greenlet_time measurement because it doesn't
            # take into account the last switch's time. This is why we force a last switch.
            # This does cause a performance overhead. Instead, we should print the
            # last timing directly from the trace() function in context?

            # pylint: disable=protected-access

            gevent.sleep(0)
            current_greenlet = gevent.getcurrent()
            t = (datetime.datetime.utcnow() - self.datestarted).total_seconds()

            context.log.debug(
                "Job %s success: %0.6fs total, %0.6fs in greenlet, %s switches" %
                (self.id,
                 t,
                 current_greenlet._trace_time,
                 current_greenlet._trace_switches - 1)
            )

        else:
            context.log.debug("Job %s success: %0.6fs total" % (
                self.id, (datetime.datetime.utcnow() -
                          self.datestarted).total_seconds()
            ))

        return result

    def wait(self, poll_interval=1, timeout=None, full_data=False):
        """ Wait for this job to finish. """

        end_time = None
        if timeout:
            end_time = time.time() + timeout

        while end_time is None or time.time() < end_time:

            job_data = self.collection.find_one({
                "_id": ObjectId(self.id),
                "status": {"$nin": ["started", "queued"]}
            }, projection=({
                "_id": 0,
                "result": 1,
                "status": 1
            } if not full_data else None))

            if job_data:
                return job_data

            time.sleep(poll_interval)
        raise Exception("Waited for job result for %s seconds, timeout." % timeout)

    def kill(self, block=False, reason="unknown"):
        """ Forcefully kill all greenlets associated with this job """

        current_greenletid = id(gevent.getcurrent())

        trace = "Job killed: %s" % reason
        for greenlet, job in context._GLOBAL_CONTEXT["greenlets"].values():
            greenletid = id(greenlet)
            if job and job.id == self.id and greenletid != current_greenletid:
                greenlet.kill(block=block)
                trace += "\n\n--- Greenlet %s ---\n" % greenletid
                trace += "".join(traceback.format_stack(greenlet.gr_frame))
            context._GLOBAL_CONTEXT["greenlets"].pop(greenletid, None)

        if reason == "timeout" and self.data["status"] != "timeout":
            updates = {
                "exceptiontype": "TimeoutInterrupt",
                "traceback": trace
            }
            self._save_status("timeout", updates=updates, exception=False)

    def save_retry(self, retry_exc):

        # If delay=0, requeue right away, don't go through the "retry" status
        if retry_exc.delay == 0:
            self.requeue(queue=retry_exc.queue, retry_count=retry_exc.retry_count)

        else:

            dateretry = datetime.datetime.utcnow() + datetime.timedelta(seconds=retry_exc.delay)
            updates = {
                "dateretry": dateretry,
                "queue": retry_exc.queue,
                "retry_count": retry_exc.retry_count
            }

            self._save_status("retry", updates, exception=True)

    def _save_traceback_history(self, status, trace, job_exc):
        """ Create traceback history or add a new traceback to history. """
        failure_date = datetime.datetime.utcnow()

        new_history = {
            "date": failure_date,
            "status": status,
            "exceptiontype": job_exc.__name__
        }

        traces = trace.split("---- Original exception: -----")
        if len(traces) > 1:
            new_history["original_traceback"] = traces[1]
        worker = context.get_current_worker()
        if worker:
            new_history["worker"] = worker.id
        new_history["traceback"] = traces[0]
        self.collection.update({
            "_id": self.id
        }, {"$push": {"traceback_history": new_history}})

    def save_success(self, result=None):

        dateexpires = datetime.datetime.utcnow() + datetime.timedelta(seconds=self.result_ttl)
        updates = {
            "dateexpires": dateexpires
        }
        if result is not None:
            updates["result"] = result
        if "progress" in self.data:
            updates["progress"] = 1

        self._save_status("success", updates)

    def save_cancel(self):

        dateexpires = datetime.datetime.utcnow() + datetime.timedelta(seconds=self.cancel_ttl)
        updates = {
            "dateexpires": dateexpires
        }

        self._save_status("cancel", updates)

    def save_abort(self):
        dateexpires = datetime.datetime.utcnow() + datetime.timedelta(seconds=self.abort_ttl)
        updates = {
            "dateexpires": dateexpires
        }

        self._save_status("abort", updates, exception=True)

    def _save_status(self, status, updates=None, exception=False, w=None, j=None):

        if self.id is None:
            return

        # Forbid some status transitions
        if self.data and self.data.get("status") in FINAL_STATUSES and status not in TRANSIENT_STATUSES:
            context.log.error("Can't go from status %s to %s" % (self.data["status"], status))
            return

        context.metric("jobs.status.%s" % status)

        if self.stored is False and self.statuses_no_storage is not None and status in self.statuses_no_storage:
            return

        now = datetime.datetime.utcnow()
        db_updates = {
            "status": status,
            "dateupdated": now
        }

        # we don't want started jobs to expire unexpectedly
        if status not in ["started", "success", "abort", "cancel"] and hasattr(self, "default_ttl") and self.default_ttl is not None:
            db_updates["dateexpires"] = (self.data.get("datequeued") or now) + datetime.timedelta(seconds=self.default_ttl)

        db_updates.update(updates or {})

        if self.datestarted:
            db_updates["totaltime"] = (now - self.datestarted).total_seconds()

        if context.get_current_config().get("trace_greenlets"):
            current_greenlet = gevent.getcurrent()

            # TODO are we sure the current job is doing the save_status() on itself?
            if hasattr(current_greenlet, "_trace_time"):
                # pylint: disable=protected-access
                db_updates["time"] = current_greenlet._trace_time
                db_updates["switches"] = current_greenlet._trace_switches

        if exception:
            trace = traceback.format_exc()
            context.log.error(trace)
            exc, value = sys.exc_info()[0:2]
            if hasattr(value, "subpool_traceback"):
                trace = "Exception first caught in a subpool. Traceback:\n%s\n%s" % (value.subpool_traceback, trace)
            db_updates["traceback"] = trace
            db_updates["exceptiontype"] = exc.__name__

        if self.data:
            self.data.update(db_updates)
            # get all data before updating them
            current_queue = (db_updates or {}).get("queue") or self.data["queue"]
            old_queue = self.data.get("queue")
            old_status = self.data.get("status")
            raw_queue = self.data.get("raw_queue")
            retry_count = self.data.get("retry_count", 0)

        # In the most common case, we allow an optimization on Mongo writes
        if status == "success":
            if w is None:
                w = getattr(self.task, "status_success_update_w", None)
            if j is None:
                j = getattr(self.task, "status_success_update_j", None)

        # This job wasn't inserted because "started" is in statuses_no_storage
        # So we must insert it for the first time instead of updating it.
        if self.stored is False:
            db_updates["queue"] = self.data["queue"]
            db_updates["params"] = self.data["params"]
            db_updates["path"] = self.data["path"]
            self.collection.insert(db_updates, w=w, j=j, manipulate=True)
            self.id = db_updates["_id"]  # Persistent ID assigned by the server
            self.stored = True

        else:
            self.collection.update({
                "_id": self.id
            }, {"$set": db_updates}, w=w, j=j, manipulate=False)

        if exception:
            self._save_traceback_history(status, trace, exc)

        if self.data:
            with context.connections.redis.pipeline(transaction=False) as pipe:
                if status != "started":
                    # Queue change
                    if current_queue != old_queue:
                        pipe.decr("queuesize:%s" % old_queue)
                        if status == "queued":
                            pipe.incr("queuesize:%s" % current_queue)

                    # Regular queues
                    elif status == "queued" and old_status != "started":
                        pipe.incr("queuesize:%s" % current_queue)

                    elif status != "queued" and not raw_queue:
                        pipe.decr("queuesize:%s" % current_queue)

                    # Raw queues retries
                    elif (db_updates or {}).get("retry_count", 0) > retry_count:
                        pipe.incr("queuesize:%s" % current_queue)

                    pipe.expire("queuesize:%s" % current_queue, context.get_current_config().get("queue_ttl"))
                pipe.execute()

    def set_current_io(self, io_data):

        # pylint: disable=protected-access

        if io_data is None:
            if not self._current_io:
                return

            t = time.time() - self._current_io["started"]
            if self.worker and self.data.get("path"):
                self.worker._traced_io["types"][self._current_io["type"]] += t
                self.worker._traced_io["tasks"][self.data["path"]] += t
                self.worker._traced_io["total"] += t

            self._current_io = None

        else:
            io_data["started"] = time.time()
            self._current_io = io_data

    def trace_memory_clean_caches(self):
        """ Avoid polluting results with some builtin python caches """

        urllib.parse.clear_cache()
        re.purge()
        linecache.clearcache()
        copyreg.clear_extension_cache()

        if hasattr(fnmatch, "purge"):
            fnmatch.purge()  # pylint: disable=no-member
        elif hasattr(fnmatch, "_purge"):
            fnmatch._purge()  # pylint: disable=no-member

        if hasattr(encodings, "_cache") and len(encodings._cache) > 0:
            encodings._cache = {}

        for handler in context.log.handlers:
            handler.flush()

    def trace_memory_start(self):
        """ Starts measuring memory consumption """

        self.trace_memory_clean_caches()

        objgraph.show_growth(limit=30)

        gc.collect()
        self._memory_start = self.worker.get_memory()["total"]

    def trace_memory_stop(self):
        """ Stops measuring memory consumption """

        self.trace_memory_clean_caches()

        objgraph.show_growth(limit=30)

        trace_type = context.get_current_config()["trace_memory_type"]
        if trace_type:

            filename = '%s/%s-%s.png' % (
                context.get_current_config()["trace_memory_output_dir"],
                trace_type,
                self.id)

            chain = objgraph.find_backref_chain(
                random.choice(
                    objgraph.by_type(trace_type)
                ),
                objgraph.is_proper_module
            )
            objgraph.show_chain(chain, filename=filename)
            del filename
            del chain

        gc.collect()
        self._memory_stop = self.worker.get_memory()["total"]

        diff = self._memory_stop - self._memory_start

        context.log.debug("Memory diff for job %s : %s" % (self.id, diff))

        # We need to update it later than the results, we need them off memory
        # already.
        self.collection.update(
            {"_id": self.id},
            {"$set": {
                "memory_diff": diff
            }},
            w=1
        ) class TimeoutInterrupt(_MrqInterrupt):
    pass class StopRequested(GreenletExit):
    """ Thrown in the mail greenlet to stop dequeuing jobs. """
    pass class JobInterrupt(GreenletExit):
    """ Interrupts that stop a job in its execution, e.g. when responding to a SIGTERM. """
    pass class AbortInterrupt(_MrqInterrupt):
    pass class RetryInterrupt(_MrqInterrupt):
    delay = None
    queue = None
    retry_count = 0

    def _get_exception_name(self):
        return "%s #%s: %s seconds, %s queue" % (
            self.__class__.__name__, self.retry_count, self.delay, self.queue
        ) class MaxRetriesInterrupt(_MrqInterrupt):
    pass class MaxConcurrencyInterrupt(_MrqInterrupt):
    pass _GLOBAL_CONTEXT = {

    # Contains all the running greenlets for this worker. greenletid => Job object
    "greenlets": {},

    # pointer to the current worker
    "worker": None,

    # pointer to the current config
    "config": {}
}
def setup_context(**kwargs):
def set_current_job(job):
def get_current_job(greenlet_id=None):
def set_current_worker(worker):
def get_current_worker():
def set_logger_config():
def set_current_config(config):
def get_current_config():
def retry_current_job(delay=None, max_retries=None, queue=None):
def abort_current_job():
def _connections_factory(attr):
    def versiontuple(v):
def enable_greenlet_tracing():
    def trace(*args):
def run_task(path, params):
def set_current_job_progress(ratio, save=False):
def queue_raw_jobs(*args, **kwargs):
def queue_job(*args, **kwargs):
def queue_jobs(*args, **kwargs):
def metric(*args, **kwargs): class Queue(object):
    """ A Queue for Jobs. """

    is_raw = False
    is_timed = False
    is_sorted = False
    is_set = False

    is_reverse = False

    # root_id will contain the root queue id without any trailing subqueue delimiter
    # e.g. if self.id is "some_queue/" then self.root_id will contain "some_queue"
    # and if self.id is "some_queue/some_subqueue" then self.root_id will contain "some_queue"
    root_id = None
    is_subqueue = False

    use_large_ids = False

    # This is a mutable type so it is shared by all instances
    # of Queue in the current process
    paused_queues = set()

    def __new__(cls, queue_id, **kwargs):
        """ Creates a new instance of the right queue type """

        if cls is not Queue:
            return object.__new__(cls)

        if isinstance(queue_id, Queue):
            queue_id = queue_id.id

        queue_type = Queue.get_queue_type(queue_id)

        if queue_type == "regular":
            from .queue_regular import QueueRegular
            return QueueRegular(queue_id, **kwargs)
        else:
            from .queue_raw import QueueRaw
            return QueueRaw(queue_id, **kwargs)

    def __init__(self, queue_id):

        if queue_id[-8:] == "_reverse":
            self.is_reverse = True
            queue_id = queue_id[:-8]

        self.id = queue_id

        self.root_id = self.id

        if "/" in self.id:
            # Get the root queue id with no trailing delimiter
            self.root_id = self.id.split("/")[0]

            if self.id.split("/")[1]:
                self.is_subqueue = True

        self.use_large_ids = context.get_current_config()["use_large_job_ids"]

    @classmethod
    def get_queue_type(cls, queue_id):
        """ Return the queue type, currently determined only by its suffix. """

        for queue_type in ("timed_set", "sorted_set", "set", "raw"):
            if queue_id.split("/")[0].endswith("_%s" % queue_type):
                return queue_type

        return "regular"

    @classmethod
    def get_queues_config(cls):
        """ Returns the queues configuration dict """
        _config = context.get_current_config()
        return _config.get("raw_queues") or _config.get("queues_config") or {}

    def get_config(self):
        """ Returns the specific configuration for this queue """
        return Queue.get_queues_config().get(self.root_id) or {}

    def serialize_job_ids(self, job_ids):
        """ Returns job_ids serialized for storage in Redis """
        if len(job_ids) == 0 or self.use_large_ids:
            return job_ids
        elif isinstance(job_ids[0], ObjectId):
            return [x.binary for x in job_ids]
        else:
            return [bytes.fromhex(str(x)) for x in job_ids]

    def unserialize_job_ids(self, job_ids):
        """ Unserialize job_ids stored in Redis """
        if len(job_ids) == 0 or self.use_large_ids:
            return job_ids
        else:
            return [binascii.hexlify(x.encode('utf-8') if (PY3 and isinstance(x, str)) else x).decode('ascii')
                    for x in job_ids]

    def _get_pausable_id(self):
      """
          Get the queue id (either id or root_id) that should be used to pause/unpause the current queue
          TODO: handle subqueues with more than one level, e.g. "queue/subqueue/"
      """
      queue = self.id
      if self.id.endswith("/"):
          queue = self.root_id
      return queue

    def pause(self):
        """ Adds this queue to the set of paused queues """
        context.connections.redis.sadd(redis_key("paused_queues"), self._get_pausable_id())

    def is_paused(self):
        """
            Returns wether the queue is paused or not.
            Warning: this does NOT ensure that the queue was effectively added to
            the set of paused queues. See the 'paused_queues_refresh_interval' option.
        """
        root_is_paused = False
        if self.root_id != self.id:
            root_is_paused = context.connections.redis.sismember(redis_key("paused_queues"), self.root_id)

        return root_is_paused or context.connections.redis.sismember(redis_key("paused_queues"), self.id)

    def resume(self):
        """ Resumes a paused queue """
        context.connections.redis.srem(redis_key("paused_queues"), self._get_pausable_id())

    def count_jobs_to_dequeue(self):
        """ Returns the number of jobs that can be dequeued right now from the queue. """

        return self.size()

    @classmethod
    def all_active(cls):
        """ List active queues, based on their lengths in Redis. Warning, uses the unscalable KEYS redis command """

        prefix = context.get_current_config()["redis_prefix"]
        queues = []
        for key in context.connections.redis.keys():
            if key.startswith(prefix):
                queues.append(Queue(key[len(prefix) + 3:]))

        return queues

    @classmethod
    def all_known(cls, sources=None, prefixes=None):
        """ List all currently known queues """

        sources = sources or ("config", "jobs", "raw_subqueues")

        queues = set()

        if "config" in sources and not prefixes:
            # Some queues are explicitly declared in the config (including all root raw queues)
            cfg = context.get_current_config()

            queues_from_config = [
                t.get("queue")
                for t in (cfg.get("tasks") or {}).values()
                if t.get("queue")
            ]

            queues_from_config += Queue.get_queues_config().keys()

            queues_from_config += [
                t.get("retry_queue")
                for t in Queue.get_queues_config().values()
                if t.get("retry_queue")
            ]

            queues |= set(queues_from_config)

        if "jobs" in sources:

            # This will get all queues from mongodb, including those where we have only non-queued jobs
            for q in context.connections.mongodb_jobs.mrq_jobs.distinct("queue"):
                if prefixes and not any(q.startswith(p) for p in prefixes):
                    continue
                queues.add(q)

        if "raw_subqueues" in sources:
            for q in Queue.get_queues_config():
                if prefixes and not any(q + "/" == p for p in prefixes):
                    continue
                queue_obj = Queue(q)
                if queue_obj.is_raw and queue_obj.has_subqueues:
                    # TODO: optimize this with a single SUNION on all keys
                    queues |= queue_obj.get_known_subqueues()

        return queues

    @classmethod
    def all(cls):
        """ List all queues in MongoDB via aggregation, with their queued jobs counts. Might be slow. """

        # Start with raw queues we know exist from the config
        queues = {x: 0 for x in Queue.get_queues_config()}

        stats = list(context.connections.mongodb_jobs.mrq_jobs.aggregate([
            {"$match": {"status": "queued"}},
            {"$group": {"_id": "$queue", "jobs": {"$sum": 1}}}
        ]))

        queues.update({x["_id"]: x["jobs"] for x in stats})

        return queues

    def use_notify(self):
        """ Does this queue use notifications? """
        return bool(self.get_config().get("notify"))

    def notify(self, new_jobs_count):
        """ We just queued new_jobs_count jobs on this queue, wake up the workers if needed """

        if not self.use_notify():
            return

        # Not really useful to send more than 100 notifs (to be configured)
        count = min(new_jobs_count, 100)

        notify_key = redis_key("notify", self)

        context.connections.redis.lpush(notify_key, *([1] * count))
        context.connections.redis.expire(notify_key, max(1, int(context.get_current_config()["max_latency"] * 2))) class MongoJSONEncoder(json.JSONEncoder):

    def default(self, obj):  # pylint: disable=E0202
        if isinstance(obj, (datetime.datetime, datetime.date)):
            return obj.isoformat()
        elif isinstance(obj, ObjectId):
            return str(obj)
        elif isinstance(obj, uuid.UUID):
            return str(obj)
        elif isinstance(obj, bytes):
            return obj.decode('utf-8')
        return json.JSONEncoder.default(self, obj) class MovingAverage(object):

    def __init__(self, size):
        self.__size = size
        self.__sum = 0
        self.__q = deque([])

    def next(self, val):
        if len(self.__q) == self.__size:
            self.__sum -= self.__q.popleft()
        self.__sum += val
        self.__q.append(val)
        return 1.0 * self.__sum / len(self.__q) class Process(object):
    """ The parent class of Worker, Agent and Supervisor """

    exitcode = 0

    def install_signal_handlers(self):
        """ Handle events like Ctrl-C from the command line. """

        self.graceful_stop = False

        def request_shutdown_now():
            self.shutdown_now()

        def request_shutdown_graceful():

            # Second time CTRL-C, shutdown now
            if self.graceful_stop:
                self.shutdown_now()
            else:
                self.graceful_stop = True
                self.shutdown_graceful()

        # First time CTRL-C, try to shutdown gracefully
        gevent.signal_handler(signal.SIGINT, request_shutdown_graceful)

        # User (or Heroku) requests a stop now, just mark tasks as interrupted.
        gevent.signal_handler(signal.SIGTERM, request_shutdown_now) def redis_key(name, *args):
  prefix = context.get_current_config()["redis_prefix"]
  if name == "known_subqueues":
    return "%s:ksq:%s" % (prefix, args[0].root_id)
  elif name == "queue":
     return "%s:q:%s" % (prefix, args[0].id)
  elif name == "started_jobs":
    return "%s:s:started" % prefix
  elif name == "paused_queues":
    return "%s:s:paused" % prefix
  elif name == "notify":
     return "%s:notify:%s" % (prefix, args[0].root_id)
from future import standard_library
from future.builtins import str, bytes
from future.utils import iteritems
from bson import ObjectId
    from redis.lock import LuaLock
    from redis.lock import Lock as LuaLock
from collections import defaultdict
from mrq.utils import load_class_by_path
from .job import Job
from .exceptions import (TimeoutInterrupt, StopRequested, JobInterrupt, AbortInterrupt,
                         RetryInterrupt, MaxRetriesInterrupt, MaxConcurrencyInterrupt)
from .context import (set_current_worker, set_current_job, get_current_job, get_current_config,
                      connections, enable_greenlet_tracing, run_task, log)
from .queue import Queue
from .utils import MongoJSONEncoder, MovingAverage
from .processes import Process
from .redishelpers import redis_key
        from gevent import pywsgi
            from .scheduler import Scheduler
import gevent
import gevent.pool
import os
import signal
import datetime
import time
import socket
import traceback
import psutil
import sys
import json as json_stdlib
import ujson as json
        import logging
standard_library.install_aliases()

try:
except ImportError:
    # Change name to avoid NameError raised when use of LuaLock at line 151
    



class Worker(Process):
    """ Main worker class """

    # Allow easy overloading
    job_class = Job

    # See the doc for valid statuses
    status = "init"

    mongodb_jobs = None
    mongodb_logs = None
    redis = None

    def __init__(self):

        set_current_worker(self)

        if self.config.get("trace_greenlets"):
            enable_greenlet_tracing()

        self.datestarted = datetime.datetime.utcnow()

        self.done_jobs = 0
        self.max_jobs = self.config["max_jobs"]
        self.max_time = datetime.timedelta(seconds=self.config["max_time"]) or None

        self.paused_queues = set()

        self.connected = False  # MongoDB + Redis

        self.process = psutil.Process(os.getpid())
        self.greenlet = gevent.getcurrent()
        self.graceful_stop = None

        self.work_lock = gevent.lock.Semaphore()

        if self.config.get("worker_id"):
            self.id = ObjectId(self.config["worker_id"])
        else:
            self.id = ObjectId()

        if self.config.get("name"):
            self.name = self.config["name"]
        else:
            # Generate a somewhat human-readable name for this worker
            self.name = "%s.%s" % (socket.gethostname().split(".")[0], os.getpid())

        self.pool_size = self.config["greenlets"]

        self.pool_usage_average = MovingAverage((60 / self.config["report_interval"] or 1))