lib/fuse_loop_mt.c

   1 /*
   2   FUSE: Filesystem in Userspace
   3   Copyright (C) 2001-2007  Miklos Szeredi <miklos@szeredi.hu>
   4
   5   This program can be distributed under the terms of the GNU LGPLv2.
   6   See the file COPYING.LIB.
   7 */
   8
   9 #include "fuse_lowlevel.h"
  10 #include "fuse_misc.h"
  11 #include "fuse_kernel.h"
  12
  13 #include <stdio.h>
  14 #include <stdlib.h>
  15 #include <string.h>
  16 #include <unistd.h>
  17 #include <signal.h>
  18 #include <semaphore.h>
  19 #include <errno.h>
  20 #include <sys/time.h>
  21
  22 /* Environment var controlling the thread stack size */
  23 #define ENVNAME_THREAD_STACK "FUSE_THREAD_STACK"
  24
  25 struct fuse_worker {
  26         struct fuse_worker *prev;
  27         struct fuse_worker *next;
  28         pthread_t thread_id;
  29         size_t bufsize;
  30         char *buf;
  31         struct fuse_mt *mt;
  32 };
  33
  34 struct fuse_mt {
  35         pthread_mutex_t lock;
  36         int numworker;
  37         int numavail;
  38         struct fuse_session *se;
  39         struct fuse_chan *prevch;
  40         struct fuse_worker main;
  41         sem_t finish;
  42         int exit;
  43         int error;
  44 };
  45
  46 static void list_add_worker(struct fuse_worker *w, struct fuse_worker *next)
  47 {
  48         struct fuse_worker *prev = next->prev;
  49         w->next = next;
  50         w->prev = prev;
  51         prev->next = w;
  52         next->prev = w;
  53 }
  54
  55 static void list_del_worker(struct fuse_worker *w)
  56 {
  57         struct fuse_worker *prev = w->prev;
  58         struct fuse_worker *next = w->next;
  59         prev->next = next;
  60         next->prev = prev;
  61 }
  62
  63 static int fuse_start_thread(struct fuse_mt *mt);
  64
  65 static void *fuse_do_work(void *data)
  66 {
  67         struct fuse_worker *w = (struct fuse_worker *) data;
  68         struct fuse_mt *mt = w->mt;
  69
  70         while (!fuse_session_exited(mt->se)) {
  71                 int isforget = 0;
  72                 struct fuse_chan *ch = mt->prevch;
  73                 int res;
  74
  75                 pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);
  76                 res = fuse_chan_recv(&ch, w->buf, w->bufsize);
  77                 pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL);
  78                 if (res == -EINTR)
  79                         continue;
  80                 if (res <= 0) {
  81                         if (res < 0) {
  82                                 fuse_session_exit(mt->se);
  83                                 mt->error = -1;
  84                         }
  85                         break;
  86                 }
  87
  88                 pthread_mutex_lock(&mt->lock);
  89                 if (mt->exit) {
  90                         pthread_mutex_unlock(&mt->lock);
  91                         return NULL;
  92                 }
  93
  94                 /*
  95                  * This disgusting hack is needed so that zillions of threads
  96                  * are not created on a burst of FORGET messages
  97                  */
  98                 if (((struct fuse_in_header *) w->buf)->opcode == FUSE_FORGET)
  99                         isforget = 1;
 100
 101                 if (!isforget)
 102                         mt->numavail--;
 103                 if (mt->numavail == 0)
 104                         fuse_start_thread(mt);
 105                 pthread_mutex_unlock(&mt->lock);
 106
 107                 fuse_session_process(mt->se, w->buf, res, ch);
 108
 109                 pthread_mutex_lock(&mt->lock);
 110                 if (!isforget)
 111                         mt->numavail++;
 112                 if (mt->numavail > 10) {
 113                         if (mt->exit) {
 114                                 pthread_mutex_unlock(&mt->lock);
 115                                 return NULL;
 116                         }
 117                         list_del_worker(w);
 118                         mt->numavail--;
 119                         mt->numworker--;
 120                         pthread_mutex_unlock(&mt->lock);
 121
 122                         pthread_detach(w->thread_id);
 123                         free(w->buf);
 124                         free(w);
 125                         return NULL;
 126                 }
 127                 pthread_mutex_unlock(&mt->lock);
 128         }
 129
 130         sem_post(&mt->finish);
 131         pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);
 132         pause();
 133
 134         return NULL;
 135 }
 136
 137 static int fuse_start_thread(struct fuse_mt *mt)
 138 {
 139         sigset_t oldset;
 140         sigset_t newset;
 141         int res;
 142         pthread_attr_t attr;
 143         char *stack_size;
 144         struct fuse_worker *w = malloc(sizeof(struct fuse_worker));
 145         if (!w) {
 146                 fprintf(stderr, "fuse: failed to allocate worker structure\n");
 147                 return -1;
 148         }
 149         memset(w, 0, sizeof(struct fuse_worker));
 150         w->bufsize = fuse_chan_bufsize(mt->prevch);
 151         w->buf = malloc(w->bufsize);
 152         w->mt = mt;
 153         if (!w->buf) {
 154                 fprintf(stderr, "fuse: failed to allocate read buffer\n");
 155                 free(w);
 156                 return -1;
 157         }
 158
 159         /* Override default stack size */
 160         pthread_attr_init(&attr);
 161         stack_size = getenv(ENVNAME_THREAD_STACK);
 162         if (stack_size && pthread_attr_setstacksize(&attr, atoi(stack_size)))
 163                 fprintf(stderr, "fuse: invalid stack size: %s\n", stack_size);
 164
 165         /* Disallow signal reception in worker threads */
 166         sigemptyset(&newset);
 167         sigaddset(&newset, SIGTERM);
 168         sigaddset(&newset, SIGINT);
 169         sigaddset(&newset, SIGHUP);
 170         sigaddset(&newset, SIGQUIT);
 171         pthread_sigmask(SIG_BLOCK, &newset, &oldset);
 172         res = pthread_create(&w->thread_id, &attr, fuse_do_work, w);
 173         pthread_sigmask(SIG_SETMASK, &oldset, NULL);
 174         pthread_attr_destroy(&attr);
 175         if (res != 0) {
 176                 fprintf(stderr, "fuse: error creating thread: %s\n",
 177                         strerror(res));
 178                 free(w->buf);
 179                 free(w);
 180                 return -1;
 181         }
 182         list_add_worker(w, &mt->main);
 183         mt->numavail ++;
 184         mt->numworker ++;
 185
 186         return 0;
 187 }
 188
 189 static void fuse_join_worker(struct fuse_mt *mt, struct fuse_worker *w)
 190 {
 191         pthread_join(w->thread_id, NULL);
 192         pthread_mutex_lock(&mt->lock);
 193         list_del_worker(w);
 194         pthread_mutex_unlock(&mt->lock);
 195         free(w->buf);
 196         free(w);
 197 }
 198
 199 int fuse_session_loop_mt(struct fuse_session *se)
 200 {
 201         int err;
 202         struct fuse_mt mt;
 203         struct fuse_worker *w;
 204
 205         memset(&mt, 0, sizeof(struct fuse_mt));
 206         mt.se = se;
 207         mt.prevch = fuse_session_next_chan(se, NULL);
 208         mt.error = 0;
 209         mt.numworker = 0;
 210         mt.numavail = 0;
 211         mt.main.thread_id = pthread_self();
 212         mt.main.prev = mt.main.next = &mt.main;
 213         sem_init(&mt.finish, 0, 0);
 214         fuse_mutex_init(&mt.lock);
 215
 216         pthread_mutex_lock(&mt.lock);
 217         err = fuse_start_thread(&mt);
 218         pthread_mutex_unlock(&mt.lock);
 219         if (!err) {
 220                 /* sem_wait() is interruptible */
 221                 while (!fuse_session_exited(se))
 222                         sem_wait(&mt.finish);
 223
 224                 for (w = mt.main.next; w != &mt.main; w = w->next)
 225                         pthread_cancel(w->thread_id);
 226                 mt.exit = 1;
 227                 pthread_mutex_unlock(&mt.lock);
 228
 229                 while (mt.main.next != &mt.main)
 230                         fuse_join_worker(&mt, mt.main.next);
 231
 232                 err = mt.error;
 233         }
 234
 235         pthread_mutex_destroy(&mt.lock);
 236         sem_destroy(&mt.finish);
 237         fuse_session_reset(se);
 238         return err;
 239 }