celery/celery/concurrency/solo.py

# -*- coding: utf-8 -*-
"""Single-threaded execution pool."""
from __future__ import absolute_import, unicode_literals

import os

from .base import BasePool, apply_target

__all__ = ['TaskPool']


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

    body_can_be_buffer = True

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

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