environment/lib/python2.7/site-packages/IPython/parallel/controller/scheduler.py
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.unserialize(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']
        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 = header.get('targets', [])
        targets = map(cast_bytes, targets)
        targets = set(targets)

        retries = header.get('retries', 0)
        self.retries[msg_id] = retries

        # time dependencies
        after = header.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(header.get('follow', []))

        # turn timeouts into datetime objects:
        timeout = header.get('timeout', None)
        if timeout:
            # cast to float, because jsonlib returns floats as decimal.Decimal,
            # which timedelta does not accept
            timeout = datetime.now() + timedelta(0,float(timeout),0)

        job = Job(msg_id=msg_id, raw_msg=raw_msg, idents=idents, msg=msg,
                 header=header, targets=targets, after=after, follow=follow,
                 timeout=timeout,
        )

        # 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.depending[msg_id] = job
                return self.fail_unreachable(msg_id, error.InvalidDependency)
            # check if unreachable:
            if dep.unreachable(self.all_completed, self.all_failed):
                self.depending[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)
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.unserialize(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']
        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 = header.get('targets', [])
        targets = map(cast_bytes, targets)
        targets = set(targets)

        retries = header.get('retries', 0)
        self.retries[msg_id] = retries

        # time dependencies
        after = header.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(header.get('follow', []))

        # turn timeouts into datetime objects:
        timeout = header.get('timeout', None)
        if timeout:
            # cast to float, because jsonlib returns floats as decimal.Decimal,
            # which timedelta does not accept
            timeout = datetime.now() + timedelta(0,float(timeout),0)

        job = Job(msg_id=msg_id, raw_msg=raw_msg, idents=idents, msg=msg,
                 header=header, targets=targets, after=after, follow=follow,
                 timeout=timeout,
        )

        # 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.depending[msg_id] = job
                return self.fail_unreachable(msg_id, error.InvalidDependency)
            # check if unreachable:
            if dep.unreachable(self.all_completed, self.all_failed):
                self.depending[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)
Dispatch job submission to appropriate handlers.