Drop the "shared" parameter.

Author: ericsnowcurrently

Doc/library/concurrent.futures.rst

@@ -287,7 +287,7 @@ efficient alternative is to serialize with :mod:`pickle` and then send
 the bytes over a shared :mod:`socket <socket>` or
 :func:`pipe <os.pipe>`.
 
-.. class:: InterpreterPoolExecutor(max_workers=None, thread_name_prefix='', initializer=None, initargs=(), shared=None)
+.. class:: InterpreterPoolExecutor(max_workers=None, thread_name_prefix='', initializer=None, initargs=())
 
    A :class:`ThreadPoolExecutor` subclass that executes calls asynchronously
    using a pool of at most *max_workers* threads.  Each thread runs
@@ -308,13 +308,6 @@ the bytes over a shared :mod:`socket <socket>` or
       The executor may replace uncaught exceptions from *initializer*
       with :class:`~concurrent.futures.interpreter.ExecutionFailed`.
 
-   The optional *shared* argument is a :class:`dict` of objects that all
-   interpreters in the pool share.  The *shared* items are added to each
-   interpreter's ``__main__`` module.  Not all objects are shareable.
-   Shareable objects include the builtin singletons, :class:`str`
-   and :class:`bytes`, and :class:`memoryview`.  See :pep:`734`
-   for more info.
-
    Other caveats from parent :class:`ThreadPoolExecutor` apply here.
 
 :meth:`~Executor.submit` and :meth:`~Executor.map` work like normal,

Lib/concurrent/futures/interpreter.py

@@ -39,7 +39,7 @@ def __str__(self):
 class WorkerContext(_thread.WorkerContext):
 
     @classmethod
-    def prepare(cls, initializer, initargs, shared):
+    def prepare(cls, initializer, initargs):
         def resolve_task(fn, args, kwargs):
             if isinstance(fn, str):
                 # XXX Circle back to this later.
@@ -58,12 +58,11 @@ def resolve_task(fn, args, kwargs):
         else:
             initdata = None
         def create_context():
-            return cls(initdata, shared)
+            return cls(initdata)
         return create_context, resolve_task
 
-    def __init__(self, initdata, shared=None):
+    def __init__(self, initdata):
         self.initdata = initdata
-        self.shared = dict(shared) if shared else None
         self.interpid = None
         self.resultsid = None
 
@@ -131,10 +130,6 @@ def initialize(self):
             maxsize = 0
             self.resultsid = _interpqueues.create(maxsize)
 
-            if self.shared:
-                _interpreters.set___main___attrs(
-                                    self.interpid, self.shared, restrict=True)
-
             if self.initdata:
                 self.run(self.initdata)
         except BaseException:
@@ -180,11 +175,11 @@ class InterpreterPoolExecutor(_thread.ThreadPoolExecutor):
     BROKEN = BrokenInterpreterPool
 
     @classmethod
-    def prepare_context(cls, initializer, initargs, shared):
-        return WorkerContext.prepare(initializer, initargs, shared)
+    def prepare_context(cls, initializer, initargs):
+        return WorkerContext.prepare(initializer, initargs)
 
     def __init__(self, max_workers=None, thread_name_prefix='',
-                 initializer=None, initargs=(), shared=None):
+                 initializer=None, initargs=()):
         """Initializes a new InterpreterPoolExecutor instance.
 
         Args:
@@ -194,8 +189,6 @@ def __init__(self, max_workers=None, thread_name_prefix='',
             initializer: A callable or script used to initialize
                 each worker interpreter.
             initargs: A tuple of arguments to pass to the initializer.
-            shared: A mapping of shareabled objects to be inserted into
-                each worker interpreter.
         """
         super().__init__(max_workers, thread_name_prefix,
-                         initializer, initargs, shared=shared)
+                         initializer, initargs)

Lib/test/test_concurrent_futures/test_interpreter_pool.py

@@ -224,24 +224,6 @@ def initializer(self):
         with self.assertRaises(BrokenInterpreterPool):
             fut.result()
 
-    def test_init_shared(self):
-        msg = b'eggs'
-        r, w = self.pipe()
-        script = f"""if True:
-            import os
-            if __name__ != '__main__':
-                import __main__
-                spam = __main__.spam
-            os.write({w}, spam + b'\\0')
-            """
-
-        executor = self.executor_type(shared={'spam': msg})
-        fut = executor.submit(exec, script)
-        fut.result()
-        after = read_msg(r)
-
-        self.assertEqual(after, msg)
-
     @unittest.expectedFailure
     def test_init_exception_in_script(self):
         executor = self.executor_type(initializer='raise Exception("spam")')
@@ -363,16 +345,39 @@ def test_submit_exception_in_func(self):
 
     def test_saturation(self):
         blocker = queues.create()
-        executor = self.executor_type(4, shared=dict(blocker=blocker))
+        executor = self.executor_type(4)
 
         for i in range(15 * executor._max_workers):
-            executor.submit(exec, 'import __main__; __main__.blocker.get()')
-            #executor.submit('blocker.get()')
+            executor.submit(blocker.get)
         self.assertEqual(len(executor._threads), executor._max_workers)
         for i in range(15 * executor._max_workers):
             blocker.put_nowait(None)
         executor.shutdown(wait=True)
 
+    def test_blocking(self):
+        ready = queues.create()
+        blocker = queues.create()
+
+        def run(ready, blocker):
+            ready.put(None)
+            blocker.get()  # blocking
+
+        numtasks = 10
+        futures = []
+        executor = self.executor_type()
+        try:
+            for i in range(numtasks):
+                fut = executor.submit(run, ready, blocker)
+                futures.append(fut)
+            # Wait for them all to be ready.
+            for i in range(numtasks):
+                ready.get()  # blocking
+            # Unblock the workers.
+            for i in range(numtasks):
+                blocker.put_nowait(None)
+        finally:
+            executor.shutdown(wait=True)
+
     @support.requires_gil_enabled("gh-117344: test is flaky without the GIL")
     def test_idle_thread_reuse(self):
         executor = self.executor_type()