This document describes the current stable version of Celery (4.0). For development docs, go here.
Source code for celery.concurrency.solo
# -*- coding: utf-8 -*-
"""Single-threaded execution pool."""
from __future__ import absolute_import, unicode_literals
import os
from .base import BasePool, apply_target
__all__ = ['TaskPool']
[docs]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': (),
}