| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308 |
- from __future__ import absolute_import
- import errno
- import numbers
- import os
- import sys
- from .five import range, 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
- try:
- from io import UnsupportedOperation
- FILENO_ERRORS = (AttributeError, ValueError, UnsupportedOperation)
- except ImportError: # pragma: no cover
- # Py2
- FILENO_ERRORS = (AttributeError, ValueError) # noqa
- if sys.version_info > (2, 7, 5):
- buf_t, is_new_buffer = memoryview, True # noqa
- else:
- buf_t, is_new_buffer = buffer, False # noqa
- if hasattr(os, 'write'):
- __write__ = os.write
- if is_new_buffer:
- def send_offset(fd, buf, offset):
- return __write__(fd, buf[offset:])
- else: # Py<2.7.6
- def send_offset(fd, buf, offset): # noqa
- return __write__(fd, buf_t(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
- if sys.version_info[0] == 3:
- bytes = bytes
- else:
- _bytes = bytes
- # the 'bytes' alias in Python2 does not support an encoding argument.
- class bytes(_bytes): # noqa
- def __new__(cls, *args):
- if len(args) > 1:
- return _bytes(args[0]).encode(*args[1:])
- return _bytes(*args)
- 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
- 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)
- try:
- closerange = os.closerange
- except AttributeError:
- def closerange(fd_low, fd_high): # noqa
- for fd in reversed(range(fd_low, fd_high)):
- try:
- os.close(fd)
- except OSError as exc:
- if exc.errno != errno.EBADF:
- raise
- 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
- )))
- 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:
- closerange(low + 1, high)
- else:
- def close_open_fds(keep=None): # noqa
- keep = [maybe_fileno(f)
- for f in (keep or []) if maybe_fileno(f) is not None]
- for fd in reversed(range(get_fdmax(default=2048))):
- if fd not in keep:
- try:
- os.close(fd)
- except OSError as exc:
- if exc.errno != errno.EBADF:
- raise
- def get_errno(exc):
- """:exc:`socket.error` and :exc:`IOError` first got
- the ``.errno`` attribute in Py2.7"""
- try:
- return exc.errno
- except AttributeError:
- try:
- # e.args = (errno, reason)
- if isinstance(exc.args, tuple) and len(exc.args) == 2:
- return exc.args[0]
- except AttributeError:
- pass
- 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:
- return _posixsubprocess.fork_exec(
- args, [fsencode(path)], True, tuple(passfds), None, None,
- -1, -1, -1, -1, -1, -1, errpipe_read, errpipe_write,
- False, False, None)
- 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)
|