:mod:`_dummy_thread` module is available. It duplicates this module's interface
and can be used as a drop-in replacement.
-It defines the following constant and functions:
+It defines the following constants and functions:
.. exception:: error
Availability: Windows, systems with POSIX threads.
+.. data:: TIMEOUT_MAX
+
+ The maximum value allowed for the *timeout* parameter of
+ :meth:`Lock.acquire`. Specifiying a timeout greater than this value will
+ raise an :exc:`OverflowError`.
+
+
Lock objects have the following methods:
-.. method:: lock.acquire([waitflag])
+.. method:: lock.acquire(waitflag=1, timeout=-1)
- Without the optional argument, this method acquires the lock unconditionally, if
+ Without any optional argument, this method acquires the lock unconditionally, if
necessary waiting until it is released by another thread (only one thread at a
- time can acquire a lock --- that's their reason for existence). If the integer
- *waitflag* argument is present, the action depends on its value: if it is zero,
- the lock is only acquired if it can be acquired immediately without waiting,
- while if it is nonzero, the lock is acquired unconditionally as before. The
- return value is ``True`` if the lock is acquired successfully, ``False`` if not.
+ time can acquire a lock --- that's their reason for existence).
+ If the integer *waitflag* argument is present, the action depends on its
+ value: if it is zero, the lock is only acquired if it can be acquired
+ immediately without waiting, while if it is nonzero, the lock is acquired
+ unconditionally as above.
+
+ If the floating-point *timeout* argument is present and positive, it
+ specifies the maximum wait time in seconds before returning. A negative
+ *timeout* argument specifies an unbounded wait. You cannot specify
+ a *timeout* if *waitflag* is zero.
+
+ The return value is ``True`` if the lock is acquired successfully,
+ ``False`` if not.
.. method:: lock.release()
Availability: Windows, systems with POSIX threads.
+This module also defines the following constant:
+
+.. data:: TIMEOUT_MAX
+
+ The maximum value allowed for the *timeout* parameter of blocking functions
+ (:meth:`Lock.acquire`, :meth:`RLock.acquire`, :meth:`Condition.wait`, etc.).
+ Specifiying a timeout greater than this value will raise an
+ :exc:`OverflowError`.
+
+
Detailed interfaces for the objects are documented below.
The design of this module is loosely based on Java's threading model. However,
All methods are executed atomically.
-.. method:: Lock.acquire(blocking=True)
+.. method:: Lock.acquire(blocking=True, timeout=-1)
Acquire a lock, blocking or non-blocking.
without an argument would block, return false immediately; otherwise, do the
same thing as when called without arguments, and return true.
+ When invoked with the floating-point *timeout* argument set to a positive
+ value, block for at most the number of seconds specified by *timeout*
+ and as long as the lock cannot be acquired. A negative *timeout* argument
+ specifies an unbounded wait. It is forbidden to specify a *timeout*
+ when *blocking* is false.
+
+ The return value is ``True`` if the lock is acquired successfully,
+ ``False`` if not (for example if the *timeout* expired).
+
.. method:: Lock.release()
:meth:`acquire` to proceed.
-.. method:: RLock.acquire(blocking=True)
+.. method:: RLock.acquire(blocking=True, timeout=-1)
Acquire a lock, blocking or non-blocking.
without an argument would block, return false immediately; otherwise, do the
same thing as when called without arguments, and return true.
+ When invoked with the floating-point *timeout* argument set to a positive
+ value, block for at most the number of seconds specified by *timeout*
+ and as long as the lock cannot be acquired. Return true if the lock has
+ been acquired, false if the timeout has elapsed.
+
.. method:: RLock.release()
PyAPI_FUNC(int) PyThread_acquire_lock(PyThread_type_lock, int);
#define WAIT_LOCK 1
#define NOWAIT_LOCK 0
+
+/* PY_TIMEOUT_T is the integral type used to specify timeouts when waiting
+ on a lock (see PyThread_acquire_lock_timed() below).
+ PY_TIMEOUT_MAX is the highest usable value (in microseconds) of that
+ type, and depends on the system threading API.
+
+ NOTE: this isn't the same value as `_thread.TIMEOUT_MAX`. The _thread
+ module exposes a higher-level API, with timeouts expressed in seconds
+ and floating-point numbers allowed.
+*/
+#if defined(HAVE_LONG_LONG)
+#define PY_TIMEOUT_T PY_LONG_LONG
+#define PY_TIMEOUT_MAX PY_LLONG_MAX
+#else
+#define PY_TIMEOUT_T long
+#define PY_TIMEOUT_MAX LONG_MAX
+#endif
+
+/* In the NT API, the timeout is a DWORD and is expressed in milliseconds */
+#if defined (NT_THREADS)
+#if (0xFFFFFFFFLL * 1000 < PY_TIMEOUT_MAX)
+#undef PY_TIMEOUT_MAX
+#define PY_TIMEOUT_MAX (0xFFFFFFFFLL * 1000)
+#endif
+#endif
+
+/* If microseconds == 0, the call is non-blocking: it returns immediately
+ even when the lock can't be acquired.
+ If microseconds > 0, the call waits up to the specified duration.
+ If microseconds < 0, the call waits until success (or abnormal failure)
+
+ microseconds must be less than PY_TIMEOUT_MAX. Behaviour otherwise is
+ undefined. */
+PyAPI_FUNC(int) PyThread_acquire_lock_timed(PyThread_type_lock,
+ PY_TIMEOUT_T microseconds);
PyAPI_FUNC(void) PyThread_release_lock(PyThread_type_lock);
PyAPI_FUNC(size_t) PyThread_get_stacksize(void);
'interrupt_main', 'LockType']
import traceback as _traceback
+import time
+
+# A dummy value
+TIMEOUT_MAX = 2**31
class error(Exception):
"""Dummy implementation of _thread.error."""
def __init__(self):
self.locked_status = False
- def acquire(self, waitflag=None):
+ def acquire(self, waitflag=None, timeout=-1):
"""Dummy implementation of acquire().
For blocking calls, self.locked_status is automatically set to
self.locked_status = True
return True
else:
+ if timeout > 0:
+ time.sleep(timeout)
return False
__enter__ = acquire
p.terminate()
debug('joining task handler')
- task_handler.join(1e100)
+ task_handler.join()
debug('joining result handler')
- result_handler.join(1e100)
+ task_handler.join()
if pool and hasattr(pool[0], 'terminate'):
debug('joining pool workers')
import sys
import time
-from _thread import start_new_thread, get_ident
+from _thread import start_new_thread, get_ident, TIMEOUT_MAX
import threading
import unittest
support.threading_cleanup(*self._threads)
support.reap_children()
+ def assertTimeout(self, actual, expected):
+ # The waiting and/or time.time() can be imprecise, which
+ # is why comparing to the expected value would sometimes fail
+ # (especially under Windows).
+ self.assertGreaterEqual(actual, expected * 0.6)
+ # Test nothing insane happened
+ self.assertLess(actual, expected * 10.0)
+
class BaseLockTests(BaseTestCase):
"""
Bunch(f, 15).wait_for_finished()
self.assertEqual(n, len(threading.enumerate()))
+ def test_timeout(self):
+ lock = self.locktype()
+ # Can't set timeout if not blocking
+ self.assertRaises(ValueError, lock.acquire, 0, 1)
+ # Invalid timeout values
+ self.assertRaises(ValueError, lock.acquire, timeout=-100)
+ self.assertRaises(OverflowError, lock.acquire, timeout=1e100)
+ self.assertRaises(OverflowError, lock.acquire, timeout=TIMEOUT_MAX + 1)
+ # TIMEOUT_MAX is ok
+ lock.acquire(timeout=TIMEOUT_MAX)
+ lock.release()
+ t1 = time.time()
+ self.assertTrue(lock.acquire(timeout=5))
+ t2 = time.time()
+ # Just a sanity test that it didn't actually wait for the timeout.
+ self.assertLess(t2 - t1, 5)
+ results = []
+ def f():
+ t1 = time.time()
+ results.append(lock.acquire(timeout=0.5))
+ t2 = time.time()
+ results.append(t2 - t1)
+ Bunch(f, 1).wait_for_finished()
+ self.assertFalse(results[0])
+ self.assertTimeout(results[1], 0.5)
+
class LockTests(BaseLockTests):
"""
def f():
results1.append(evt.wait(0.0))
t1 = time.time()
- r = evt.wait(0.2)
+ r = evt.wait(0.5)
t2 = time.time()
results2.append((r, t2 - t1))
Bunch(f, N).wait_for_finished()
self.assertEqual(results1, [False] * N)
for r, dt in results2:
self.assertFalse(r)
- self.assertTrue(dt >= 0.2, dt)
+ self.assertTimeout(dt, 0.5)
# The event is set
results1 = []
results2 = []
def f():
cond.acquire()
t1 = time.time()
- cond.wait(0.2)
+ cond.wait(0.5)
t2 = time.time()
cond.release()
results.append(t2 - t1)
Bunch(f, N).wait_for_finished()
self.assertEqual(len(results), 5)
for dt in results:
- self.assertTrue(dt >= 0.2, dt)
+ self.assertTimeout(dt, 0.5)
class BaseSemaphoreTests(BaseTestCase):
_CRLock = _thread.RLock
except AttributeError:
_CRLock = None
+TIMEOUT_MAX = _thread.TIMEOUT_MAX
del _thread
return "<%s owner=%r count=%d>" % (
self.__class__.__name__, owner, self._count)
- def acquire(self, blocking=True):
+ def acquire(self, blocking=True, timeout=-1):
me = _get_ident()
if self._owner == me:
self._count = self._count + 1
if __debug__:
self._note("%s.acquire(%s): recursive success", self, blocking)
return 1
- rc = self._block.acquire(blocking)
+ rc = self._block.acquire(blocking, timeout)
if rc:
self._owner = me
self._count = 1
if __debug__:
self._note("%s.wait(): got it", self)
else:
- # Balancing act: We can't afford a pure busy loop, so we
- # have to sleep; but if we sleep the whole timeout time,
- # we'll be unresponsive. The scheme here sleeps very
- # little at first, longer as time goes on, but never longer
- # than 20 times per second (or the timeout time remaining).
- endtime = _time() + timeout
- delay = 0.0005 # 500 us -> initial delay of 1 ms
- while True:
- gotit = waiter.acquire(0)
- if gotit:
- break
- remaining = endtime - _time()
- if remaining <= 0:
- break
- delay = min(delay * 2, remaining, .05)
- _sleep(delay)
+ if timeout > 0:
+ gotit = waiter.acquire(True, timeout)
+ else:
+ gotit = waiter.acquire(False)
if not gotit:
if __debug__:
self._note("%s.wait(%s): timed out", self, timeout)
Library
-------
+- Issue #7316: the acquire() method of lock objects in the :mod:`threading`
+ module now takes an optional timeout argument in seconds. Timeout support
+ relies on the system threading library, so as to avoid a semi-busy wait
+ loop.
+
- Issue #8383: pickle and pickletools use surrogatepass error handler when
encoding unicode as utf8 to support lone surrogates and stay compatible with
Python 2.x and 3.0
}
static PyObject *
-lock_PyThread_acquire_lock(lockobject *self, PyObject *args)
+lock_PyThread_acquire_lock(lockobject *self, PyObject *args, PyObject *kwds)
{
- int i = 1;
+ char *kwlist[] = {"blocking", "timeout", NULL};
+ int blocking = 1;
+ double timeout = -1;
+ PY_TIMEOUT_T microseconds;
+ int r;
- if (!PyArg_ParseTuple(args, "|i:acquire", &i))
+ if (!PyArg_ParseTupleAndKeywords(args, kwds, "|id:acquire", kwlist,
+ &blocking, &timeout))
return NULL;
+ if (!blocking && timeout != -1) {
+ PyErr_SetString(PyExc_ValueError, "can't specify a timeout "
+ "for a non-blocking call");
+ return NULL;
+ }
+ if (timeout < 0 && timeout != -1) {
+ PyErr_SetString(PyExc_ValueError, "timeout value must be "
+ "strictly positive");
+ return NULL;
+ }
+ if (!blocking)
+ microseconds = 0;
+ else if (timeout == -1)
+ microseconds = -1;
+ else {
+ timeout *= 1e6;
+ if (timeout >= (double) PY_TIMEOUT_MAX) {
+ PyErr_SetString(PyExc_OverflowError,
+ "timeout value is too large");
+ return NULL;
+ }
+ microseconds = (PY_TIMEOUT_T) timeout;
+ }
+
Py_BEGIN_ALLOW_THREADS
- i = PyThread_acquire_lock(self->lock_lock, i);
+ r = PyThread_acquire_lock_timed(self->lock_lock, microseconds);
Py_END_ALLOW_THREADS
- return PyBool_FromLong((long)i);
+ return PyBool_FromLong(r);
}
PyDoc_STRVAR(acquire_doc,
static PyMethodDef lock_methods[] = {
{"acquire_lock", (PyCFunction)lock_PyThread_acquire_lock,
- METH_VARARGS, acquire_doc},
+ METH_VARARGS | METH_KEYWORDS, acquire_doc},
{"acquire", (PyCFunction)lock_PyThread_acquire_lock,
- METH_VARARGS, acquire_doc},
+ METH_VARARGS | METH_KEYWORDS, acquire_doc},
{"release_lock", (PyCFunction)lock_PyThread_release_lock,
METH_NOARGS, release_doc},
{"release", (PyCFunction)lock_PyThread_release_lock,
{"locked", (PyCFunction)lock_locked_lock,
METH_NOARGS, locked_doc},
{"__enter__", (PyCFunction)lock_PyThread_acquire_lock,
- METH_VARARGS, acquire_doc},
+ METH_VARARGS | METH_KEYWORDS, acquire_doc},
{"__exit__", (PyCFunction)lock_PyThread_release_lock,
METH_VARARGS, release_doc},
{NULL, NULL} /* sentinel */
static PyObject *
rlock_acquire(rlockobject *self, PyObject *args, PyObject *kwds)
{
- char *kwlist[] = {"blocking", NULL};
+ char *kwlist[] = {"blocking", "timeout", NULL};
int blocking = 1;
+ double timeout = -1;
+ PY_TIMEOUT_T microseconds;
long tid;
int r = 1;
- if (!PyArg_ParseTupleAndKeywords(args, kwds, "|i:acquire", kwlist,
- &blocking))
+ if (!PyArg_ParseTupleAndKeywords(args, kwds, "|id:acquire", kwlist,
+ &blocking, &timeout))
return NULL;
+ if (!blocking && timeout != -1) {
+ PyErr_SetString(PyExc_ValueError, "can't specify a timeout "
+ "for a non-blocking call");
+ return NULL;
+ }
+ if (timeout < 0 && timeout != -1) {
+ PyErr_SetString(PyExc_ValueError, "timeout value must be "
+ "strictly positive");
+ return NULL;
+ }
+ if (!blocking)
+ microseconds = 0;
+ else if (timeout == -1)
+ microseconds = -1;
+ else {
+ timeout *= 1e6;
+ if (timeout >= (double) PY_TIMEOUT_MAX) {
+ PyErr_SetString(PyExc_OverflowError,
+ "timeout value is too large");
+ return NULL;
+ }
+ microseconds = (PY_TIMEOUT_T) timeout;
+ }
+
tid = PyThread_get_thread_ident();
if (self->rlock_count > 0 && tid == self->rlock_owner) {
unsigned long count = self->rlock_count + 1;
if (self->rlock_count > 0 ||
!PyThread_acquire_lock(self->rlock_lock, 0)) {
- if (!blocking) {
+ if (microseconds == 0) {
Py_RETURN_FALSE;
}
Py_BEGIN_ALLOW_THREADS
- r = PyThread_acquire_lock(self->rlock_lock, blocking);
+ r = PyThread_acquire_lock_timed(self->rlock_lock, microseconds);
Py_END_ALLOW_THREADS
}
if (r) {
PyMODINIT_FUNC
PyInit__thread(void)
{
- PyObject *m, *d;
+ PyObject *m, *d, *timeout_max;
/* Initialize types: */
if (PyType_Ready(&localtype) < 0)
if (m == NULL)
return NULL;
+ timeout_max = PyFloat_FromDouble(PY_TIMEOUT_MAX / 1000000);
+ if (!timeout_max)
+ return NULL;
+ if (PyModule_AddObject(m, "TIMEOUT_MAX", timeout_max) < 0)
+ return NULL;
+
/* Add a symbolic constant */
d = PyModule_GetDict(m);
ThreadError = PyErr_NewException("_thread.error", NULL, NULL);
}
DWORD
-EnterNonRecursiveMutex(PNRMUTEX mutex, BOOL wait)
+EnterNonRecursiveMutex(PNRMUTEX mutex, DWORD milliseconds)
{
/* Assume that the thread waits successfully */
DWORD ret ;
/* InterlockedIncrement(&mutex->owned) == 0 means that no thread currently owns the mutex */
- if (!wait)
+ if (milliseconds == 0)
{
if (InterlockedCompareExchange(&mutex->owned, 0, -1) != -1)
return WAIT_TIMEOUT ;
else
ret = InterlockedIncrement(&mutex->owned) ?
/* Some thread owns the mutex, let's wait... */
- WaitForSingleObject(mutex->hevent, INFINITE) : WAIT_OBJECT_0 ;
+ WaitForSingleObject(mutex->hevent, milliseconds) : WAIT_OBJECT_0 ;
mutex->thread_id = GetCurrentThreadId() ; /* We own it */
return ret ;
* if the lock has already been acquired by this thread!
*/
int
-PyThread_acquire_lock(PyThread_type_lock aLock, int waitflag)
+PyThread_acquire_lock_timed(PyThread_type_lock aLock, PY_TIMEOUT_T microseconds)
{
int success ;
+ PY_TIMEOUT_T milliseconds;
+
+ if (microseconds >= 0) {
+ milliseconds = microseconds / 1000;
+ if (microseconds % 1000 > 0)
+ ++milliseconds;
+ if ((DWORD) milliseconds != milliseconds)
+ Py_FatalError("Timeout too large for a DWORD, "
+ "please check PY_TIMEOUT_MAX");
+ }
+ else
+ milliseconds = INFINITE;
- dprintf(("%ld: PyThread_acquire_lock(%p, %d) called\n", PyThread_get_thread_ident(),aLock, waitflag));
+ dprintf(("%ld: PyThread_acquire_lock_timed(%p, %lld) called\n",
+ PyThread_get_thread_ident(), aLock, microseconds));
- success = aLock && EnterNonRecursiveMutex((PNRMUTEX) aLock, (waitflag ? INFINITE : 0)) == WAIT_OBJECT_0 ;
+ success = aLock && EnterNonRecursiveMutex((PNRMUTEX) aLock, (DWORD) milliseconds) == WAIT_OBJECT_0 ;
- dprintf(("%ld: PyThread_acquire_lock(%p, %d) -> %d\n", PyThread_get_thread_ident(),aLock, waitflag, success));
+ dprintf(("%ld: PyThread_acquire_lock(%p, %lld) -> %d\n",
+ PyThread_get_thread_ident(), aLock, microseconds, success));
return success;
}
+int
+PyThread_acquire_lock(PyThread_type_lock aLock, int waitflag)
+{
+ return PyThread_acquire_lock_timed(aLock, waitflag ? -1 : 0);
+}
void
PyThread_release_lock(PyThread_type_lock aLock)
#endif
+/* We assume all modern POSIX systems have gettimeofday() */
+#ifdef GETTIMEOFDAY_NO_TZ
+#define GETTIMEOFDAY(ptv) gettimeofday(ptv)
+#else
+#define GETTIMEOFDAY(ptv) gettimeofday(ptv, (struct timezone *)NULL)
+#endif
+
+#define MICROSECONDS_TO_TIMESPEC(microseconds, ts) \
+do { \
+ struct timeval tv; \
+ GETTIMEOFDAY(&tv); \
+ tv.tv_usec += microseconds % 1000000; \
+ tv.tv_sec += microseconds / 1000000; \
+ tv.tv_sec += tv.tv_usec / 1000000; \
+ tv.tv_usec %= 1000000; \
+ ts.tv_sec = tv.tv_sec; \
+ ts.tv_nsec = tv.tv_usec * 1000; \
+} while(0)
+
+
/* A pthread mutex isn't sufficient to model the Python lock type
* because, according to Draft 5 of the docs (P1003.4a/D5), both of the
* following are undefined:
return (status == -1) ? errno : status;
}
-int
-PyThread_acquire_lock(PyThread_type_lock lock, int waitflag)
+int
+PyThread_acquire_lock_timed(PyThread_type_lock lock, PY_TIMEOUT_T microseconds)
{
int success;
sem_t *thelock = (sem_t *)lock;
int status, error = 0;
+ struct timespec ts;
- dprintf(("PyThread_acquire_lock(%p, %d) called\n", lock, waitflag));
+ dprintf(("PyThread_acquire_lock_timed(%p, %lld) called\n",
+ lock, microseconds));
+ if (microseconds > 0)
+ MICROSECONDS_TO_TIMESPEC(microseconds, ts);
do {
- if (waitflag)
- status = fix_status(sem_wait(thelock));
- else
+ if (microseconds > 0)
+ status = fix_status(sem_timedwait(thelock, &ts));
+ else if (microseconds == 0)
status = fix_status(sem_trywait(thelock));
+ else
+ status = fix_status(sem_wait(thelock));
} while (status == EINTR); /* Retry if interrupted by a signal */
- if (waitflag) {
+ if (microseconds > 0) {
+ if (status != ETIMEDOUT)
+ CHECK_STATUS("sem_timedwait");
+ }
+ else if (microseconds == 0) {
+ if (status != EAGAIN)
+ CHECK_STATUS("sem_trywait");
+ }
+ else {
CHECK_STATUS("sem_wait");
- } else if (status != EAGAIN) {
- CHECK_STATUS("sem_trywait");
}
success = (status == 0) ? 1 : 0;
- dprintf(("PyThread_acquire_lock(%p, %d) -> %d\n", lock, waitflag, success));
+ dprintf(("PyThread_acquire_lock_timed(%p, %lld) -> %d\n",
+ lock, microseconds, success));
return success;
}
+int
+PyThread_acquire_lock(PyThread_type_lock lock, int waitflag)
+{
+ return PyThread_acquire_lock_timed(lock, waitflag ? -1 : 0);
+}
+
void
PyThread_release_lock(PyThread_type_lock lock)
{
free((void *)thelock);
}
-int
-PyThread_acquire_lock(PyThread_type_lock lock, int waitflag)
+int
+PyThread_acquire_lock_timed(PyThread_type_lock lock, PY_TIMEOUT_T microseconds)
{
int success;
pthread_lock *thelock = (pthread_lock *)lock;
int status, error = 0;
- dprintf(("PyThread_acquire_lock(%p, %d) called\n", lock, waitflag));
+ dprintf(("PyThread_acquire_lock_timed(%p, %lld) called\n",
+ lock, microseconds));
status = pthread_mutex_lock( &thelock->mut );
CHECK_STATUS("pthread_mutex_lock[1]");
success = thelock->locked == 0;
- if ( !success && waitflag ) {
+ if (!success && microseconds != 0) {
+ struct timespec ts;
+ if (microseconds > 0)
+ MICROSECONDS_TO_TIMESPEC(microseconds, ts);
/* continue trying until we get the lock */
/* mut must be locked by me -- part of the condition
* protocol */
- while ( thelock->locked ) {
- status = pthread_cond_wait(&thelock->lock_released,
- &thelock->mut);
- CHECK_STATUS("pthread_cond_wait");
+ while (thelock->locked) {
+ if (microseconds > 0) {
+ status = pthread_cond_timedwait(
+ &thelock->lock_released,
+ &thelock->mut, &ts);
+ if (status == ETIMEDOUT)
+ break;
+ CHECK_STATUS("pthread_cond_timed_wait");
+ }
+ else {
+ status = pthread_cond_wait(
+ &thelock->lock_released,
+ &thelock->mut);
+ CHECK_STATUS("pthread_cond_wait");
+ }
}
- success = 1;
+ success = (status == 0);
}
if (success) thelock->locked = 1;
status = pthread_mutex_unlock( &thelock->mut );
CHECK_STATUS("pthread_mutex_unlock[1]");
if (error) success = 0;
- dprintf(("PyThread_acquire_lock(%p, %d) -> %d\n", lock, waitflag, success));
+ dprintf(("PyThread_acquire_lock_timed(%p, %lld) -> %d\n",
+ lock, microseconds, success));
return success;
}
+int
+PyThread_acquire_lock(PyThread_type_lock lock, int waitflag)
+{
+ return PyThread_acquire_lock_timed(lock, waitflag ? -1 : 0);
+}
+
void
PyThread_release_lock(PyThread_type_lock lock)
{
projects / python / commitdiff