# 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
    async def _consume_in_subprocess(self, job, kwargs):
        params = dict(self.json_params())
        loop = job._loop

        transport, protocol = await loop.subprocess_exec(
            lambda: StreamProtocol(job),
            sys.executable,
            PROCESS_FILE,
            json.dumps(sys.path),
            json.dumps(params),
            json.dumps(job.task.tojson())
        )
        process = Process(transport, protocol, loop)
        await process.communicate()
        if job.task.stacktrace:
            raise RemoteStackTrace
        return job.task.result

    def _should_retry(self, job):
        return (job.task.status != states.SUCCESS and
                job.task.queue and
                job.max_retries and
                job.task.retry < job.max_retries)

    def _requeue_task(self, job):
        task = job.task
        meta_params = task.meta.copy()
        meta_params['retry'] = task.retry + 1
        return job.queue(
            job.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

    async def _consume_in_subprocess(self, job, kwargs):
        params = dict(self.json_params())
        loop = job._loop

        transport, protocol = await loop.subprocess_exec(
            lambda: StreamProtocol(job),
            sys.executable,
            PROCESS_FILE,
            json.dumps(sys.path),
            json.dumps(params),
            json.dumps(job.task.tojson())
        )
        process = Process(transport, protocol, loop)
        await process.communicate()
        if job.task.stacktrace:
            raise RemoteStackTrace
        return job.task.result

    def _should_retry(self, job):
        return (job.task.status != states.SUCCESS and
                job.task.queue and
                job.max_retries and
                job.task.retry < job.max_retries)

    def _requeue_task(self, job):
        task = job.task
        meta_params = task.meta.copy()
        meta_params['retry'] = task.retry + 1
        return job.queue(
            job.name,

            callback=False,