class_docstr: An extension for the scheduler to manage queues
  
  This adds the following routes to the scheduler
  
  *  queue_create
  *  queue_release
  *  queue_put
  *  queue_get
  *  queue_size
class_docstr_tok: ["An","extension","for","the","scheduler","to","manage","queues","This","adds","the","following","routes","to","the","scheduler","*","queue_create","*","queue_release","*","queue_put","*","queue_get","*","queue_size"]
code: def __init__(self, scheduler):
      self.scheduler = scheduler
      self.queues = dict()
      self.client_refcount = dict()
      self.future_refcount = defaultdict(int)
  
      self.scheduler.handlers.update(
          {
              "queue_create": self.create,
              "queue_put": self.put,
              "queue_get": self.get,
              "queue_qsize": self.qsize,
          }
      )
  
      self.scheduler.stream_handlers.update(
          {
              "queue-future-release": self.future_release,
              "queue_release": self.release,
          }
      )
  
imports: from __future__ import annotations
  import asyncio
  import logging
  import uuid
  from collections import defaultdict
  from dask.utils import parse_timedelta
  from distributed.client import Future
  from distributed.utils import wait_for
  from distributed.worker import get_client
label_desc: Use image node_id 1 to create a new QueueExtension object  with example: obj = QueueExtension(scheduler)