Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions Doc/library/concurrent.futures.rst
Original file line number Diff line number Diff line change
Expand Up @@ -190,6 +190,11 @@
pending jobs will raise a :exc:`~concurrent.futures.thread.BrokenThreadPool`,
as well as any attempt to submit more jobs to the pool.

Each callable submitted with :meth:`submit` runs in a copy of the

Check warning on line 193 in Doc/library/concurrent.futures.rst

View workflow job for this annotation

GitHub Actions / Docs / Docs

py:meth reference target not found: submit [ref.meth]
:mod:`contextvars` context of the thread that submits it. Changes to
context-variable bindings made by a task do not affect the submitting
thread's context or the context used by later tasks.

.. versionchanged:: 3.5
If *max_workers* is ``None`` or
not given, it will default to the number of processors on the machine,
Expand Down Expand Up @@ -219,6 +224,9 @@
Default value of *max_workers* is changed to
``min(32, (os.process_cpu_count() or 1) + 4)``.

.. versionchanged:: 3.16
Tasks run in a copy of the submitting thread's context.


.. _threadpoolexecutor-example:

Expand Down
7 changes: 4 additions & 3 deletions Lib/concurrent/futures/thread.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
__author__ = 'Brian Quinlan (brian@sweetapp.com)'

from concurrent.futures import _base
import contextvars
import itertools
import queue
import threading
Expand Down Expand Up @@ -54,7 +55,7 @@ def prepare(cls, initializer, initargs):
def create_context():
return cls(initializer, initargs)
def resolve_task(fn, args, kwargs):
return (fn, args, kwargs)
return (contextvars.copy_context(), fn, args, kwargs)
return create_context, resolve_task

def __init__(self, initializer, initargs):
Expand All @@ -69,8 +70,8 @@ def finalize(self):
pass

def run(self, task):
fn, args, kwargs = task
return fn(*args, **kwargs)
context, fn, args, kwargs = task
return context.run(fn, *args, **kwargs)


class _WorkItem:
Expand Down
29 changes: 29 additions & 0 deletions Lib/test/test_concurrent_futures/test_thread_pool.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
import contextlib
import contextvars
import multiprocessing as mp
import multiprocessing.process
import multiprocessing.util
Expand All @@ -14,6 +15,34 @@


class ThreadPoolExecutorTest(ThreadPoolMixin, ExecutorTest, BaseTestCase):
def test_contextvars_are_copied_for_each_task(self):
value = contextvars.ContextVar("value", default="default")

def read_value():
return value.get()

def change_value():
value.set("worker")

with self.executor_type(max_workers=1) as executor:
token = value.set("first")
try:
self.assertEqual(executor.submit(read_value).result(), "first")
finally:
value.reset(token)

self.assertEqual(executor.submit(read_value).result(), "default")

token = value.set("third")
try:
self.assertEqual(executor.submit(read_value).result(), "third")
finally:
value.reset(token)

# Changes made by one task must not leak into the next task.
executor.submit(change_value).result()
self.assertEqual(executor.submit(read_value).result(), "default")

def test_map_submits_without_iteration(self):
"""Tests verifying issue 11777."""
finished = []
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
Fix :class:`concurrent.futures.ThreadPoolExecutor` to run each submitted callable in a copy of the submitting thread's :mod:`contextvars` context.
Loading