# Path: pq/tasks/task.py
# class TaskError(PulsarException):
#     status = states.FAILURE
# 
# class TaskTimeout(TaskError):
#     pass
# 
# Path: pq/cpubound.py
# class StreamProtocol(subprocess.SubprocessStreamProtocol):
# 
#     def __init__(self, job):
#         super().__init__(streams._DEFAULT_LIMIT, job._loop)
#         self.info = CpuTaskInfo(job)
#         self.error = CpuTaskInfo(job)
# 
#     def pipe_data_received(self, fd, data):
#         if fd == 1:
#             self.info.feed(data)
#         elif fd == 2:
#             self.error.feed(data)
#         super().pipe_data_received(fd, data)
# 
# PROCESS_FILE = os.path.join(PQPATH, "cpubound.py")
# 
# Path: pq/utils/exc.py
# def string_exception(exc):
#     args = []
#     for arg in exc.args:
#         if isinstance(arg, str):
#             args.append(arg)
#     if args:
#         return args[0] if len(args) == 1 else args
#     return str(exc)
# 
# Path: pq/utils/concurrency.py
# ASYNC_IO = 1        # run in the worker event loop
# THREAD_IO = 3       # run in the event loop executor
# CPUBOUND = 4        # run in a subprocess


import sys
import time
import traceback
import json
from asyncio import wait_for
from asyncio import CancelledError, TimeoutError
from asyncio.subprocess import Process
from .task import TaskError, TaskTimeout
from . import states
from ..cpubound import StreamProtocol, PROCESS_FILE
from ..utils.exc import string_exception
from ..utils import concurrency
        JobClass = self.registry.get(task.name)

        try:
            if not JobClass:
                raise RuntimeError('%s not in registry' % task.name)

            if task.status > states.STARTED:
                queued = task.time_queued
                timeout = task.timeout
                delay = task.delay or 0
                start = queued + delay

                if delay:  # Task with delay
                    gap = start - time_ended
                    if gap > 0:
                        self._loop.call_later(gap, self._queue_again, task)
                        if worker:
                            self._concurrent_tasks.pop(task_id, None)
                        return task

                if timeout:  # Handle timeout
                    timeout = timeout + start - time_ended
                    if timeout <= 0:
                        raise TaskTimeout

                task.node = self.backend.node_name
                if worker:
                    task.consumer = worker.aid
                    concurrent = await self.broker.incr(JobClass.name)


# Path: pq/tasks/task.py
# class TaskError(PulsarException):
#     status = states.FAILURE
# 
# class TaskTimeout(TaskError):
#     pass
# 
# Path: pq/cpubound.py
# class StreamProtocol(subprocess.SubprocessStreamProtocol):
# 
#     def __init__(self, job):
#         super().__init__(streams._DEFAULT_LIMIT, job._loop)
#         self.info = CpuTaskInfo(job)
#         self.error = CpuTaskInfo(job)
# 
#     def pipe_data_received(self, fd, data):
#         if fd == 1:
#             self.info.feed(data)
#         elif fd == 2:
#             self.error.feed(data)
#         super().pipe_data_received(fd, data)
# 
# PROCESS_FILE = os.path.join(PQPATH, "cpubound.py")
# 
# Path: pq/utils/exc.py
# def string_exception(exc):
#     args = []
#     for arg in exc.args:
#         if isinstance(arg, str):
#             args.append(arg)
#     if args:
#         return args[0] if len(args) == 1 else args
#     return str(exc)
# 
# Path: pq/utils/concurrency.py
# ASYNC_IO = 1        # run in the worker event loop
# THREAD_IO = 3       # run in the event loop executor
# CPUBOUND = 4        # run in a subprocess


# Path: pq/tasks/executor.py
import sys
import time
import traceback
import json
from asyncio import wait_for
from asyncio import CancelledError, TimeoutError
from asyncio.subprocess import Process
from .task import TaskError, TaskTimeout
from . import states
from ..cpubound import StreamProtocol, PROCESS_FILE
from ..utils.exc import string_exception
from ..utils import concurrency

        JobClass = self.registry.get(task.name)

        try:
            if not JobClass:
                raise RuntimeError('%s not in registry' % task.name)

            if task.status > states.STARTED:
                queued = task.time_queued
                timeout = task.timeout
                delay = task.delay or 0
                start = queued + delay

                if delay:  # Task with delay
                    gap = start - time_ended
                    if gap > 0:
                        self._loop.call_later(gap, self._queue_again, task)
                        if worker:
                            self._concurrent_tasks.pop(task_id, None)
                        return task

                if timeout:  # Handle timeout
                    timeout = timeout + start - time_ended
                    if timeout <= 0:
                        raise TaskTimeout

                task.node = self.backend.node_name
                if worker:
                    task.consumer = worker.aid
                    concurrent = await self.broker.incr(JobClass.name)


                job = JobClass(self, task)