You have:
with concurrent.futures.ThreadPoolExecutor(16) as executor:
executor.submit(f1)
The main thread submits a task specifying worker function f1. Then the main thread exits the block and an implicit call to executor.shutdown() is made. Any tasks already submitted and have started to execute will complete but onceshutdown is called submitted tasks that have not yet started execution will be thrown away. In your code the call to shutdown occurs before worker function f1 has had a chance to submit the new task with f2 as the worker function and get its execution started. This can be demonstrated as follows:
with concurrent.futures.ThreadPoolExecutor(16) as executor:
executor.submit(f1)
import time
time.sleep(.1)
We have delayed the call to shutdown by .1 seconds giving f1 a chance to get f2 started. But even this has a race condition: Is .1 seconds always enough time to allow f1 to submit the second task and for that task to start? We cannot depend on this method.
TL;DR
You can skip to the final section Solution if you wish and not read the following solutions for simpler cases.
Attempts
To remove that race condition we can use a multithreading.Event that gets set only after all tasks that we need to submit have started executing:
import concurrent.futures
from threading import Event
all_tasks_submitted = Event()
def f2():
all_tasks_submitted.set()
print("hello, f2")
return 3
def f1():
print("hello, f1")
print(executor.submit(f2).result())
with concurrent.futures.ThreadPoolExecutor(16) as executor:
executor.submit(f1)
all_tasks_submitted.wait()
Prints:
hello, f1
hello, f2
3
So now let's look at your actual case. First, there is a slight bug: f2 takes only two arguments but f1 is trying to invoke it with 3 arguments.
This is far more complicated case in that we are ultimately trying to start 10 * 10 * 10 = 1000 f3 tasks. So we now need a counter to keep track of how many f3 have been started:
import concurrent.futures
from threading import Event, Lock
all_tasks_started = Event()
lock = Lock()
NUM_F3_TASKS = 1_000
total_f3_tasks_started = 0
def f3(arg1, arg2):
global total_f3_tasks_started
with lock:
total_f3_tasks_started += 1
n = total_f3_tasks_started
if n == NUM_F3_TASKS:
all_tasks_started.set()
print(f"hello, f3, {arg1}, {arg2}, f3 tasks started = {n}")
def f2(arg1, arg2):
print(f"hello, f2 {arg1}")
for i in range(10):
executor.submit(f3, arg2, i)
def f1(arg1, arg2, arg3):
print(f"hello, f1 {arg1}")
for i in range(10):
executor.submit(f2, arg2, arg3)
with concurrent.futures.ThreadPoolExecutor(16) as executor:
for i in range(10):
executor.submit(f1, i, 1, 2)
all_tasks_started.wait()
Prints:
hello, f1 0
hello, f1 1
hello, f1 2
hello, f1 3
hello, f1 4
hello, f1 5
hello, f1 6
hello, f1 7
hello, f1 8
hello, f1 9
hello, f2 1
hello, f2 1
hello, f2 1
hello, f2 1
hello, f2 1
hello, f2 1
hello, f2 1
hello, f2 1
hello, f2 1
hello, f2 1
hello, f2 1
hello, f2 1
hello, f2 1
hello, f2 1
hello, f2 1
hello, f2 1
hello, f2 1
hello, f2 1
hello, f2 1
hello, f2 1
hello, f2 1
hello, f2 1
hello, f2 1
...
hello, f3, 2, 2, f3 tasks started = 993
hello, f3, 2, 4, f3 tasks started = 995
hello, f3, 2, 6, f3 tasks started = 997
hello, f3, 2, 8, f3 tasks started = 999
hello, f3, 2, 3, f3 tasks started = 994
hello, f3, 2, 7, f3 tasks started = 998
hello, f3, 2, 5, f3 tasks started = 996
hello, f3, 2, 9, f3 tasks started = 1000
But this means that you need to know in advance exactly how many f3 tasks need to be created. You might be tempted to solve the problem by having f1 not return until all tasks it has submitted complete and having f2 not return until all tasks it has submitted complete. You would thus be having a 10 f1 tasks, 100 f2 tasks and 1000 f3 tasks running concurrently for which you would need a thread pool of size 1110.
Solution
We use an explicit task queue and a task executor as follows:
import concurrent.futures
from queue import Queue
from threading import Lock
task_queue = Queue()
lock = Lock()
task_number = 0
def f3(arg1, arg2):
global task_number
with lock:
task_number += 1
n = task_number
print(f"hello, f3, {arg1}, {arg2}, task_number = {n}")
def f2(arg1, arg2):
print(f"hello, f2 {arg1}")
for i in range(10):
task_queue.put((f3, arg2, i))
def f1(arg1, arg2, arg3):
print(f"hello, f1 {arg1}")
for i in range(10):
task_queue.put((f2, arg2, arg3))
def pool_executor():
while True:
task = task_queue.get()
if task is None:
# sentinel to terminate
return
fn, *args = task
fn(*args)
# Show this work has been completed:
task_queue.task_done()
POOL_SIZE = 16
with concurrent.futures.ThreadPoolExecutor(POOL_SIZE) as executor:
for _ in range(POOL_SIZE):
executor.submit(pool_executor)
for i in range(10):
task_queue.put((f1, i, 1, 2))
# Wait for all tasks to complete
task_queue.join()
# Now we need to terminate the running pool_executor tasks:
# Add sentinels:
for _ in range(POOL_SIZE):
task_queue.put(None)
Prints:
hello, f1 0
hello, f1 1
hello, f1 3
hello, f1 5
hello, f1 7
hello, f1 9
hello, f1 2
hello, f2 1
hello, f2 1
hello, f2 1
hello, f1 4
hello, f1 6
hello, f1 8
hello, f2 1
hello, f2 1
hello, f2 1
hello, f2 1
hello, f2 1
...
hello, f3, 2, 1, task_number = 992
hello, f3, 2, 2, task_number = 993
hello, f3, 2, 4, task_number = 995
hello, f3, 2, 6, task_number = 997
hello, f3, 2, 8, task_number = 999
hello, f3, 2, 3, task_number = 994
hello, f3, 2, 7, task_number = 998
hello, f3, 2, 5, task_number = 996
hello, f3, 2, 9, task_number = 1000
Perhaps you should consider creating your own thread pool with dameon threads, which will terminate when the main process terminates (you could still use the technique of adding sentinel values to signal these threads to terminate when we no longer require them in which case the threads need not be daemon threads).
from queue import Queue
from threading import Lock, Thread
...
def pool_executor():
while True:
fn, *args = task_queue.get()
fn(*args)
# Show this work has been completed:
task_queue.task_done()
POOL_SIZE = 16
for _ in range(POOL_SIZE):
Thread(target=pool_executor, daemon=True).start()
for i in range(10):
task_queue.put((f1, i, 1, 2))
# Wait for all tasks to complete
task_queue.join()
A New Type of Multithreading Pool
We can abstract a multithreading pool that allows running tasks to continue to arbitrarily submit additional tasks and then be able to wait for all tasks to complete. That is, we wait until the task queue has quiesced, the condition where the task queue is empty and no new tasks will be added because there are no tasks currently running:
from queue import Queue
from threading import Thread
class ThreadPool:
def __init__(self, pool_size):
self._pool_size = pool_size
self._task_queue = Queue()
self._shutting_down = False
for _ in range(self._pool_size):
Thread(target=self._executor, daemon=True).start()
def __enter__(self):
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.shutdown()
def _terminate_threads(self):
"""Tell threads to terminate."""
# No new tasks in case this is an immediate shutdown:
self._shutting_down = True
for _ in range(self._pool_size):
self._task_queue.put(None)
self._task_queue.join() # Wait for all threads to terminate
def shutdown(self, wait=True):
if wait:
# Wait until the task queue quiesces (becomes empty).
# Running tasks may be continuing to submit tasks to the queue but
# the expectation is that at some point no more tasks will be added
# and we wait for the queue to become empty:
self._task_queue.join()
self._terminate_threads()
def submit(self, fn, *args):
if self._shutting_down:
return
self._task_queue.put((fn, args))
def _executor(self):
while True:
task = self._task_queue.get()
if task is None: # sentinel
self._task_queue.task_done()
return
fn, args = task
try:
fn(*args)
except Exception as e:
print(e)
# Show this work has been completed:
self._task_queue.task_done()
###############################################
from threading import Lock
lock = Lock()
task_number = 0
results = []
def f3(arg1, arg2):
global task_number
with lock:
task_number += 1
n = task_number
#print(f"hello, f3, {arg1}, {arg2}, task_number = {n}")
results.append(f"hello, f3, {arg1}, {arg2}, task_number = {n}")
def f2(arg1, arg2):
for i in range(10):
pool.submit(f3, arg2, i)
def f1(arg1, arg2, arg3):
for i in range(10):
pool.submit(f2, arg2, arg3)
with ThreadPool(16) as pool:
for i in range(10):
pool.submit(f1, i, 1, 2)
for result in results:
print(result)
Another Way That Uses Standard concurrent.futures Methods
As you have observed, in the above solution an f1 task will complete before the f2 tasks it has submitted has completed and f2 tasks will terminate before f3 tasks have terminated. The problem with your original code was due to a shutdown being implicitly called before all 1000 f3 tasks were submitted. We can prevent this premature shutdown from occuring by having each worker function return a list of Future instance whose results we await:
from concurrent.futures import ThreadPoolExecutor, Future
from threading import Lock
task_number = 0
lock = Lock()
futures = []
def f3(arg1, arg2):
global task_number
with lock:
task_number += 1
n = task_number
print(f"hello, f3, {arg1}, {arg2}, f3 tasks started = {n}")
def f2(arg1, arg2):
print(f"hello, f2 {arg1}")
futures.extend(
executor.submit(f3, arg2, i)
for i in range(10)
)
def f1(arg1, arg2, arg3):
print(f"hello, f1 {arg1}")
futures.extend(
executor.submit(f2, arg2, arg3)
for i in range(10)
)
with ThreadPoolExecutor(16) as executor:
futures.extend(
executor.submit(f1, i, 1, 2)
for i in range(10)
)
cnt = 0
for future in futures:
future.result()
cnt += 1
print(cnt, 'tasks completed.')
Prints:
...
hello, f3, 2, 4, f3 tasks started = 995
hello, f3, 2, 6, f3 tasks started = 997
hello, f3, 2, 8, f3 tasks started = 999
hello, f3, 2, 3, f3 tasks started = 994
hello, f3, 2, 7, f3 tasks started = 998
hello, f3, 2, 5, f3 tasks started = 996
hello, f3, 2, 9, f3 tasks started = 1000
1110 tasks completed.
Answer from Booboo on Stack OverflowYou have:
with concurrent.futures.ThreadPoolExecutor(16) as executor:
executor.submit(f1)
The main thread submits a task specifying worker function f1. Then the main thread exits the block and an implicit call to executor.shutdown() is made. Any tasks already submitted and have started to execute will complete but onceshutdown is called submitted tasks that have not yet started execution will be thrown away. In your code the call to shutdown occurs before worker function f1 has had a chance to submit the new task with f2 as the worker function and get its execution started. This can be demonstrated as follows:
with concurrent.futures.ThreadPoolExecutor(16) as executor:
executor.submit(f1)
import time
time.sleep(.1)
We have delayed the call to shutdown by .1 seconds giving f1 a chance to get f2 started. But even this has a race condition: Is .1 seconds always enough time to allow f1 to submit the second task and for that task to start? We cannot depend on this method.
TL;DR
You can skip to the final section Solution if you wish and not read the following solutions for simpler cases.
Attempts
To remove that race condition we can use a multithreading.Event that gets set only after all tasks that we need to submit have started executing:
import concurrent.futures
from threading import Event
all_tasks_submitted = Event()
def f2():
all_tasks_submitted.set()
print("hello, f2")
return 3
def f1():
print("hello, f1")
print(executor.submit(f2).result())
with concurrent.futures.ThreadPoolExecutor(16) as executor:
executor.submit(f1)
all_tasks_submitted.wait()
Prints:
hello, f1
hello, f2
3
So now let's look at your actual case. First, there is a slight bug: f2 takes only two arguments but f1 is trying to invoke it with 3 arguments.
This is far more complicated case in that we are ultimately trying to start 10 * 10 * 10 = 1000 f3 tasks. So we now need a counter to keep track of how many f3 have been started:
import concurrent.futures
from threading import Event, Lock
all_tasks_started = Event()
lock = Lock()
NUM_F3_TASKS = 1_000
total_f3_tasks_started = 0
def f3(arg1, arg2):
global total_f3_tasks_started
with lock:
total_f3_tasks_started += 1
n = total_f3_tasks_started
if n == NUM_F3_TASKS:
all_tasks_started.set()
print(f"hello, f3, {arg1}, {arg2}, f3 tasks started = {n}")
def f2(arg1, arg2):
print(f"hello, f2 {arg1}")
for i in range(10):
executor.submit(f3, arg2, i)
def f1(arg1, arg2, arg3):
print(f"hello, f1 {arg1}")
for i in range(10):
executor.submit(f2, arg2, arg3)
with concurrent.futures.ThreadPoolExecutor(16) as executor:
for i in range(10):
executor.submit(f1, i, 1, 2)
all_tasks_started.wait()
Prints:
hello, f1 0
hello, f1 1
hello, f1 2
hello, f1 3
hello, f1 4
hello, f1 5
hello, f1 6
hello, f1 7
hello, f1 8
hello, f1 9
hello, f2 1
hello, f2 1
hello, f2 1
hello, f2 1
hello, f2 1
hello, f2 1
hello, f2 1
hello, f2 1
hello, f2 1
hello, f2 1
hello, f2 1
hello, f2 1
hello, f2 1
hello, f2 1
hello, f2 1
hello, f2 1
hello, f2 1
hello, f2 1
hello, f2 1
hello, f2 1
hello, f2 1
hello, f2 1
hello, f2 1
...
hello, f3, 2, 2, f3 tasks started = 993
hello, f3, 2, 4, f3 tasks started = 995
hello, f3, 2, 6, f3 tasks started = 997
hello, f3, 2, 8, f3 tasks started = 999
hello, f3, 2, 3, f3 tasks started = 994
hello, f3, 2, 7, f3 tasks started = 998
hello, f3, 2, 5, f3 tasks started = 996
hello, f3, 2, 9, f3 tasks started = 1000
But this means that you need to know in advance exactly how many f3 tasks need to be created. You might be tempted to solve the problem by having f1 not return until all tasks it has submitted complete and having f2 not return until all tasks it has submitted complete. You would thus be having a 10 f1 tasks, 100 f2 tasks and 1000 f3 tasks running concurrently for which you would need a thread pool of size 1110.
Solution
We use an explicit task queue and a task executor as follows:
import concurrent.futures
from queue import Queue
from threading import Lock
task_queue = Queue()
lock = Lock()
task_number = 0
def f3(arg1, arg2):
global task_number
with lock:
task_number += 1
n = task_number
print(f"hello, f3, {arg1}, {arg2}, task_number = {n}")
def f2(arg1, arg2):
print(f"hello, f2 {arg1}")
for i in range(10):
task_queue.put((f3, arg2, i))
def f1(arg1, arg2, arg3):
print(f"hello, f1 {arg1}")
for i in range(10):
task_queue.put((f2, arg2, arg3))
def pool_executor():
while True:
task = task_queue.get()
if task is None:
# sentinel to terminate
return
fn, *args = task
fn(*args)
# Show this work has been completed:
task_queue.task_done()
POOL_SIZE = 16
with concurrent.futures.ThreadPoolExecutor(POOL_SIZE) as executor:
for _ in range(POOL_SIZE):
executor.submit(pool_executor)
for i in range(10):
task_queue.put((f1, i, 1, 2))
# Wait for all tasks to complete
task_queue.join()
# Now we need to terminate the running pool_executor tasks:
# Add sentinels:
for _ in range(POOL_SIZE):
task_queue.put(None)
Prints:
hello, f1 0
hello, f1 1
hello, f1 3
hello, f1 5
hello, f1 7
hello, f1 9
hello, f1 2
hello, f2 1
hello, f2 1
hello, f2 1
hello, f1 4
hello, f1 6
hello, f1 8
hello, f2 1
hello, f2 1
hello, f2 1
hello, f2 1
hello, f2 1
...
hello, f3, 2, 1, task_number = 992
hello, f3, 2, 2, task_number = 993
hello, f3, 2, 4, task_number = 995
hello, f3, 2, 6, task_number = 997
hello, f3, 2, 8, task_number = 999
hello, f3, 2, 3, task_number = 994
hello, f3, 2, 7, task_number = 998
hello, f3, 2, 5, task_number = 996
hello, f3, 2, 9, task_number = 1000
Perhaps you should consider creating your own thread pool with dameon threads, which will terminate when the main process terminates (you could still use the technique of adding sentinel values to signal these threads to terminate when we no longer require them in which case the threads need not be daemon threads).
from queue import Queue
from threading import Lock, Thread
...
def pool_executor():
while True:
fn, *args = task_queue.get()
fn(*args)
# Show this work has been completed:
task_queue.task_done()
POOL_SIZE = 16
for _ in range(POOL_SIZE):
Thread(target=pool_executor, daemon=True).start()
for i in range(10):
task_queue.put((f1, i, 1, 2))
# Wait for all tasks to complete
task_queue.join()
A New Type of Multithreading Pool
We can abstract a multithreading pool that allows running tasks to continue to arbitrarily submit additional tasks and then be able to wait for all tasks to complete. That is, we wait until the task queue has quiesced, the condition where the task queue is empty and no new tasks will be added because there are no tasks currently running:
from queue import Queue
from threading import Thread
class ThreadPool:
def __init__(self, pool_size):
self._pool_size = pool_size
self._task_queue = Queue()
self._shutting_down = False
for _ in range(self._pool_size):
Thread(target=self._executor, daemon=True).start()
def __enter__(self):
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.shutdown()
def _terminate_threads(self):
"""Tell threads to terminate."""
# No new tasks in case this is an immediate shutdown:
self._shutting_down = True
for _ in range(self._pool_size):
self._task_queue.put(None)
self._task_queue.join() # Wait for all threads to terminate
def shutdown(self, wait=True):
if wait:
# Wait until the task queue quiesces (becomes empty).
# Running tasks may be continuing to submit tasks to the queue but
# the expectation is that at some point no more tasks will be added
# and we wait for the queue to become empty:
self._task_queue.join()
self._terminate_threads()
def submit(self, fn, *args):
if self._shutting_down:
return
self._task_queue.put((fn, args))
def _executor(self):
while True:
task = self._task_queue.get()
if task is None: # sentinel
self._task_queue.task_done()
return
fn, args = task
try:
fn(*args)
except Exception as e:
print(e)
# Show this work has been completed:
self._task_queue.task_done()
###############################################
from threading import Lock
lock = Lock()
task_number = 0
results = []
def f3(arg1, arg2):
global task_number
with lock:
task_number += 1
n = task_number
#print(f"hello, f3, {arg1}, {arg2}, task_number = {n}")
results.append(f"hello, f3, {arg1}, {arg2}, task_number = {n}")
def f2(arg1, arg2):
for i in range(10):
pool.submit(f3, arg2, i)
def f1(arg1, arg2, arg3):
for i in range(10):
pool.submit(f2, arg2, arg3)
with ThreadPool(16) as pool:
for i in range(10):
pool.submit(f1, i, 1, 2)
for result in results:
print(result)
Another Way That Uses Standard concurrent.futures Methods
As you have observed, in the above solution an f1 task will complete before the f2 tasks it has submitted has completed and f2 tasks will terminate before f3 tasks have terminated. The problem with your original code was due to a shutdown being implicitly called before all 1000 f3 tasks were submitted. We can prevent this premature shutdown from occuring by having each worker function return a list of Future instance whose results we await:
from concurrent.futures import ThreadPoolExecutor, Future
from threading import Lock
task_number = 0
lock = Lock()
futures = []
def f3(arg1, arg2):
global task_number
with lock:
task_number += 1
n = task_number
print(f"hello, f3, {arg1}, {arg2}, f3 tasks started = {n}")
def f2(arg1, arg2):
print(f"hello, f2 {arg1}")
futures.extend(
executor.submit(f3, arg2, i)
for i in range(10)
)
def f1(arg1, arg2, arg3):
print(f"hello, f1 {arg1}")
futures.extend(
executor.submit(f2, arg2, arg3)
for i in range(10)
)
with ThreadPoolExecutor(16) as executor:
futures.extend(
executor.submit(f1, i, 1, 2)
for i in range(10)
)
cnt = 0
for future in futures:
future.result()
cnt += 1
print(cnt, 'tasks completed.')
Prints:
...
hello, f3, 2, 4, f3 tasks started = 995
hello, f3, 2, 6, f3 tasks started = 997
hello, f3, 2, 8, f3 tasks started = 999
hello, f3, 2, 3, f3 tasks started = 994
hello, f3, 2, 7, f3 tasks started = 998
hello, f3, 2, 5, f3 tasks started = 996
hello, f3, 2, 9, f3 tasks started = 1000
1110 tasks completed.
Looks like, in your first example, maybe the program terminates before the "f2" thread gets its chance to print.
Adding a time.sleep call to the very end of your first example allows it to print both "hello" messages when I run it.*
import concurrent.futures
import time
def f2():
print("hello, f2")
def f1():
print("hello, f1")
executor.submit(f2)
with concurrent.futures.ThreadPoolExecutor(16) as executor:
executor.submit(f1)
time.sleep(3)
I have not ever used concurrent.futures. Are its worker threads daemons?โ
Update:
I wonder why main thread cannot find the f2 future...
Not sure what you're asking, but try this:
import concurrent.futures
import queue
q = queue.Queue(2)
def f2():
print("hello, f2")
def f1():
print("hello, f1")
future_2 = executor.submit(f2)
q.put(future_2)
with concurrent.futures.ThreadPoolExecutor(16) as executor:
future_1 = executor.submit(f1)
future_2 = q.get()
future_1.result()
future_2.result()
* Python 3.12.4 running on macOS Sonoma 14.4.1.
โ I can't find "daemon" or "demon" anywhere in The documentation, and I can't find out by using Threading.current_thread().daemon because the current_thread function only works in threads that were created directly by the Threading module. current_thread returns a bogus object when called by a concurrent.futures worker thread.
dictionary - Python concurrent executor.map() and submit() - Stack Overflow
multithreading - Python - Why ThreadPoolExecutor().submit a thread with queue blocks and Thread().start doesn't? - Stack Overflow
python - passing multiple arguments to a thread pool executor - Stack Overflow
using python concurrent.futures submit with processes numbering - Stack Overflow
Here is the map version of your existing code. Note that the callback now accepts a tuple as a parameter. I added an try\except in the callback so the results will not throw an error. The results are ordered according to the input list.
from concurrent.futures import ThreadPoolExecutor
import urllib.request
URLS = ['http://www.foxnews.com/',
'http://www.cnn.com/',
'http://www.wsj.com/',
'http://www.bbc.co.uk/',
'http://some-made-up-domain.com/']
# Retrieve a single page and report the url and contents
def load_url(tt): # (url,timeout)
url, timeout = tt
try:
with urllib.request.urlopen(url, timeout=timeout) as conn:
return (url, conn.read())
except Exception as ex:
print("Error:", url, ex)
return(url,"") # error, return empty string
with ThreadPoolExecutor(max_workers=5) as executor:
results = executor.map(load_url, [(u,60) for u in URLS]) # pass url and timeout as tuple to callback
executor.shutdown(wait=True) # wait for all complete
print("Results:")
for r in results: # ordered results, will throw exception here if not handled in callback
print(' %r page is %d bytes' % (r[0], len(r[1])))
Output
Error: http://www.wsj.com/ HTTP Error 404: Not Found
Results:
'http://www.foxnews.com/' page is 320028 bytes
'http://www.cnn.com/' page is 1144916 bytes
'http://www.wsj.com/' page is 0 bytes
'http://www.bbc.co.uk/' page is 279418 bytes
'http://some-made-up-domain.com/' page is 64668 bytes
Without using the map method, you can use enumerate to build the future_to_url dict with not just the URLs as values, but also their indices in the list. You can then build a dict from the future objects returned by the call to concurrent.futures.as_completed(future_to_url) with indices as the keys, so that you can iterate an index over the length of the dict to read the dict in the same order as the corresponding items in the original list:
with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
# Start the load operations and mark each future with its URL
future_to_url = {
executor.submit(load_url, url, 60): (i, url) for i, url in enumerate(URLS)
}
futures = {}
for future in concurrent.futures.as_completed(future_to_url):
i, url = future_to_url[future]
futures[i] = url, future
for i in range(len(futures)):
url, future = futures[i]
try:
data = future.result()
except Exception as exc:
print('%r generated an exception: %s' % (url, exc))
else:
print('%r page is %d bytes' % (url, len(data)))
Q-1) Why does submit() block the code?
A-1) No submit() method not blocking, it is schedules the callable mytest to be executed as mytest(task_queue) and returns a Future object. Look at below code, you will see that submit() method will not block the main thread
if __name__ == '__main__':
import queue
task_queue = queue.Queue()
task_queue.put(0)
executor = ThreadPoolExecutor()
executor.submit(mytest, task_queue)
task_queue.put(1)
task_queue.put(2)
print("hello")
>> 0
hello
1
2
Or you can do like :
if __name__ == '__main__':
import queue
task_queue = queue.Queue()
task_queue.put(0)
with concurrent.futures.ThreadPoolExecutor() as executor:
executor.submit(mytest, task_queue)
task_queue.put(1)
task_queue.put(2)
You will see that task_queue.put(1) and other will be called immediately
As you see above examples, submit() method not blocking, but when you use with statement with concurrent.futures.ThreadPoolExecutor(), __exit__() method will be called end of the with statement. This __exit__() method will call shutdown(wait=True) method of ThreadPoolExecutor() class. When we look at the doc about shutdown(wait=True) method :
If wait is True then this method will not return until all the pending futures are done executing and the resources associated with the executor have been freed. If wait is False then this method will return immediately and the resources associated with the executor will be freed when all pending futures are done executing. Regardless of the value of wait, the entire Python program will not exit until all pending futures are done executing.
That's why your main thread blocked end of the with statement.
I want to give an answer to your second question, but i am confused with something about main thread exit or not. I will edit this answer later (for second question)
Thread(...).start() creates a new thread. End of story. You can always create a new thread if there's still some memory left in which to create it.
executor.submit(mytest, task_queue) creates a new task and adds it to the task_queue. But adding the task to the queue will force the caller to wait until there is room for it in the queue.
Some time later, when the task eventually reaches the head of the queue, a worker thread will take the task and execute it.