Освобождение памяти после окончания потоков

Question

Я работаю над приложением, которое в первую очередь является API, но также имеет многопоточную систему фоновой обработки заданий, используемую для выполнения запланированных заданий, а также специальных заданий, которые требуют слишком много времени для мгновенного ответа API.

Это будет разветвлено 10 раз через огнестрельное оружие.Любой отдельный разветвленный процесс способен подбирать задание для выполнения, поэтому обработка заданий сбалансирована между процессами и обслуживанием запросов API.

Моя задача состоит в том, чтобы каждый процесс продолжал требовать пиковое количествопамяти, которая когда-либо была необходима для обработки задания.Для некоторых заданий требуется память объемом 1,5–2 ГБ.

При наличии достаточного количества времени все 10 процессов должны будут выполнять такие задания, и каждый из них будет цепляться за объем, превышающий 2 ГБ памяти.Даже если среднее использование памяти процессом редко превышает 100 МБ.

Эти интенсивные задания выполняются только через выделенные потоки внутри процесса.

Существует ли ЛЮБОЙ механизм, чтобы заставить Python освободить требуемую памятьспециально для потока при его закрытии?Или какой-нибудь общий механизм, заставляющий процесс Python сбрасывать память только на то, что активно требуется в данный момент?

Примечание: я также изучаю разветвление вместо многопоточности, но пока это создает другие проблемы.уверен, что я могу обойти.

Answer 1

Просто чтобы доказать, что потоки уничтожаются после завершения их работы, вы можете запустить этот код:

def job(o: dict):
    count = 1
    r = random.randrange(10, 20)
    while count < r:
        print(f"{o['name']}={count}/{r}")
        count += 1
        time.sleep(1)

    print(f"{o['name']} finished.")


def run_thread(o: dict):
    threading.Thread(target=job, args=(o,)).start()


if __name__ == '__main__':
    obj1 = {"name": "A"}
    run_thread(obj1)

    obj2 = {"name": "B"}
    run_thread(obj2)

    while True:
        time.sleep(1)
        print(f"THREADS: {len(threading.enumerate())}")

Вывод будет выглядеть примерно так:

A=1/14
B=1/10
THREADS: 3
B=2/10
A=2/14    
THREADS: 3
...
B finished.
A=10/14
A=11/14
THREADS: 2
A=12/14
THREADS: 2
A=13/14
THREADS: 2
A finished.
THREADS: 1
THREADS: 1
THREADS: 1

Как вы можетесм., когда поток заканчивается, общее количество потоков уменьшается.

ОБНОВЛЕНИЕ:

ОК.Я надеюсь, что этот сценарий удовлетворит вас.

from typing import List
import random
import threading
import time
import os
import psutil


def get_mem_usage():
    return PROCESS.memory_info().rss // 1024


def show_mem_usage():
    global MAX_MEMORY
    while True:
        mem = get_mem_usage()
        print(f"Currently used memory={mem} KB")
        MAX_MEMORY = max(mem, MAX_MEMORY)
        time.sleep(5)


def job(name: str):
    print(f"{name} started.")
    job_memory: List[int] = []
    total_bit_length = 0
    while command['stop_thread'] is False:
        num = random.randrange(100000, 999999)
        job_memory.append(num)
        total_bit_length += int.bit_length(num)
        time.sleep(0.0000001)
        if len(job_memory) % 100000 == 0:
            print(f"{name} Memory={total_bit_length//1024} KB")

    print(f"{name} finished.")


def start_thread(name: str):
    threading.Thread(target=job, args=(name,), daemon=True).start()


if __name__ == '__main__':
    command = {'stop_thread': False}

    STOP_THREAD = False
    PROCESS = psutil.Process(os.getpid())

    mem_before_threads = get_mem_usage()

    MAX_MEMORY = 0

    print(f"Starting memory={mem_before_threads} KB")

    threading.Thread(target=show_mem_usage, daemon=True).start()

    input("Press enter to START threads...\n")

    for i in range(20):
        start_thread("Job" + str(i + 1))

    input("Press enter to STOP threads...\n")
    print("Stopping threads...")

    command['stop_thread'] = True

    time.sleep(2)  # give some time to stop threads

    print("Threads stopped.")

    mem_after_threads = get_mem_usage()

    print(f"Memory before threads={mem_before_threads} KB")
    print(f"Max Memory while threads running={MAX_MEMORY} KB")
    print(f"Memory after threads stopped={mem_after_threads} KB")

    input("Press enter to exit.")

И это вывод:

Starting memory=12980 KB
Currently used memory=13020 KB
Press enter to START threads...

Job1 started.
Job2 started.
Job3 started.
Job4 started.
Job5 started.
Job6 started.
Job7 started.
Job8 started.
Job9 started.
Job10 started.
Job11 started.
Job12 started.
Job13 started.
Job14 started.
Job15 started.
Job16 started.
Job17 started.
Job18 started.
Job19 started.
Job20 started.

Press enter to STOP threads...
Currently used memory=16740 KB
Currently used memory=19764 KB
Currently used memory=22516 KB
Currently used memory=25420 KB
Currently used memory=28340 KB

Stopping threads...
Job12 finished.
Job20 finished.
Job11 finished.
Job7 finished.
Job18 finished.
Job2 finished.
Job4 finished.
Job19 finished.
Job16 finished.
Job10 finished.
Job1 finished.
Job9 finished.
Job6 finished.
Job13 finished.
Job15 finished.
Job17 finished.
Job3 finished.
Job5 finished.
Job8 finished.
Job14 finished.

Threads stopped.

Memory before threads=12980 KB
Max Memory while threads running=28340 KB
Memory after threads stopped=13384 KB
Press enter to exit.
Currently used memory=13388 KB

Я действительно не знаю, почему существует разница в 408 КБ, покупка может быть чрезмернойза использование 15 МБ памяти.

Answer 2

Без конкретного примера того, что делают ваши API и рабочие процессы / потоки, трудно дать конкретный ответ.

Python - это язык с подсчетом ссылок: когда на объект не ссылаются никакие другие, он являетсябесплатно собирать мусор.Можно заставить сборщик мусора работать (см. https://docs.python.org/3/library/gc.html),, но почти всегда лучше всего позволить ему делать свое дело.

Когда ваши рабочие потоки выходят, любые объекты, созданные в потоке, вероятно, будутсборщик мусора, и исключением из этого могут быть объекты, помещенные в некоторую глобальную структуру данных (но ваш вариант использования не похож на то, что вы делаете).