def cpu_count_logical():
"""Return the number of logical CPUs in the system."""
try:
return os.sysconf("SC_NPROCESSORS_ONLN")
except ValueError:
# as a second fallback we try to parse /proc/cpuinfo
num = 0
with open_binary('%s/cpuinfo' % get_procfs_path()) as f:
for line in f:
if line.lower().startswith(b'processor'):
num += 1
# unknown format (e.g. amrel/sparc architectures), see:
# https://github.com/giampaolo/psutil/issues/200
# try to parse /proc/stat as a last resort
if num == 0:
search = re.compile('cpu\d')
with open_text('%s/stat' % get_procfs_path()) as f:
for line in f:
line = line.split(' ')[0]
if search.match(line):
num += 1
if num == 0:
# mimic os.cpu_count()
return None
return num
python类cpu_count()的实例源码
def cpu_count_physical():
"""Return the number of physical CPUs in the system."""
# From the C module we'll get an XML string similar to this:
# http://manpages.ubuntu.com/manpages/precise/man4/smp.4freebsd.html
# We may get None in case "sysctl kern.sched.topology_spec"
# is not supported on this BSD version, in which case we'll mimic
# os.cpu_count() and return None.
ret = None
s = cext.cpu_count_phys()
if s is not None:
# get rid of padding chars appended at the end of the string
index = s.rfind("</groups>")
if index != -1:
s = s[:index + 9]
root = ET.fromstring(s)
try:
ret = len(root.findall('group/children/group/cpu')) or None
finally:
# needed otherwise it will memleak
root.clear()
if not ret:
# If logical CPUs are 1 it's obvious we'll have only 1
# physical CPU.
if cpu_count_logical() == 1:
return 1
return ret
def cpu_count_logical():
"""Return the number of logical CPUs in the system."""
try:
return os.sysconf("SC_NPROCESSORS_ONLN")
except ValueError:
# mimic os.cpu_count() behavior
return None
def cpu_count_physical():
"""Return the number of physical CPUs in the system."""
# From the C module we'll get an XML string similar to this:
# http://manpages.ubuntu.com/manpages/precise/man4/smp.4freebsd.html
# We may get None in case "sysctl kern.sched.topology_spec"
# is not supported on this BSD version, in which case we'll mimic
# os.cpu_count() and return None.
ret = None
s = cext.cpu_count_phys()
if s is not None:
# get rid of padding chars appended at the end of the string
index = s.rfind("</groups>")
if index != -1:
s = s[:index + 9]
root = ET.fromstring(s)
try:
ret = len(root.findall('group/children/group/cpu')) or None
finally:
# needed otherwise it will memleak
root.clear()
if not ret:
# If logical CPUs are 1 it's obvious we'll have only 1
# physical CPU.
if cpu_count_logical() == 1:
return 1
return ret
def cpu_count_logical():
"""Return the number of logical CPUs in the system."""
try:
return os.sysconf("SC_NPROCESSORS_ONLN")
except ValueError:
# mimic os.cpu_count() behavior
return None
def download_songs(song_ids: typing.Iterator[str]):
songs = filter(None, map(_lookup_gmusic_song, song_ids))
def _load(song):
print("Loading song", song)
song.load()
with ThreadPoolExecutor(os.cpu_count() * 2) as thread_pool:
loading = thread_pool.map(_load, songs)
for _ in loading:
pass
def defaultNumberOfMakeJobs():
makeJobs = os.cpu_count()
if makeJobs > 24:
# don't use up all the resources on shared build systems
# (you can still override this with the -j command line option)
makeJobs = 16
return makeJobs
def task_add_missing_blocks(missing_block_nums,
max_procs,
max_threads,
database_url,
steemd_http_url,
task_num=5):
task_message = fmt_task_message(
'Adding missing blocks to db, this may take a while',
emoji_code_point=u'\U0001F4DD',
task_num=task_num)
click.echo(task_message)
max_workers = max_procs or os.cpu_count() or 1
chunksize = len(missing_block_nums) // max_workers
if chunksize <= 0:
chunksize = 1
#counter = Value('L',0)
map_func = partial(
block_adder_process_worker,
database_url,
steemd_http_url,
max_threads=max_threads)
chunks = chunkify(missing_block_nums, 10000)
with Pool(processes=max_workers) as pool:
results = pool.map(map_func, chunks)
success_msg = fmt_success_message('added missing blocks')
click.echo(success_msg)
def __init__(self, max_workers=0, thread_name_prefix='', max_queue_size=0):
#super().__init__(max_workers or os.cpu_count() or 1, thread_name_prefix)
super().__init__(max_workers or os.cpu_count() or 1)
self.max_queue_size = max_queue_size or self._max_workers * 10
if self.max_queue_size > 0:
self._work_queue = queue.Queue(self.max_queue_size)
self.max_queue_size_reached = 0
def __init__(self, map_func, reduce_func, workers=None):
self.map_func = map_func
self.reduce_func = reduce_func
self._workers = os.cpu_count() or 1 if workers is None else workers
def cpu_count_logical():
"""Return the number of logical CPUs in the system."""
try:
return os.sysconf("SC_NPROCESSORS_ONLN")
except ValueError:
# as a second fallback we try to parse /proc/cpuinfo
num = 0
with open_binary('%s/cpuinfo' % get_procfs_path()) as f:
for line in f:
if line.lower().startswith(b'processor'):
num += 1
# unknown format (e.g. amrel/sparc architectures), see:
# https://github.com/giampaolo/psutil/issues/200
# try to parse /proc/stat as a last resort
if num == 0:
search = re.compile('cpu\d')
with open_text('%s/stat' % get_procfs_path()) as f:
for line in f:
line = line.split(' ')[0]
if search.match(line):
num += 1
if num == 0:
# mimic os.cpu_count()
return None
return num
def cpu_count_physical():
"""Return the number of physical CPUs in the system."""
# From the C module we'll get an XML string similar to this:
# http://manpages.ubuntu.com/manpages/precise/man4/smp.4freebsd.html
# We may get None in case "sysctl kern.sched.topology_spec"
# is not supported on this BSD version, in which case we'll mimic
# os.cpu_count() and return None.
ret = None
s = cext.cpu_count_phys()
if s is not None:
# get rid of padding chars appended at the end of the string
index = s.rfind("</groups>")
if index != -1:
s = s[:index + 9]
root = ET.fromstring(s)
try:
ret = len(root.findall('group/children/group/cpu')) or None
finally:
# needed otherwise it will memleak
root.clear()
if not ret:
# If logical CPUs are 1 it's obvious we'll have only 1
# physical CPU.
if cpu_count_logical() == 1:
return 1
return ret
def cpu_count_logical():
"""Return the number of logical CPUs in the system."""
try:
return os.sysconf("SC_NPROCESSORS_ONLN")
except ValueError:
# mimic os.cpu_count() behavior
return None
def cpu_count_physical():
"""Return the number of physical CPUs in the system."""
# From the C module we'll get an XML string similar to this:
# http://manpages.ubuntu.com/manpages/precise/man4/smp.4freebsd.html
# We may get None in case "sysctl kern.sched.topology_spec"
# is not supported on this BSD version, in which case we'll mimic
# os.cpu_count() and return None.
ret = None
s = cext.cpu_count_phys()
if s is not None:
# get rid of padding chars appended at the end of the string
index = s.rfind("</groups>")
if index != -1:
s = s[:index + 9]
root = ET.fromstring(s)
try:
ret = len(root.findall('group/children/group/cpu')) or None
finally:
# needed otherwise it will memleak
root.clear()
if not ret:
# If logical CPUs are 1 it's obvious we'll have only 1
# physical CPU.
if cpu_count_logical() == 1:
return 1
return ret
def cpu_count_logical():
"""Return the number of logical CPUs in the system."""
try:
return os.sysconf("SC_NPROCESSORS_ONLN")
except ValueError:
# mimic os.cpu_count() behavior
return None
def test_system(json_request, accept):
result = await json_request('/system', {}, accept=accept)
assert result['success']
assert result['system']['cpu_count'] == os.cpu_count()
assert isinstance(result['system']['cpu_mhz'], float)
assert isinstance(result['system']['memory'], int)
def _workers_count():
cpu_count = 0
try:
cpu_count = len(os.sched_getaffinity(0))
except AttributeError:
cpu_count = os.cpu_count()
return cpu_count * 4
def cpu_count_logical():
"""Return the number of logical CPUs in the system."""
try:
return os.sysconf("SC_NPROCESSORS_ONLN")
except ValueError:
# as a second fallback we try to parse /proc/cpuinfo
num = 0
with open_binary('%s/cpuinfo' % get_procfs_path()) as f:
for line in f:
if line.lower().startswith(b'processor'):
num += 1
# unknown format (e.g. amrel/sparc architectures), see:
# https://github.com/giampaolo/psutil/issues/200
# try to parse /proc/stat as a last resort
if num == 0:
search = re.compile('cpu\d')
with open_text('%s/stat' % get_procfs_path()) as f:
for line in f:
line = line.split(' ')[0]
if search.match(line):
num += 1
if num == 0:
# mimic os.cpu_count()
return None
return num
def test_cpu_count_logical_w_nproc(self):
num = int(sh("nproc --all"))
self.assertEqual(psutil.cpu_count(logical=True), num)
def test_cpu_count_logical_w_lscpu(self):
out = sh("lscpu -p")
num = len([x for x in out.split('\n') if not x.startswith('#')])
self.assertEqual(psutil.cpu_count(logical=True), num)
