I have a list with a length of 100. and I am running it concurrently with threadpool. I can add the time delay inside the executing function, but I would like to have a code that automatically sleeps for X seconds when 10 successful thread execution happen.
import time
from concurrent.futures import ThreadPoolExecutor
user_list = [1,2,3,4,5,6,7,8,9,10,11,12,13,...,100]
def parse(user):
return str(user) + "parsed!"
with ThreadPoolExecutor(max_workers=10) as exe:
start = time.time()
result = exe.map(parse,user_list)
output = list(result)
end = time.time()
print('taken time' end-start)
I want to add a time delay between each 10 successful thread executions. I hope my question is clear, and it can be solved with some sheduling
The straightforward way is to just submit up to 10 jobs at a time, then sleep between each chunk:
import itertools
import time
from concurrent.futures import ThreadPoolExecutor
# See https://stackoverflow.com/a/8991553/51685
def chunker(n, iterable):
it = iter(iterable)
while True:
chunk = tuple(itertools.islice(it, n))
if not chunk:
return
yield chunk
def parse(user):
return f"{user} parsed!"
def main():
user_list = list(range(100))
with ThreadPoolExecutor(max_workers=10) as exe:
for chunk in chunker(10, user_list):
start = time.time()
result = exe.map(parse, chunk)
output = list(result)
end = time.time()
print(output, "taken time", end - start)
time.sleep(1)
if __name__ == "__main__":
main()
This prints out e.g.
['0 parsed!', '1 parsed!', '2 parsed!', '3 parsed!', '4 parsed!', '5 parsed!', '6 parsed!', '7 parsed!', '8 parsed!', '9 parsed!'] taken time 0.0006809234619140625
['10 parsed!', '11 parsed!', '12 parsed!', '13 parsed!', '14 parsed!', '15 parsed!', '16 parsed!', '17 parsed!', '18 parsed!', '19 parsed!'] taken time 0.0008037090301513672
['20 parsed!', '21 parsed!', '22 parsed!', '23 parsed!', '24 parsed!', '25 parsed!', '26 parsed!', '27 parsed!', '28 parsed!', '29 parsed!'] taken time 0.0008540153503417969
...
To use tqdm
with this approach so it gets updated on each parse
step, you'll need something like the below (bits identical to the above replaced with ...
).
(tqdm
won't update the screen unless enough time has passed since the last time it did, hence the random sleep to represent work done.)
def parse(user, prog):
time.sleep(random.uniform(.1, 1.3)) # Do work here...
prog.update() # Step the progress bar.
return f"{user} parsed!"
def main():
# ...
with ThreadPoolExecutor(max_workers=10) as exe, tqdm.tqdm(total=len(user_list)) as prog:
for chunk in chunker(10, user_list):
# ...
result = exe.map(parse, chunk, [prog] * len(chunk))
# ...