docs: 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