From aef863591b9ce7265e630cd3969cfa79674c5943 Mon Sep 17 00:00:00 2001 From: Robert Schmidt Date: Thu, 6 Jun 2024 08:32:14 +0200 Subject: [PATCH] CI: increase max number of concurrent workers to 64 According to the ThreadPoolExecutor documentation [1], the default number of maximum workers min(32, os.cpu_count() + 4). The problem is that we execute many pipelines from nano, on which os.cpu_count() reports 4, so we can only have up to 8 concurrent workers by default. This leads to the fact that on pipelines with many UEs (e.g., AW2S-AmariUE), we don't ping/iperf on all UEs at the same time. Fix this by increasing the number of workers to 64. This should be fine, as there is almost no real work being done in the Workers (only opening a shell and execute programs such as ping/iperf). [1] https://docs.python.org/3/library/concurrent.futures.html#threadpoolexecutor --- ci-scripts/cls_oaicitest.py | 18 +++++++++--------- 1 file changed, 9 insertions(+), 9 deletions(-) diff --git a/ci-scripts/cls_oaicitest.py b/ci-scripts/cls_oaicitest.py index 2a23f43dd7..fb862b132c 100644 --- a/ci-scripts/cls_oaicitest.py +++ b/ci-scripts/cls_oaicitest.py @@ -394,7 +394,7 @@ class OaiCiTest(): def InitializeUE(self, HTML): ues = [cls_module.Module_UE(n.strip()) for n in self.ue_ids] messages = [] - with concurrent.futures.ThreadPoolExecutor() as executor: + with concurrent.futures.ThreadPoolExecutor(max_workers=64) as executor: futures = [executor.submit(ue.initialize) for ue in ues] for f, ue in zip(futures, ues): uename = f'UE {ue.getName()}' @@ -609,7 +609,7 @@ class OaiCiTest(): def AttachUE(self, HTML, RAN, EPC, CONTAINERS): ues = [cls_module.Module_UE(ue_id, server_name) for ue_id, server_name in zip(self.ue_ids, self.nodes)] - with concurrent.futures.ThreadPoolExecutor() as executor: + with concurrent.futures.ThreadPoolExecutor(max_workers=64) as executor: futures = [executor.submit(ue.attach) for ue in ues] attached = [f.result() for f in futures] futures = [executor.submit(ue.checkMTU) for ue in ues] @@ -624,7 +624,7 @@ class OaiCiTest(): def DetachUE(self, HTML): ues = [cls_module.Module_UE(ue_id, server_name) for ue_id, server_name in zip(self.ue_ids, self.nodes)] - with concurrent.futures.ThreadPoolExecutor() as executor: + with concurrent.futures.ThreadPoolExecutor(max_workers=64) as executor: futures = [executor.submit(ue.detach) for ue in ues] [f.result() for f in futures] messages = [f"UE {ue.getName()}: detached" for ue in ues] @@ -632,7 +632,7 @@ class OaiCiTest(): def DataDisableUE(self, HTML): ues = [cls_module.Module_UE(n.strip()) for n in self.ue_ids] - with concurrent.futures.ThreadPoolExecutor() as executor: + with concurrent.futures.ThreadPoolExecutor(max_workers=64) as executor: futures = [executor.submit(ue.dataDisable) for ue in ues] status = [f.result() for f in futures] if all(status): @@ -645,7 +645,7 @@ class OaiCiTest(): def DataEnableUE(self, HTML): ues = [cls_module.Module_UE(n.strip()) for n in self.ue_ids] logging.debug(f'disabling data for UEs {ues}') - with concurrent.futures.ThreadPoolExecutor() as executor: + with concurrent.futures.ThreadPoolExecutor(max_workers=64) as executor: futures = [executor.submit(ue.dataEnable) for ue in ues] status = [f.result() for f in futures] if all(status): @@ -659,7 +659,7 @@ class OaiCiTest(): ues = [cls_module.Module_UE(n.strip()) for n in self.ue_ids] logging.debug(f'checking status of UEs {ues}') messages = [] - with concurrent.futures.ThreadPoolExecutor() as executor: + with concurrent.futures.ThreadPoolExecutor(max_workers=64) as executor: futures = [executor.submit(ue.check) for ue in ues] messages = [f.result() for f in futures] HTML.CreateHtmlTestRowQueue('NA', 'OK', messages) @@ -766,7 +766,7 @@ class OaiCiTest(): ues = [cls_module.Module_UE(ue_id, server_name) for ue_id, server_name in zip(self.ue_ids, self.nodes)] logging.debug(ues) pingLock = Lock() - with concurrent.futures.ThreadPoolExecutor() as executor: + with concurrent.futures.ThreadPoolExecutor(max_workers=64) as executor: futures = [executor.submit(self.Ping_common, EPC, ue, RAN, CONTAINERS, pingLock) for ue in ues] results = [f.result() for f in futures] # each result in results is a tuple, first member goes to successes, second to messages @@ -855,7 +855,7 @@ class OaiCiTest(): ues = [cls_module.Module_UE(ue_id, server_name) for ue_id, server_name in zip(self.ue_ids, self.nodes)] svr = cls_module.Module_UE(self.svr_id,self.svr_node) logging.debug(ues) - with concurrent.futures.ThreadPoolExecutor() as executor: + with concurrent.futures.ThreadPoolExecutor(max_workers=64) as executor: futures = [executor.submit(self.Iperf_Module, EPC, ue, svr, RAN, i, len(ues), CONTAINERS) for i, ue in enumerate(ues)] results = [f.result() for f in futures] # each result in results is a tuple, first member goes to successes, second to messages @@ -1179,7 +1179,7 @@ class OaiCiTest(): def TerminateUE(self, HTML): ues = [cls_module.Module_UE(n.strip()) for n in self.ue_ids] - with concurrent.futures.ThreadPoolExecutor() as executor: + with concurrent.futures.ThreadPoolExecutor(max_workers=64) as executor: futures = [executor.submit(ue.terminate) for ue in ues] archives = [f.result() for f in futures] archive_info = [f'Log at: {a}' if a else 'No log available' for a in archives]