Сообщить об ошибке.

Функции для работы c PID модуля psutil в Python

Функции для работы с запущенными процессами в Python

Материал содержит описание функций модуля psutil с примерами, которые возвращают список текущих запущенных процессов PID, а также удобные функции проверки существования процесса и ожидания завершения списка экземпляров процесса.

Содержание:

psutil.pids():

Функция psutil.pids() возвращает отсортированный список текущих запущенных PID. Чтобы перебрать все процессы и избежать условий гонки, следует предпочесть функцию psutil.process_iter().

Пример использования psutil.pids():

>>> import psutil
>>> psutil.pids()
# [1, 2, 3, 5, 7, 8, 9, 10, 11, 12, 13, 14, 15, 17, 18, 19, ..., 32498]

psutil.process_iter(attrs=None, ad_value=None):

Функция psutil.process_iter() возвращает итератор, возвращающий экземпляр класса psutil.Process() для всех запущенных процессов на локальном компьютере. Это должно быть предпочтительнее, чем psutil.pids() для перебора процессов, т.к. это безопасно от состояния гонки.

Каждый экземпляр psutil.Process() создается только один раз, а затем кэшируется для следующего вызова psutil.process_iter() (если PID еще жив). Также это гарантирует, что PID процессов не будут использоваться повторно.

Аргументы attrs и ad_value имеют то же значение, что и в Process.as_dict(). Если attrs указан, то результат Process.as_dict() будет сохранен как информационный атрибут, прикрепленный к возвращенным экземплярам процесса. Если attrs является пустым списком, то он медленно будет получать всю информацию о процессе.

Порядок сортировки, в котором возвращаются процессы, основан на их PID.

Пример использования psutil.process_iter():

>>> import psutil
>>> for proc in psutil.process_iter(['pid', 'name', 'username']):
...     print(proc.info)

# {'name': 'systemd', 'pid': 1, 'username': 'root'}
# {'name': 'kthreadd', 'pid': 2, 'username': 'root'}
# {'name': 'ksoftirqd/0', 'pid': 3, 'username': 'root'}
# ...

Пример использования выражения генератора словаря для создания структуры данных {pid: info, ...}:

>>> import psutil, pprint
>>> procs = {p.pid: p.info for p in psutil.process_iter(['name', 'username'])}
>>> pprint.pprint(prosc)
# {
#    1: {'name': 'systemd', 'username': 'root'},
#    2: {'name': 'kthreadd', 'username': 'root'},
#    3: {'name': 'ksoftirqd/0', 'username': 'root'},
#  ...
# }

psutil.pid_exists(pid):

Функция psutil.pid_exists() проверяет, существует ли данный PID в текущем списке процессов. Это быстрее, чем выполнение выражения pid in psutil.pids() и должно быть предпочтительным.

psutil.wait_procs(procs, timeout=None, callback=None):

Удобная функция psutil.wait_procs(), ожидающая завершения списка экземпляров процесса. Возвращает кортеж (gone, live), указывающий, какие процессы завершились, а какие еще живы. У завершенных будет новый атрибут returncode, указывающий статус завершения процесса, возвращаемый Process.wait().

Аргумент callback - это функция, которая вызывается, когда один из ожидающих процессов завершается и экземпляр Process передается в качестве аргумента обратного вызова (экземпляр также будет иметь установленный атрибут returncode). Эта функция возвращает результат, как только все процессы завершатся или когда завершиться timeout (в секундах). В отличие от Process.wait(), если завершиться тайм-аут, то функция не будет вызывать исключение TimeoutExpired.

Типичным вариантом использования может быть:

  • отправить SIGTERM в список процессов;
  • дать какое-то время процессам для их корректного завершения;
  • отправить SIGKILL тем процессам, которые еще живы.

Пример, который завершает и ожидает всех дочерних элементов этого процесса:

import psutil

def on_terminate(proc):
    print(f"Процесс {proc} завершается с кодом {proc.returncode}")

procs = psutil.Process().children()
for p in procs:
    p.terminate()
gone, alive = psutil.wait_procs(procs, timeout=3, callback=on_terminate)
for p in alive:
    p.kill()

Пример поиска процесса по имени.

Для поиска процесса сервера по имени необходимо проверить строку с именем на Process.info['name']. Из документации видно, что функция psutil.process_iter() возвращает объект Process, следовательно функция поиска процесса по имени может быть следующая:

import psutil

def find_procs_by_name(name):
    "Возвращает список процессов, соответствующих 'name'."
    ls = []
    for p in psutil.process_iter(['name']):
        if p.info['name'] == name:
            ls.append(p)
    return ls

Более продвинутая функция поиска процесса по имени может быть с дополнительными проверками на Process.name(), Process.exe() и Process.cmdline():

import os
import psutil

def find_procs_by_name(name):
    "Return a list of processes matching 'name'."
    ls = []
    for p in psutil.process_iter(["name", "exe", "cmdline"]):
        if name == p.info['name'] or \
                p.info['exe'] and os.path.basename(p.info['exe']) == name or \
                p.info['cmdline'] and p.info['cmdline'][0] == name:
            ls.append(p)
    return ls

Примеры фильтрации и сортировки процессов.

Коллекция примеров кода, показывающих, как использовать psutil.process_iter() для фильтрации процессов и их сортировки.

Процессы, принадлежащие пользователю:

>>> import psutil
>>> from pprint import pprint as pp
>>> import getpass
>>> pp([
...     (p.pid, p.info['name']) 
...     for p in psutil.process_iter(['name', 'username']) 
...     if p.info['username'] == getpass.getuser()
...    ])

# (16832, 'bash'),
# (19772, 'ssh'),
# (20492, 'python')]

Вывод всех запущенных/активных процессов:

>>> import psutil
>>> from pprint import pprint as pp
>>> pp([
...     (p.pid, p.info) 
...     for p in psutil.process_iter(['name', 'status']) 
...     if p.info['status'] == psutil.STATUS_RUNNING
...    ])

# [
#   (1150, {'name': 'Xorg', 'status': 'running'}),
#   (1776, {'name': 'unity-panel-service', 'status': 'running'}),
#   (20492, {'name': 'python', 'status': 'running'})
# ]

Процессы, использующие лог-файлы:

>>> import psutil
>>> from pprint import pprint as pp
>>> for p in psutil.process_iter(['name', 'open_files']):
...      for file in p.info['open_files'] or []:
...          if file.path.endswith('.log'):
...               print("%-5s %-10s %s" % (p.pid, p.info['name'][:10], file.path))

# 1510  upstart    ~/.cache/upstart/unity-settings-daemon.log
# 2174  nautilus   ~/.local/share/gvfs-metadata/home-ce08efac.log
# 2650  chrome     ~/.config/google-chrome/Default/data_reduction_proxy_leveldb/000003.log

Процессы, потребляющие более 500 МБ памяти:

>>> import psutil
>>> from pprint import pprint as pp
>>> pp([
...     (p.pid, p.info['name'], p.info['memory_info'].rss) 
...     for p in psutil.process_iter(['name', 'memory_info']) 
...     if p.info['memory_info'].rss > 500 * 1024 * 1024
...    ])

# [
#   (2650, 'chrome', 532324352),
#   (3038, 'chrome', 1120088064),
#   (21915, 'sublime_text', 615407616)
# ]

Топ-3 процесса, потребляющих больше всего процессорного времени:

>>> import psutil
>>> from pprint import pprint as pp
>>> pp([
...     (p.pid, p.info['name'], sum(p.info['cpu_times'])) 
...     for p in sorted(psutil.process_iter(['name', 'cpu_times']), 
...                     key=lambda p: sum(p.info['cpu_times'][:2]))
...    ][-3:])

# [
#   (2721, 'chrome', 10219.73),
#   (1150, 'Xorg', 11116.989999999998),
#   (2650, 'chrome', 18451.97)
# ]