1 files changed, 46 insertions, 30 deletions

diff --git a/fuse/fuse_loop_mt.c b/fuse/fuse_loop_mt.c
index a73c399b7..7e400c2a4 100644
--- a/fuse/fuse_loop_mt.c
+++ b/fuse/fuse_loop_mt.c
@@ -9,6 +9,7 @@
 #include "fuse_lowlevel.h"
 #include "fuse_misc.h"
 #include "fuse_kernel.h"
+#include "fuse_i.h"
 
 #include <stdio.h>
 #include <stdlib.h>
@@ -18,8 +19,7 @@
 #include <semaphore.h>
 #include <errno.h>
 #include <sys/time.h>
-
-#ifdef __MULTI_THREAD
+#include <pthread.h>
 
 /* Environment var controlling the thread stack size */
 #define ENVNAME_THREAD_STACK "FUSE_THREAD_STACK"
@@ -65,7 +65,7 @@ static void list_del_worker(struct fuse_worker *w)
 #define PTHREAD_CANCEL_ENABLE 0
 #define PTHREAD_CANCEL_DISABLE 1
 
-static int fuse_start_thread(struct fuse_mt *mt);
+static int fuse_loop_start_thread(struct fuse_mt *mt);
 
 static void *fuse_do_work(void *data)
 {
@@ -75,10 +75,14 @@ static void *fuse_do_work(void *data)
 	while (!fuse_session_exited(mt->se)) {
 		int isforget = 0;
 		struct fuse_chan *ch = mt->prevch;
+		struct fuse_buf fbuf = {
+			.mem = w->buf,
+			.size = w->bufsize,
+		};
 		int res;
 
 		pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);
-		res = fuse_chan_recv(&ch, w->buf, w->bufsize);
+		res = fuse_session_receive_buf(mt->se, &fbuf, &ch);
 		pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL);
 		if (res == -EINTR)
 			continue;
@@ -100,16 +104,21 @@ static void *fuse_do_work(void *data)
 		 * This disgusting hack is needed so that zillions of threads
 		 * are not created on a burst of FORGET messages
 		 */
-		if (((struct fuse_in_header *) w->buf)->opcode == FUSE_FORGET)
-			isforget = 1;
+		if (!(fbuf.flags & FUSE_BUF_IS_FD)) {
+			struct fuse_in_header *in = fbuf.mem;
+
+			if (in->opcode == FUSE_FORGET ||
+			    in->opcode == FUSE_BATCH_FORGET)
+				isforget = 1;
+		}
 
 		if (!isforget)
 			mt->numavail--;
 		if (mt->numavail == 0)
-			fuse_start_thread(mt);
+			fuse_loop_start_thread(mt);
 		pthread_mutex_unlock(&mt->lock);
 
-		fuse_session_process(mt->se, w->buf, res, ch);
+		fuse_session_process_buf(mt->se, &fbuf, ch);
 
 		pthread_mutex_lock(&mt->lock);
 		if (!isforget)
@@ -133,33 +142,17 @@ static void *fuse_do_work(void *data)
 	}
 
 	sem_post(&mt->finish);
-	pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);
-	pause();
 
 	return NULL;
 }
 
-static int fuse_start_thread(struct fuse_mt *mt)
+int fuse_start_thread(pthread_t *thread_id, void *(*func)(void *), void *arg)
 {
 	sigset_t oldset;
 	sigset_t newset;
 	int res;
 	pthread_attr_t attr;
 	char *stack_size;
-	struct fuse_worker *w = malloc(sizeof(struct fuse_worker));
-	if (!w) {
-		fprintf(stderr, "fuse: failed to allocate worker structure\n");
-		return -1;
-	}
-	memset(w, 0, sizeof(struct fuse_worker));
-	w->bufsize = fuse_chan_bufsize(mt->prevch);
-	w->buf = malloc(w->bufsize);
-	w->mt = mt;
-	if (!w->buf) {
-		fprintf(stderr, "fuse: failed to allocate read buffer\n");
-		free(w);
-		return -1;
-	}
 
 	/* Override default stack size */
 	pthread_attr_init(&attr);
@@ -174,12 +167,38 @@ static int fuse_start_thread(struct fuse_mt *mt)
 	sigaddset(&newset, SIGHUP);
 	sigaddset(&newset, SIGQUIT);
 	pthread_sigmask(SIG_BLOCK, &newset, &oldset);
-	res = pthread_create(&w->thread_id, &attr, fuse_do_work, w);
+	res = pthread_create(thread_id, &attr, func, arg);
 	pthread_sigmask(SIG_SETMASK, &oldset, NULL);
 	pthread_attr_destroy(&attr);
 	if (res != 0) {
 		fprintf(stderr, "fuse: error creating thread: %s\n",
 			strerror(res));
+		return -1;
+	}
+
+	return 0;
+}
+
+static int fuse_loop_start_thread(struct fuse_mt *mt)
+{
+	int res;
+	struct fuse_worker *w = malloc(sizeof(struct fuse_worker));
+	if (!w) {
+		fprintf(stderr, "fuse: failed to allocate worker structure\n");
+		return -1;
+	}
+	memset(w, 0, sizeof(struct fuse_worker));
+	w->bufsize = fuse_chan_bufsize(mt->prevch);
+	w->buf = malloc(w->bufsize);
+	w->mt = mt;
+	if (!w->buf) {
+		fprintf(stderr, "fuse: failed to allocate read buffer\n");
+		free(w);
+		return -1;
+	}
+
+	res = fuse_start_thread(&w->thread_id, fuse_do_work, w);
+	if (res == -1) {
 		free(w->buf);
 		free(w);
 		return -1;
@@ -219,7 +238,7 @@ int fuse_session_loop_mt(struct fuse_session *se)
 	fuse_mutex_init(&mt.lock);
 
 	pthread_mutex_lock(&mt.lock);
-	err = fuse_start_thread(&mt);
+	err = fuse_loop_start_thread(&mt);
 	pthread_mutex_unlock(&mt.lock);
 	if (!err) {
 		/* sem_wait() is interruptible */
@@ -229,7 +248,6 @@ int fuse_session_loop_mt(struct fuse_session *se)
 		for (w = mt.main.next; w != &mt.main; w = w->next)
 			pthread_cancel(w->thread_id);
 		mt.exit = 1;
-		pthread_mutex_unlock(&mt.lock);
 
 		while (mt.main.next != &mt.main)
 			fuse_join_worker(&mt, mt.main.next);
@@ -242,5 +260,3 @@ int fuse_session_loop_mt(struct fuse_session *se)
 	fuse_session_reset(se);
 	return err;
 }
-
-#endif