diff --git a/Doc/api/init.tex b/Doc/api/init.tex
index 9225f69..e380bdb 100644
--- a/Doc/api/init.tex
+++ b/Doc/api/init.tex
@@ -696,15 +696,15 @@
 \end{cfuncdesc}
 
 \begin{cfuncdesc}{int}{PyThreadState_SetAsyncExc}{long id, PyObject *exc}
-  Asynchronously raise an exception in a thread. 
+  Asynchronously raise an exception in a thread.
   The \var{id} argument is the thread id of the target thread;
   \var{exc} is the exception object to be raised.
   This function does not steal any references to \var{exc}.
-  To prevent naive misuse, you must write your own C extension 
-  to call this.  Must be called with the GIL held. 
-  Returns the number of thread states modified; if it returns a number 
-  greater than one, you're in trouble, and you should call it again 
-  with \var{exc} set to \constant{NULL} to revert the effect.
+  To prevent naive misuse, you must write your own C extension
+  to call this.  Must be called with the GIL held.
+  Returns the number of thread states modified; this is normally one, but
+  will be zero if the thread id isn't found.  If \var{exc} is
+  \constant{NULL}, the pending exception (if any) for the thread is cleared.
   This raises no exceptions.
   \versionadded{2.3}
 \end{cfuncdesc}
diff --git a/Lib/test/test_threading.py b/Lib/test/test_threading.py
index 79335ea..1d99fa7 100644
--- a/Lib/test/test_threading.py
+++ b/Lib/test/test_threading.py
@@ -131,6 +131,75 @@
                                 threading._DummyThread))
         del threading._active[tid]
 
+    # PyThreadState_SetAsyncExc() is a CPython-only gimmick, not (currently)
+    # exposed at the Python level.  This test relies on ctypes to get at it.
+    def test_PyThreadState_SetAsyncExc(self):
+        try:
+            import ctypes
+        except ImportError:
+            if verbose:
+                print "test_PyThreadState_SetAsyncExc can't import ctypes"
+            return  # can't do anything
+
+        set_async_exc = ctypes.pythonapi.PyThreadState_SetAsyncExc
+
+        class AsyncExc(Exception):
+            pass
+
+        exception = ctypes.py_object(AsyncExc)
+
+        # `worker_started` is set by the thread when it's inside a try/except
+        # block waiting to catch the asynchronously set AsyncExc exception.
+        # `worker_saw_exception` is set by the thread upon catching that
+        # exception.
+        worker_started = threading.Event()
+        worker_saw_exception = threading.Event()
+
+        class Worker(threading.Thread):
+            def run(self):
+                self.id = thread.get_ident()
+                self.finished = False
+
+                try:
+                    while True:
+                        worker_started.set()
+                        time.sleep(0.1)
+                except AsyncExc:
+                    self.finished = True
+                    worker_saw_exception.set()
+
+        t = Worker()
+        if verbose:
+            print "    started worker thread"
+        t.start()
+
+        # Try a thread id that doesn't make sense.
+        if verbose:
+            print "    trying nonsensical thread id"
+        result = set_async_exc(-1, exception)
+        self.assertEqual(result, 0)  # no thread states modified
+
+        # Now raise an exception in the worker thread.
+        if verbose:
+            print "    waiting for worker thread to get started"
+        worker_started.wait()
+        if verbose:
+            print "    verifying worker hasn't exited"
+        self.assert_(not t.finished)
+        if verbose:
+            print "    attempting to raise asynch exception in worker"
+        result = set_async_exc(t.id, exception)
+        self.assertEqual(result, 1) # one thread state modified
+        if verbose:
+            print "    waiting for worker to say it caught the exception"
+        worker_saw_exception.wait(timeout=10)
+        self.assert_(t.finished)
+        if verbose:
+            print "    all OK -- joining worker"
+        if t.finished:
+            t.join()
+        # else the thread is still running, and we have no way to kill it
+
 def test_main():
     test.test_support.run_unittest(ThreadTests)
 
diff --git a/Misc/NEWS b/Misc/NEWS
index db417c9..25d0294 100644
--- a/Misc/NEWS
+++ b/Misc/NEWS
@@ -78,6 +78,15 @@
 - Bug #1530448, ctypes buld failure on Solaris 10 was fixed.
 
 
+C API
+-----
+
+- Bug #1069160.  Internal correctness changes were made to
+  ``PyThreadState_SetAsyncExc()``.  A test case was added, and
+  the documentation was changed to state that the return value
+  is always 1 (normal) or 0 (if the specified thread wasn't found).
+
+
 Mac
 ---
 
@@ -148,7 +157,7 @@
 - os.urandom no longer masks unrelated exceptions like SystemExit or
   KeyboardInterrupt.
 
-- Bug #1525866: Don't copy directory stat times in 
+- Bug #1525866: Don't copy directory stat times in
   shutil.copytree on Windows
 
 - Bug #1002398: The documentation for os.path.sameopenfile now correctly
@@ -281,7 +290,7 @@
 
 - Bug #1527397: PythonLauncher now launches scripts with the working directory
   set to the directory that contains the script instead of the user home
-  directory. That latter was an implementation accident and not what users 
+  directory. That latter was an implementation accident and not what users
   expect.
 
 
diff --git a/Python/pystate.c b/Python/pystate.c
index 3fae85b..f591a59 100644
--- a/Python/pystate.c
+++ b/Python/pystate.c
@@ -342,28 +342,43 @@
 /* Asynchronously raise an exception in a thread.
    Requested by Just van Rossum and Alex Martelli.
    To prevent naive misuse, you must write your own extension
-   to call this.  Must be called with the GIL held.
-   Returns the number of tstates modified; if it returns a number
-   greater than one, you're in trouble, and you should call it again
-   with exc=NULL to revert the effect.  This raises no exceptions. */
+   to call this, or use ctypes.  Must be called with the GIL held.
+   Returns the number of tstates modified (normally 1, but 0 if `id` didn't
+   match any known thread id).  Can be called with exc=NULL to clear an
+   existing async exception.  This raises no exceptions. */
 
 int
 PyThreadState_SetAsyncExc(long id, PyObject *exc) {
 	PyThreadState *tstate = PyThreadState_GET();
 	PyInterpreterState *interp = tstate->interp;
 	PyThreadState *p;
-	int count = 0;
+
+	/* Although the GIL is held, a few C API functions can be called
+	 * without the GIL held, and in particular some that create and
+	 * destroy thread and interpreter states.  Those can mutate the
+	 * list of thread states we're traversing, so to prevent that we lock
+	 * head_mutex for the duration.
+	 */
 	HEAD_LOCK();
 	for (p = interp->tstate_head; p != NULL; p = p->next) {
-		if (p->thread_id != id)
-			continue;
-		Py_CLEAR(p->async_exc);
-		Py_XINCREF(exc);
-		p->async_exc = exc;
-		count += 1;
+		if (p->thread_id == id) {
+			/* Tricky:  we need to decref the current value
+			 * (if any) in p->async_exc, but that can in turn
+			 * allow arbitrary Python code to run, including
+			 * perhaps calls to this function.  To prevent
+			 * deadlock, we need to release head_mutex before
+			 * the decref.
+			 */
+			PyObject *old_exc = p->async_exc;
+			Py_XINCREF(exc);
+			p->async_exc = exc;
+			HEAD_UNLOCK();
+			Py_XDECREF(old_exc);
+			return 1;
+		}
 	}
 	HEAD_UNLOCK();
-	return count;
+	return 0;
 }