Merged revisions 68425,68461,68498 via svnmerge from

svn+ssh://pythondev@svn.python.org/python/trunk

........
  r68425 | benjamin.peterson | 2009-01-08 20:56:32 -0600 (Thu, 08 Jan 2009) | 1 line

  fix markup
........
  r68461 | kristjan.jonsson | 2009-01-09 15:35:16 -0600 (Fri, 09 Jan 2009) | 2 lines

  Issue 4293:  Make Py_AddPendingCall() thread safe
  Add test cases and documentation
........
  r68498 | benjamin.peterson | 2009-01-10 13:08:49 -0600 (Sat, 10 Jan 2009) | 1 line

  fix encoding
........

Doc/c-api/init.rst

@@ -765,6 +765,50 @@ created.
    :cfunc:`PyGILState_Release` on the same thread.
 
 
+
+Asynchronous Notifications
+==========================
+
+A mechanism is provided to make asynchronous notifications to the the main
+interpreter thread.  These notifications take the form of a function
+pointer and a void argument.
+
+.. index:: single: setcheckinterval() (in module sys)
+
+Every check interval, when the interpreter lock is released and reacquired,
+python will also call any such provided functions.  This can be used for
+example by asynchronous IO handlers.  The notification can be scheduled
+from a worker thread and the actual call than made at the earliest
+convenience by the main thread where it has possession of the global
+interpreter lock and can perform any Python API calls.
+
+.. cfunction:: void Py_AddPendingCall( int (*func)(void *), void *arg) )
+
+   .. index:: single: Py_AddPendingCall()
+
+   Post a notification to the Python main thread.  If successful,
+   \*:attr`func` will be called with the argument :attr:`arg` at the earliest
+   convenience.  \*:attr:`func` will be called having the global interpreter
+   lock held and can thus use the full Python API and can take any
+   action such as setting object attributes to signal IO completion.
+   It must return 0 on success, or -1 signalling an exception.
+   The notification function won't be interrupted to perform another
+   asynchronous notification recursively,
+   but it can still be interrupted to switch threads if the interpreter
+   lock is released, for example, if it calls back into python code.
+
+   This function returns 0 on success in which case the notification has been
+   scheduled.  Otherwise, for example if the notification buffer is full,
+   it returns -1 without setting any exception.
+
+   This function can be called on any thread, be it a Python thread or
+   some other system thread.  If it is a Python thread, it doesen't matter if
+   it holds the global interpreter lock or not.
+
+   .. versionadded:: 2.7
+
+
+
 .. _profiling:
 
 Profiling and Tracing

Doc/whatsnew/2.7.rst

@@ -60,6 +60,11 @@ No release schedule has been decided yet for 2.7.
 .. ========================================================================
 
 
+Kristján Valur Jónsson, issue 4293
+Py_AddPendingCall is now thread safe.  This allows any worker thread
+to submit notifications to the python main thread.  This is particularly
+useful for asynchronous IO operations.
+
 
 Other Language Changes
 ======================
@@ -121,11 +126,10 @@ changes, or look through the Subversion logs for all the details.
   (Contributed by Gregory P. Smith.)
 
 * It is not mandatory anymore to store clear text passwords in the
-  :file:`.pypirc` file when registering and uploading packages to PyPI. As
-  long as the username is present in that file, the :mod:`distutils` package
-  will prompt for the password if not present.
-  (Added by tarek, with the initial contribution of Nathan Van Gheem;
-   :issue:`4394`.)
+  :file:`.pypirc` file when registering and uploading packages to PyPI. As long
+  as the username is present in that file, the :mod:`distutils` package will
+  prompt for the password if not present.  (Added by tarek, with the initial
+  contribution of Nathan Van Gheem; :issue:`4394`.)
 
 .. ======================================================================
 .. whole new modules get described in subsections here

Lib/test/test_capi.py

@@ -2,10 +2,14 @@
 # these are all functions _testcapi exports whose name begins with 'test_'.
 
 import sys
+import time
+import random
 import unittest
+import threading
 from test import support
 import _testcapi
 
+
 def testfunction(self):
     """some doc"""
     return self
@@ -28,6 +32,67 @@ class CAPITest(unittest.TestCase):
         self.assertRaises(AttributeError, setattr, inst.testfunction, "attribute", "test")
 
 
+class TestPendingCalls(unittest.TestCase):
+
+    def pendingcalls_submit(self, l, n):
+        def callback():
+            #this function can be interrupted by thread switching so let's
+            #use an atomic operation
+            l.append(None)
+
+        for i in range(n):
+            time.sleep(random.random()*0.02) #0.01 secs on average
+            #try submitting callback until successful.
+            #rely on regular interrupt to flush queue if we are
+            #unsuccessful.
+            while True:
+                if _testcapi._pending_threadfunc(callback):
+                    break;
+
+    def pendingcalls_wait(self, l, n):
+        #now, stick around until l[0] has grown to 10
+        count = 0;
+        while len(l) != n:
+            #this busy loop is where we expect to be interrupted to
+            #run our callbacks.  Note that callbacks are only run on the
+            #main thread
+            if False and test_support.verbose:
+                print("(%i)"%(len(l),),)
+            for i in range(1000):
+                a = i*i
+            count += 1
+            self.failUnless(count < 10000,
+                "timeout waiting for %i callbacks, got %i"%(n, len(l)))
+        if False and test_support.verbose:
+            print("(%i)"%(len(l),))
+
+    def test_pendingcalls_threaded(self):
+        l = []
+
+        #do every callback on a separate thread
+        n = 32
+        threads = []
+        for i in range(n):
+            t = threading.Thread(target=self.pendingcalls_submit, args = (l, 1))
+            t.start()
+            threads.append(t)
+
+        self.pendingcalls_wait(l, n)
+
+        for t in threads:
+            t.join()
+
+    def test_pendingcalls_non_threaded(self):
+        #again, just using the main thread, likely they will all be dispathced at
+        #once.  It is ok to ask for too many, because we loop until we find a slot.
+        #the loop can be interrupted to dispatch.
+        #there are only 32 dispatch slots, so we go for twice that!
+        l = []
+        n = 64
+        self.pendingcalls_submit(l, n)
+        self.pendingcalls_wait(l, n)
+
+
 def test_main():
     support.run_unittest(CAPITest)
 
@@ -71,6 +136,8 @@ def test_main():
         t.start()
         t.join()
 
+    support.run_unittest(TestPendingCalls)
+
 
 if __name__ == "__main__":
     test_main()

Modules/_testcapimodule.c

@@ -919,6 +919,43 @@ test_thread_state(PyObject *self, PyObject *args)
 		return NULL;
 	Py_RETURN_NONE;
 }
+
+/* test Py_AddPendingCalls using threads */
+static int _pending_callback(void *arg)
+{
+	/* we assume the argument is callable object to which we own a reference */
+	PyObject *callable = (PyObject *)arg;
+	PyObject *r = PyObject_CallObject(callable, NULL);
+	Py_DECREF(callable);
+	Py_XDECREF(r);
+	return r != NULL ? 0 : -1;
+}
+
+/* The following requests n callbacks to _pending_callback.  It can be
+ * run from any python thread.
+ */
+PyObject *pending_threadfunc(PyObject *self, PyObject *arg)
+{
+	PyObject *callable;
+	int r;
+	if (PyArg_ParseTuple(arg, "O", &callable) == 0)
+		return NULL;
+
+	/* create the reference for the callbackwhile we hold the lock */
+	Py_INCREF(callable);
+
+	Py_BEGIN_ALLOW_THREADS
+	r = Py_AddPendingCall(&_pending_callback, callable);
+	Py_END_ALLOW_THREADS
+
+	if (r<0) {
+		Py_DECREF(callable); /* unsuccessful add, destroy the extra reference */
+		Py_INCREF(Py_False);
+		return Py_False;
+	}
+	Py_INCREF(Py_True);
+	return Py_True;
+}
 #endif
 
 /* Some tests of PyUnicode_FromFormat().  This needs more tests. */
@@ -1171,6 +1208,7 @@ static PyMethodDef TestMethods[] = {
 	{"test_Z_code",		(PyCFunction)test_Z_code,	 METH_NOARGS},
 #ifdef WITH_THREAD
 	{"_test_thread_state",  test_thread_state, 		 METH_VARARGS},
+	{"_pending_threadfunc",	pending_threadfunc,		 METH_VARARGS},
 #endif
 #ifdef HAVE_GETTIMEOFDAY
 	{"profile_int",		profile_int,			METH_NOARGS},