This document describes the current stable version of Celery (5.6). For development docs, go here.
Source code for billiard.compat
import numbers
import os
import subprocess
import sys
from itertools import zip_longest
if sys.platform == 'win32':
try:
import _winapi # noqa
except ImportError: # pragma: no cover
from _multiprocessing import win32 as _winapi # noqa
else:
_winapi = None # noqa
try:
import resource
except ImportError: # pragma: no cover
resource = None
from io import UnsupportedOperation
FILENO_ERRORS = (AttributeError, ValueError, UnsupportedOperation)
if hasattr(os, 'write'):
__write__ = os.write
def send_offset(fd, buf, offset):
return __write__(fd, buf[offset:])
else: # non-posix platform
def send_offset(fd, buf, offset): # noqa
raise NotImplementedError('send_offset')
try:
fsencode = os.fsencode
fsdecode = os.fsdecode
except AttributeError:
def _fscodec():
encoding = sys.getfilesystemencoding()
if encoding == 'mbcs':
errors = 'strict'
else:
errors = 'surrogateescape'
def fsencode(filename):
"""
Encode filename to the filesystem encoding with 'surrogateescape'
error handler, return bytes unchanged. On Windows, use 'strict'
error handler if the file system encoding is 'mbcs' (which is the
default encoding).
"""
if isinstance(filename, bytes):
return filename
elif isinstance(filename, str):
return filename.encode(encoding, errors)
else:
raise TypeError("expect bytes or str, not %s"
% type(filename).__name__)
def fsdecode(filename):
"""
Decode filename from the filesystem encoding with 'surrogateescape'
error handler, return str unchanged. On Windows, use 'strict' error
handler if the file system encoding is 'mbcs' (which is the default
encoding).
"""
if isinstance(filename, str):
return filename
elif isinstance(filename, bytes):
return filename.decode(encoding, errors)
else:
raise TypeError("expect bytes or str, not %s"
% type(filename).__name__)
return fsencode, fsdecode
fsencode, fsdecode = _fscodec()
del _fscodec
def maybe_fileno(f):
"""Get object fileno, or :const:`None` if not defined."""
if isinstance(f, numbers.Integral):
return f
try:
return f.fileno()
except FILENO_ERRORS:
pass
[docs]
def get_fdmax(default=None):
"""Return the maximum number of open file descriptors
on this system.
:keyword default: Value returned if there's no file
descriptor limit.
"""
try:
return os.sysconf('SC_OPEN_MAX')
except:
pass
if resource is None: # Windows
return default
fdmax = resource.getrlimit(resource.RLIMIT_NOFILE)[1]
if fdmax == resource.RLIM_INFINITY:
return default
return fdmax
def uniq(it):
"""Return all unique elements in ``it``, preserving order."""
seen = set()
return (seen.add(obj) or obj for obj in it if obj not in seen)
# Directory listing the descriptors open in the calling process. Same
# choice as CPython's FD_DIR (Modules/_posixsubprocess.c, Cygwin added in
# gh-148575): /dev/fd on macOS, Cygwin, FreeBSD and DragonFly, /proc/self/fd
# elsewhere. On FreeBSD and DragonFly it is only trusted when fdescfs is
# mounted, see _open_fds().
_BSD_WITH_FDESCFS = ('freebsd', 'dragonfly')
if sys.platform.startswith(('cygwin', 'darwin') + _BSD_WITH_FDESCFS):
_FD_DIR = '/dev/fd'
else:
_FD_DIR = '/proc/self/fd'
def _dev_fd_is_fdescfs():
# devfs alone creates only /dev/fd/0-2, while fdescfs creates entries for
# every descriptor the process has open. Same check as CPython's
# _is_fdescfs_mounted_on_dev_fd().
try:
return os.stat('/dev').st_dev != os.stat(_FD_DIR).st_dev
except OSError:
return False
def _open_fds():
"""Return the descriptors open in this process, or None.
None means this platform has no directory listing them, and the caller
has to fall back to a numeric range.
"""
if sys.platform.startswith(_BSD_WITH_FDESCFS) and not _dev_fd_is_fdescfs():
return None
try:
names = os.listdir(_FD_DIR)
except OSError:
return None
return sorted(int(name) for name in names if name.isdigit())
def close_open_fds(keep=None):
# must make sure this is 0-inclusive (Issue #celery/1882)
keep = list(uniq(sorted(
f for f in map(maybe_fileno, keep or []) if f is not None
)))
fds = _open_fds()
if fds is not None:
# Only the descriptors actually open are touched, so the cost does
# not depend on RLIMIT_NOFILE (celery/celery#9886).
keep = set(keep)
for fd in fds:
if fd in keep:
continue
try:
os.close(fd)
except OSError:
# Same as os.closerange() below: closing is best effort.
pass
return
maxfd = get_fdmax(default=2048)
kL, kH = iter([-1] + keep), iter(keep + [maxfd])
for low, high in zip_longest(kL, kH):
if low + 1 != high:
os.closerange(low + 1, high)
def get_errno(exc):
""":exc:`socket.error` and :exc:`IOError` first got
the ``.errno`` attribute in Py2.7"""
try:
return exc.errno
except AttributeError:
return 0
try:
import _posixsubprocess
except ImportError:
def spawnv_passfds(path, args, passfds):
if sys.platform != 'win32':
# when not using _posixsubprocess (on earlier python) and not on
# windows, we want to keep stdout/stderr open...
passfds = passfds + [
maybe_fileno(sys.stdout),
maybe_fileno(sys.stderr),
]
pid = os.fork()
if not pid:
close_open_fds(keep=sorted(f for f in passfds if f))
os.execv(fsencode(path), args)
return pid
else:
def spawnv_passfds(path, args, passfds):
passfds = sorted(passfds)
errpipe_read, errpipe_write = os.pipe()
try:
args = [
args, [fsencode(path)], True, tuple(passfds), None, None,
-1, -1, -1, -1, -1, -1, errpipe_read, errpipe_write,
False, False]
if sys.version_info >= (3, 11):
args.append(-1) # process_group
if sys.version_info >= (3, 9):
args.extend((None, None, None, -1)) # group, extra_groups, user, umask
args.append(None) # preexec_fn
if (3, 11) <= sys.version_info < (3, 14):
args.append(subprocess._USE_VFORK)
return _posixsubprocess.fork_exec(*args)
finally:
os.close(errpipe_read)
os.close(errpipe_write)
if sys.platform == 'win32':
def setblocking(handle, blocking):
raise NotImplementedError('setblocking not implemented on win32')
def isblocking(handle):
raise NotImplementedError('isblocking not implemented on win32')
else:
from os import O_NONBLOCK
from fcntl import fcntl, F_GETFL, F_SETFL
def isblocking(handle): # noqa
return not (fcntl(handle, F_GETFL) & O_NONBLOCK)
def setblocking(handle, blocking): # noqa
flags = fcntl(handle, F_GETFL, 0)
fcntl(
handle, F_SETFL,
flags & (~O_NONBLOCK) if blocking else flags | O_NONBLOCK,
)
E_PSUTIL_MISSING = """
On Windows, the ability to inspect memory usage requires the psutil library.
You can install it using pip:
$ pip install psutil
"""
E_RESOURCE_MISSING = """
Your platform ({0}) does not seem to have the `resource.getrusage' function.
Please open an issue so that we can add support for this platform.
"""
if sys.platform == 'win32':
try:
import psutil
except ImportError: # pragma: no cover
psutil = None # noqa
def mem_rss():
# type () -> int
if psutil is None:
raise ImportError(E_PSUTIL_MISSING.strip())
return int(psutil.Process(os.getpid()).memory_info()[0] / 1024.0)
else:
try:
from resource import getrusage, RUSAGE_SELF
except ImportError: # pragma: no cover
getrusage = RUSAGE_SELF = None # noqa
if 'bsd' in sys.platform or sys.platform == 'darwin':
# On BSD platforms :man:`getrusage(2)` ru_maxrss field is in bytes.
def maxrss_to_kb(v):
# type: (SupportsInt) -> int
return int(v) / 1024.0
else:
# On Linux it's kilobytes.
def maxrss_to_kb(v):
# type: (SupportsInt) -> int
return int(v)
def mem_rss():
# type () -> int
if resource is None:
raise ImportError(E_RESOURCE_MISSING.strip().format(sys.platform))
return maxrss_to_kb(getrusage(RUSAGE_SELF).ru_maxrss)
