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)
def dispatch_submission ( self , raw_msg ) : # 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.
Dispatch job submission to appropriate handlers .