for the FetchConnectionCache use a queue where each thread can get an unsed connection_cache that is properly initialized before we fireup the ThreadPoolExecutor.
for the progress bar we need an adictional task counter that is protected with thread lock as it runs inside the ThreadPoolExecutor. Fixes [YOCTO #14775] -- https://bugzilla.yoctoproject.org/show_bug.cgi?id=14775 Signed-off-by: Jose Quaresma <[email protected]> --- meta/classes/sstate.bbclass | 32 +++++++++++++++++++------------- 1 file changed, 19 insertions(+), 13 deletions(-) diff --git a/meta/classes/sstate.bbclass b/meta/classes/sstate.bbclass index b9e9ef8e26..7a18d9d554 100644 --- a/meta/classes/sstate.bbclass +++ b/meta/classes/sstate.bbclass @@ -997,15 +997,19 @@ def sstate_checkhashes(sq_data, d, siginfo=False, currentcount=0, summary=True, localdata.delVar('BB_NO_NETWORK') from bb.fetch2 import FetchConnectionCache - def checkstatus_init(thread_worker): - thread_worker.connection_cache = FetchConnectionCache() + def checkstatus_init(): + while not connection_cache_pool.full(): + connection_cache_pool.put(FetchConnectionCache()) - def checkstatus_end(thread_worker): - thread_worker.connection_cache.close_connections() + def checkstatus_end(): + while not connection_cache_pool.empty(): + connection_cache = connection_cache_pool.get() + connection_cache.close_connections() - def checkstatus(thread_worker, arg): + def checkstatus(arg): (tid, sstatefile) = arg + connection_cache = connection_cache_pool.get() localdata2 = bb.data.createCopy(localdata) srcuri = "file://" + sstatefile localdata2.setVar('SRC_URI', srcuri) @@ -1015,7 +1019,7 @@ def sstate_checkhashes(sq_data, d, siginfo=False, currentcount=0, summary=True, try: fetcher = bb.fetch2.Fetch(srcuri.split(), localdata2, - connection_cache=thread_worker.connection_cache) + connection_cache=connection_cache) fetcher.checkstatus() bb.debug(2, "SState: Successful fetch test for %s" % srcuri) found.add(tid) @@ -1025,6 +1029,8 @@ def sstate_checkhashes(sq_data, d, siginfo=False, currentcount=0, summary=True, except Exception as e: bb.error("SState: cannot test %s: %s\n%s" % (srcuri, repr(e), traceback.format_exc())) + connection_cache_pool.put(connection_cache) + if progress: tasks.remove(arg) bb.event.fire(bb.event.ProcessProgress(msg, total_tasks - len(tasks)), d) @@ -1047,13 +1053,13 @@ def sstate_checkhashes(sq_data, d, siginfo=False, currentcount=0, summary=True, fetcherenv = bb.fetch2.get_fetcher_environment(d) with bb.utils.environment(**fetcherenv): bb.event.enable_threadlock() - pool = oe.utils.ThreadedPool(nproc, total_tasks, - worker_init=checkstatus_init, worker_end=checkstatus_end, - name="sstate_checkhashes-") - for t in tasks: - pool.add_task(checkstatus, t) - pool.start() - pool.wait_completion() + import concurrent.futures + from queue import Queue + connection_cache_pool = Queue(nproc) + checkstatus_init() + with concurrent.futures.ThreadPoolExecutor(max_workers=nproc) as executor: + executor.map(checkstatus, tasks) + checkstatus_end() bb.event.disable_threadlock() if progress: -- 2.35.3
-=-=-=-=-=-=-=-=-=-=-=- Links: You receive all messages sent to this group. View/Reply Online (#164565): https://lists.openembedded.org/g/openembedded-core/message/164565 Mute This Topic: https://lists.openembedded.org/mt/90514093/21656 Group Owner: [email protected] Unsubscribe: https://lists.openembedded.org/g/openembedded-core/unsub [[email protected]] -=-=-=-=-=-=-=-=-=-=-=-
