gh-132775: Use New Cross-Interpreter Internal C-API Functions

Include/internal/pycore_crossinterp.h

@@ -335,71 +335,29 @@ typedef struct _sharedexception {
 PyAPI_FUNC(PyObject *) _PyXI_ApplyError(_PyXI_error *err);
 
 
-typedef struct xi_session _PyXI_session;
-typedef struct _sharedns _PyXI_namespace;
-
-PyAPI_FUNC(void) _PyXI_FreeNamespace(_PyXI_namespace *ns);
-PyAPI_FUNC(_PyXI_namespace *) _PyXI_NamespaceFromNames(PyObject *names);
-PyAPI_FUNC(int) _PyXI_FillNamespaceFromDict(
-    _PyXI_namespace *ns,
-    PyObject *nsobj,
-    _PyXI_session *session);
-PyAPI_FUNC(int) _PyXI_ApplyNamespace(
-    _PyXI_namespace *ns,
-    PyObject *nsobj,
-    PyObject *dflt);
-
-
 // A cross-interpreter session involves entering an interpreter
-// (_PyXI_Enter()), doing some work with it, and finally exiting
-// that interpreter (_PyXI_Exit()).
+// with _PyXI_Enter(), doing some work with it, and finally exiting
+// that interpreter with _PyXI_Exit().
 //
 // At the boundaries of the session, both entering and exiting,
 // data may be exchanged between the previous interpreter and the
 // target one in a thread-safe way that does not violate the
 // isolation between interpreters.  This includes setting objects
 // in the target's __main__ module on the way in, and capturing
 // uncaught exceptions on the way out.
-struct xi_session {
-    // Once a session has been entered, this is the tstate that was
-    // current before the session.  If it is different from cur_tstate
-    // then we must have switched interpreters.  Either way, this will
-    // be the current tstate once we exit the session.
-    PyThreadState *prev_tstate;
-    // Once a session has been entered, this is the current tstate.
-    // It must be current when the session exits.
-    PyThreadState *init_tstate;
-    // This is true if init_tstate needs cleanup during exit.
-    int own_init_tstate;
-
-    // This is true if, while entering the session, init_thread took
-    // "ownership" of the interpreter's __main__ module.  This means
-    // it is the only thread that is allowed to run code there.
-    // (Caveat: for now, users may still run exec() against the
-    // __main__ module's dict, though that isn't advisable.)
-    int running;
-    // This is a cached reference to the __dict__ of the entered
-    // interpreter's __main__ module.  It is looked up when at the
-    // beginning of the session as a convenience.
-    PyObject *main_ns;
-
-    // This is set if the interpreter is entered and raised an exception
-    // that needs to be handled in some special way during exit.
-    _PyXI_errcode *error_override;
-    // This is set if exit captured an exception to propagate.
-    _PyXI_error *error;
-
-    // -- pre-allocated memory --
-    _PyXI_error _error;
-    _PyXI_errcode _error_override;
-};
+typedef struct xi_session _PyXI_session;
+
+PyAPI_FUNC(_PyXI_session *) _PyXI_NewSession(void);
+PyAPI_FUNC(void) _PyXI_FreeSession(_PyXI_session *);
 
 PyAPI_FUNC(int) _PyXI_Enter(
     _PyXI_session *session,
     PyInterpreterState *interp,
     PyObject *nsupdates);
 PyAPI_FUNC(void) _PyXI_Exit(_PyXI_session *session);
 
+PyAPI_FUNC(PyObject *) _PyXI_GetMainNamespace(_PyXI_session *);
+
 PyAPI_FUNC(PyObject *) _PyXI_ApplyCapturedException(_PyXI_session *session);
 PyAPI_FUNC(int) _PyXI_HasCapturedException(_PyXI_session *session);
 

Include/internal/pycore_pyerrors.h

@@ -94,13 +94,13 @@ extern void _PyErr_Fetch(
     PyObject **value,
     PyObject **traceback);
 
-extern PyObject* _PyErr_GetRaisedException(PyThreadState *tstate);
+PyAPI_FUNC(PyObject*) _PyErr_GetRaisedException(PyThreadState *tstate);
 
 PyAPI_FUNC(int) _PyErr_ExceptionMatches(
     PyThreadState *tstate,
     PyObject *exc);
 
-extern void _PyErr_SetRaisedException(PyThreadState *tstate, PyObject *exc);
+PyAPI_FUNC(void) _PyErr_SetRaisedException(PyThreadState *tstate, PyObject *exc);
 
 extern void _PyErr_Restore(
     PyThreadState *tstate,

Lib/concurrent/futures/interpreter.py

@@ -36,9 +36,6 @@ def __str__(self):
                 """.strip())
 
 
-UNBOUND = 2  # error; this should not happen.
-
-
 class WorkerContext(_thread.WorkerContext):
 
     @classmethod
@@ -47,23 +44,13 @@ def resolve_task(fn, args, kwargs):
             if isinstance(fn, str):
                 # XXX Circle back to this later.
                 raise TypeError('scripts not supported')
-                if args or kwargs:
-                    raise ValueError(f'a script does not take args or kwargs, got {args!r} and {kwargs!r}')
-                data = textwrap.dedent(fn)
-                kind = 'script'
-                # Make sure the script compiles.
-                # Ideally we wouldn't throw away the resulting code
-                # object.  However, there isn't much to be done until
-                # code objects are shareable and/or we do a better job
-                # of supporting code objects in _interpreters.exec().
-                compile(data, '<string>', 'exec')
             else:
                 # Functions defined in the __main__ module can't be pickled,
                 # so they can't be used here.  In the future, we could possibly
                 # borrow from multiprocessing to work around this.
-                data = pickle.dumps((fn, args, kwargs))
-                kind = 'function'
-            return (data, kind)
+                task = (fn, args, kwargs)
+                data = pickle.dumps(task)
+            return data
 
         if initializer is not None:
             try:
@@ -86,24 +73,20 @@ def _capture_exc(cls, resultsid):
         except BaseException as exc:
             # Send the captured exception out on the results queue,
             # but still leave it unhandled for the interpreter to handle.
-            err = pickle.dumps(exc)
-            _interpqueues.put(resultsid, (None, err), 1, UNBOUND)
+            _interpqueues.put(resultsid, (None, exc))
             raise  # re-raise
 
     @classmethod
     def _send_script_result(cls, resultsid):
-        _interpqueues.put(resultsid, (None, None), 0, UNBOUND)
+        _interpqueues.put(resultsid, (None, None))
 
     @classmethod
     def _call(cls, func, args, kwargs, resultsid):
         with cls._capture_exc(resultsid):
             res = func(*args or (), **kwargs or {})
         # Send the result back.
-        try:
-            _interpqueues.put(resultsid, (res, None), 0, UNBOUND)
-        except _interpreters.NotShareableError:
-            res = pickle.dumps(res)
-            _interpqueues.put(resultsid, (res, None), 1, UNBOUND)
+        with cls._capture_exc(resultsid):
+            _interpqueues.put(resultsid, (res, None))
 
     @classmethod
     def _call_pickled(cls, pickled, resultsid):
@@ -134,8 +117,7 @@ def initialize(self):
             _interpreters.incref(self.interpid)
 
             maxsize = 0
-            fmt = 0
-            self.resultsid = _interpqueues.create(maxsize, fmt, UNBOUND)
+            self.resultsid = _interpqueues.create(maxsize)
 
             self._exec(f'from {__name__} import WorkerContext')
 
@@ -166,17 +148,8 @@ def finalize(self):
                 pass
 
     def run(self, task):
-        data, kind = task
-        if kind == 'script':
-            raise NotImplementedError('script kind disabled')
-            script = f"""
-with WorkerContext._capture_exc({self.resultsid}):
-{textwrap.indent(data, '    ')}
-WorkerContext._send_script_result({self.resultsid})"""
-        elif kind == 'function':
-            script = f'WorkerContext._call_pickled({data!r}, {self.resultsid})'
-        else:
-            raise NotImplementedError(kind)
+        data = task
+        script = f'WorkerContext._call_pickled({data!r}, {self.resultsid})'
 
         try:
             self._exec(script)
@@ -199,15 +172,13 @@ def run(self, task):
                 continue
             else:
                 break
-        (res, excdata), pickled, unboundop = obj
+        (res, exc), unboundop = obj
         assert unboundop is None, unboundop
-        if excdata is not None:
+        if exc is not None:
             assert res is None, res
-            assert pickled
             assert exc_wrapper is not None
-            exc = pickle.loads(excdata)
             raise exc from exc_wrapper
-        return pickle.loads(res) if pickled else res
+        return res
 
 
 class BrokenInterpreterPool(_thread.BrokenThreadPool):

Lib/test/support/interpreters/channels.py

@@ -55,15 +55,23 @@ def create(*, unbounditems=UNBOUND):
     """
     unbound = _serialize_unbound(unbounditems)
     unboundop, = unbound
-    cid = _channels.create(unboundop)
-    recv, send = RecvChannel(cid), SendChannel(cid, _unbound=unbound)
+    cid = _channels.create(unboundop, -1)
+    recv, send = RecvChannel(cid), SendChannel(cid)
+    send._set_unbound(unboundop, unbounditems)
     return recv, send
 
 
 def list_all():
     """Return a list of (recv, send) for all open channels."""
-    return [(RecvChannel(cid), SendChannel(cid, _unbound=unbound))
-            for cid, unbound in _channels.list_all()]
+    channels = []
+    for cid, unboundop, _ in _channels.list_all():
+        chan = _, send = RecvChannel(cid), SendChannel(cid)
+        if not hasattr(send, '_unboundop'):
+            send._set_unbound(unboundop)
+        else:
+            assert send._unbound[0] == unboundop
+        channels.append(chan)
+    return channels
 
 
 class _ChannelEnd:
@@ -175,78 +183,95 @@ class SendChannel(_ChannelEnd):
 
     _end = 'send'
 
-    def __new__(cls, cid, *, _unbound=None):
-        if _unbound is None:
-            try:
-                op = _channels.get_channel_defaults(cid)
-                _unbound = (op,)
-            except ChannelNotFoundError:
-                _unbound = _serialize_unbound(UNBOUND)
-        self = super().__new__(cls, cid)
-        self._unbound = _unbound
-        return self
+#    def __new__(cls, cid, *, _unbound=None):
+#        if _unbound is None:
+#            try:
+#                op = _channels.get_channel_defaults(cid)
+#                _unbound = (op,)
+#            except ChannelNotFoundError:
+#                _unbound = _serialize_unbound(UNBOUND)
+#        self = super().__new__(cls, cid)
+#        self._unbound = _unbound
+#        return self
+
+    def _set_unbound(self, op, items=None):
+        assert not hasattr(self, '_unbound')
+        if items is None:
+            items = _resolve_unbound(op)
+        unbound = (op, items)
+        self._unbound = unbound
+        return unbound
+
+    @property
+    def unbounditems(self):
+        try:
+            _, items = self._unbound
+        except AttributeError:
+            op, _ = _channels.get_queue_defaults(self._id)
+            _, items = self._set_unbound(op)
+        return items
 
     @property
     def is_closed(self):
         info = self._info
         return info.closed or info.closing
 
     def send(self, obj, timeout=None, *,
-             unbound=None,
+             unbounditems=None,
              ):
         """Send the object (i.e. its data) to the channel's receiving end.
 
         This blocks until the object is received.
         """
-        if unbound is None:
-            unboundop, = self._unbound
+        if unbounditems is None:
+            unboundop = -1
         else:
-            unboundop, = _serialize_unbound(unbound)
+            unboundop, = _serialize_unbound(unbounditems)
         _channels.send(self._id, obj, unboundop, timeout=timeout, blocking=True)
 
     def send_nowait(self, obj, *,
-                    unbound=None,
+                    unbounditems=None,
                     ):
         """Send the object to the channel's receiving end.
 
         If the object is immediately received then return True
         (else False).  Otherwise this is the same as send().
         """
-        if unbound is None:
-            unboundop, = self._unbound
+        if unbounditems is None:
+            unboundop = -1
         else:
-            unboundop, = _serialize_unbound(unbound)
+            unboundop, = _serialize_unbound(unbounditems)
         # XXX Note that at the moment channel_send() only ever returns
         # None.  This should be fixed when channel_send_wait() is added.
         # See bpo-32604 and gh-19829.
         return _channels.send(self._id, obj, unboundop, blocking=False)
 
     def send_buffer(self, obj, timeout=None, *,
-                    unbound=None,
+                    unbounditems=None,
                     ):
         """Send the object's buffer to the channel's receiving end.
 
         This blocks until the object is received.
         """
-        if unbound is None:
-            unboundop, = self._unbound
+        if unbounditems is None:
+            unboundop = -1
         else:
-            unboundop, = _serialize_unbound(unbound)
+            unboundop, = _serialize_unbound(unbounditems)
         _channels.send_buffer(self._id, obj, unboundop,
                               timeout=timeout, blocking=True)
 
     def send_buffer_nowait(self, obj, *,
-                           unbound=None,
+                           unbounditems=None,
                            ):
         """Send the object's buffer to the channel's receiving end.
 
         If the object is immediately received then return True
         (else False).  Otherwise this is the same as send().
         """
-        if unbound is None:
-            unboundop, = self._unbound
+        if unbounditems is None:
+            unboundop = -1
         else:
-            unboundop, = _serialize_unbound(unbound)
+            unboundop, = _serialize_unbound(unbounditems)
         return _channels.send_buffer(self._id, obj, unboundop, blocking=False)
 
     def close(self):