Обычно программы пишут так, чтобы при работе они использовали как можно меньше ресурсов компьютера, ну или хотя бы использовали их максимально эффективно, если речь идёт о компьютерных играх или специальном ПО для работы с графикой. Но сегодня мы поступим иначе: создадим программу, которая использует процессор по полной, причём задействуя все ядра.
Зачем нагружать процессор на 100%
Чаще всего мы не хотим, чтобы наш процессор был сильно нагружен, но есть ситуации, в которых это требуется.
Тестирование производительности при разработке или оптимизации. Так можно определить, насколько эффективны изменения в коде и как они влияют на производительность системы.
Имитация высокой нагрузки. Это позволяет узнать, как программа или веб-сервис работают при высокой нагрузке и есть ли проблемы с производительностью и стабильностью.
Тестирование охлаждения. Это помогает проверить, как система охлаждения справляется со своей задачей.
Тестирование инструментов мониторинга производительности системы. Так можно убедиться, что они правильно собирают и анализируют данные.
Криптографические вычисления. Так можно проверить производительность алгоритмов шифрования и дешифрования, которые могут быть процессороёмкими.
⚠️ На всякий случай предупреждаем, что при нагрузке процессора на 100% компьютер может потреблять больше энергии и сильно нагреваться. Советуем убедиться, что охлаждение работает как следует. Например, если у вас ноутбук в защитных накладках, их желательно снять.
Установим модуль многопроцессорности
Для начала установим модуль multiprocessing
с помощью команды
pip install multiprocessing
Если этот способ не работает, пробуем команду pip3:
pip3 install multiprocessing
Используем класс Process
Класс Process
не создавался для целенаправленной нагрузки процессора, но с его помощью можно создать несколько параллельных процессов, которые будут выполнять вычислительно интенсивные задачи.
Подключим нужные модули и классы:
# подключаем модуль многопроцессорности
import multiprocessing
# подключаем класс Process
from multiprocessing import Process
# подключаем модуль времени
import time
Создаём функцию для нагрузки процессора
Создадим функцию, которая будет выполняться в каждом отдельном процессе, и зададим бесконечный цикл нагрузки:
# определяем функцию, которая будет выполняться в каждом отдельном процессе
def cp_100():
# задаём бесконечный цикл, который будет нагружать процессор
while True:
pass
На всякий случай проверим, что код выполняется в основном потоке, а не дочернем процессе.
if __name__ == "__main__":
Помимо этого, учтём, что в современных компьютерах центральный процессор многоядерный, а нам нужно нагрузить их все.
num_processes = multiprocessing.cpu_count()
Теперь нам нужно, чтобы процессы где-то хранили свои объекты, а также создавались и запускались по какому-то циклу. Наконец, создадим объект процесса, который будет выполнять нашу функцию, нагружающую процессор.
# создаём пустой список для хранения объектов процессов
processes = []
# задаём цикл, который будет создавать и запускать новые процессы
for _ in range(num_processes):
# создаём объект процесса, который будет выполнять функцию cp_100
p = Process(target=cp_100)
Запускаем программу
Теперь у нас всё готово для запуска. Запустим процесс, добавим объекты процесса в список, а затем используем команду time.sleep(10), чтобы приостановить выполнение основного потока на 10 секунд. После этого зададим цикл, чтобы завершить все процессы и основной процесс, который их создаёт. Продолжим писать код в цикле:
# запускаем процесс
p.start()
# добавляем объекты процесса в список
processes.append(p)
# приостанавливаем выполнение основного потока на 10 секунд
time.sleep(10)
# задаём цикл, который завершает все созданные процессы
for p in processes:
# завершаем процесс
p.terminate()
После запуска проверим загрузку всех ядер процессора в мониторинге производительности. Мы используем 8-ядерный компьютер, видно, что нагрузка создана на все. Она не составляет 100%, но приближается к этому показателю:
# подключаем модуль многопроцессорности
import multiprocessing
# подключаем класс Process
from multiprocessing import Process
# подключаем модуль времени
import time
# определяем функцию, которая будет выполняться в каждом отдельном процессе
def cp_100():
# задаём бесконечный цикл, который будет нагружать процессор
while True:
pass
# проверяем, что код выполняется в основном потоке, а не дочернем процессе
if __name__ == "__main__":
# получаем количество доступных процессоров на компьютере
num_processes = multiprocessing.cpu_count()
# создаём пустой список для хранения объектов процессов
processes = []
# задаём цикл, который будет создавать и запускать новые процессы
for _ in range(num_processes):
# создаём объект процесса, который будет выполнять функцию cp_100
p = Process(target=cp_100)
# запускаем процесс
p.start()
# добавляем объекты процесса в список
processes.append(p)
# приостанавливаем выполнение основного потока на 10 секунд
time.sleep(10)
# задаём цикл, который завершает все созданные процессы
for p in processes:
# завершаем процесс
p.terminate()