clearlinux/uwsgi
2
0
Fork 0
mirror of https://github.com/clearlinux/uwsgi.git synced 2026-09-07 06:06:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

fixed locking in multithread

Browse Source
This commit is contained in:
roberto@precise64
2012-03-01 09:06:02 +01:00
parent 57bade7eae
commit af8bb8fd05
2 changed files with 136 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lock.c
+45 -27
View File
@@ -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")) {
plugins/python/uwsgi_pymodule.c
+91 -18
View File
@@ -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;i<uwsgi.farms_cnt;i++) {
if (!strcmp(farm_name, uwsgi.farms[i].name)) {
UWSGI_RELEASE_GIL
len = write(uwsgi.farms[i].queue_pipe[0], message, message_len);
UWSGI_GET_GIL
if (len <= 0) {
uwsgi_error("write()");
}
@@ -1095,7 +1103,9 @@ PyObject *py_uwsgi_mule_msg(PyObject * self, PyObject * args) {
return PyErr_Format(PyExc_ValueError, "no mule configured");
if (mule_obj == NULL) {
UWSGI_RELEASE_GIL
mule_send_msg(uwsgi.shared->mule_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;
}