Как написать потребитель для огромного генератора, который не пропускает память?

Question

TL / DR: ThreadPoolExecutor был причиной. Использование памяти с concurrent.futures.ThreadPoolExecutor в Python3

Вот скрипт на языке Python (значительно упрощенный), который выполняет алгоритм маршрутизации «все-ко-всему», и в процессе он съедает всю память.

Я понимаю, что проблема в том, что основная функция не возвращается, и объекты, созданные внутри нее, не очищаются сборщиком мусора.

Мой главный вопрос: можно ли написать получателя для возвращаемого генератора, чтобы данные были очищены?Или я должен просто вызвать утилиту сборщика мусора?

# thread pool executor like in python documentation example
def table_process(callable, total):
    with ThreadPoolExecutor(max_workers=threads) as e:
    future_map = {
        e.submit(callable, i): i
        for i in range(total)
    }

    for future in as_completed(future_map):
        if future.exception() is None:
            yield future.result()
        else:
            raise future.exception()

@argh.dispatch_command
def main():
    threads = 10
    data = pd.DataFrame(...)  # about 12K rows

    # this function routes only one slice of sources/destinations
    def _process_chunk(x:int) -> gpd.GeoDataFrame:
        # slicing is more complex, but simplified here for presentation
        # do cross-product and an http request to process the result
        result_df = _do_process(grid[x], grid)
        return result_df

    # writing to geopackage
    with fiona.open('/tmp/some_file.gpkg', 'w', driver='GPKG', schema=...) as f:
        for results_df in table_process(_process_chunk, len(data)):
            aggregated_df = results_df.groupby('...').aggregate({...})
            f.writerecords(aggregated_df)

Answer 1

Оказалось, что именно ThreadPoolExecutor сохраняет работников и не освобождает память.

Решения здесь: Использование памяти с concurrent.futures.ThreadPoolExecutor в Python3