A function annotation that adds a worker request. A worker request
           is a POST request that is computed asynchronously. That is, the
           actual task is performed in a different thread and the network
           request returns immediately. The client side uses polling to fetch
           the result and can also cancel the task. The worker javascript
           client side must be linked and used for accessing the request.

        Parameters
        ----------
        mask : string
            The URL that must be matched to perform this request.

        cache_id : function(args) or None
            Optional function for caching the result. If set the worker must be
            idempotent. Requires a `cache` object for the server. The function
            needs to return an object constructed from the function arguments
            to uniquely identify the result. Results are cached verbatim.

        cache_method : string or None
            Optional cache method string. Gets passed to get_hnd() of the
            cache. Defaults to "string" which requires a JSON serializable
            cache_id.

        cache_section : string or None
            Optional cache section string. Gets passed to get_hnd() of the
            cache. Defaults to "www".

        fun : function(args); (The annotated function)
            A function returning a (JSON-able) object. The function takes one
            argument which is the dictionary containing the payload from the
            client side. If the result is None a 404 error is sent.

def json_worker(self, mask, cache_id=None, cache_method="string",
                    cache_section="www"):
        """A function annotation that adds a worker request. A worker request
           is a POST request that is computed asynchronously. That is, the
           actual task is performed in a different thread and the network
           request returns immediately. The client side uses polling to fetch
           the result and can also cancel the task. The worker javascript
           client side must be linked and used for accessing the request.

        Parameters
        ----------
        mask : string
            The URL that must be matched to perform this request.

        cache_id : function(args) or None
            Optional function for caching the result. If set the worker must be
            idempotent. Requires a `cache` object for the server. The function
            needs to return an object constructed from the function arguments
            to uniquely identify the result. Results are cached verbatim.

        cache_method : string or None
            Optional cache method string. Gets passed to get_hnd() of the
            cache. Defaults to "string" which requires a JSON serializable
            cache_id.

        cache_section : string or None
            Optional cache section string. Gets passed to get_hnd() of the
            cache. Defaults to "www".

        fun : function(args); (The annotated function)
            A function returning a (JSON-able) object. The function takes one
            argument which is the dictionary containing the payload from the
            client side. If the result is None a 404 error is sent.
        """
        use_cache = cache_id is not None

        def wrapper(fun):
            lock = threading.RLock()
            tasks = {}
            cargo = {}
            cargo_cleaner = [None]

            def is_done(cur_key):
                with lock:
                    if cur_key not in tasks:
                        return True
                    if "running" not in tasks[cur_key]:
                        return False
                    return not tasks[cur_key]["running"]

            def start_cargo_cleaner():

                def get_next_cargo():
                    with lock:
                        next_ttl = None
                        for value in cargo.values():
                            ttl, _ = value
                            if next_ttl is None or ttl < next_ttl:
                                next_ttl = ttl
                        return next_ttl

                def clean_for(timestamp):
                    with lock:
                        keys = []
                        for (key, value) in cargo.items():
                            ttl, _ = value
                            if ttl > timestamp:
                                continue
                            keys.append(key)
                        for k in keys:
                            cargo.pop(k)
                            msg("purged cargo that was never read ({0})", k)

                def remove_cleaner():
                    with lock:
                        if get_next_cargo() is not None:
                            return False
                        cargo_cleaner[0] = None
                        return True

                def clean():
                    while True:
                        next_ttl = get_next_cargo()
                        if next_ttl is None:
                            if remove_cleaner():
                                break
                            else:
                                continue
                        time_until = next_ttl - time.time()
                        if time_until > 0:
                            time.sleep(time_until)
                        clean_for(time.time())

                with lock:
                    if cargo_cleaner[0] is not None:
                        return
                    cleaner = self._thread_factory(
                        target=clean,
                        name="{0}-Cargo-Cleaner".format(self.__class__))
                    cleaner.daemon = True
                    cargo_cleaner[0] = cleaner
                    cleaner.start()

            def add_cargo(content):
                with lock:
                    mcs = self.max_chunk_size
                    if mcs < 1:
                        raise ValueError("invalid chunk size: {0}".format(mcs))
                    ttl = time.time() + 10 * 60  # 10 minutes
                    chunks = []
                    while len(content) > 0:
                        chunk = content[:mcs]
                        content = content[mcs:]
                        cur_key = get_key()
                        cargo[cur_key] = (ttl, chunk)
                        chunks.append(cur_key)
                    start_cargo_cleaner()
                    return chunks

            def remove_cargo(cur_key):
                with lock:
                    _, result = cargo.pop(cur_key)
                    return result

            def remove_worker(cur_key):
                with lock:
                    task = tasks.pop(cur_key, None)
                    if task is None:
                        err_msg = "Task {0} not found!".format(cur_key)
                        return None, (ValueError(err_msg), None)
                    if task["running"]:
                        th = task["thread"]
                        if th.is_alive():
                            # kill the thread
                            tid = None
                            for tk, tobj in threading._active.items():
                                if tobj is th:
                                    tid = tk
                                    break
                            if tid is not None:
                                papi = ctypes.pythonapi
                                pts_sae = papi.PyThreadState_SetAsyncExc
                                res = pts_sae(ctypes.c_long(tid),
                                              ctypes.py_object(WorkerDeath))
                                if res == 0:
                                    # invalid thread id -- the thread might
                                    # be done already
                                    msg("invalid thread id for " +
                                        "killing worker {0}", cur_key)
                                elif res != 1:
                                    # roll back
                                    pts_sae(ctypes.c_long(tid), None)
                                    msg("killed too many ({0}) workers? {1}",
                                        res, cur_key)
                                else:
                                    if self.verbose_workers:
                                        msg("killed worker {0}", cur_key)
                        err_msg = "Task {0} is still running!".format(cur_key)
                        return None, (ValueError(err_msg), None)
                    return task["result"], task["exception"]

            def start_worker(args, cur_key, get_thread):
                try:
                    with lock:
                        task = {
                            "running": True,
                            "result": None,
                            "exception": None,
                            "thread": get_thread(),
                        }
                        tasks[cur_key] = task
                    if use_cache:
                        cache_obj = cache_id(args)
                        if cache_obj is not None and self.cache is not None:
                            with self.cache.get_hnd(
                                    cache_obj,
                                    section=cache_section,
                                    method=cache_method) as hnd:
                                if hnd.has():
                                    result = hnd.read()
                                else:
                                    result = hnd.write(json_dumps(fun(args)))
                        else:
                            result = json_dumps(fun(args))
                    else:
                        result = json_dumps(fun(args))
                    with lock:
                        task["running"] = False
                        task["result"] = result
                except (KeyboardInterrupt, SystemExit):
                    raise
                except Exception as e:
                    with lock:
                        task["running"] = False
                        task["exception"] = (e, traceback.format_exc())
                    return
                # make sure the result does not get stored forever
                try:
                    # remove 2 minutes after not reading the result
                    time.sleep(120)
                finally:
                    _result, err = remove_worker(cur_key)
                    if err is not None:
                        e, tb = err
                        if tb is not None:
                            msg("Error in purged worker for {0}: {1}\n{2}",
                                cur_key, e, tb)
                        return
                    msg("purged result that was never read ({0})", cur_key)

            def get_key():
                with lock:
                    crc32 = zlib.crc32(repr(get_time()).encode('utf8'))
                    cur_key = int(crc32 & 0xFFFFFFFF)
                    while cur_key in tasks or cur_key in cargo:
                        key = int(cur_key + 1)
                        if key == cur_key:
                            key = 0
                        cur_key = key
                    return cur_key

            def reserve_worker():
                with lock:
                    cur_key = get_key()
                    tasks[cur_key] = {}  # put marker
                    return cur_key

            def run_worker(req, args):
                post = args["post"]
                try:
                    action = post["action"]
                    cur_key = None
                    if action == "stop":
                        cur_key = post["token"]
                        remove_worker(cur_key)  # throw away the result
                        return {
                            "token": cur_key,
                            "done": True,
                            "result": None,
                            "continue": False,
                        }
                    if action == "start":
                        cur_key = reserve_worker()
                        inner_post = post.get("payload", {})
                        th = []
                        wname = "{0}-Worker-{1}".format(self.__class__,
                                                        cur_key)
                        worker = self._thread_factory(
                            target=start_worker,
                            name=wname,
                            args=(inner_post, cur_key, lambda: th[0]))
                        th.append(worker)
                        worker.start()
                        # give fast tasks a way to immediately return results
                        time.sleep(0.1)
                    if action == "cargo":
                        cur_key = post["token"]
                        result = remove_cargo(cur_key)
                        return {
                            "token": cur_key,
                            "result": result,
                        }
                    if action == "get":
                        cur_key = post["token"]
                    if cur_key is None:
                        raise ValueError("invalid action: {0}".format(action))
                    if is_done(cur_key):
                        result, exception = remove_worker(cur_key)
                        if exception is not None:
                            e, tb = exception
                            if tb is None:
                                # token does not exist anymore
                                return {
                                    "token": cur_key,
                                    "done": False,
                                    "result": None,
                                    "continue": False,
                                }
                            if isinstance(e, PreventDefaultResponse):
                                raise e
                            msg("Error in worker for {0}: {1}\n{2}",
                                cur_key, e, tb)
                            raise PreventDefaultResponse(500, "worker error")
                        if len(result) > self.max_chunk_size:
                            cargo_keys = add_cargo(result)
                            return {
                                "token": cur_key,
                                "done": True,
                                "result": cargo_keys,
                                "continue": True,
                            }
                        return {
                            "token": cur_key,
                            "done": True,
                            "result": result,
                            "continue": False,
                        }
                    return {
                        "token": cur_key,
                        "done": False,
                        "result": None,
                        "continue": True,
                    }
                except:  # nopep8
                    msg("Error processing worker command: {0}", post)
                    raise

            self.add_json_post_mask(mask, run_worker)
            self.set_file_argc(mask, 0)
            return fun
        return wrapper