Modules/_multiprocessing/semaphore.c

   1 /*
   2  * A type which wraps a semaphore
   3  *
   4  * semaphore.c
   5  *
   6  * Copyright (c) 2006-2008, R Oudkerk --- see COPYING.txt
   7  */
   8
   9 #include "multiprocessing.h"
  10
  11 enum { RECURSIVE_MUTEX, SEMAPHORE };
  12
  13 typedef struct {
  14         PyObject_HEAD
  15         SEM_HANDLE handle;
  16         long last_tid;
  17         int count;
  18         int maxvalue;
  19         int kind;
  20 } SemLockObject;
  21
  22 #define ISMINE(o) (o->count > 0 && PyThread_get_thread_ident() == o->last_tid)
  23
  24
  25 #ifdef MS_WINDOWS
  26
  27 /*
  28  * Windows definitions
  29  */
  30
  31 #define SEM_FAILED NULL
  32
  33 #define SEM_CLEAR_ERROR() SetLastError(0)
  34 #define SEM_GET_LAST_ERROR() GetLastError()
  35 #define SEM_CREATE(name, val, max) CreateSemaphore(NULL, val, max, NULL)
  36 #define SEM_CLOSE(sem) (CloseHandle(sem) ? 0 : -1)
  37 #define SEM_GETVALUE(sem, pval) _GetSemaphoreValue(sem, pval)
  38 #define SEM_UNLINK(name) 0
  39
  40 static int
  41 _GetSemaphoreValue(HANDLE handle, long *value)
  42 {
  43         long previous;
  44
  45         switch (WaitForSingleObject(handle, 0)) {
  46         case WAIT_OBJECT_0:
  47                 if (!ReleaseSemaphore(handle, 1, &previous))
  48                         return MP_STANDARD_ERROR;
  49                 *value = previous + 1;
  50                 return 0;
  51         case WAIT_TIMEOUT:
  52                 *value = 0;
  53                 return 0;
  54         default:
  55                 return MP_STANDARD_ERROR;
  56         }
  57 }
  58
  59 static PyObject *
  60 semlock_acquire(SemLockObject *self, PyObject *args, PyObject *kwds)
  61 {
  62         int blocking = 1;
  63         double timeout;
  64         PyObject *timeout_obj = Py_None;
  65         DWORD res, full_msecs, msecs, start, ticks;
  66
  67         static char *kwlist[] = {"block", "timeout", NULL};
  68
  69         if (!PyArg_ParseTupleAndKeywords(args, kwds, "|iO", kwlist,
  70                                          &blocking, &timeout_obj))
  71                 return NULL;
  72
  73         /* calculate timeout */
  74         if (!blocking) {
  75                 full_msecs = 0;
  76         } else if (timeout_obj == Py_None) {
  77                 full_msecs = INFINITE;
  78         } else {
  79                 timeout = PyFloat_AsDouble(timeout_obj);
  80                 if (PyErr_Occurred())
  81                         return NULL;
  82                 timeout *= 1000.0;      /* convert to millisecs */
  83                 if (timeout < 0.0) {
  84                         timeout = 0.0;
  85                 } else if (timeout >= 0.5 * INFINITE) { /* 25 days */
  86                         PyErr_SetString(PyExc_OverflowError,
  87                                         "timeout is too large");
  88                         return NULL;
  89                 }
  90                 full_msecs = (DWORD)(timeout + 0.5);
  91         }
  92
  93         /* check whether we already own the lock */
  94         if (self->kind == RECURSIVE_MUTEX && ISMINE(self)) {
  95                 ++self->count;
  96                 Py_RETURN_TRUE;
  97         }
  98
  99         /* check whether we can acquire without blocking */
 100         if (WaitForSingleObject(self->handle, 0) == WAIT_OBJECT_0) {
 101                 self->last_tid = GetCurrentThreadId();
 102                 ++self->count;
 103                 Py_RETURN_TRUE;
 104         }
 105
 106         msecs = full_msecs;
 107         start = GetTickCount();
 108
 109         for ( ; ; ) {
 110                 HANDLE handles[2] = {self->handle, sigint_event};
 111
 112                 /* do the wait */
 113                 Py_BEGIN_ALLOW_THREADS
 114                 ResetEvent(sigint_event);
 115                 res = WaitForMultipleObjects(2, handles, FALSE, msecs);
 116                 Py_END_ALLOW_THREADS
 117
 118                 /* handle result */
 119                 if (res != WAIT_OBJECT_0 + 1)
 120                         break;
 121
 122                 /* got SIGINT so give signal handler a chance to run */
 123                 Sleep(1);
 124
 125                 /* if this is main thread let KeyboardInterrupt be raised */
 126                 if (PyErr_CheckSignals())
 127                         return NULL;
 128
 129                 /* recalculate timeout */
 130                 if (msecs != INFINITE) {
 131                         ticks = GetTickCount();
 132                         if ((DWORD)(ticks - start) >= full_msecs)
 133                                 Py_RETURN_FALSE;
 134                         msecs = full_msecs - (ticks - start);
 135                 }
 136         }
 137
 138         /* handle result */
 139         switch (res) {
 140         case WAIT_TIMEOUT:
 141                 Py_RETURN_FALSE;
 142         case WAIT_OBJECT_0:
 143                 self->last_tid = GetCurrentThreadId();
 144                 ++self->count;
 145                 Py_RETURN_TRUE;
 146         case WAIT_FAILED:
 147                 return PyErr_SetFromWindowsErr(0);
 148         default:
 149                 PyErr_Format(PyExc_RuntimeError, "WaitForSingleObject() or "
 150                              "WaitForMultipleObjects() gave unrecognized "
 151                              "value %d", res);
 152                 return NULL;
 153         }
 154 }
 155
 156 static PyObject *
 157 semlock_release(SemLockObject *self, PyObject *args)
 158 {
 159         if (self->kind == RECURSIVE_MUTEX) {
 160                 if (!ISMINE(self)) {
 161                         PyErr_SetString(PyExc_AssertionError, "attempt to "
 162                                         "release recursive lock not owned "
 163                                         "by thread");
 164                         return NULL;
 165                 }
 166                 if (self->count > 1) {
 167                         --self->count;
 168                         Py_RETURN_NONE;
 169                 }
 170                 assert(self->count == 1);
 171         }
 172
 173         if (!ReleaseSemaphore(self->handle, 1, NULL)) {
 174                 if (GetLastError() == ERROR_TOO_MANY_POSTS) {
 175                         PyErr_SetString(PyExc_ValueError, "semaphore or lock "
 176                                         "released too many times");
 177                         return NULL;
 178                 } else {
 179                         return PyErr_SetFromWindowsErr(0);
 180                 }
 181         }
 182
 183         --self->count;
 184         Py_RETURN_NONE;
 185 }
 186
 187 #else /* !MS_WINDOWS */
 188
 189 /*
 190  * Unix definitions
 191  */
 192
 193 #define SEM_CLEAR_ERROR()
 194 #define SEM_GET_LAST_ERROR() 0
 195 #define SEM_CREATE(name, val, max) sem_open(name, O_CREAT | O_EXCL, 0600, val)
 196 #define SEM_CLOSE(sem) sem_close(sem)
 197 #define SEM_GETVALUE(sem, pval) sem_getvalue(sem, pval)
 198 #define SEM_UNLINK(name) sem_unlink(name)
 199
 200 #if HAVE_BROKEN_SEM_UNLINK
 201 #  define sem_unlink(name) 0
 202 #endif
 203
 204 #if !HAVE_SEM_TIMEDWAIT
 205 #  define sem_timedwait(sem,deadline) sem_timedwait_save(sem,deadline,_save)
 206
 207 int
 208 sem_timedwait_save(sem_t *sem, struct timespec *deadline, PyThreadState *_save)
 209 {
 210         int res;
 211         unsigned long delay, difference;
 212         struct timeval now, tvdeadline, tvdelay;
 213
 214         errno = 0;
 215         tvdeadline.tv_sec = deadline->tv_sec;
 216         tvdeadline.tv_usec = deadline->tv_nsec / 1000;
 217
 218         for (delay = 0 ; ; delay += 1000) {
 219                 /* poll */
 220                 if (sem_trywait(sem) == 0)
 221                         return 0;
 222                 else if (errno != EAGAIN)
 223                         return MP_STANDARD_ERROR;
 224
 225                 /* get current time */
 226                 if (gettimeofday(&now, NULL) < 0)
 227                         return MP_STANDARD_ERROR;
 228
 229                 /* check for timeout */
 230                 if (tvdeadline.tv_sec < now.tv_sec ||
 231                     (tvdeadline.tv_sec == now.tv_sec &&
 232                      tvdeadline.tv_usec <= now.tv_usec)) {
 233                         errno = ETIMEDOUT;
 234                         return MP_STANDARD_ERROR;
 235                 }
 236
 237                 /* calculate how much time is left */
 238                 difference = (tvdeadline.tv_sec - now.tv_sec) * 1000000 +
 239                         (tvdeadline.tv_usec - now.tv_usec);
 240
 241                 /* check delay not too long -- maximum is 20 msecs */
 242                 if (delay > 20000)
 243                         delay = 20000;
 244                 if (delay > difference)
 245                         delay = difference;
 246
 247                 /* sleep */
 248                 tvdelay.tv_sec = delay / 1000000;
 249                 tvdelay.tv_usec = delay % 1000000;
 250                 if (select(0, NULL, NULL, NULL, &tvdelay) < 0)
 251                         return MP_STANDARD_ERROR;
 252
 253                 /* check for signals */
 254                 Py_BLOCK_THREADS
 255                 res = PyErr_CheckSignals();
 256                 Py_UNBLOCK_THREADS
 257
 258                 if (res) {
 259                         errno = EINTR;
 260                         return MP_EXCEPTION_HAS_BEEN_SET;
 261                 }
 262         }
 263 }
 264
 265 #endif /* !HAVE_SEM_TIMEDWAIT */
 266
 267 static PyObject *
 268 semlock_acquire(SemLockObject *self, PyObject *args, PyObject *kwds)
 269 {
 270         int blocking = 1, res;
 271         double timeout;
 272         PyObject *timeout_obj = Py_None;
 273         struct timespec deadline = {0};
 274         struct timeval now;
 275         long sec, nsec;
 276
 277         static char *kwlist[] = {"block", "timeout", NULL};
 278
 279         if (!PyArg_ParseTupleAndKeywords(args, kwds, "|iO", kwlist,
 280                                          &blocking, &timeout_obj))
 281                 return NULL;
 282
 283         if (self->kind == RECURSIVE_MUTEX && ISMINE(self)) {
 284                 ++self->count;
 285                 Py_RETURN_TRUE;
 286         }
 287
 288         if (timeout_obj != Py_None) {
 289                 timeout = PyFloat_AsDouble(timeout_obj);
 290                 if (PyErr_Occurred())
 291                         return NULL;
 292                 if (timeout < 0.0)
 293                         timeout = 0.0;
 294
 295                 if (gettimeofday(&now, NULL) < 0) {
 296                         PyErr_SetFromErrno(PyExc_OSError);
 297                         return NULL;
 298                 }
 299                 sec = (long) timeout;
 300                 nsec = (long) (1e9 * (timeout - sec) + 0.5);
 301                 deadline.tv_sec = now.tv_sec + sec;
 302                 deadline.tv_nsec = now.tv_usec * 1000 + nsec;
 303                 deadline.tv_sec += (deadline.tv_nsec / 1000000000);
 304                 deadline.tv_nsec %= 1000000000;
 305         }
 306
 307         do {
 308                 Py_BEGIN_ALLOW_THREADS
 309                 if (blocking && timeout_obj == Py_None)
 310                         res = sem_wait(self->handle);
 311                 else if (!blocking)
 312                         res = sem_trywait(self->handle);
 313                 else
 314                         res = sem_timedwait(self->handle, &deadline);
 315                 Py_END_ALLOW_THREADS
 316                 if (res == MP_EXCEPTION_HAS_BEEN_SET)
 317                         break;
 318         } while (res < 0 && errno == EINTR && !PyErr_CheckSignals());
 319
 320         if (res < 0) {
 321                 if (errno == EAGAIN || errno == ETIMEDOUT)
 322                         Py_RETURN_FALSE;
 323                 else if (errno == EINTR)
 324                         return NULL;
 325                 else
 326                         return PyErr_SetFromErrno(PyExc_OSError);
 327         }
 328
 329         ++self->count;
 330         self->last_tid = PyThread_get_thread_ident();
 331
 332         Py_RETURN_TRUE;
 333 }
 334
 335 static PyObject *
 336 semlock_release(SemLockObject *self, PyObject *args)
 337 {
 338         if (self->kind == RECURSIVE_MUTEX) {
 339                 if (!ISMINE(self)) {
 340                         PyErr_SetString(PyExc_AssertionError, "attempt to "
 341                                         "release recursive lock not owned "
 342                                         "by thread");
 343                         return NULL;
 344                 }
 345                 if (self->count > 1) {
 346                         --self->count;
 347                         Py_RETURN_NONE;
 348                 }
 349                 assert(self->count == 1);
 350         } else {
 351 #if HAVE_BROKEN_SEM_GETVALUE
 352                 /* We will only check properly the maxvalue == 1 case */
 353                 if (self->maxvalue == 1) {
 354                         /* make sure that already locked */
 355                         if (sem_trywait(self->handle) < 0) {
 356                                 if (errno != EAGAIN) {
 357                                         PyErr_SetFromErrno(PyExc_OSError);
 358                                         return NULL;
 359                                 }
 360                                 /* it is already locked as expected */
 361                         } else {
 362                                 /* it was not locked so undo wait and raise  */
 363                                 if (sem_post(self->handle) < 0) {
 364                                         PyErr_SetFromErrno(PyExc_OSError);
 365                                         return NULL;
 366                                 }
 367                                 PyErr_SetString(PyExc_ValueError, "semaphore "
 368                                                 "or lock released too many "
 369                                                 "times");
 370                                 return NULL;
 371                         }
 372                 }
 373 #else
 374                 int sval;
 375
 376                 /* This check is not an absolute guarantee that the semaphore
 377                    does not rise above maxvalue. */
 378                 if (sem_getvalue(self->handle, &sval) < 0) {
 379                         return PyErr_SetFromErrno(PyExc_OSError);
 380                 } else if (sval >= self->maxvalue) {
 381                         PyErr_SetString(PyExc_ValueError, "semaphore or lock "
 382                                         "released too many times");
 383                         return NULL;
 384                 }
 385 #endif
 386         }
 387
 388         if (sem_post(self->handle) < 0)
 389                 return PyErr_SetFromErrno(PyExc_OSError);
 390
 391         --self->count;
 392         Py_RETURN_NONE;
 393 }
 394
 395 #endif /* !MS_WINDOWS */
 396
 397 /*
 398  * All platforms
 399  */
 400
 401 static PyObject *
 402 newsemlockobject(PyTypeObject *type, SEM_HANDLE handle, int kind, int maxvalue)
 403 {
 404         SemLockObject *self;
 405
 406         self = PyObject_New(SemLockObject, type);
 407         if (!self)
 408                 return NULL;
 409         self->handle = handle;
 410         self->kind = kind;
 411         self->count = 0;
 412         self->last_tid = 0;
 413         self->maxvalue = maxvalue;
 414         return (PyObject*)self;
 415 }
 416
 417 static PyObject *
 418 semlock_new(PyTypeObject *type, PyObject *args, PyObject *kwds)
 419 {
 420         char buffer[256];
 421         SEM_HANDLE handle = SEM_FAILED;
 422         int kind, maxvalue, value;
 423         PyObject *result;
 424         static char *kwlist[] = {"kind", "value", "maxvalue", NULL};
 425         static int counter = 0;
 426
 427         if (!PyArg_ParseTupleAndKeywords(args, kwds, "iii", kwlist,
 428                                          &kind, &value, &maxvalue))
 429                 return NULL;
 430
 431         if (kind != RECURSIVE_MUTEX && kind != SEMAPHORE) {
 432                 PyErr_SetString(PyExc_ValueError, "unrecognized kind");
 433                 return NULL;
 434         }
 435
 436         PyOS_snprintf(buffer, sizeof(buffer), "/mp%d-%d", getpid(), counter++);
 437
 438         SEM_CLEAR_ERROR();
 439         handle = SEM_CREATE(buffer, value, maxvalue);
 440         /* On Windows we should fail if GetLastError()==ERROR_ALREADY_EXISTS */
 441         if (handle == SEM_FAILED || SEM_GET_LAST_ERROR() != 0)
 442                 goto failure;
 443
 444         if (SEM_UNLINK(buffer) < 0)
 445                 goto failure;
 446
 447         result = newsemlockobject(type, handle, kind, maxvalue);
 448         if (!result)
 449                 goto failure;
 450
 451         return result;
 452
 453   failure:
 454         if (handle != SEM_FAILED)
 455                 SEM_CLOSE(handle);
 456         mp_SetError(NULL, MP_STANDARD_ERROR);
 457         return NULL;
 458 }
 459
 460 static PyObject *
 461 semlock_rebuild(PyTypeObject *type, PyObject *args)
 462 {
 463         SEM_HANDLE handle;
 464         int kind, maxvalue;
 465
 466         if (!PyArg_ParseTuple(args, F_SEM_HANDLE "ii",
 467                               &handle, &kind, &maxvalue))
 468                 return NULL;
 469
 470         return newsemlockobject(type, handle, kind, maxvalue);
 471 }
 472
 473 static void
 474 semlock_dealloc(SemLockObject* self)
 475 {
 476         if (self->handle != SEM_FAILED)
 477                 SEM_CLOSE(self->handle);
 478         PyObject_Del(self);
 479 }
 480
 481 static PyObject *
 482 semlock_count(SemLockObject *self)
 483 {
 484         return PyInt_FromLong((long)self->count);
 485 }
 486
 487 static PyObject *
 488 semlock_ismine(SemLockObject *self)
 489 {
 490         /* only makes sense for a lock */
 491         return PyBool_FromLong(ISMINE(self));
 492 }
 493
 494 static PyObject *
 495 semlock_getvalue(SemLockObject *self)
 496 {
 497 #if HAVE_BROKEN_SEM_GETVALUE
 498         PyErr_SetNone(PyExc_NotImplementedError);
 499         return NULL;
 500 #else
 501         int sval;
 502         if (SEM_GETVALUE(self->handle, &sval) < 0)
 503                 return mp_SetError(NULL, MP_STANDARD_ERROR);
 504         /* some posix implementations use negative numbers to indicate
 505            the number of waiting threads */
 506         if (sval < 0)
 507                 sval = 0;
 508         return PyInt_FromLong((long)sval);
 509 #endif
 510 }
 511
 512 static PyObject *
 513 semlock_iszero(SemLockObject *self)
 514 {
 515         int sval;
 516 #if HAVE_BROKEN_SEM_GETVALUE
 517         if (sem_trywait(self->handle) < 0) {
 518                 if (errno == EAGAIN)
 519                         Py_RETURN_TRUE;
 520                 return mp_SetError(NULL, MP_STANDARD_ERROR);
 521         } else {
 522                 if (sem_post(self->handle) < 0)
 523                         return mp_SetError(NULL, MP_STANDARD_ERROR);
 524                 Py_RETURN_FALSE;
 525         }
 526 #else
 527         if (SEM_GETVALUE(self->handle, &sval) < 0)
 528                 return mp_SetError(NULL, MP_STANDARD_ERROR);
 529         return PyBool_FromLong((long)sval == 0);
 530 #endif
 531 }
 532
 533 static PyObject *
 534 semlock_afterfork(SemLockObject *self)
 535 {
 536         self->count = 0;
 537         Py_RETURN_NONE;
 538 }
 539
 540 /*
 541  * Semaphore methods
 542  */
 543
 544 static PyMethodDef semlock_methods[] = {
 545         {"acquire", (PyCFunction)semlock_acquire, METH_VARARGS | METH_KEYWORDS,
 546          "acquire the semaphore/lock"},
 547         {"release", (PyCFunction)semlock_release, METH_NOARGS,
 548          "release the semaphore/lock"},
 549         {"__enter__", (PyCFunction)semlock_acquire, METH_VARARGS,
 550          "enter the semaphore/lock"},
 551         {"__exit__", (PyCFunction)semlock_release, METH_VARARGS,
 552          "exit the semaphore/lock"},
 553         {"_count", (PyCFunction)semlock_count, METH_NOARGS,
 554          "num of `acquire()`s minus num of `release()`s for this process"},
 555         {"_is_mine", (PyCFunction)semlock_ismine, METH_NOARGS,
 556          "whether the lock is owned by this thread"},
 557         {"_get_value", (PyCFunction)semlock_getvalue, METH_NOARGS,
 558          "get the value of the semaphore"},
 559         {"_is_zero", (PyCFunction)semlock_iszero, METH_NOARGS,
 560          "returns whether semaphore has value zero"},
 561         {"_rebuild", (PyCFunction)semlock_rebuild, METH_VARARGS | METH_CLASS,
 562          ""},
 563         {"_after_fork", (PyCFunction)semlock_afterfork, METH_NOARGS,
 564          "rezero the net acquisition count after fork()"},
 565         {NULL}
 566 };
 567
 568 /*
 569  * Member table
 570  */
 571
 572 static PyMemberDef semlock_members[] = {
 573         {"handle", T_SEM_HANDLE, offsetof(SemLockObject, handle), READONLY,
 574          ""},
 575         {"kind", T_INT, offsetof(SemLockObject, kind), READONLY,
 576          ""},
 577         {"maxvalue", T_INT, offsetof(SemLockObject, maxvalue), READONLY,
 578          ""},
 579         {NULL}
 580 };
 581
 582 /*
 583  * Semaphore type
 584  */
 585
 586 PyTypeObject SemLockType = {
 587         PyVarObject_HEAD_INIT(NULL, 0)
 588         /* tp_name           */ "_multiprocessing.SemLock",
 589         /* tp_basicsize      */ sizeof(SemLockObject),
 590         /* tp_itemsize       */ 0,
 591         /* tp_dealloc        */ (destructor)semlock_dealloc,
 592         /* tp_print          */ 0,
 593         /* tp_getattr        */ 0,
 594         /* tp_setattr        */ 0,
 595         /* tp_compare        */ 0,
 596         /* tp_repr           */ 0,
 597         /* tp_as_number      */ 0,
 598         /* tp_as_sequence    */ 0,
 599         /* tp_as_mapping     */ 0,
 600         /* tp_hash           */ 0,
 601         /* tp_call           */ 0,
 602         /* tp_str            */ 0,
 603         /* tp_getattro       */ 0,
 604         /* tp_setattro       */ 0,
 605         /* tp_as_buffer      */ 0,
 606         /* tp_flags          */ Py_TPFLAGS_DEFAULT | Py_TPFLAGS_BASETYPE,
 607         /* tp_doc            */ "Semaphore/Mutex type",
 608         /* tp_traverse       */ 0,
 609         /* tp_clear          */ 0,
 610         /* tp_richcompare    */ 0,
 611         /* tp_weaklistoffset */ 0,
 612         /* tp_iter           */ 0,
 613         /* tp_iternext       */ 0,
 614         /* tp_methods        */ semlock_methods,
 615         /* tp_members        */ semlock_members,
 616         /* tp_getset         */ 0,
 617         /* tp_base           */ 0,
 618         /* tp_dict           */ 0,
 619         /* tp_descr_get      */ 0,
 620         /* tp_descr_set      */ 0,
 621         /* tp_dictoffset     */ 0,
 622         /* tp_init           */ 0,
 623         /* tp_alloc          */ 0,
 624         /* tp_new            */ semlock_new,
 625 };