use a semaphore to avoid fetching complete library to memory

-> this happens if the processing of items is slower as the fetching of
new
-> if a big library is synced, the old behavior could lead to extensive
use of memory
-> the semaphore acts like a buffer that only allows fetching of new
items from the library if old ones are processed
-> the current size of the 'buffer' is hard coded to 2 * [max. item
fetch limit] * [number of download threads]
This commit is contained in:
mammo0 2020-10-14 08:24:37 +02:00
parent 65f400b08d
commit 09b0bdbc48
1 changed files with 8 additions and 1 deletions

View File

@ -281,7 +281,13 @@ def _get_items(query, server_id=None):
# threads. Dont be a dummy.Pool, be a ThreadPoolExecutor
p = concurrent.futures.ThreadPoolExecutor(DTHREADS)
results = p.map(lambda params: _get(url, params, server_id=server_id), query_params)
thread_buffer = threading.Semaphore(2 * LIMIT * DTHREADS)
def get_wrapper(params):
thread_buffer.acquire()
return _get(url, params, server_id=server_id)
results = p.map(get_wrapper, query_params)
for params, result in zip(query_params, results):
query['params'] = params
@ -302,6 +308,7 @@ def _get_items(query, server_id=None):
items['RestorePoint'] = query
yield items
del items['Items'][:]
thread_buffer.release()
class GetItemWorker(threading.Thread):