def merge(input_bams, output_bam, threads=1):
''' Merge the sorted bam chunks hierarchically to conserve open file handles '''
soft, _ = resource.getrlimit(resource.RLIMIT_NOFILE)
soft -= 100
tmp_dir = os.path.dirname(output_bam)
while len(input_bams) > 1:
new_bams = []
for i in range(0, len(input_bams), soft):
bam_chunk = input_bams[i:i+soft]
if len(bam_chunk) > 1:
new_bam = os.path.join(tmp_dir, "%d-%d.bam" % (i, len(input_bams)))
tk_bam.merge(new_bam, bam_chunk, threads)
new_bams.append(new_bam)
else:
new_bams.append(input_bams[i])
input_bams = new_bams
cr_utils.move(input_bams[0], output_bam)
python类getrlimit()的实例源码
def check_open_fh():
_, hard = resource.getrlimit(resource.RLIMIT_NOFILE)
if 0 <= hard and hard < tk_constants.MIN_PROCESS_NOFILE:
return False, "On machine: %s, process open file handle hard limit (%d) is less than %d. Please run 'ulimit -n %d' before restarting the pipeline." % (
socket.gethostname(), hard, tk_constants.MIN_PROCESS_NOFILE, tk_constants.MIN_PROCESS_NOFILE)
if not os.path.exists(tk_constants.GLOBAL_NOFILE_PATH):
return False, "On machine: %s, %s does not exist." % (socket.gethostname(), tk_constants.GLOBAL_NOFILE_PATH)
with open(tk_constants.GLOBAL_NOFILE_PATH) as f:
glob_str = f.read().strip()
if not glob_str.isdigit():
return False, "On machine: %s, %s contains a non-integer global open file handle limit: %s." % (
socket.gethostname(), tk_constants.GLOBAL_NOFILE_PATH, glob_str)
glob = int(glob_str)
if glob < tk_constants.MIN_GLOBAL_NOFILE:
return False, "On machine: %s, global open file handle limit (%d) is less than %d. Please set the global file handle limit to %d before restarting the pipeline." % (
socket.gethostname(), glob, tk_constants.MIN_GLOBAL_NOFILE, tk_constants.MIN_GLOBAL_NOFILE)
return True, None
def get_num_processes(num_servers):
# Since each process is not resource heavy and simply takes time waiting for pings, maximise the number of processes (within constraints of the current configuration)
# Maximum open file descriptors of current configuration
soft_limit, _ = resource.getrlimit(resource.RLIMIT_NOFILE)
# Find how many file descriptors are already in use by the parent process
ppid = os.getppid()
used_file_descriptors = int(subprocess.run('ls -l /proc/' + str(ppid) + '/fd | wc -l', shell=True, stdout=subprocess.PIPE).stdout.decode('utf-8'))
# Max processes is the number of file descriptors left, before the sof limit (configuration maximum) is reached
max_processes = int((soft_limit - used_file_descriptors) / 2)
if num_servers > max_processes:
return max_processes
else:
return num_servers
def set_open_file_limit_up_to(limit=65536):
import settings
soft, hard = resource.getrlimit(resource.RLIMIT_NOFILE)
limit = max(soft, limit)
limit = min(limit, hard)
while limit > soft:
try:
resource.setrlimit(resource.RLIMIT_NOFILE, (limit, hard))
break
except ValueError:
limit -= 256
except:
settings.CONFIG_LOG.exception('unexpected exception')
soft, hard = resource.getrlimit(resource.RLIMIT_NOFILE)
settings.CONFIG_LOG.info('open file limit set to %d:%d', soft, hard)
return (soft, hard)
def prevent_core_dump():
""" Prevent this process from generating a core dump.
:return: ``None``.
Set the soft and hard limits for core dump size to zero. On Unix,
this entirely prevents the process from creating core dump.
"""
core_resource = resource.RLIMIT_CORE
try:
# Ensure the resource limit exists on this platform, by requesting
# its current value.
core_limit_prev = resource.getrlimit(core_resource)
except ValueError as exc:
error = DaemonOSEnvironmentError(
"System does not support RLIMIT_CORE resource limit"
" ({exc})".format(exc=exc))
raise error
# Set hard and soft limits to zero, i.e. no core dump at all.
core_limit = (0, 0)
resource.setrlimit(core_resource, core_limit)
def get_maximum_file_descriptors():
""" Get the maximum number of open file descriptors for this process.
:return: The number (integer) to use as the maximum number of open
files for this process.
The maximum is the process hard resource limit of maximum number of
open file descriptors. If the limit is “infinity”, a default value
of ``MAXFD`` is returned.
"""
limits = resource.getrlimit(resource.RLIMIT_NOFILE)
result = limits[1]
if result == resource.RLIM_INFINITY:
result = MAXFD
return result
def main(number=1024, **kwargs):
for name, desc in [
('RLIMIT_NOFILE', 'number of open files'),
]:
limit_num = getattr(resource, name)
soft, hard = resource.getrlimit(limit_num)
print('Maximum %-25s (%-15s) : %20s %20s' % (desc, name, soft, hard))
files = []
try:
for i in range(0, number):
files.append(_create_file(i))
finally:
for f in files:
f.close()
def main(number=128, **kwargs):
for name, desc in [
('RLIMIT_NPROC', 'number of processes'),
]:
limit_num = getattr(resource, name)
soft, hard = resource.getrlimit(limit_num)
print('Maximum %-25s (%-15s) : %20s %20s' % (desc, name, soft, hard))
processes = []
for i in range(0, number):
p = Process(
target=_sleep,
args=(i,)
)
p.start()
processes.append(p)
for p in processes:
p.join()
def __enter__(self):
"""Try to save previous ulimit, then set it to (0, 0)."""
if resource is not None:
try:
self.old_limit = resource.getrlimit(resource.RLIMIT_CORE)
resource.setrlimit(resource.RLIMIT_CORE, (0, 0))
except (ValueError, resource.error):
pass
if sys.platform == 'darwin':
# Check if the 'Crash Reporter' on OSX was configured
# in 'Developer' mode and warn that it will get triggered
# when it is.
#
# This assumes that this context manager is used in tests
# that might trigger the next manager.
value = subprocess.Popen(['/usr/bin/defaults', 'read',
'com.apple.CrashReporter', 'DialogType'],
stdout=subprocess.PIPE).communicate()[0]
if value.strip() == b'developer':
print("this tests triggers the Crash Reporter, "
"that is intentional", end='')
sys.stdout.flush()
def test_rlimit_get(self):
import resource
p = psutil.Process(os.getpid())
names = [x for x in dir(psutil) if x.startswith('RLIMIT')]
assert names, names
for name in names:
value = getattr(psutil, name)
self.assertGreaterEqual(value, 0)
if name in dir(resource):
self.assertEqual(value, getattr(resource, name))
# XXX - On PyPy RLIMIT_INFINITY returned by
# resource.getrlimit() is reported as a very big long
# number instead of -1. It looks like a bug with PyPy.
if PYPY:
continue
self.assertEqual(p.rlimit(value), resource.getrlimit(value))
else:
ret = p.rlimit(value)
self.assertEqual(len(ret), 2)
self.assertGreaterEqual(ret[0], -1)
self.assertGreaterEqual(ret[1], -1)
def test_urandom_failure(self):
# Check urandom() failing when it is not able to open /dev/random.
# We spawn a new process to make the test more robust (if getrlimit()
# failed to restore the file descriptor limit after this, the whole
# test suite would crash; this actually happened on the OS X Tiger
# buildbot).
code = """if 1:
import errno
import os
import resource
soft_limit, hard_limit = resource.getrlimit(resource.RLIMIT_NOFILE)
resource.setrlimit(resource.RLIMIT_NOFILE, (1, hard_limit))
try:
os.urandom(16)
except OSError as e:
assert e.errno == errno.EMFILE, e.errno
else:
raise AssertionError("OSError not raised")
"""
assert_python_ok('-c', code)
def __enter__(self):
"""Try to save previous ulimit, then set it to (0, 0)."""
if resource is not None:
try:
self.old_limit = resource.getrlimit(resource.RLIMIT_CORE)
resource.setrlimit(resource.RLIMIT_CORE, (0, 0))
except (ValueError, resource.error):
pass
if sys.platform == 'darwin':
# Check if the 'Crash Reporter' on OSX was configured
# in 'Developer' mode and warn that it will get triggered
# when it is.
#
# This assumes that this context manager is used in tests
# that might trigger the next manager.
value = subprocess.Popen(['/usr/bin/defaults', 'read',
'com.apple.CrashReporter', 'DialogType'],
stdout=subprocess.PIPE).communicate()[0]
if value.strip() == b'developer':
print "this tests triggers the Crash Reporter, that is intentional"
sys.stdout.flush()
def test_urandom_failure(self):
# Check urandom() failing when it is not able to open /dev/random.
# We spawn a new process to make the test more robust (if getrlimit()
# failed to restore the file descriptor limit after this, the whole
# test suite would crash; this actually happened on the OS X Tiger
# buildbot).
code = """if 1:
import errno
import os
import resource
soft_limit, hard_limit = resource.getrlimit(resource.RLIMIT_NOFILE)
resource.setrlimit(resource.RLIMIT_NOFILE, (1, hard_limit))
try:
os.urandom(16)
except OSError as e:
assert e.errno == errno.EMFILE, e.errno
else:
raise AssertionError("OSError not raised")
"""
assert_python_ok('-c', code)
def __enter__(self):
"""Try to save previous ulimit, then set it to (0, 0)."""
if resource is not None:
try:
self.old_limit = resource.getrlimit(resource.RLIMIT_CORE)
resource.setrlimit(resource.RLIMIT_CORE, (0, 0))
except (ValueError, resource.error):
pass
if sys.platform == 'darwin':
# Check if the 'Crash Reporter' on OSX was configured
# in 'Developer' mode and warn that it will get triggered
# when it is.
#
# This assumes that this context manager is used in tests
# that might trigger the next manager.
value = subprocess.Popen(['/usr/bin/defaults', 'read',
'com.apple.CrashReporter', 'DialogType'],
stdout=subprocess.PIPE).communicate()[0]
if value.strip() == b'developer':
print "this tests triggers the Crash Reporter, that is intentional"
sys.stdout.flush()
def test_urandom_failure(self):
# Check urandom() failing when it is not able to open /dev/random.
# We spawn a new process to make the test more robust (if getrlimit()
# failed to restore the file descriptor limit after this, the whole
# test suite would crash; this actually happened on the OS X Tiger
# buildbot).
code = """if 1:
import errno
import os
import resource
soft_limit, hard_limit = resource.getrlimit(resource.RLIMIT_NOFILE)
resource.setrlimit(resource.RLIMIT_NOFILE, (1, hard_limit))
try:
os.urandom(16)
except OSError as e:
assert e.errno == errno.EMFILE, e.errno
else:
raise AssertionError("OSError not raised")
"""
assert_python_ok('-c', code)
def __enter__(self):
"""Try to save previous ulimit, then set it to (0, 0)."""
if resource is not None:
try:
self.old_limit = resource.getrlimit(resource.RLIMIT_CORE)
resource.setrlimit(resource.RLIMIT_CORE, (0, 0))
except (ValueError, resource.error):
pass
if sys.platform == 'darwin':
# Check if the 'Crash Reporter' on OSX was configured
# in 'Developer' mode and warn that it will get triggered
# when it is.
#
# This assumes that this context manager is used in tests
# that might trigger the next manager.
value = subprocess.Popen(['/usr/bin/defaults', 'read',
'com.apple.CrashReporter', 'DialogType'],
stdout=subprocess.PIPE).communicate()[0]
if value.strip() == b'developer':
print("this tests triggers the Crash Reporter, "
"that is intentional", end='')
sys.stdout.flush()
def test_urandom_failure(self):
# Check urandom() failing when it is not able to open /dev/random.
# We spawn a new process to make the test more robust (if getrlimit()
# failed to restore the file descriptor limit after this, the whole
# test suite would crash; this actually happened on the OS X Tiger
# buildbot).
code = """if 1:
import errno
import os
import resource
soft_limit, hard_limit = resource.getrlimit(resource.RLIMIT_NOFILE)
resource.setrlimit(resource.RLIMIT_NOFILE, (1, hard_limit))
try:
os.urandom(16)
except OSError as e:
assert e.errno == errno.EMFILE, e.errno
else:
raise AssertionError("OSError not raised")
"""
assert_python_ok('-c', code)
def __enter__(self):
"""Try to save previous ulimit, then set it to (0, 0)."""
if resource is not None:
try:
self.old_limit = resource.getrlimit(resource.RLIMIT_CORE)
resource.setrlimit(resource.RLIMIT_CORE, (0, 0))
except (ValueError, resource.error):
pass
if sys.platform == 'darwin':
# Check if the 'Crash Reporter' on OSX was configured
# in 'Developer' mode and warn that it will get triggered
# when it is.
#
# This assumes that this context manager is used in tests
# that might trigger the next manager.
value = subprocess.Popen(['/usr/bin/defaults', 'read',
'com.apple.CrashReporter', 'DialogType'],
stdout=subprocess.PIPE).communicate()[0]
if value.strip() == b'developer':
print "this tests triggers the Crash Reporter, that is intentional"
sys.stdout.flush()
def test_urandom_failure(self):
# Check urandom() failing when it is not able to open /dev/random.
# We spawn a new process to make the test more robust (if getrlimit()
# failed to restore the file descriptor limit after this, the whole
# test suite would crash; this actually happened on the OS X Tiger
# buildbot).
code = """if 1:
import errno
import os
import resource
soft_limit, hard_limit = resource.getrlimit(resource.RLIMIT_NOFILE)
resource.setrlimit(resource.RLIMIT_NOFILE, (1, hard_limit))
try:
os.urandom(16)
except OSError as e:
assert e.errno == errno.EMFILE, e.errno
else:
raise AssertionError("OSError not raised")
"""
assert_python_ok('-c', code)
def __enter__(self):
"""Try to save previous ulimit, then set it to (0, 0)."""
if resource is not None:
try:
self.old_limit = resource.getrlimit(resource.RLIMIT_CORE)
resource.setrlimit(resource.RLIMIT_CORE, (0, 0))
except (ValueError, resource.error):
pass
if sys.platform == 'darwin':
# Check if the 'Crash Reporter' on OSX was configured
# in 'Developer' mode and warn that it will get triggered
# when it is.
#
# This assumes that this context manager is used in tests
# that might trigger the next manager.
value = subprocess.Popen(['/usr/bin/defaults', 'read',
'com.apple.CrashReporter', 'DialogType'],
stdout=subprocess.PIPE).communicate()[0]
if value.strip() == b'developer':
print "this tests triggers the Crash Reporter, that is intentional"
sys.stdout.flush()
def setup_limit(self):
"""set up the process limit"""
assert currentThread().getName() == 'MainThread'
os.setpgrp()
if self._limit_set <= 0:
if self.max_time is not None:
self._old_usr2_hdlr = signal(SIGUSR2, self._hangle_sig_timeout)
self._timer = Timer(max(1, int(self.max_time) - self._elapse_time),
self._time_out)
self._start_time = int(time())
self._timer.start()
if self.max_cpu_time is not None:
self._old_max_cpu_time = getrlimit(RLIMIT_CPU)
cpu_limit = (int(self.max_cpu_time), self._old_max_cpu_time[1])
self._old_sigxcpu_hdlr = signal(SIGXCPU, self._handle_sigxcpu)
setrlimit(RLIMIT_CPU, cpu_limit)
if self.max_memory is not None:
self._msentinel = MemorySentinel(1, int(self.max_memory) )
self._old_max_memory = getrlimit(RLIMIT_AS)
self._old_usr1_hdlr = signal(SIGUSR1, self._hangle_sig_memory)
as_limit = (int(self.max_memory), self._old_max_memory[1])
setrlimit(RLIMIT_AS, as_limit)
self._msentinel.start()
self._limit_set += 1
def setup_limit(self):
"""set up the process limit"""
assert currentThread().getName() == 'MainThread'
os.setpgrp()
if self._limit_set <= 0:
if self.max_time is not None:
self._old_usr2_hdlr = signal(SIGUSR2, self._hangle_sig_timeout)
self._timer = Timer(max(1, int(self.max_time) - self._elapse_time),
self._time_out)
self._start_time = int(time())
self._timer.start()
if self.max_cpu_time is not None:
self._old_max_cpu_time = getrlimit(RLIMIT_CPU)
cpu_limit = (int(self.max_cpu_time), self._old_max_cpu_time[1])
self._old_sigxcpu_hdlr = signal(SIGXCPU, self._handle_sigxcpu)
setrlimit(RLIMIT_CPU, cpu_limit)
if self.max_memory is not None:
self._msentinel = MemorySentinel(1, int(self.max_memory) )
self._old_max_memory = getrlimit(RLIMIT_AS)
self._old_usr1_hdlr = signal(SIGUSR1, self._hangle_sig_memory)
as_limit = (int(self.max_memory), self._old_max_memory[1])
setrlimit(RLIMIT_AS, as_limit)
self._msentinel.start()
self._limit_set += 1
def _fallbackFDImplementation(self):
"""
Fallback implementation where either the resource module can inform us
about the upper bound of how many FDs to expect, or where we just guess
a constant maximum if there is no resource module.
All possible file descriptors from 0 to that upper bound are returned
with no attempt to exclude invalid file descriptor values.
"""
try:
import resource
except ImportError:
maxfds = 1024
else:
# OS-X reports 9223372036854775808. That's a lot of fds to close.
# OS-X should get the /dev/fd implementation instead, so mostly
# this check probably isn't necessary.
maxfds = min(1024, resource.getrlimit(resource.RLIMIT_NOFILE)[1])
return range(maxfds)
def test_rlimit_get(self):
import resource
p = psutil.Process(os.getpid())
names = [x for x in dir(psutil) if x.startswith('RLIMIT')]
assert names, names
for name in names:
value = getattr(psutil, name)
self.assertGreaterEqual(value, 0)
if name in dir(resource):
self.assertEqual(value, getattr(resource, name))
# XXX - On PyPy RLIMIT_INFINITY returned by
# resource.getrlimit() is reported as a very big long
# number instead of -1. It looks like a bug with PyPy.
if PYPY:
continue
self.assertEqual(p.rlimit(value), resource.getrlimit(value))
else:
ret = p.rlimit(value)
self.assertEqual(len(ret), 2)
self.assertGreaterEqual(ret[0], -1)
self.assertGreaterEqual(ret[1], -1)
def test_rlimit_get(self):
import resource
p = psutil.Process(os.getpid())
names = [x for x in dir(psutil) if x.startswith('RLIMIT')]
assert names, names
for name in names:
value = getattr(psutil, name)
self.assertGreaterEqual(value, 0)
if name in dir(resource):
self.assertEqual(value, getattr(resource, name))
# XXX - On PyPy RLIMIT_INFINITY returned by
# resource.getrlimit() is reported as a very big long
# number instead of -1. It looks like a bug with PyPy.
if PYPY:
continue
self.assertEqual(p.rlimit(value), resource.getrlimit(value))
else:
ret = p.rlimit(value)
self.assertEqual(len(ret), 2)
self.assertGreaterEqual(ret[0], -1)
self.assertGreaterEqual(ret[1], -1)
def test_urandom_failure(self):
# Check urandom() failing when it is not able to open /dev/random.
# We spawn a new process to make the test more robust (if getrlimit()
# failed to restore the file descriptor limit after this, the whole
# test suite would crash; this actually happened on the OS X Tiger
# buildbot).
code = """if 1:
import errno
import os
import resource
soft_limit, hard_limit = resource.getrlimit(resource.RLIMIT_NOFILE)
resource.setrlimit(resource.RLIMIT_NOFILE, (1, hard_limit))
try:
os.urandom(16)
except OSError as e:
assert e.errno == errno.EMFILE, e.errno
else:
raise AssertionError("OSError not raised")
"""
assert_python_ok('-c', code)
def setup_limit(self):
"""set up the process limit"""
assert currentThread().getName() == 'MainThread'
os.setpgrp()
if self._limit_set <= 0:
if self.max_time is not None:
self._old_usr2_hdlr = signal(SIGUSR2, self._hangle_sig_timeout)
self._timer = Timer(max(1, int(self.max_time) - self._elapse_time),
self._time_out)
self._start_time = int(time())
self._timer.start()
if self.max_cpu_time is not None:
self._old_max_cpu_time = getrlimit(RLIMIT_CPU)
cpu_limit = (int(self.max_cpu_time), self._old_max_cpu_time[1])
self._old_sigxcpu_hdlr = signal(SIGXCPU, self._handle_sigxcpu)
setrlimit(RLIMIT_CPU, cpu_limit)
if self.max_memory is not None:
self._msentinel = MemorySentinel(1, int(self.max_memory) )
self._old_max_memory = getrlimit(RLIMIT_AS)
self._old_usr1_hdlr = signal(SIGUSR1, self._hangle_sig_memory)
as_limit = (int(self.max_memory), self._old_max_memory[1])
setrlimit(RLIMIT_AS, as_limit)
self._msentinel.start()
self._limit_set += 1
def setup_limit(self):
"""set up the process limit"""
assert currentThread().getName() == 'MainThread'
os.setpgrp()
if self._limit_set <= 0:
if self.max_time is not None:
self._old_usr2_hdlr = signal(SIGUSR2, self._hangle_sig_timeout)
self._timer = Timer(max(1, int(self.max_time) - self._elapse_time),
self._time_out)
self._start_time = int(time())
self._timer.start()
if self.max_cpu_time is not None:
self._old_max_cpu_time = getrlimit(RLIMIT_CPU)
cpu_limit = (int(self.max_cpu_time), self._old_max_cpu_time[1])
self._old_sigxcpu_hdlr = signal(SIGXCPU, self._handle_sigxcpu)
setrlimit(RLIMIT_CPU, cpu_limit)
if self.max_memory is not None:
self._msentinel = MemorySentinel(1, int(self.max_memory) )
self._old_max_memory = getrlimit(RLIMIT_AS)
self._old_usr1_hdlr = signal(SIGUSR1, self._hangle_sig_memory)
as_limit = (int(self.max_memory), self._old_max_memory[1])
setrlimit(RLIMIT_AS, as_limit)
self._msentinel.start()
self._limit_set += 1
def close_open_files():
'''Closes all open files. Useful after a fork.'''
maxfd = resource.getrlimit(resource.RLIMIT_NOFILE)[1]
if maxfd == resource.RLIM_INFINITY:
maxfd = MAXFD
for fd in reversed(range(maxfd)):
try:
os.close(fd)
except OSError, e:
if e.errno == errno.EBADF:
pass # File not open
else:
raise Exception("Failed to close file descriptor %d: %s" % (fd, e))
def config_max_files(self):
soft, _ = resource.getrlimit(resource.RLIMIT_NOFILE)
self.maxfiles = soft - 100
