From af8bb8fd05e469beee767bae2ec60e7b5ee7dde6 Mon Sep 17 00:00:00 2001 From: "roberto@precise64" Date: Thu, 1 Mar 2012 09:06:02 +0100 Subject: [PATCH] fixed locking in multithread --- lock.c | 72 +++++++++++++-------- plugins/python/uwsgi_pymodule.c | 109 ++++++++++++++++++++++++++------ 2 files changed, 136 insertions(+), 45 deletions(-) diff --git a/lock.c b/lock.c index 80007e1c..126b1b58 100644 --- a/lock.c +++ b/lock.c @@ -47,33 +47,39 @@ static struct uwsgi_lock_item *uwsgi_register_lock(char *id, int rw) { #define UWSGI_LOCK_ENGINE_NAME "pthread mutexes" -#define UWSGI_LOCK_SIZE sizeof(pthread_mutexattr_t) + sizeof(pthread_mutex_t) +#define UWSGI_LOCK_SIZE sizeof(pthread_mutex_t) #ifdef OBSOLETE_LINUX_KERNEL -#define UWSGI_RWLOCK_SIZE sizeof(pthread_mutexattr_t) + sizeof(pthread_mutex_t) +#define UWSGI_RWLOCK_SIZE sizeof(pthread_mutex_t) #else -#define UWSGI_RWLOCK_SIZE sizeof(pthread_rwlockattr_t) + sizeof(pthread_rwlock_t) +#define UWSGI_RWLOCK_SIZE sizeof(pthread_rwlock_t) #endif // REMEMBER lock must contains space for both pthread_mutex_t and pthread_mutexattr_t !!! struct uwsgi_lock_item *uwsgi_lock_fast_init(char *id) { + + pthread_mutexattr_t attr; struct uwsgi_lock_item *uli = uwsgi_register_lock(id, 0); - if (pthread_mutexattr_init((pthread_mutexattr_t *) uli->lock_ptr)) { + uwsgi_log("initializing lock %s\n", id); + + if (pthread_mutexattr_init(&attr)) { uwsgi_log("unable to allocate mutexattr structure\n"); exit(1); } - if (pthread_mutexattr_setpshared((pthread_mutexattr_t *) uli->lock_ptr, PTHREAD_PROCESS_SHARED)) { + if (pthread_mutexattr_setpshared(&attr, PTHREAD_PROCESS_SHARED)) { uwsgi_log("unable to share mutex\n"); exit(1); } - if (pthread_mutex_init((pthread_mutex_t *) (uli->lock_ptr + sizeof(pthread_mutexattr_t)), (pthread_mutexattr_t *) uli->lock_ptr)) { + if (pthread_mutex_init((pthread_mutex_t *) uli->lock_ptr, &attr)) { uwsgi_log("unable to initialize mutex\n"); exit(1); } + pthread_mutexattr_destroy(&attr); + uli->can_deadlock = 1; return uli; @@ -81,8 +87,8 @@ struct uwsgi_lock_item *uwsgi_lock_fast_init(char *id) { pid_t uwsgi_lock_fast_check(struct uwsgi_lock_item *uli) { - if (pthread_mutex_trylock((pthread_mutex_t *) (uli->lock_ptr + sizeof(pthread_mutexattr_t))) == 0 ) { - pthread_mutex_unlock((pthread_mutex_t *) (uli->lock_ptr + sizeof(pthread_mutexattr_t))); + if (pthread_mutex_trylock((pthread_mutex_t *) uli->lock_ptr) == 0 ) { + pthread_mutex_unlock((pthread_mutex_t *) uli->lock_ptr); return 0; } return uli->pid; @@ -93,8 +99,8 @@ pid_t uwsgi_rwlock_fast_check(struct uwsgi_lock_item *uli) { return uwsgi_lock_fast_check(uli); #else - if (pthread_rwlock_trywrlock((pthread_rwlock_t *) (uli->lock_ptr + sizeof(pthread_rwlockattr_t)) ) == 0 ) { - pthread_rwlock_unlock((pthread_rwlock_t *) (uli->lock_ptr + sizeof(pthread_rwlockattr_t))); + if (pthread_rwlock_trywrlock((pthread_rwlock_t *) uli->lock_ptr) == 0 ) { + pthread_rwlock_unlock((pthread_rwlock_t *) uli->lock_ptr); return 0; } return uli->pid; @@ -105,7 +111,7 @@ void uwsgi_rlock_fast(struct uwsgi_lock_item *uli) { #ifdef OBSOLETE_LINUX_KERNEL uwsgi_lock_fast(uli); #else - pthread_rwlock_rdlock((pthread_rwlock_t *) (uli->lock_ptr + sizeof(pthread_rwlockattr_t))); + pthread_rwlock_rdlock((pthread_rwlock_t *) uli->lock_ptr); uli->pid = uwsgi.mypid; #endif } @@ -114,7 +120,7 @@ void uwsgi_wlock_fast(struct uwsgi_lock_item *uli) { #ifdef OBSOLETE_LINUX_KERNEL uwsgi_lock_fast(uli); #else - pthread_rwlock_wrlock((pthread_rwlock_t *) (uli->lock_ptr + sizeof(pthread_rwlockattr_t))); + pthread_rwlock_wrlock((pthread_rwlock_t *) uli->lock_ptr); uli->pid = uwsgi.mypid; #endif } @@ -123,20 +129,25 @@ void uwsgi_rwunlock_fast(struct uwsgi_lock_item *uli) { #ifdef OBSOLETE_LINUX_KERNEL uwsgi_unlock_fast(uli); #else - pthread_rwlock_unlock((pthread_rwlock_t *) (uli->lock_ptr + sizeof(pthread_rwlockattr_t))); + pthread_rwlock_unlock((pthread_rwlock_t *) uli->lock_ptr); uli->pid = 0; #endif } void uwsgi_lock_fast(struct uwsgi_lock_item *uli) { - pthread_mutex_lock((pthread_mutex_t *) (uli->lock_ptr + sizeof(pthread_mutexattr_t))); + uwsgi_log("locking\n"); + pthread_mutex_lock((pthread_mutex_t *) uli->lock_ptr); + uwsgi_log("in-locking\n"); uli->pid = uwsgi.mypid; + uwsgi_log("in-locking 2\n"); } void uwsgi_unlock_fast(struct uwsgi_lock_item *uli) { - pthread_mutex_unlock((pthread_mutex_t *) (uli->lock_ptr + sizeof(pthread_mutexattr_t))); + uwsgi_log("unlock !!!\n"); + pthread_mutex_unlock((pthread_mutex_t *) uli->lock_ptr); + uwsgi_log("unlocked !!!\n"); uli->pid = 0; } @@ -147,22 +158,26 @@ struct uwsgi_lock_item *uwsgi_rwlock_fast_init(char *id) { return uwsgi_lock_fast_init(uli); #else + pthread_rwlockattr_t attr; + struct uwsgi_lock_item *uli = uwsgi_register_lock(id, 1); - if (pthread_rwlockattr_init((pthread_rwlockattr_t *) uli->lock_ptr)) { + if (pthread_rwlockattr_init(&attr)) { uwsgi_log("unable to allocate rwlock structure\n"); exit(1); } - if (pthread_rwlockattr_setpshared((pthread_rwlockattr_t *) uli->lock_ptr, PTHREAD_PROCESS_SHARED)) { + if (pthread_rwlockattr_setpshared(&attr, PTHREAD_PROCESS_SHARED)) { uwsgi_log("unable to share rwlock\n"); exit(1); } - if (pthread_rwlock_init((pthread_rwlock_t *) (uli->lock_ptr + sizeof(pthread_rwlockattr_t)), (pthread_rwlockattr_t *) uli->lock_ptr)) { + if (pthread_rwlock_init((pthread_rwlock_t *) uli->lock_ptr, &attr)) { uwsgi_log("unable to initialize rwlock\n"); exit(1); } + pthread_rwlockattr_destroy(&attr); + uli->can_deadlock = 1; return uli; @@ -290,15 +305,15 @@ void uwsgi_rwunlock_fast(struct uwsgi_lock_item *uli) { uwsgi_unlock_fast(uli); struct uwsgi_lock_item *uwsgi_lock_flock_init(char *id) { struct uwsgi_lock_item *uli = uwsgi_register_lock(id, 0); + FILE *tf = tmpfile(); - int fd; if (!tf) { uwsgi_error_open("temp lock file"); exit(1); } - fd = fileno(tf); + int fd = fileno(tf); memcpy(uli->lock_ptr, &fd, sizeof(int)); return uli; } @@ -310,7 +325,9 @@ void uwsgi_lock_flock(struct uwsgi_lock_item *uli) { #ifdef __sun__ if (lockf(fd, F_LOCK, 0)) { uwsgi_error("lockf()"); } #else + uwsgi_log("flocking %d\n", fd); if (flock(fd, LOCK_EX)) { uwsgi_error("flock()"); } + uwsgi_log("inflocking %d\n", fd); #endif } @@ -320,6 +337,7 @@ void uwsgi_unlock_flock(struct uwsgi_lock_item *uli) { #ifdef __sun__ if (lockf(fd, F_ULOCK, 0)) { uwsgi_error("lockf()"); } #else + uwsgi_log("un-flocking %d\n", fd); if (flock(fd, LOCK_UN)) { uwsgi_error("flock()"); } #endif } @@ -362,15 +380,15 @@ void uwsgi_setup_locking() { uwsgi_log("lock engine: flock\n"); uwsgi.lock_ops.lock_init = uwsgi_lock_flock_init; uwsgi.lock_ops.lock_check = uwsgi_lock_flock_check; - uwsgi.lock_ops.lock = uwsgi_lock_fast; - uwsgi.lock_ops.unlock = uwsgi_unlock_fast; + uwsgi.lock_ops.lock = uwsgi_lock_flock; + uwsgi.lock_ops.unlock = uwsgi_unlock_flock; uwsgi.lock_ops.rwlock_init = uwsgi_rwlock_flock_init; uwsgi.lock_ops.rwlock_check = uwsgi_rwlock_flock_check; - uwsgi.lock_ops.rlock = uwsgi_rlock_fast; - uwsgi.lock_ops.wlock = uwsgi_wlock_fast; - uwsgi.lock_ops.rwunlock = uwsgi_rwunlock_fast; - uwsgi.lock_size = UWSGI_LOCK_SIZE; - uwsgi.rwlock_size = UWSGI_RWLOCK_SIZE; + uwsgi.lock_ops.rlock = uwsgi_rlock_flock; + uwsgi.lock_ops.wlock = uwsgi_wlock_flock; + uwsgi.lock_ops.rwunlock = uwsgi_rwunlock_flock; + uwsgi.lock_size = 8; + uwsgi.rwlock_size = 8; return; } else if (!strcmp(uwsgi.lock_engine, "ipcsem")) { diff --git a/plugins/python/uwsgi_pymodule.c b/plugins/python/uwsgi_pymodule.c index 42e5b1f8..68880444 100644 --- a/plugins/python/uwsgi_pymodule.c +++ b/plugins/python/uwsgi_pymodule.c @@ -913,11 +913,16 @@ PyObject *py_uwsgi_is_locked(PyObject * self, PyObject * args) { return PyErr_Format(PyExc_ValueError, "Invalid lock number"); } + UWSGI_RELEASE_GIL + if (uwsgi_lock_check(uwsgi.user_lock[lock_num]) == 0) { + UWSGI_GET_GIL Py_INCREF(Py_False); return Py_False; } + UWSGI_GET_GIL + Py_INCREF(Py_True); return Py_True; } @@ -942,7 +947,9 @@ PyObject *py_uwsgi_lock(PyObject * self, PyObject * args) { return PyErr_Format(PyExc_ValueError, "Invalid lock number"); } + UWSGI_RELEASE_GIL uwsgi_lock(uwsgi.user_lock[lock_num]); + UWSGI_GET_GIL Py_INCREF(Py_None); return Py_None; @@ -966,7 +973,6 @@ PyObject *py_uwsgi_unlock(PyObject * self, PyObject * args) { return PyErr_Format(PyExc_ValueError, "Invalid lock number"); } - uwsgi_unlock(uwsgi.user_lock[lock_num]); Py_INCREF(Py_None); @@ -1064,7 +1070,9 @@ PyObject *py_uwsgi_farm_msg(PyObject * self, PyObject * args) { for(i=0;imule_queue_pipe[0], message, message_len); + UWSGI_GET_GIL } else { if (PyString_Check(mule_obj)) { @@ -1122,7 +1132,9 @@ PyObject *py_uwsgi_mule_msg(PyObject * self, PyObject * args) { } if (fd > -1) { + UWSGI_RELEASE_GIL mule_send_msg(fd, message, message_len); + UWSGI_GET_GIL } } @@ -1267,6 +1279,8 @@ PyObject *py_uwsgi_sharedarea_inclong(PyObject * self, PyObject * args) { Py_INCREF(Py_None); return Py_None; } + + UWSGI_RELEASE_GIL uwsgi_wlock(uwsgi.sa_lock); @@ -1274,11 +1288,10 @@ PyObject *py_uwsgi_sharedarea_inclong(PyObject * self, PyObject * args) { value = current_value + value; memcpy(uwsgi.sharedarea + pos, &value, 8); - PyObject *ret = PyInt_FromLong(value); - uwsgi_rwunlock(uwsgi.sa_lock); + UWSGI_GET_GIL - return ret; + return PyInt_FromLong(value); } @@ -1300,16 +1313,17 @@ PyObject *py_uwsgi_sharedarea_writelong(PyObject * self, PyObject * args) { return Py_None; } + UWSGI_RELEASE_GIL + uwsgi_wlock(uwsgi.sa_lock); memcpy(uwsgi.sharedarea + pos, &value, 8); - - PyObject *ret = PyInt_FromLong(value); - uwsgi_rwunlock(uwsgi.sa_lock); - return ret; + UWSGI_GET_GIL + + return PyInt_FromLong(value); } @@ -1332,15 +1346,19 @@ PyObject *py_uwsgi_sharedarea_write(PyObject * self, PyObject * args) { return Py_None; } + UWSGI_RELEASE_GIL + uwsgi_wlock(uwsgi.sa_lock); memcpy(uwsgi.sharedarea + pos, value, value_len); - PyObject *ret = PyInt_FromLong(value_len); uwsgi_rwunlock(uwsgi.sa_lock); + + UWSGI_GET_GIL + + return PyInt_FromLong(value_len); - return ret; } @@ -1363,15 +1381,18 @@ PyObject *py_uwsgi_sharedarea_writebyte(PyObject * self, PyObject * args) { return Py_None; } + UWSGI_RELEASE_GIL + uwsgi_wlock(uwsgi.sa_lock); uwsgi.sharedarea[pos] = value; - PyObject *ret = PyInt_FromLong(uwsgi.sharedarea[pos]); uwsgi_rwunlock(uwsgi.sa_lock); - return ret; + UWSGI_GET_GIL + + return PyInt_FromLong(value); } @@ -1393,15 +1414,18 @@ PyObject *py_uwsgi_sharedarea_readlong(PyObject * self, PyObject * args) { return Py_None; } + UWSGI_RELEASE_GIL + uwsgi_wlock(uwsgi.sa_lock); memcpy(&value, uwsgi.sharedarea + pos, 8); - PyObject *ret = PyLong_FromLong(value); uwsgi_rwunlock(uwsgi.sa_lock); - return ret; + UWSGI_GET_GIL + + return PyLong_FromLong(value); } @@ -1423,13 +1447,17 @@ PyObject *py_uwsgi_sharedarea_readbyte(PyObject * self, PyObject * args) { return Py_None; } + UWSGI_RELEASE_GIL + uwsgi_wlock(uwsgi.sa_lock); - PyObject *ret = PyInt_FromLong(uwsgi.sharedarea[pos]); + char value = uwsgi.sharedarea[pos]; uwsgi_rwunlock(uwsgi.sa_lock); - return ret; + UWSGI_GET_GIL + + return PyInt_FromLong(value); } @@ -1451,12 +1479,20 @@ PyObject *py_uwsgi_sharedarea_read(PyObject * self, PyObject * args) { return Py_None; } + UWSGI_RELEASE_GIL + uwsgi_wlock(uwsgi.sa_lock); - PyObject *ret = PyString_FromStringAndSize(uwsgi.sharedarea + pos, len); + char *chunk = uwsgi_concat2n(uwsgi.sharedarea + pos, len, "", 0); uwsgi_rwunlock(uwsgi.sa_lock); + UWSGI_GET_GIL + + PyObject *ret = PyString_FromStringAndSize( chunk, len); + + free(chunk); + return ret; } @@ -1907,7 +1943,9 @@ PyObject *py_uwsgi_multicast(PyObject * self, PyObject * args) { uwsgi_message = uwsgi_malloc(message_len+4); memcpy(uwsgi_message+4, message, message_len); + UWSGI_RELEASE_GIL ret = send_udp_message(UWSGI_MODIFIER_MULTICAST, 0, host, uwsgi_message, message_len); + UWSGI_GET_GIL free(uwsgi_message); if (ret <= 0) { @@ -2277,9 +2315,12 @@ PyObject *py_uwsgi_route(PyObject * self, PyObject * args) { return PyErr_Format(PyExc_IOError, "unable to connect to host %s", addr); } + UWSGI_RELEASE_GIL if (uwsgi_send_message(uwsgi_fd, wsgi_req->uh.modifier1, wsgi_req->uh.modifier2, wsgi_req->buffer, wsgi_req->uh.pktsize, wsgi_req->poll.fd, wsgi_req->post_cl, 0) < 0) { + UWSGI_GET_GIL return PyErr_Format(PyExc_IOError, "unable to send uwsgi request to host %s", addr); } + UWSGI_GET_GIL // request sent, return the iterator response uwsgi_Iter *ui = PyObject_New(uwsgi_Iter, &uwsgi_IterType); @@ -2329,7 +2370,9 @@ PyObject *py_uwsgi_send_message(PyObject * self, PyObject * args) { uwsgi_fd = PyInt_AsLong(destination); } else if (PyString_Check(destination)) { + UWSGI_RELEASE_GIL uwsgi_fd = uwsgi_connect(PyString_AsString(destination), timeout, 0); + UWSGI_GET_GIL close_fd = 1; } @@ -2693,30 +2736,38 @@ PyObject *py_uwsgi_parse_file(PyObject * self, PyObject * args) { return NULL; } + UWSGI_RELEASE_GIL + fd = open(filename, O_RDONLY); if (fd < 0) { uwsgi_error_open(filename); + UWSGI_GET_GIL goto clear; } len = read(fd, &uh, 4); if (len != 4) { uwsgi_error("read()"); + UWSGI_GET_GIL goto clear2; } buffer = malloc(uh.pktsize); if (!buffer) { uwsgi_error("malloc()"); + UWSGI_GET_GIL goto clear2; } len = read(fd, buffer, uh.pktsize); if (len != uh.pktsize) { uwsgi_error("read()"); free(buffer); + UWSGI_GET_GIL goto clear2; } + UWSGI_GET_GIL + ptrbuf = buffer; bufferend = ptrbuf + uh.pktsize; @@ -3066,16 +3117,21 @@ PyObject *py_uwsgi_cache_del(PyObject * self, PyObject * args) { } if (remote && strlen(remote) > 0) { + UWSGI_RELEASE_GIL uwsgi_simple_send_string(remote, 111, 2, key, keylen, uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT]); + UWSGI_GET_GIL } else if (uwsgi.cache_max_items) { + UWSGI_RELEASE_GIL uwsgi_wlock(uwsgi.cache_lock); if (uwsgi_cache_del(key, keylen)) { uwsgi_rwunlock(uwsgi.cache_lock); + UWSGI_GET_GIL Py_INCREF(Py_None); return Py_None; } uwsgi_rwunlock(uwsgi.cache_lock); + UWSGI_GET_GIL } Py_INCREF(Py_True); @@ -3103,16 +3159,21 @@ PyObject *py_uwsgi_cache_set(PyObject * self, PyObject * args) { } if (remote && strlen(remote) > 0) { + UWSGI_RELEASE_GIL uwsgi_simple_send_string2(remote, 111, 1, key, keylen, value, vallen, uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT]); + UWSGI_GET_GIL } else if (uwsgi.cache_max_items) { + UWSGI_RELEASE_GIL uwsgi_wlock(uwsgi.cache_lock); if (uwsgi_cache_set(key, keylen, value, vallen, expires, 0)) { uwsgi_rwunlock(uwsgi.cache_lock); + UWSGI_GET_GIL Py_INCREF(Py_None); return Py_None; } uwsgi_rwunlock(uwsgi.cache_lock); + UWSGI_GET_GIL } Py_INCREF(Py_True); @@ -3139,16 +3200,21 @@ PyObject *py_uwsgi_cache_update(PyObject * self, PyObject * args) { } if (remote && strlen(remote) > 0) { + UWSGI_RELEASE_GIL uwsgi_simple_send_string2(remote, 111, 1, key, keylen, value, vallen, uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT]); + UWSGI_GET_GIL } else if (uwsgi.cache_max_items) { + UWSGI_RELEASE_GIL uwsgi_wlock(uwsgi.cache_lock); if (uwsgi_cache_set(key, keylen, value, vallen, expires, UWSGI_CACHE_FLAG_UPDATE)) { uwsgi_rwunlock(uwsgi.cache_lock); + UWSGI_GET_GIL Py_INCREF(Py_None); return Py_None; } uwsgi_rwunlock(uwsgi.cache_lock); + UWSGI_GET_GIL } Py_INCREF(Py_True); @@ -3431,7 +3497,9 @@ PyObject *py_uwsgi_cache_get(PyObject * self, PyObject * args) { } if (remote && strlen(remote) > 0) { + UWSGI_RELEASE_GIL uwsgi_simple_message_string(remote, 111, 0, key, keylen, buffer, &valsize16, uwsgi.shared->options[UWSGI_OPTION_SOCKET_TIMEOUT]); + UWSGI_GET_GIL if (valsize16 > 0) { value = buffer; valsize = valsize16; @@ -3441,14 +3509,16 @@ PyObject *py_uwsgi_cache_get(PyObject * self, PyObject * args) { #ifdef UWSGI_DEBUG gettimeofday(&tv, NULL); #endif + UWSGI_RELEASE_GIL uwsgi_rlock(uwsgi.cache_lock); value = uwsgi_cache_get(key, keylen, &valsize); if (!value) { uwsgi_rwunlock(uwsgi.cache_lock); + UWSGI_GET_GIL Py_INCREF(Py_None); return Py_None; } - res = PyString_FromStringAndSize(value, valsize); + char *chunk = uwsgi_concat2n(value, valsize, "", 0); #ifdef UWSGI_DEBUG gettimeofday(&tv2, NULL); if ((tv2.tv_sec* (1000*1000) + tv2.tv_usec) - (tv.tv_sec* (1000*1000) + tv.tv_usec) > 30000) { @@ -3456,6 +3526,9 @@ PyObject *py_uwsgi_cache_get(PyObject * self, PyObject * args) { } #endif uwsgi_rwunlock(uwsgi.cache_lock); + UWSGI_GET_GIL + res = PyString_FromStringAndSize(chunk, valsize); + free(chunk); return res; }