# 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
            task.result = string_exception(exc)
            task.status = states.FAILURE
            task.stacktrace = traceback.format_tb(exc_info[2])
            task.exception = traceback.format_exception_only(
                exc_info[0], exc_info[1])[0]
            logger.exception(task.lazy_info())
        else:
            task.status = states.SUCCESS
            logger.info(task.lazy_info())
        #
        task.time_ended = time.time()
        if worker:
            self._concurrent_tasks.pop(task_id, None)

        await self.backend.publish('done', task)

        if job:
            if worker:
                await self.broker.decr(job.name)
            if self._should_retry(job):
                await self._requeue_task(job)

        return task

    def _consume(self, job, kwargs):
        model = job.get_concurrency()

        if model == concurrency.THREAD_IO:
            return job._loop.run_in_executor(None, lambda: job(**kwargs))


# 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

            task.result = string_exception(exc)
            task.status = states.FAILURE
            task.stacktrace = traceback.format_tb(exc_info[2])
            task.exception = traceback.format_exception_only(
                exc_info[0], exc_info[1])[0]
            logger.exception(task.lazy_info())
        else:
            task.status = states.SUCCESS
            logger.info(task.lazy_info())
        #
        task.time_ended = time.time()
        if worker:
            self._concurrent_tasks.pop(task_id, None)

        await self.backend.publish('done', task)

        if job:
            if worker:
                await self.broker.decr(job.name)
            if self._should_retry(job):
                await self._requeue_task(job)

        return task

    def _consume(self, job, kwargs):
        model = job.get_concurrency()

        if model == concurrency.THREAD_IO:
            return job._loop.run_in_executor(None, lambda: job(**kwargs))


        elif model == concurrency.CPUBOUND: