Measure wait time of threadpool tasks in python

Question

I'm running a threadpool executor to which I submit some tasks.

executor = ThreadPoolExecutor(thread_name_prefix='OMS.oms_thread_', max_workers=16)
task = executor.submit(method_to_run, args)

I know I can get the status of these by calling task.running() to know if it is completed or not. What I can't figure out is how to measure how much time a task spent waiting to be started. One way could be to store the time when a task was created, and pass some task Id to method_to_run and have it store the time when it started running and get the difference between those times.

But that's a lot of trouble and would need me to change the method_to_run. Is there a better way?

Answer 1

The main idea is to subclass the Executor, overload the submit method in such a way to record the time before returning a Future object representing the task and make a custom chrono_submit2running which query the Future objects until running is True.

from concurrent.futures import ThreadPoolExecutor, Future

max_workers = 10
n_tasks = 20 # amount of tasks
task_time_scale = 2 # seconds

############################ Test tasks - create random tasks
import random
random.seed(12345)

def func_body(func_name, time_sleep):
    start = time()
    #print(f"{func_name} start at {start}")
    sleep(time_sleep)
    end = time()
    #print(f"{func_name} end at {end}, tot {end - start:.3f} sec")

class TestFunctions:
    pass

times = [task_time_scale*random.random() for _ in range(n_tasks)]
# add dynamically task-function to class as static methods
apply_func_name_pattern = "func_{}".format
for i, t in enumerate(times):
    setattr(TestFunctions,
            apply_func_name_pattern(i), 
            lambda i=i, t=t: staticmethod(func_body(i, t)))

############################# main


class ExeStartChrono(ThreadPoolExecutor):
    # measure the time between submission and running of each thread

    def __init__(self, max_workers):
        super().__init__(max_workers, thread_name_prefix, initializer, initargs)
        self.time = time()
        self.TIME = {}

    def submit(self, i, fn, /, *args, **kwargs) -> Future:
        self.TIME[i] = [time()-self.time]
        return super().submit(fn, *args, **kwargs)

    def chrono_submit2running(self, funcs):
        futures = {i: self.submit(i, func) for i, func in enumerate(funcs)}

        while futures:
            
            futures_it = iter(futures.copy().items())
            while futures_it:
                p = next(futures_it, None)
                if p is None:
                    break
                i, f = p
                if f.running():
                    self.TIME[i].append(time()-self.time)
                    del futures[i]
    
    def __str__(self):
        out = 'fname\ttsubmit trunning twait\n'
        for k, (t_submit, t_running) in self.TIME.items():
            out += f"{k}\t{t_submit:.5f} {t_running:.5f}  {t_running-t_submit:.5f}\n"
            
        return out.rstrip()

    def __exit__(self, *args, **kwargs):
        super().__exit__(*args, **kwargs)
        print(self)
        print(f"Total execution time {time()-self.time:.5f} seconds")


funcs = [getattr(TestFunctions, apply_func_name_pattern(i)) for i in range(n_tasks)]            
with ExeStartChrono(max_workers=max_workers) as exe:
    exe.chrono_submit2running(funcs)

Output

fname   tsubmit trunning twait
0       0.00001 0.00366  0.00365
1       0.00039 0.00367  0.00328
2       0.00071 0.00367  0.00297
3       0.00099 0.00368  0.00268
4       0.00125 0.00369  0.00244
5       0.00192 0.00369  0.00177
6       0.00224 0.00370  0.00146
7       0.00254 0.00371  0.00117
8       0.00284 0.00371  0.00087
9       0.00315 0.00372  0.00057
10      0.00336 0.02638  0.02303
11      0.00339 0.25694  0.25355
12      0.00342 0.33146  0.32805
13      0.00345 0.39496  0.39151
14      0.00349 0.60386  0.60037
15      0.00352 0.61087  0.60735
16      0.00355 0.74407  0.74053
17      0.00357 0.79863  0.79506
18      0.00360 0.83886  0.83526
19      0.00362 0.87449  0.87086
Total execution time 2.70298 seconds