#include "multiprocessing.h"

#include "pycore_object.h" // _PyObject_VisitType()

#ifdef HAVE_SYS_TIME_H

# include <sys/time.h> // gettimeofday()

#ifdef HAVE_MP_SEMAPHORE

enum { RECURSIVE_MUTEX, SEMAPHORE };

typedef struct {

PyObject_HEAD

SEM_HANDLE handle;

unsigned long last_tid;

int count;

int maxvalue;

int kind;

char *name;

} SemLockObject;

#define _SemLockObject_CAST(op) ((SemLockObject *)(op))

#include "clinic/semaphore.c.h"

#define ISMINE(o) (o->count > 0 && PyThread_get_thread_ident() == o->last_tid)

#ifdef MS_WINDOWS

#define SEM_FAILED NULL

#define SEM_CLEAR_ERROR() SetLastError(0)

#define SEM_GET_LAST_ERROR() GetLastError()

#define SEM_CREATE(name, val, max) CreateSemaphore(NULL, val, max, NULL)

#define SEM_CLOSE(sem) (CloseHandle(sem) ? 0 : -1)

#define SEM_GETVALUE(sem, pval) _GetSemaphoreValue(sem, pval)

#define SEM_UNLINK(name) 0

static int

_GetSemaphoreValue(HANDLE handle, int *value)

{

long previous;

switch (WaitForSingleObjectEx(handle, 0, FALSE)) {

case WAIT_OBJECT_0:

if (!ReleaseSemaphore(handle, 1, &previous))

return MP_STANDARD_ERROR;

*value = previous + 1;

return 0;

case WAIT_TIMEOUT:

*value = 0;

return 0;

default:

return MP_STANDARD_ERROR;

}

}

static PyObject *

_multiprocessing_SemLock_acquire_impl(SemLockObject *self, int blocking,

PyObject *timeout_obj)

{

double timeout;

DWORD res, full_msecs, nhandles;

HANDLE handles[2], sigint_event;

/* calculate timeout */

if (!blocking) {

full_msecs = 0;

} else if (timeout_obj == Py_None) {

full_msecs = INFINITE;

} else {

timeout = PyFloat_AsDouble(timeout_obj);

if (PyErr_Occurred())

return NULL;

timeout *= 1000.0; /* convert to millisecs */

if (timeout < 0.0) {

timeout = 0.0;

} else if (timeout >= 0.5 * INFINITE) { /* 25 days */

PyErr_SetString(PyExc_OverflowError,

"timeout is too large");

return NULL;

}

full_msecs = (DWORD)(timeout + 0.5);

}

/* check whether we already own the lock */

if (self->kind == RECURSIVE_MUTEX && ISMINE(self)) {

++self->count;

Py_RETURN_TRUE;

}

/* check whether we can acquire without releasing the GIL and blocking */

if (WaitForSingleObjectEx(self->handle, 0, FALSE) == WAIT_OBJECT_0) {

self->last_tid = GetCurrentThreadId();

++self->count;

Py_RETURN_TRUE;

}

/* prepare list of handles */

nhandles = 0;

handles[nhandles++] = self->handle;

if (_PyOS_IsMainThread()) {

sigint_event = _PyOS_SigintEvent();

assert(sigint_event != NULL);

handles[nhandles++] = sigint_event;

}

else {

sigint_event = NULL;

}

/* do the wait */

Py_BEGIN_ALLOW_THREADS

if (sigint_event != NULL)

ResetEvent(sigint_event);

res = WaitForMultipleObjectsEx(nhandles, handles, FALSE, full_msecs, FALSE);

Py_END_ALLOW_THREADS

/* handle result */

switch (res) {

case WAIT_TIMEOUT:

Py_RETURN_FALSE;

case WAIT_OBJECT_0 + 0:

self->last_tid = GetCurrentThreadId();

++self->count;

Py_RETURN_TRUE;

case WAIT_OBJECT_0 + 1:

errno = EINTR;

return PyErr_SetFromErrno(PyExc_OSError);

case WAIT_FAILED:

return PyErr_SetFromWindowsErr(0);

default:

PyErr_Format(PyExc_RuntimeError, "WaitForSingleObject() or "

"WaitForMultipleObjects() gave unrecognized "

"value %u", res);

return NULL;

}

}

static PyObject *

_multiprocessing_SemLock_release_impl(SemLockObject *self)

{

if (self->kind == RECURSIVE_MUTEX) {

if (!ISMINE(self)) {

PyErr_SetString(PyExc_AssertionError, "attempt to "

"release recursive lock not owned "

"by thread");

return NULL;

}

if (self->count > 1) {

--self->count;

Py_RETURN_NONE;

}

assert(self->count == 1);

}

if (!ReleaseSemaphore(self->handle, 1, NULL)) {

if (GetLastError() == ERROR_TOO_MANY_POSTS) {

PyErr_SetString(PyExc_ValueError, "semaphore or lock "

"released too many times");

return NULL;

} else {

return PyErr_SetFromWindowsErr(0);

}

}

--self->count;

Py_RETURN_NONE;

}

#else /* !MS_WINDOWS */

#define SEM_CLEAR_ERROR()

#define SEM_GET_LAST_ERROR() 0

#define SEM_CREATE(name, val, max) sem_open(name, O_CREAT | O_EXCL, 0600, val)

#define SEM_CLOSE(sem) sem_close(sem)

#define SEM_GETVALUE(sem, pval) sem_getvalue(sem, pval)

#define SEM_UNLINK(name) sem_unlink(name)

#ifdef __APPLE__

# undef SEM_FAILED

# define SEM_FAILED ((sem_t *)-1)

#ifndef HAVE_SEM_UNLINK

# define sem_unlink(name) 0

#ifndef HAVE_SEM_TIMEDWAIT

# define sem_timedwait(sem,deadline) sem_timedwait_save(sem,deadline,_save)

static int

sem_timedwait_save(sem_t *sem, struct timespec *deadline, PyThreadState *_save)

{

int res;

unsigned long delay, difference;

struct timeval now, tvdeadline, tvdelay;

errno = 0;

tvdeadline.tv_sec = deadline->tv_sec;

tvdeadline.tv_usec = deadline->tv_nsec / 1000;

for (delay = 0 ; ; delay += 1000) {

/* poll */

if (sem_trywait(sem) == 0)

return 0;

else if (errno != EAGAIN)

return MP_STANDARD_ERROR;

/* get current time */

if (gettimeofday(&now, NULL) < 0)

return MP_STANDARD_ERROR;

/* check for timeout */

if (tvdeadline.tv_sec < now.tv_sec ||

(tvdeadline.tv_sec == now.tv_sec &&

tvdeadline.tv_usec <= now.tv_usec)) {

errno = ETIMEDOUT;

return MP_STANDARD_ERROR;

}

/* calculate how much time is left */

difference = (tvdeadline.tv_sec - now.tv_sec) * 1000000 +

(tvdeadline.tv_usec - now.tv_usec);

/* check delay not too long -- maximum is 20 msecs */

if (delay > 20000)

delay = 20000;

if (delay > difference)

delay = difference;

/* sleep */

tvdelay.tv_sec = delay / 1000000;

tvdelay.tv_usec = delay % 1000000;

if (select(0, NULL, NULL, NULL, &tvdelay) < 0)

return MP_STANDARD_ERROR;

/* check for signals */

Py_BLOCK_THREADS

res = PyErr_CheckSignals();

Py_UNBLOCK_THREADS

if (res) {

errno = EINTR;

return MP_EXCEPTION_HAS_BEEN_SET;

}

}

}

#endif /* !HAVE_SEM_TIMEDWAIT */

static PyObject *

_multiprocessing_SemLock_acquire_impl(SemLockObject *self, int blocking,

PyObject *timeout_obj)

{

int res, err = 0;

struct timespec deadline = {0};

if (self->kind == RECURSIVE_MUTEX && ISMINE(self)) {

++self->count;

Py_RETURN_TRUE;

}

int use_deadline = (timeout_obj != Py_None);

if (use_deadline) {

double timeout = PyFloat_AsDouble(timeout_obj);

if (PyErr_Occurred()) {

return NULL;

}

if (timeout < 0.0) {

timeout = 0.0;

}

struct timeval now;

if (gettimeofday(&now, NULL) < 0) {

PyErr_SetFromErrno(PyExc_OSError);

return NULL;

}

long sec = (long) timeout;

long nsec = (long) (1e9 * (timeout - sec) + 0.5);

deadline.tv_sec = now.tv_sec + sec;

deadline.tv_nsec = now.tv_usec * 1000 + nsec;

deadline.tv_sec += (deadline.tv_nsec / 1000000000);

deadline.tv_nsec %= 1000000000;

}

/* Check whether we can acquire without releasing the GIL and blocking */

do {

res = sem_trywait(self->handle);

err = errno;

} while (res < 0 && errno == EINTR && !PyErr_CheckSignals());

errno = err;

if (res < 0 && errno == EAGAIN && blocking) {

/* Couldn't acquire immediately, need to block */

do {

Py_BEGIN_ALLOW_THREADS

if (!use_deadline) {

res = sem_wait(self->handle);

}

else {

res = sem_timedwait(self->handle, &deadline);

}

Py_END_ALLOW_THREADS

err = errno;

if (res == MP_EXCEPTION_HAS_BEEN_SET)

break;

} while (res < 0 && errno == EINTR && !PyErr_CheckSignals());

}

if (res < 0) {

errno = err;

if (errno == EAGAIN || errno == ETIMEDOUT)

Py_RETURN_FALSE;

else if (errno == EINTR)

return NULL;

else

return PyErr_SetFromErrno(PyExc_OSError);

}

++self->count;

self->last_tid = PyThread_get_thread_ident();

Py_RETURN_TRUE;

}

static PyObject *

_multiprocessing_SemLock_release_impl(SemLockObject *self)

{

if (self->kind == RECURSIVE_MUTEX) {

if (!ISMINE(self)) {

PyErr_SetString(PyExc_AssertionError, "attempt to "

"release recursive lock not owned "

"by thread");

return NULL;

}

if (self->count > 1) {

--self->count;

Py_RETURN_NONE;

}

assert(self->count == 1);

} else {

#ifdef HAVE_BROKEN_SEM_GETVALUE

/* We will only check properly the maxvalue == 1 case */

if (self->maxvalue == 1) {

/* make sure that already locked */

if (sem_trywait(self->handle) < 0) {

if (errno != EAGAIN) {

PyErr_SetFromErrno(PyExc_OSError);

return NULL;

}

/* it is already locked as expected */

} else {

/* it was not locked so undo wait and raise */

if (sem_post(self->handle) < 0) {

PyErr_SetFromErrno(PyExc_OSError);

return NULL;

}

PyErr_SetString(PyExc_ValueError, "semaphore "

"or lock released too many "

"times");

return NULL;

}

}

int sval;

/* This check is not an absolute guarantee that the semaphore

if (sem_getvalue(self->handle, &sval) < 0) {

return PyErr_SetFromErrno(PyExc_OSError);

} else if (sval >= self->maxvalue) {

PyErr_SetString(PyExc_ValueError, "semaphore or lock "

"released too many times");

return NULL;

}

}

if (sem_post(self->handle) < 0)

return PyErr_SetFromErrno(PyExc_OSError);

--self->count;

Py_RETURN_NONE;

}

#endif /* !MS_WINDOWS */

static PyObject *

newsemlockobject(PyTypeObject *type, SEM_HANDLE handle, int kind, int maxvalue,

char *name)

{

SemLockObject *self = (SemLockObject *)type->tp_alloc(type, 0);

if (!self)

return NULL;

self->handle = handle;

self->kind = kind;

self->count = 0;

self->last_tid = 0;

self->maxvalue = maxvalue;

self->name = name;

return (PyObject*)self;

}

static PyObject *

_multiprocessing_SemLock_impl(PyTypeObject *type, int kind, int value,

int maxvalue, const char *name, int unlink)

{

SEM_HANDLE handle = SEM_FAILED;

PyObject *result;

char *name_copy = NULL;

if (kind != RECURSIVE_MUTEX && kind != SEMAPHORE) {

PyErr_SetString(PyExc_ValueError, "unrecognized kind");

return NULL;

}

if (!unlink) {

name_copy = PyMem_Malloc(strlen(name) + 1);

if (name_copy == NULL) {

return PyErr_NoMemory();

}

strcpy(name_copy, name);

}

SEM_CLEAR_ERROR();

handle = SEM_CREATE(name, value, maxvalue);

/* On Windows we should fail if GetLastError()==ERROR_ALREADY_EXISTS */

if (handle == SEM_FAILED || SEM_GET_LAST_ERROR() != 0)

goto failure;

if (unlink && SEM_UNLINK(name) < 0)

goto failure;

result = newsemlockobject(type, handle, kind, maxvalue, name_copy);

if (!result)

goto failure;

return result;

failure:

if (!PyErr_Occurred()) {

_PyMp_SetError(NULL, MP_STANDARD_ERROR);

}

if (handle != SEM_FAILED)

SEM_CLOSE(handle);

PyMem_Free(name_copy);

return NULL;

}

static PyObject *

_multiprocessing_SemLock__rebuild_impl(PyTypeObject *type, SEM_HANDLE handle,

int kind, int maxvalue,

const char *name)

{

char *name_copy = NULL;

if (name != NULL) {

name_copy = PyMem_Malloc(strlen(name) + 1);

if (name_copy == NULL)

return PyErr_NoMemory();

strcpy(name_copy, name);

}

#ifndef MS_WINDOWS

if (name != NULL) {

handle = sem_open(name, 0);

if (handle == SEM_FAILED) {

PyErr_SetFromErrno(PyExc_OSError);

PyMem_Free(name_copy);

return NULL;

}

}

return newsemlockobject(type, handle, kind, maxvalue, name_copy);

}

static void

semlock_dealloc(PyObject *op)

{

SemLockObject *self = _SemLockObject_CAST(op);

PyTypeObject *tp = Py_TYPE(self);

PyObject_GC_UnTrack(self);

if (self->handle != SEM_FAILED)

SEM_CLOSE(self->handle);

PyMem_Free(self->name);

tp->tp_free(self);

Py_DECREF(tp);

}

static PyObject *

_multiprocessing_SemLock__count_impl(SemLockObject *self)

{

return PyLong_FromLong((long)self->count);

}

static PyObject *

_multiprocessing_SemLock__is_mine_impl(SemLockObject *self)

{

/* only makes sense for a lock */

return PyBool_FromLong(ISMINE(self));

}

static PyObject *

_multiprocessing_SemLock__get_value_impl(SemLockObject *self)

{

#ifdef HAVE_BROKEN_SEM_GETVALUE

PyErr_SetNone(PyExc_NotImplementedError);

return NULL;

int sval;

if (SEM_GETVALUE(self->handle, &sval) < 0)

return _PyMp_SetError(NULL, MP_STANDARD_ERROR);

/* some posix implementations use negative numbers to indicate

if (sval < 0)

sval = 0;

return PyLong_FromLong((long)sval);

}

static PyObject *

_multiprocessing_SemLock__is_zero_impl(SemLockObject *self)

{

#ifdef HAVE_BROKEN_SEM_GETVALUE

if (sem_trywait(self->handle) < 0) {

if (errno == EAGAIN)

Py_RETURN_TRUE;

return _PyMp_SetError(NULL, MP_STANDARD_ERROR);

} else {

if (sem_post(self->handle) < 0)

return _PyMp_SetError(NULL, MP_STANDARD_ERROR);

Py_RETURN_FALSE;

}

int sval;

if (SEM_GETVALUE(self->handle, &sval) < 0)

return _PyMp_SetError(NULL, MP_STANDARD_ERROR);

return PyBool_FromLong((long)sval == 0);

}

static PyObject *

_multiprocessing_SemLock__after_fork_impl(SemLockObject *self)

{

self->count = 0;

Py_RETURN_NONE;

}

static PyObject *

_multiprocessing_SemLock___enter___impl(SemLockObject *self)

{

return _multiprocessing_SemLock_acquire_impl(self, 1, Py_None);

}

static PyObject *

_multiprocessing_SemLock___exit___impl(SemLockObject *self,

PyObject *exc_type,

PyObject *exc_value, PyObject *exc_tb)

{

return _multiprocessing_SemLock_release_impl(self);

}

static PyMethodDef semlock_methods[] = {

_MULTIPROCESSING_SEMLOCK_ACQUIRE_METHODDEF

_MULTIPROCESSING_SEMLOCK_RELEASE_METHODDEF

_MULTIPROCESSING_SEMLOCK___ENTER___METHODDEF

_MULTIPROCESSING_SEMLOCK___EXIT___METHODDEF

_MULTIPROCESSING_SEMLOCK__COUNT_METHODDEF

_MULTIPROCESSING_SEMLOCK__IS_MINE_METHODDEF

_MULTIPROCESSING_SEMLOCK__GET_VALUE_METHODDEF

_MULTIPROCESSING_SEMLOCK__IS_ZERO_METHODDEF

_MULTIPROCESSING_SEMLOCK__REBUILD_METHODDEF

_MULTIPROCESSING_SEMLOCK__AFTER_FORK_METHODDEF

{NULL}

};

static PyMemberDef semlock_members[] = {

{"handle", T_SEM_HANDLE, offsetof(SemLockObject, handle), Py_READONLY,

""},

{"kind", Py_T_INT, offsetof(SemLockObject, kind), Py_READONLY,

""},

{"maxvalue", Py_T_INT, offsetof(SemLockObject, maxvalue), Py_READONLY,

""},

{"name", Py_T_STRING, offsetof(SemLockObject, name), Py_READONLY,

""},

{NULL}

};

static PyType_Slot _PyMp_SemLockType_slots[] = {

{Py_tp_dealloc, semlock_dealloc},

{Py_tp_getattro, PyObject_GenericGetAttr},

{Py_tp_setattro, PyObject_GenericSetAttr},

{Py_tp_methods, semlock_methods},

{Py_tp_members, semlock_members},

{Py_tp_alloc, PyType_GenericAlloc},

{Py_tp_new, _multiprocessing_SemLock},

{Py_tp_traverse, _PyObject_VisitType},

{Py_tp_free, PyObject_GC_Del},

{Py_tp_doc, (void *)PyDoc_STR("Semaphore/Mutex type")},

{0, 0},

};

PyType_Spec _PyMp_SemLockType_spec = {

.name = "_multiprocessing.SemLock",

.basicsize = sizeof(SemLockObject),

.flags = (Py_TPFLAGS_DEFAULT | Py_TPFLAGS_BASETYPE |

Py_TPFLAGS_HAVE_GC | Py_TPFLAGS_IMMUTABLETYPE),

.slots = _PyMp_SemLockType_slots,

};

PyObject *

_PyMp_sem_unlink(const char *name)

{

if (SEM_UNLINK(name) < 0) {

_PyMp_SetError(NULL, MP_STANDARD_ERROR);

return NULL;

}

Py_RETURN_NONE;

}

#endif // HAVE_MP_SEMAPHORE