lib/fuse_loop_mt.c

   1 /*
   2     FUSE: Filesystem in Userspace
   3     Copyright (C) 2001-2006  Miklos Szeredi <miklos@szeredi.hu>
   4
   5     This program can be distributed under the terms of the GNU LGPL.
   6     See the file COPYING.LIB.
   7 */
   8
   9 #include "fuse_lowlevel.h"
  10 #include "fuse_misc.h"
  11
  12 #include <stdio.h>
  13 #include <stdlib.h>
  14 #include <string.h>
  15 #include <unistd.h>
  16 #include <signal.h>
  17 #include <errno.h>
  18 #include <sys/time.h>
  19
  20 struct fuse_worker {
  21     struct fuse_worker *prev;
  22     struct fuse_worker *next;
  23     pthread_t thread_id;
  24     size_t bufsize;
  25     char *buf;
  26     struct fuse_mt *mt;
  27 };
  28
  29 struct fuse_mt {
  30     pthread_mutex_t lock;
  31     int numworker;
  32     int numavail;
  33     struct fuse_session *se;
  34     struct fuse_chan *prevch;
  35     struct fuse_worker main;
  36     int exit;
  37     int error;
  38 };
  39
  40 static void list_add_worker(struct fuse_worker *w, struct fuse_worker *next)
  41 {
  42     struct fuse_worker *prev = next->prev;
  43     w->next = next;
  44     w->prev = prev;
  45     prev->next = w;
  46     next->prev = w;
  47 }
  48
  49 static void list_del_worker(struct fuse_worker *w)
  50 {
  51     struct fuse_worker *prev = w->prev;
  52     struct fuse_worker *next = w->next;
  53     prev->next = next;
  54     next->prev = prev;
  55 }
  56
  57 static int fuse_start_thread(struct fuse_mt *mt);
  58
  59 static void *fuse_do_work(void *data)
  60 {
  61     struct fuse_worker *w = (struct fuse_worker *) data;
  62     struct fuse_mt *mt = w->mt;
  63
  64     while (!fuse_session_exited(mt->se)) {
  65         struct fuse_chan *ch = mt->prevch;
  66         int res = fuse_chan_recv(&ch, w->buf, w->bufsize);
  67         if (res == -EINTR)
  68             continue;
  69         if (res <= 0) {
  70             if (res < 0) {
  71                 fuse_session_exit(mt->se);
  72                 mt->error = -1;
  73             }
  74             break;
  75         }
  76
  77         pthread_mutex_lock(&mt->lock);
  78         if (mt->exit) {
  79             pthread_mutex_unlock(&mt->lock);
  80             return NULL;
  81         }
  82         mt->numavail--;
  83         if (mt->numavail == 0)
  84             fuse_start_thread(mt);
  85         pthread_mutex_unlock(&mt->lock);
  86
  87         fuse_session_process(mt->se, w->buf, res, ch);
  88
  89         pthread_mutex_lock(&mt->lock);
  90         mt->numavail ++;
  91         if (mt->numavail > 10) {
  92             if (mt->exit) {
  93                 pthread_mutex_unlock(&mt->lock);
  94                 return NULL;
  95             }
  96             list_del_worker(w);
  97             mt->numavail--;
  98             mt->numworker--;
  99             pthread_mutex_unlock(&mt->lock);
 100
 101             pthread_detach(w->thread_id);
 102             free(w->buf);
 103             free(w);
 104             return NULL;
 105         }
 106         pthread_mutex_unlock(&mt->lock);
 107     }
 108
 109     pthread_kill(mt->main.thread_id, SIGHUP);
 110     pause();
 111
 112     return NULL;
 113 }
 114
 115 static int fuse_start_thread(struct fuse_mt *mt)
 116 {
 117     sigset_t oldset;
 118     sigset_t newset;
 119     int res;
 120     struct fuse_worker *w = malloc(sizeof(struct fuse_worker));
 121     if (!w) {
 122         fprintf(stderr, "fuse: failed to allocate worker structure\n");
 123         return -1;
 124     }
 125     memset(w, 0, sizeof(struct fuse_worker));
 126     w->bufsize = fuse_chan_bufsize(mt->prevch);
 127     w->buf = malloc(w->bufsize);
 128     w->mt = mt;
 129     if (!w->buf) {
 130         fprintf(stderr, "fuse: failed to allocate read buffer\n");
 131         free(w);
 132         return -1;
 133     }
 134
 135     /* Disallow signal reception in worker threads */
 136     sigemptyset(&newset);
 137     sigaddset(&newset, SIGTERM);
 138     sigaddset(&newset, SIGINT);
 139     sigaddset(&newset, SIGHUP);
 140     sigaddset(&newset, SIGQUIT);
 141     pthread_sigmask(SIG_BLOCK, &newset, &oldset);
 142     res = pthread_create(&w->thread_id, NULL, fuse_do_work, w);
 143     pthread_sigmask(SIG_SETMASK, &oldset, NULL);
 144     if (res != 0) {
 145         fprintf(stderr, "fuse: error creating thread: %s\n", strerror(res));
 146         return -1;
 147     }
 148     list_add_worker(w, &mt->main);
 149     mt->numavail ++;
 150     mt->numworker ++;
 151
 152     return 0;
 153 }
 154
 155 static void fuse_join_worker(struct fuse_mt *mt, struct fuse_worker *w)
 156 {
 157     pthread_join(w->thread_id, NULL);
 158     pthread_mutex_lock(&mt->lock);
 159     list_del_worker(w);
 160     pthread_mutex_unlock(&mt->lock);
 161     free(w->buf);
 162     free(w);
 163 }
 164
 165 int fuse_session_loop_mt(struct fuse_session *se)
 166 {
 167     int err;
 168     struct fuse_mt mt;
 169     struct fuse_worker *w;
 170
 171     memset(&mt, 0, sizeof(struct fuse_mt));
 172     mt.se = se;
 173     mt.prevch = fuse_session_next_chan(se, NULL);
 174     mt.error = 0;
 175     mt.numworker = 0;
 176     mt.numavail = 0;
 177     mt.main.thread_id = pthread_self();
 178     mt.main.prev = mt.main.next = &mt.main;
 179     fuse_mutex_init(&mt.lock);
 180
 181     pthread_mutex_lock(&mt.lock);
 182     err = fuse_start_thread(&mt);
 183     pthread_mutex_unlock(&mt.lock);
 184     if (!err) {
 185         while (!fuse_session_exited(se))
 186             pause();
 187
 188         for (w = mt.main.next; w != &mt.main; w = w->next)
 189             pthread_cancel(w->thread_id);
 190         mt.exit = 1;
 191         pthread_mutex_unlock(&mt.lock);
 192
 193         while (mt.main.next != &mt.main)
 194             fuse_join_worker(&mt, mt.main.next);
 195
 196         err = mt.error;
 197     }
 198
 199     pthread_mutex_destroy(&mt.lock);
 200     fuse_session_reset(se);
 201     return err;
 202 }