celery/celery/concurrency/solo.py

# -*- coding: utf-8 -*-
from __future__ import absolute_import

import os

from .base import BasePool, apply_target


class TaskPool(BasePool):
    """Solo task pool (blocking, inline)."""

    def __init__(self, *args, **kwargs):
        super(TaskPool, self).__init__(*args, **kwargs)
        self.on_apply = apply_target

    def _get_info(self):
        return {"max-concurrency": 1,
                "processes": [os.getpid()],
                "max-tasks-per-child": None,
                "put-guarded-by-semaphore": True,
                "timeouts": ()}