concurrent.py 文件源码

def _execute(self, context, sub_args):
        executable = self._get_executable(context)
        results = []
        for args in sub_args:
            if self._error_break:  # ??????????
                return
            try:
                result = None
                if args is None:
                    result = executable(context)
                elif isinstance(args, SequenceCollectionType):
                    result = executable(context, *args)
                elif isinstance(args, types.DictType):
                    result = executable(context, **args)
            except Exception as e:
                context.logger.exception(
                    u"????'{}'???????????????'{}'".format(
                        self.name, self._error_action))
                if self._error_action == "stop":
                    # rethrow????????
                    self._error_break = True  # ???????
                    raise e
                elif self._error_action == "continue":
                    # ???????error_handler
                    if self._error_handler is not None:
                        exc_type, exc_value, tb = sys.exc_info()
                        self._error_handler(context, exc_type, exc_value, tb)
                    # ??????????None???????
                    results.append(self._error_default_value)
            else:
                results.append(result)

        if self._sub_join is not None:
            return self._sub_join(context, results)
        return results