# 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
                    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)

                if job.max_concurrency and concurrent > job.max_concurrency:
                    raise TooManyTasksForJob('max concurrency %d reached',
                                             job.max_concurrency)

                kwargs = task.kwargs or {}
                task.status = states.STARTED
                task.time_started = time_ended
                if worker:
                    task.worker = worker.aid
                logger.info(task.lazy_info())
                await self.backend.publish('started', task)
                future = self._consume(job, kwargs)
                #
                # record future for cancellation
                if worker:
                    self._concurrent_tasks[task_id].future = future
                #
                # This may block until timeout
                task.result = await wait_for(future, timeout)
            else:

# 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

                    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)

                if job.max_concurrency and concurrent > job.max_concurrency:
                    raise TooManyTasksForJob('max concurrency %d reached',
                                             job.max_concurrency)

                kwargs = task.kwargs or {}
                task.status = states.STARTED
                task.time_started = time_ended
                if worker:
                    task.worker = worker.aid
                logger.info(task.lazy_info())
                await self.backend.publish('started', task)
                future = self._consume(job, kwargs)
                #
                # record future for cancellation
                if worker:
                    self._concurrent_tasks[task_id].future = future
                #
                # This may block until timeout
                task.result = await wait_for(future, timeout)
            else:

                raise TaskError('Invalid status %s' % task.status_string)