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.
This commit is contained in:
Antoine Pitrou 2010-04-14 15:44:10 +00:00
parent e53de3dc4a
commit 7c3e577395
11 changed files with 326 additions and 77 deletions

View file

@ -28,7 +28,7 @@ implementation. For systems lacking the :mod:`_thread` module, the
:mod:`_dummy_thread` module is available. It duplicates this module's interface :mod:`_dummy_thread` module is available. It duplicates this module's interface
and can be used as a drop-in replacement. 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 .. exception:: error
@ -103,19 +103,34 @@ It defines the following constant and functions:
Availability: Windows, systems with POSIX threads. 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: 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 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 time can acquire a lock --- that's their reason for existence).
*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.
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() .. method:: lock.release()

View file

@ -155,6 +155,16 @@ This module defines the following functions and objects:
Availability: Windows, systems with POSIX threads. 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. Detailed interfaces for the objects are documented below.
The design of this module is loosely based on Java's threading model. However, The design of this module is loosely based on Java's threading model. However,
@ -349,7 +359,7 @@ and may vary across implementations.
All methods are executed atomically. All methods are executed atomically.
.. method:: Lock.acquire(blocking=True) .. method:: Lock.acquire(blocking=True, timeout=-1)
Acquire a lock, blocking or non-blocking. Acquire a lock, blocking or non-blocking.
@ -363,6 +373,15 @@ All methods are executed atomically.
without an argument would block, return false immediately; otherwise, do the without an argument would block, return false immediately; otherwise, do the
same thing as when called without arguments, and return true. 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() .. method:: Lock.release()
@ -396,7 +415,7 @@ pair) resets the lock to unlocked and allows another thread blocked in
:meth:`acquire` to proceed. :meth:`acquire` to proceed.
.. method:: RLock.acquire(blocking=True) .. method:: RLock.acquire(blocking=True, timeout=-1)
Acquire a lock, blocking or non-blocking. Acquire a lock, blocking or non-blocking.
@ -415,6 +434,11 @@ pair) resets the lock to unlocked and allows another thread blocked in
without an argument would block, return false immediately; otherwise, do the without an argument would block, return false immediately; otherwise, do the
same thing as when called without arguments, and return true. 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() .. method:: RLock.release()

View file

@ -19,6 +19,41 @@ PyAPI_FUNC(void) PyThread_free_lock(PyThread_type_lock);
PyAPI_FUNC(int) PyThread_acquire_lock(PyThread_type_lock, int); PyAPI_FUNC(int) PyThread_acquire_lock(PyThread_type_lock, int);
#define WAIT_LOCK 1 #define WAIT_LOCK 1
#define NOWAIT_LOCK 0 #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(void) PyThread_release_lock(PyThread_type_lock);
PyAPI_FUNC(size_t) PyThread_get_stacksize(void); PyAPI_FUNC(size_t) PyThread_get_stacksize(void);

View file

@ -17,6 +17,10 @@ __all__ = ['error', 'start_new_thread', 'exit', 'get_ident', 'allocate_lock',
'interrupt_main', 'LockType'] 'interrupt_main', 'LockType']
import traceback as _traceback import traceback as _traceback
import time
# A dummy value
TIMEOUT_MAX = 2**31
class error(Exception): class error(Exception):
"""Dummy implementation of _thread.error.""" """Dummy implementation of _thread.error."""
@ -92,7 +96,7 @@ class LockType(object):
def __init__(self): def __init__(self):
self.locked_status = False self.locked_status = False
def acquire(self, waitflag=None): def acquire(self, waitflag=None, timeout=-1):
"""Dummy implementation of acquire(). """Dummy implementation of acquire().
For blocking calls, self.locked_status is automatically set to For blocking calls, self.locked_status is automatically set to
@ -111,6 +115,8 @@ class LockType(object):
self.locked_status = True self.locked_status = True
return True return True
else: else:
if timeout > 0:
time.sleep(timeout)
return False return False
__enter__ = acquire __enter__ = acquire

View file

@ -440,10 +440,10 @@ class Pool(object):
p.terminate() p.terminate()
debug('joining task handler') debug('joining task handler')
task_handler.join(1e100) task_handler.join()
debug('joining result handler') debug('joining result handler')
result_handler.join(1e100) task_handler.join()
if pool and hasattr(pool[0], 'terminate'): if pool and hasattr(pool[0], 'terminate'):
debug('joining pool workers') debug('joining pool workers')

View file

@ -4,7 +4,7 @@ Various tests for synchronization primitives.
import sys import sys
import time import time
from _thread import start_new_thread, get_ident from _thread import start_new_thread, get_ident, TIMEOUT_MAX
import threading import threading
import unittest import unittest
@ -62,6 +62,14 @@ class BaseTestCase(unittest.TestCase):
support.threading_cleanup(*self._threads) support.threading_cleanup(*self._threads)
support.reap_children() 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): class BaseLockTests(BaseTestCase):
""" """
@ -143,6 +151,32 @@ class BaseLockTests(BaseTestCase):
Bunch(f, 15).wait_for_finished() Bunch(f, 15).wait_for_finished()
self.assertEqual(n, len(threading.enumerate())) 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): class LockTests(BaseLockTests):
""" """
@ -284,14 +318,14 @@ class EventTests(BaseTestCase):
def f(): def f():
results1.append(evt.wait(0.0)) results1.append(evt.wait(0.0))
t1 = time.time() t1 = time.time()
r = evt.wait(0.2) r = evt.wait(0.5)
t2 = time.time() t2 = time.time()
results2.append((r, t2 - t1)) results2.append((r, t2 - t1))
Bunch(f, N).wait_for_finished() Bunch(f, N).wait_for_finished()
self.assertEqual(results1, [False] * N) self.assertEqual(results1, [False] * N)
for r, dt in results2: for r, dt in results2:
self.assertFalse(r) self.assertFalse(r)
self.assertTrue(dt >= 0.2, dt) self.assertTimeout(dt, 0.5)
# The event is set # The event is set
results1 = [] results1 = []
results2 = [] results2 = []
@ -397,14 +431,14 @@ class ConditionTests(BaseTestCase):
def f(): def f():
cond.acquire() cond.acquire()
t1 = time.time() t1 = time.time()
cond.wait(0.2) cond.wait(0.5)
t2 = time.time() t2 = time.time()
cond.release() cond.release()
results.append(t2 - t1) results.append(t2 - t1)
Bunch(f, N).wait_for_finished() Bunch(f, N).wait_for_finished()
self.assertEqual(len(results), 5) self.assertEqual(len(results), 5)
for dt in results: for dt in results:
self.assertTrue(dt >= 0.2, dt) self.assertTimeout(dt, 0.5)
class BaseSemaphoreTests(BaseTestCase): class BaseSemaphoreTests(BaseTestCase):

View file

@ -31,6 +31,7 @@ try:
_CRLock = _thread.RLock _CRLock = _thread.RLock
except AttributeError: except AttributeError:
_CRLock = None _CRLock = None
TIMEOUT_MAX = _thread.TIMEOUT_MAX
del _thread del _thread
@ -107,14 +108,14 @@ class _RLock(_Verbose):
return "<%s owner=%r count=%d>" % ( return "<%s owner=%r count=%d>" % (
self.__class__.__name__, owner, self._count) self.__class__.__name__, owner, self._count)
def acquire(self, blocking=True): def acquire(self, blocking=True, timeout=-1):
me = _get_ident() me = _get_ident()
if self._owner == me: if self._owner == me:
self._count = self._count + 1 self._count = self._count + 1
if __debug__: if __debug__:
self._note("%s.acquire(%s): recursive success", self, blocking) self._note("%s.acquire(%s): recursive success", self, blocking)
return 1 return 1
rc = self._block.acquire(blocking) rc = self._block.acquire(blocking, timeout)
if rc: if rc:
self._owner = me self._owner = me
self._count = 1 self._count = 1
@ -234,22 +235,10 @@ class _Condition(_Verbose):
if __debug__: if __debug__:
self._note("%s.wait(): got it", self) self._note("%s.wait(): got it", self)
else: else:
# Balancing act: We can't afford a pure busy loop, so we if timeout > 0:
# have to sleep; but if we sleep the whole timeout time, gotit = waiter.acquire(True, timeout)
# we'll be unresponsive. The scheme here sleeps very else:
# little at first, longer as time goes on, but never longer gotit = waiter.acquire(False)
# 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 not gotit: if not gotit:
if __debug__: if __debug__:
self._note("%s.wait(%s): timed out", self, timeout) self._note("%s.wait(%s): timed out", self, timeout)

View file

@ -312,6 +312,11 @@ C-API
Library 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 - Issue #8383: pickle and pickletools use surrogatepass error handler when
encoding unicode as utf8 to support lone surrogates and stay compatible with encoding unicode as utf8 to support lone surrogates and stay compatible with
Python 2.x and 3.0 Python 2.x and 3.0

View file

@ -40,18 +40,47 @@ lock_dealloc(lockobject *self)
} }
static PyObject * 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; 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 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 Py_END_ALLOW_THREADS
return PyBool_FromLong((long)i); return PyBool_FromLong(r);
} }
PyDoc_STRVAR(acquire_doc, PyDoc_STRVAR(acquire_doc,
@ -106,9 +135,9 @@ Return whether the lock is in the locked state.");
static PyMethodDef lock_methods[] = { static PyMethodDef lock_methods[] = {
{"acquire_lock", (PyCFunction)lock_PyThread_acquire_lock, {"acquire_lock", (PyCFunction)lock_PyThread_acquire_lock,
METH_VARARGS, acquire_doc}, METH_VARARGS | METH_KEYWORDS, acquire_doc},
{"acquire", (PyCFunction)lock_PyThread_acquire_lock, {"acquire", (PyCFunction)lock_PyThread_acquire_lock,
METH_VARARGS, acquire_doc}, METH_VARARGS | METH_KEYWORDS, acquire_doc},
{"release_lock", (PyCFunction)lock_PyThread_release_lock, {"release_lock", (PyCFunction)lock_PyThread_release_lock,
METH_NOARGS, release_doc}, METH_NOARGS, release_doc},
{"release", (PyCFunction)lock_PyThread_release_lock, {"release", (PyCFunction)lock_PyThread_release_lock,
@ -118,7 +147,7 @@ static PyMethodDef lock_methods[] = {
{"locked", (PyCFunction)lock_locked_lock, {"locked", (PyCFunction)lock_locked_lock,
METH_NOARGS, locked_doc}, METH_NOARGS, locked_doc},
{"__enter__", (PyCFunction)lock_PyThread_acquire_lock, {"__enter__", (PyCFunction)lock_PyThread_acquire_lock,
METH_VARARGS, acquire_doc}, METH_VARARGS | METH_KEYWORDS, acquire_doc},
{"__exit__", (PyCFunction)lock_PyThread_release_lock, {"__exit__", (PyCFunction)lock_PyThread_release_lock,
METH_VARARGS, release_doc}, METH_VARARGS, release_doc},
{NULL, NULL} /* sentinel */ {NULL, NULL} /* sentinel */
@ -183,15 +212,41 @@ rlock_dealloc(rlockobject *self)
static PyObject * static PyObject *
rlock_acquire(rlockobject *self, PyObject *args, PyObject *kwds) rlock_acquire(rlockobject *self, PyObject *args, PyObject *kwds)
{ {
char *kwlist[] = {"blocking", NULL}; char *kwlist[] = {"blocking", "timeout", NULL};
int blocking = 1; int blocking = 1;
double timeout = -1;
PY_TIMEOUT_T microseconds;
long tid; long tid;
int r = 1; int r = 1;
if (!PyArg_ParseTupleAndKeywords(args, kwds, "|i:acquire", kwlist, if (!PyArg_ParseTupleAndKeywords(args, kwds, "|id:acquire", kwlist,
&blocking)) &blocking, &timeout))
return NULL; 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(); tid = PyThread_get_thread_ident();
if (self->rlock_count > 0 && tid == self->rlock_owner) { if (self->rlock_count > 0 && tid == self->rlock_owner) {
unsigned long count = self->rlock_count + 1; unsigned long count = self->rlock_count + 1;
@ -206,11 +261,11 @@ rlock_acquire(rlockobject *self, PyObject *args, PyObject *kwds)
if (self->rlock_count > 0 || if (self->rlock_count > 0 ||
!PyThread_acquire_lock(self->rlock_lock, 0)) { !PyThread_acquire_lock(self->rlock_lock, 0)) {
if (!blocking) { if (microseconds == 0) {
Py_RETURN_FALSE; Py_RETURN_FALSE;
} }
Py_BEGIN_ALLOW_THREADS 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 Py_END_ALLOW_THREADS
} }
if (r) { if (r) {
@ -1005,7 +1060,7 @@ static struct PyModuleDef threadmodule = {
PyMODINIT_FUNC PyMODINIT_FUNC
PyInit__thread(void) PyInit__thread(void)
{ {
PyObject *m, *d; PyObject *m, *d, *timeout_max;
/* Initialize types: */ /* Initialize types: */
if (PyType_Ready(&localtype) < 0) if (PyType_Ready(&localtype) < 0)
@ -1020,6 +1075,12 @@ PyInit__thread(void)
if (m == NULL) if (m == NULL)
return 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 */ /* Add a symbolic constant */
d = PyModule_GetDict(m); d = PyModule_GetDict(m);
ThreadError = PyErr_NewException("_thread.error", NULL, NULL); ThreadError = PyErr_NewException("_thread.error", NULL, NULL);

View file

@ -34,13 +34,13 @@ DeleteNonRecursiveMutex(PNRMUTEX mutex)
} }
DWORD DWORD
EnterNonRecursiveMutex(PNRMUTEX mutex, BOOL wait) EnterNonRecursiveMutex(PNRMUTEX mutex, DWORD milliseconds)
{ {
/* Assume that the thread waits successfully */ /* Assume that the thread waits successfully */
DWORD ret ; DWORD ret ;
/* InterlockedIncrement(&mutex->owned) == 0 means that no thread currently owns the mutex */ /* InterlockedIncrement(&mutex->owned) == 0 means that no thread currently owns the mutex */
if (!wait) if (milliseconds == 0)
{ {
if (InterlockedCompareExchange(&mutex->owned, 0, -1) != -1) if (InterlockedCompareExchange(&mutex->owned, 0, -1) != -1)
return WAIT_TIMEOUT ; return WAIT_TIMEOUT ;
@ -49,7 +49,7 @@ EnterNonRecursiveMutex(PNRMUTEX mutex, BOOL wait)
else else
ret = InterlockedIncrement(&mutex->owned) ? ret = InterlockedIncrement(&mutex->owned) ?
/* Some thread owns the mutex, let's wait... */ /* 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 */ mutex->thread_id = GetCurrentThreadId() ; /* We own it */
return ret ; return ret ;
@ -239,18 +239,37 @@ PyThread_free_lock(PyThread_type_lock aLock)
* if the lock has already been acquired by this thread! * if the lock has already been acquired by this thread!
*/ */
int int
PyThread_acquire_lock(PyThread_type_lock aLock, int waitflag) PyThread_acquire_lock_timed(PyThread_type_lock aLock, PY_TIMEOUT_T microseconds)
{ {
int success ; int success ;
PY_TIMEOUT_T milliseconds;
dprintf(("%ld: PyThread_acquire_lock(%p, %d) called\n", PyThread_get_thread_ident(),aLock, waitflag)); 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;
success = aLock && EnterNonRecursiveMutex((PNRMUTEX) aLock, (waitflag ? INFINITE : 0)) == WAIT_OBJECT_0 ; dprintf(("%ld: PyThread_acquire_lock_timed(%p, %lld) called\n",
PyThread_get_thread_ident(), aLock, microseconds));
dprintf(("%ld: PyThread_acquire_lock(%p, %d) -> %d\n", PyThread_get_thread_ident(),aLock, waitflag, success)); success = aLock && EnterNonRecursiveMutex((PNRMUTEX) aLock, (DWORD) milliseconds) == WAIT_OBJECT_0 ;
dprintf(("%ld: PyThread_acquire_lock(%p, %lld) -> %d\n",
PyThread_get_thread_ident(), aLock, microseconds, success));
return success; return success;
} }
int
PyThread_acquire_lock(PyThread_type_lock aLock, int waitflag)
{
return PyThread_acquire_lock_timed(aLock, waitflag ? -1 : 0);
}
void void
PyThread_release_lock(PyThread_type_lock aLock) PyThread_release_lock(PyThread_type_lock aLock)

View file

@ -83,6 +83,26 @@
#endif #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 /* 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 * because, according to Draft 5 of the docs (P1003.4a/D5), both of the
* following are undefined: * following are undefined:
@ -295,34 +315,53 @@ fix_status(int status)
return (status == -1) ? errno : status; return (status == -1) ? errno : status;
} }
int int
PyThread_acquire_lock(PyThread_type_lock lock, int waitflag) PyThread_acquire_lock_timed(PyThread_type_lock lock, PY_TIMEOUT_T microseconds)
{ {
int success; int success;
sem_t *thelock = (sem_t *)lock; sem_t *thelock = (sem_t *)lock;
int status, error = 0; 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 { do {
if (waitflag) if (microseconds > 0)
status = fix_status(sem_wait(thelock)); status = fix_status(sem_timedwait(thelock, &ts));
else else if (microseconds == 0)
status = fix_status(sem_trywait(thelock)); status = fix_status(sem_trywait(thelock));
else
status = fix_status(sem_wait(thelock));
} while (status == EINTR); /* Retry if interrupted by a signal */ } 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"); CHECK_STATUS("sem_wait");
} else if (status != EAGAIN) {
CHECK_STATUS("sem_trywait");
} }
success = (status == 0) ? 1 : 0; 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; return success;
} }
int
PyThread_acquire_lock(PyThread_type_lock lock, int waitflag)
{
return PyThread_acquire_lock_timed(lock, waitflag ? -1 : 0);
}
void void
PyThread_release_lock(PyThread_type_lock lock) PyThread_release_lock(PyThread_type_lock lock)
{ {
@ -390,40 +429,62 @@ PyThread_free_lock(PyThread_type_lock lock)
free((void *)thelock); free((void *)thelock);
} }
int int
PyThread_acquire_lock(PyThread_type_lock lock, int waitflag) PyThread_acquire_lock_timed(PyThread_type_lock lock, PY_TIMEOUT_T microseconds)
{ {
int success; int success;
pthread_lock *thelock = (pthread_lock *)lock; pthread_lock *thelock = (pthread_lock *)lock;
int status, error = 0; 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 ); status = pthread_mutex_lock( &thelock->mut );
CHECK_STATUS("pthread_mutex_lock[1]"); CHECK_STATUS("pthread_mutex_lock[1]");
success = thelock->locked == 0; 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 */ /* continue trying until we get the lock */
/* mut must be locked by me -- part of the condition /* mut must be locked by me -- part of the condition
* protocol */ * protocol */
while ( thelock->locked ) { while (thelock->locked) {
status = pthread_cond_wait(&thelock->lock_released, if (microseconds > 0) {
&thelock->mut); status = pthread_cond_timedwait(
CHECK_STATUS("pthread_cond_wait"); &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; if (success) thelock->locked = 1;
status = pthread_mutex_unlock( &thelock->mut ); status = pthread_mutex_unlock( &thelock->mut );
CHECK_STATUS("pthread_mutex_unlock[1]"); CHECK_STATUS("pthread_mutex_unlock[1]");
if (error) success = 0; 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; return success;
} }
int
PyThread_acquire_lock(PyThread_type_lock lock, int waitflag)
{
return PyThread_acquire_lock_timed(lock, waitflag ? -1 : 0);
}
void void
PyThread_release_lock(PyThread_type_lock lock) PyThread_release_lock(PyThread_type_lock lock)
{ {