cvs/lib/fuse_mt.c

/*
    FUSE: Filesystem in Userspace
    Copyright (C) 2001-2004  Miklos Szeredi <miklos@szeredi.hu>

    This program can be distributed under the terms of the GNU LGPL.
    See the file COPYING.LIB.
*/

#include "fuse_i.h"

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <pthread.h>
#include <signal.h>
#include <errno.h>
#include <sys/time.h>

#define FUSE_MAX_WORKERS 10

struct fuse_worker {
    struct fuse *f;
    pthread_t threads[FUSE_MAX_WORKERS];
    void *data;
    fuse_processor_t proc;
};

static void start_thread(struct fuse_worker *w, pthread_t *thread_id);

static void *do_work(void *data)
{
    struct fuse_worker *w = (struct fuse_worker *) data;
    struct fuse *f = w->f;

    pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);
    pthread_setcanceltype(PTHREAD_CANCEL_ASYNCHRONOUS, NULL);

    while (1) {
        struct fuse_cmd *cmd;

        if (__fuse_exited(f))
            break;

        cmd = __fuse_read_cmd(w->f);
        if (cmd == NULL)
            continue;

        if (f->numavail == 0 && f->numworker < FUSE_MAX_WORKERS) {
            pthread_mutex_lock(&f->lock);
            if (f->numworker < FUSE_MAX_WORKERS) {
                pthread_t *thread_id = &w->threads[f->numworker];
                f->numavail ++;
                f->numworker ++;
                pthread_mutex_unlock(&f->lock);
                start_thread(w, thread_id);
            } else
                pthread_mutex_unlock(&f->lock);
        }

        w->proc(w->f, cmd, w->data);
    }

    return NULL;
}

static void start_thread(struct fuse_worker *w, pthread_t *thread_id)
{
    sigset_t oldset;
    sigset_t newset;
    int res;

    /* Disallow signal reception in worker threads */
    sigfillset(&newset);
    pthread_sigmask(SIG_SETMASK, &newset, &oldset);
    res = pthread_create(thread_id, NULL, do_work, w);
    pthread_sigmask(SIG_SETMASK, &oldset, NULL);
    if (res != 0) {
        fprintf(stderr, "Error creating thread: %s\n", strerror(res));
        exit(1);
    }
    
    pthread_detach(*thread_id);
}

static struct fuse_context *mt_getcontext(struct fuse *f)
{
    struct fuse_context *ctx;

    ctx = (struct fuse_context *) pthread_getspecific(f->context_key);
    if (ctx == NULL) {
        ctx = (struct fuse_context *) malloc(sizeof(struct fuse_context));
        pthread_setspecific(f->context_key, ctx);
    }

    return ctx;
}

static void mt_freecontext(void *data)
{
    free(data);
}

void __fuse_loop_mt(struct fuse *f, fuse_processor_t proc, void *data)
{
    struct fuse_worker *w;
    int res;
    int i;

    w = malloc(sizeof(struct fuse_worker));    
    memset(w, 0, sizeof(struct fuse_worker));
    w->f = f;
    w->data = data;
    w->proc = proc;

    f->numworker = 1;
    res = pthread_key_create(&f->context_key, mt_freecontext);
    if (res != 0) {
        fprintf(stderr, "Failed to create thread specific key\n");
        exit(1);
    }
    f->getcontext = mt_getcontext;
    do_work(w);

    pthread_mutex_lock(&f->lock);
    for (i = 1; i < f->numworker; i++)
        pthread_cancel(w->threads[i]);
    pthread_mutex_unlock(&f->lock);
    pthread_key_delete(f->context_key);
    free(w);
}

void fuse_loop_mt(struct fuse *f)
{
    if (f == NULL)
        return;

    __fuse_loop_mt(f, (fuse_processor_t) __fuse_process_cmd, NULL);
}
1	dpavlin	4	/*
2			FUSE: Filesystem in Userspace
3			Copyright (C) 2001-2004 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_i.h"
10
11			#include <stdio.h>
12			#include <stdlib.h>
13			#include <string.h>
14			#include <pthread.h>
15			#include <signal.h>
16			#include <errno.h>
17			#include <sys/time.h>
18
19			#define FUSE_MAX_WORKERS 10
20
21			struct fuse_worker {
22			struct fuse *f;
23			pthread_t threads[FUSE_MAX_WORKERS];
24			void *data;
25			fuse_processor_t proc;
26			};
27
28			static void start_thread(struct fuse_worker w, pthread_t thread_id);
29
30			static void do_work(void data)
31			{
32			struct fuse_worker w = (struct fuse_worker ) data;
33			struct fuse *f = w->f;
34
35			pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, NULL);
36			pthread_setcanceltype(PTHREAD_CANCEL_ASYNCHRONOUS, NULL);
37
38			while (1) {
39			struct fuse_cmd *cmd;
40
41			if (__fuse_exited(f))
42			break;
43
44			cmd = __fuse_read_cmd(w->f);
45			if (cmd == NULL)
46			continue;
47
48			if (f->numavail == 0 && f->numworker < FUSE_MAX_WORKERS) {
49			pthread_mutex_lock(&f->lock);
50			if (f->numworker < FUSE_MAX_WORKERS) {
51			pthread_t *thread_id = &w->threads[f->numworker];
52			f->numavail ++;
53			f->numworker ++;
54			pthread_mutex_unlock(&f->lock);
55			start_thread(w, thread_id);
56			} else
57			pthread_mutex_unlock(&f->lock);
58			}
59
60			w->proc(w->f, cmd, w->data);
61			}
62
63			return NULL;
64			}
65
66			static void start_thread(struct fuse_worker w, pthread_t thread_id)
67			{
68			sigset_t oldset;
69			sigset_t newset;
70			int res;
71
72			/* Disallow signal reception in worker threads */
73			sigfillset(&newset);
74			pthread_sigmask(SIG_SETMASK, &newset, &oldset);
75			res = pthread_create(thread_id, NULL, do_work, w);
76			pthread_sigmask(SIG_SETMASK, &oldset, NULL);
77			if (res != 0) {
78			fprintf(stderr, "Error creating thread: %s\n", strerror(res));
79			exit(1);
80			}
81
82			pthread_detach(*thread_id);
83			}
84
85			static struct fuse_context mt_getcontext(struct fuse f)
86			{
87			struct fuse_context *ctx;
88
89			ctx = (struct fuse_context *) pthread_getspecific(f->context_key);
90			if (ctx == NULL) {
91			ctx = (struct fuse_context *) malloc(sizeof(struct fuse_context));
92			pthread_setspecific(f->context_key, ctx);
93			}
94
95			return ctx;
96			}
97
98			static void mt_freecontext(void *data)
99			{
100			free(data);
101			}
102
103			void __fuse_loop_mt(struct fuse f, fuse_processor_t proc, void data)
104			{
105			struct fuse_worker *w;
106			int res;
107			int i;
108
109			w = malloc(sizeof(struct fuse_worker));
110			memset(w, 0, sizeof(struct fuse_worker));
111			w->f = f;
112			w->data = data;
113			w->proc = proc;
114
115			f->numworker = 1;
116			res = pthread_key_create(&f->context_key, mt_freecontext);
117			if (res != 0) {
118			fprintf(stderr, "Failed to create thread specific key\n");
119			exit(1);
120			}
121			f->getcontext = mt_getcontext;
122			do_work(w);
123
124			pthread_mutex_lock(&f->lock);
125			for (i = 1; i < f->numworker; i++)
126			pthread_cancel(w->threads[i]);
127			pthread_mutex_unlock(&f->lock);
128			pthread_key_delete(f->context_key);
129			free(w);
130			}
131
132			void fuse_loop_mt(struct fuse *f)
133			{
134			if (f == NULL)
135			return;
136
137			__fuse_loop_mt(f, (fuse_processor_t) __fuse_process_cmd, NULL);
138			}