class_docstr: Python TaskScheduler object.
  
  This is the simplest object that supports msg_id based
  DAG dependencies. *Only* task msg_ids are checked, not
  msg_ids of jobs submitted via the MUX queue.
class_docstr_tok: ["Python","TaskScheduler","object",".","This","is","the","simplest","object","that","supports","msg_id","based","DAG","dependencies",".","*","Only","*","task","msg_ids","are","checked",",","not","msg_ids","of","jobs","submitted","via","the","MUX","queue","."]
docstr: Dispatch job submission to appropriate handlers.
docstr_tok: ["Dispatch","job","submission","to","appropriate","handlers","."]
code: def dispatch_submission(self, raw_msg):
      """Dispatch job submission to appropriate handlers."""
      # ensure targets up to date:
      self.notifier_stream.flush()
      try:
          idents, msg = self.session.feed_identities(
              raw_msg, copy=False
          )
          msg = self.session.deserialize(msg, content=False, copy=False)
      except Exception:
          self.log.error(
              "task::Invaid task msg: %r" % raw_msg, exc_info=True
          )
          return
  
      # send to monitor
      self.mon_stream.send_multipart([b"intask"] + raw_msg, copy=False)
  
      header = msg["header"]
      md = msg["metadata"]
      msg_id = header["msg_id"]
      self.all_ids.add(msg_id)
  
      # get targets as a set of bytes objects
      # from a list of unicode objects
      targets = md.get("targets", [])
      targets = {t.encode("utf8", "replace") for t in targets}
  
      retries = md.get("retries", 0)
      self.retries[msg_id] = retries
  
      # time dependencies
      after = md.get("after", None)
      if after:
          after = Dependency(after)
          if after.all:
              if after.success:
                  after = Dependency(
                      after.difference(self.all_completed),
                      success=after.success,
                      failure=after.failure,
                      all=after.all,
                  )
              if after.failure:
                  after = Dependency(
                      after.difference(self.all_failed),
                      success=after.success,
                      failure=after.failure,
                      all=after.all,
                  )
          if after.check(self.all_completed, self.all_failed):
              # recast as empty set, if `after` already met,
              # to prevent unnecessary set comparisons
              after = MET
      else:
          after = MET
  
      # location dependencies
      follow = Dependency(md.get("follow", []))
  
      timeout = md.get("timeout", None)
      if timeout:
          timeout = float(timeout)
  
      job = Job(
          msg_id=msg_id,
          raw_msg=raw_msg,
          idents=idents,
          msg=msg,
          header=header,
          targets=targets,
          after=after,
          follow=follow,
          timeout=timeout,
          metadata=md,
      )
      # validate and reduce dependencies:
      for dep in after, follow:
          if not dep:  # empty dependency
              continue
          # check valid:
          if msg_id in dep or dep.difference(self.all_ids):
              self.queue_map[msg_id] = job
              return self.fail_unreachable(
                  msg_id, error.InvalidDependency
              )
          # check if unreachable:
          if dep.unreachable(self.all_completed, self.all_failed):
              self.queue_map[msg_id] = job
              return self.fail_unreachable(msg_id)
  
      if after.check(self.all_completed, self.all_failed):
          # time deps already met, try to run
          if not self.maybe_run(job):
              # can't run yet
              if msg_id not in self.all_failed:
                  # could have failed as unreachable
                  self.save_unmet(job)
      else:
          self.save_unmet(job)
  
imports: import time
  from collections import deque
  from random import randint, random
  from types import FunctionType
  import zmq
  from traitlets import Dict, Enum, Instance, Integer, List, observe
  from ipyparallel import Dependency, error, util
  from ipyparallel.controller.scheduler import Scheduler
label_desc: Use image node_id 10 for calling the TaskScheduler obj's underlying member method code with example usage: obj.dispatch_submission(raw_msg) and returns: self, self