spooler multiprocess fixes

lock.c

@@ -457,7 +457,7 @@ void uwsgi_setup_locking() {
 int uwsgi_fcntl_lock(int fd) {
 	struct flock fl;
 	fl.l_type = F_WRLCK;
-	fl.l_whence = SEEK_CUR;
+	fl.l_whence = SEEK_SET;
 	fl.l_start = 0;
 	fl.l_len = 0;
 	fl.l_pid = 0;
@@ -473,7 +473,7 @@ int uwsgi_fcntl_is_locked(int fd) {
 
 	struct flock fl;
         fl.l_type = F_WRLCK;
-        fl.l_whence = SEEK_CUR;
+        fl.l_whence = SEEK_SET;
         fl.l_start = 0;
         fl.l_len = 0;
         fl.l_pid = 0;

plugins/python/uwsgi_pymodule.c

@@ -1715,6 +1715,8 @@ PyObject *py_uwsgi_send_spool(PyObject * self, PyObject * args, PyObject *kw) {
 		}
 	}
 
+	UWSGI_RELEASE_GIL
+
 	if (numprio) {
 		priority = uwsgi_num2str(numprio);
 	} 
@@ -1725,6 +1727,8 @@ PyObject *py_uwsgi_send_spool(PyObject * self, PyObject * args, PyObject *kw) {
 		
 	free(spool_buffer);
 
+	UWSGI_GET_GIL
+
 	Py_DECREF(spool_vars);
 
 	if (i > 0) {

plugins/python/wsgi_handlers.c

@@ -502,12 +502,14 @@ int uwsgi_request_wsgi(struct wsgi_request *wsgi_req) {
 	}
 
 	// this object must be freed/cleared always
+	UWSGI_GET_GIL
 	if (wsgi_req->async_input) {
                 Py_DECREF((PyObject *)wsgi_req->async_input);
         }
         if (wsgi_req->async_environ) {
                 PyDict_Clear(wsgi_req->async_environ);
         }
+	UWSGI_RELEASE_GIL
 
 clear:
 

spooler.c

@@ -453,20 +453,30 @@ void spooler_manage_task(struct uwsgi_spooler *uspool, char *dir, char *task) {
 			spool_fd = open(task, O_RDWR);
 
 			if (spool_fd < 0) {
-				uwsgi_error_open(task);
+				if (errno != ENOENT)
+					uwsgi_error_open(task);
 				return;
 			}
 
-			// check if the file is locked by anther process
+			// check if the file is locked by another process
 			if (uwsgi_fcntl_is_locked(spool_fd)) {
-				close(spool_fd);
+				uwsgi_protected_close(spool_fd);
 				return;
 			}
 
-			if (read(spool_fd, &uh, 4) != 4) {
+			// unlink() can destroy the lock !!!
+			if (access(task, R_OK | W_OK)) {
+				uwsgi_protected_close(spool_fd);
+				return;
+			}
+
+			ssize_t rlen = uwsgi_protected_read(spool_fd, &uh, 4);
+
+			if (rlen != 4) {
 				// it could be here for broken file or just opened one
-				uwsgi_error("read()");
-				close(spool_fd);
+				if (rlen < 0)
+					uwsgi_error("read()");
+				uwsgi_protected_close(spool_fd);
 				return;
 			}
 
@@ -474,10 +484,10 @@ void spooler_manage_task(struct uwsgi_spooler *uspool, char *dir, char *task) {
 			uh.pktsize = uwsgi_swap16(uh.pktsize);
 #endif
 
-			if (read(spool_fd, spool_buf, uh.pktsize) != uh.pktsize) {
+			if (uwsgi_protected_read(spool_fd, spool_buf, uh.pktsize) != uh.pktsize) {
 				uwsgi_error("read()");
 				destroy_spool(dir, task);	
-				close(spool_fd);
+				uwsgi_protected_close(spool_fd);
 				return;
 			}			
 					
@@ -485,16 +495,16 @@ void spooler_manage_task(struct uwsgi_spooler *uspool, char *dir, char *task) {
 			if (sf_lstat.st_size > (uh.pktsize+4)) {
 				body_len = sf_lstat.st_size - (uh.pktsize+4);
 				body = uwsgi_malloc(body_len);
-				if ((size_t)read(spool_fd, body, body_len) != body_len) {
+				if ((size_t)uwsgi_protected_read(spool_fd, body, body_len) != body_len) {
 					uwsgi_error("read()");
 					destroy_spool(dir, task);
-					close(spool_fd);
+					uwsgi_protected_close(spool_fd);
 					free(body);
 					return;
 				}
 			}
 
-			// not the task is running and should not be waken
+			// now the task is running and should not be waken up
 			uspool->running = 1;
 
 			uwsgi_log("[spooler %s pid: %d] managing request %s ...\n", uspool->dir, (int) uwsgi.mypid, task);
@@ -533,7 +543,7 @@ void spooler_manage_task(struct uwsgi_spooler *uspool, char *dir, char *task) {
 				free(body);
 
 			// here we free and unlock the task
-			close(spool_fd);
+			uwsgi_protected_close(spool_fd);
 			uspool->running = 0;
 
 			if (!callable_found) {

utils.c

@@ -4204,3 +4204,37 @@ int uwsgi_manage_exception(char *type, char *value, char *repr) {
 
 	return 0;
 }
+
+void uwsgi_protected_close(int fd) {
+
+	sigset_t mask, oset;
+	sigfillset(&mask);
+	if (sigprocmask(SIG_BLOCK, &mask, &oset)) {
+		uwsgi_error("sigprocmask()");
+		exit(1);
+	}
+	close(fd);
+	if (sigprocmask(SIG_SETMASK, &oset, NULL)) {
+		uwsgi_error("sigprocmask()");
+		exit(1);
+	}
+}
+
+ssize_t uwsgi_protected_read(int fd, void *buf, size_t len) {
+
+        sigset_t mask, oset;
+        sigfillset(&mask);
+        if (sigprocmask(SIG_BLOCK, &mask, &oset)) {
+                uwsgi_error("sigprocmask()");
+                exit(1);
+        }
+
+	ssize_t ret = read(fd, buf, len);
+
+        if (sigprocmask(SIG_SETMASK, &oset, NULL)) {
+                uwsgi_error("sigprocmask()");
+                exit(1);
+        }
+	return ret;
+}
+

uwsgi.h

@@ -2763,6 +2763,9 @@ struct uwsgi_lock_item *uwsgi_lock_ipcsem_init(char *);
 void uwsgi_write_pidfile(char *);
 int uwsgi_manage_exception(char *, char *, char *);
 
+void uwsgi_protected_close(int);
+ssize_t uwsgi_protected_read(int, void *, size_t);
+
 #ifdef UWSGI_AS_SHARED_LIBRARY
 int uwsgi_init(int, char **, char **);
 #endif
@@ -2771,4 +2774,3 @@ int uwsgi_init(int, char **, char **);
 }
 #endif
 
-