diff options
Diffstat (limited to 'libglusterfs')
| -rw-r--r-- | libglusterfs/src/event-epoll.c | 210 | ||||
| -rw-r--r-- | libglusterfs/src/event-poll.c | 17 | ||||
| -rw-r--r-- | libglusterfs/src/event.c | 21 | ||||
| -rw-r--r-- | libglusterfs/src/event.h | 15 | ||||
| -rw-r--r-- | libglusterfs/src/glusterfs.h | 2 |
5 files changed, 235 insertions, 30 deletions
diff --git a/libglusterfs/src/event-epoll.c b/libglusterfs/src/event-epoll.c index 9082954e4e4..8d42fa71fb6 100644 --- a/libglusterfs/src/event-epoll.c +++ b/libglusterfs/src/event-epoll.c @@ -43,6 +43,10 @@ struct event_slot_epoll { gf_lock_t lock; }; +struct event_thread_data { + struct event_pool *event_pool; + int event_index; +}; static struct event_slot_epoll * __event_newtable (struct event_pool *event_pool, int table_idx) @@ -232,7 +236,7 @@ done: static struct event_pool * -event_pool_new_epoll (int count) +event_pool_new_epoll (int count, int eventthreadcount) { struct event_pool *event_pool = NULL; int epfd = -1; @@ -258,6 +262,8 @@ event_pool_new_epoll (int count) event_pool->count = count; + event_pool->eventthreadcount = eventthreadcount; + pthread_mutex_init (&event_pool->mutex, NULL); out: @@ -585,11 +591,45 @@ event_dispatch_epoll_worker (void *data) { struct epoll_event event; int ret = -1; - struct event_pool *event_pool = data; + struct event_thread_data *ev_data = data; + struct event_pool *event_pool; + int myindex = -1; + int timetodie = 0; + + GF_VALIDATE_OR_GOTO ("event", ev_data, out); + + event_pool = ev_data->event_pool; + myindex = ev_data->event_index; GF_VALIDATE_OR_GOTO ("event", event_pool, out); + gf_log ("epoll", GF_LOG_INFO, "Started thread with index %d", myindex); + for (;;) { + if (event_pool->eventthreadcount < myindex) { + /* ...time to die, thread count was decreased below + * this threads index */ + /* Start with extra safety at this point, reducing + * lock conention in normal case when threads are not + * reconfigured always */ + pthread_mutex_lock (&event_pool->mutex); + { + if (event_pool->eventthreadcount < + myindex) { + /* if found true in critical section, + * die */ + event_pool->pollers[myindex - 1] = 0; + timetodie = 1; + } + } + pthread_mutex_unlock (&event_pool->mutex); + if (timetodie) { + gf_log ("epoll", GF_LOG_INFO, + "Exited thread with index %d", myindex); + goto out; + } + } + ret = epoll_wait (event_pool->fd, &event, 1, -1); if (ret == 0) @@ -603,40 +643,164 @@ event_dispatch_epoll_worker (void *data) ret = event_dispatch_epoll_handler (event_pool, &event); } out: + if (ev_data) + GF_FREE (ev_data); return NULL; } - -#define GLUSTERFS_EPOLL_MAXTHREADS 2 - - +/* Attempts to start the # of configured pollers, ensuring at least the first + * is started in a joinable state */ static int event_dispatch_epoll (struct event_pool *event_pool) { - int i = 0; - pthread_t pollers[GLUSTERFS_EPOLL_MAXTHREADS]; - int ret = -1; - - for (i = 0; i < GLUSTERFS_EPOLL_MAXTHREADS; i++) { - ret = pthread_create (&pollers[i], NULL, - event_dispatch_epoll_worker, - event_pool); - } + int i = 0; + pthread_t t_id; + int pollercount = 0; + int ret = -1; + struct event_thread_data *ev_data = NULL; + + /* Start the configured number of pollers */ + pthread_mutex_lock (&event_pool->mutex); + { + pollercount = event_pool->eventthreadcount; + + /* Set to MAX if greater */ + if (pollercount > EVENT_MAX_THREADS) + pollercount = EVENT_MAX_THREADS; + + /* Default pollers to 1 in case this is incorrectly set */ + if (pollercount <= 0) + pollercount = 1; + + for (i = 0; i < pollercount; i++) { + ev_data = GF_CALLOC (1, sizeof (*ev_data), + gf_common_mt_event_pool); + if (!ev_data) { + gf_log ("epoll", GF_LOG_WARNING, + "Allocation failure for index %d", i); + if (i == 0) { + /* Need to suceed creating 0'th + * thread, to joinable and wait */ + break; + } else { + /* Inability to create other threads + * are a lesser evil, and ignored */ + continue; + } + } + + ev_data->event_pool = event_pool; + ev_data->event_index = i + 1; + + ret = pthread_create (&t_id, NULL, + event_dispatch_epoll_worker, + ev_data); + if (!ret) { + event_pool->pollers[i] = t_id; + + /* mark all threads other than one in index 0 + * as detachable. Errors can be ignored, they + * spend their time as zombies if not detched + * and the thread counts are decreased */ + if (i != 0) + pthread_detach (event_pool->pollers[i]); + } else { + gf_log ("epoll", GF_LOG_WARNING, + "Failed to start thread for index %d", + i); + if (i == 0) { + GF_FREE (ev_data); + break; + } else { + GF_FREE (ev_data); + continue; + } + } + } + } + pthread_mutex_unlock (&event_pool->mutex); - for (i = 0; i < GLUSTERFS_EPOLL_MAXTHREADS; i++) - pthread_join (pollers[i], NULL); + /* Just wait for the first thread, that is created in a joinable state + * and will never die, ensuring this function never returns */ + if (event_pool->pollers[0] != 0) + pthread_join (event_pool->pollers[0], NULL); return ret; } +int +event_reconfigure_threads_epoll (struct event_pool *event_pool, int value) +{ + int i; + int ret; + pthread_t t_id; + int oldthreadcount; + struct event_thread_data *ev_data = NULL; + + /* Set to MAX if greater */ + if (value > EVENT_MAX_THREADS) + value = EVENT_MAX_THREADS; + + /* Default pollers to 1 in case this is set incorrectly */ + if (value <= 0) + value = 1; + + pthread_mutex_lock (&event_pool->mutex); + { + oldthreadcount = event_pool->eventthreadcount; + + if (oldthreadcount < value) { + /* create more poll threads */ + for (i = oldthreadcount; i < value; i++) { + /* Start a thread if the index at this location + * is a 0, so that the older thread is confirmed + * as dead */ + if (event_pool->pollers[i] == 0) { + ev_data = GF_CALLOC (1, + sizeof (*ev_data), + gf_common_mt_event_pool); + if (!ev_data) { + gf_log ("epoll", GF_LOG_WARNING, + "Allocation failure for" + " index %d", i); + continue; + } + + ev_data->event_pool = event_pool; + ev_data->event_index = i + 1; + + ret = pthread_create (&t_id, NULL, + event_dispatch_epoll_worker, + ev_data); + if (ret) { + gf_log ("epoll", GF_LOG_WARNING, + "Failed to start thread for" + " index %d", i); + GF_FREE (ev_data); + } else { + pthread_detach (t_id); + event_pool->pollers[i] = t_id; + } + } + } + } + + /* if value decreases, threads will terminate, themselves */ + event_pool->eventthreadcount = value; + } + pthread_mutex_unlock (&event_pool->mutex); + + return 0; +} struct event_ops event_ops_epoll = { - .new = event_pool_new_epoll, - .event_register = event_register_epoll, - .event_select_on = event_select_on_epoll, - .event_unregister = event_unregister_epoll, - .event_unregister_close = event_unregister_close_epoll, - .event_dispatch = event_dispatch_epoll + .new = event_pool_new_epoll, + .event_register = event_register_epoll, + .event_select_on = event_select_on_epoll, + .event_unregister = event_unregister_epoll, + .event_unregister_close = event_unregister_close_epoll, + .event_dispatch = event_dispatch_epoll, + .event_reconfigure_threads = event_reconfigure_threads_epoll }; #endif diff --git a/libglusterfs/src/event-poll.c b/libglusterfs/src/event-poll.c index a7e2e663103..c91fa8487b5 100644 --- a/libglusterfs/src/event-poll.c +++ b/libglusterfs/src/event-poll.c @@ -95,7 +95,7 @@ out: static struct event_pool * -event_pool_new_poll (int count) +event_pool_new_poll (int count, int eventthreadcount) { struct event_pool *event_pool = NULL; int ret = -1; @@ -171,6 +171,12 @@ event_pool_new_poll (int count) return NULL; } + if (eventthreadcount > 1) { + gf_log ("poll", GF_LOG_INFO, + "Currently poll does not use multiple event processing" + " threads, thread count (%d) ignored", eventthreadcount); + } + return event_pool; } @@ -469,6 +475,12 @@ out: return -1; } +int +event_reconfigure_threads_poll (struct event_pool *event_pool, int value) +{ + /* No-op for poll */ + return 0; +} struct event_ops event_ops_poll = { .new = event_pool_new_poll, @@ -476,5 +488,6 @@ struct event_ops event_ops_poll = { .event_select_on = event_select_on_poll, .event_unregister = event_unregister_poll, .event_unregister_close = event_unregister_close_poll, - .event_dispatch = event_dispatch_poll + .event_dispatch = event_dispatch_poll, + .event_reconfigure_threads = event_reconfigure_threads_poll }; diff --git a/libglusterfs/src/event.c b/libglusterfs/src/event.c index 6c253df3c1a..4dd0f991700 100644 --- a/libglusterfs/src/event.c +++ b/libglusterfs/src/event.c @@ -29,7 +29,7 @@ struct event_pool * -event_pool_new (int count) +event_pool_new (int count, int eventthreadcount) { struct event_pool *event_pool = NULL; extern struct event_ops event_ops_poll; @@ -37,7 +37,7 @@ event_pool_new (int count) #ifdef HAVE_SYS_EPOLL_H extern struct event_ops event_ops_epoll; - event_pool = event_ops_epoll.new (count); + event_pool = event_ops_epoll.new (count, eventthreadcount); if (event_pool) { event_pool->ops = &event_ops_epoll; @@ -48,7 +48,7 @@ event_pool_new (int count) #endif if (!event_pool) { - event_pool = event_ops_poll.new (count); + event_pool = event_ops_poll.new (count, eventthreadcount); if (event_pool) event_pool->ops = &event_ops_poll; @@ -129,3 +129,18 @@ event_dispatch (struct event_pool *event_pool) out: return ret; } + +int +event_reconfigure_threads (struct event_pool *event_pool, int value) +{ + int ret = -1; + + GF_VALIDATE_OR_GOTO ("event", event_pool, out); + + /* call event refresh function */ + ret = event_pool->ops->event_reconfigure_threads (event_pool, + value); + +out: + return ret; +} diff --git a/libglusterfs/src/event.h b/libglusterfs/src/event.h index 3b3ab0e4b2f..930a7d1e28b 100644 --- a/libglusterfs/src/event.h +++ b/libglusterfs/src/event.h @@ -33,6 +33,7 @@ typedef int (*event_handler_t) (int fd, int idx, void *data, #define EVENT_EPOLL_TABLES 1024 #define EVENT_EPOLL_SLOTS 1024 +#define EVENT_MAX_THREADS 32 struct event_pool { struct event_ops *ops; @@ -53,10 +54,16 @@ struct event_pool { void *evcache; int evcache_size; + + /* NOTE: Currently used only when event processing is done using + * epoll. */ + int eventthreadcount; /* number of event threads to execute. */ + pthread_t pollers[EVENT_MAX_THREADS]; /* poller thread_id store, + * and live status */ }; struct event_ops { - struct event_pool * (*new) (int count); + struct event_pool * (*new) (int count, int eventthreadcount); int (*event_register) (struct event_pool *event_pool, int fd, event_handler_t handler, @@ -71,9 +78,12 @@ struct event_ops { int idx); int (*event_dispatch) (struct event_pool *event_pool); + + int (*event_reconfigure_threads) (struct event_pool *event_pool, + int newcount); }; -struct event_pool * event_pool_new (int count); +struct event_pool *event_pool_new (int count, int eventthreadcount); int event_select_on (struct event_pool *event_pool, int fd, int idx, int poll_in, int poll_out); int event_register (struct event_pool *event_pool, int fd, @@ -82,5 +92,6 @@ int event_register (struct event_pool *event_pool, int fd, int event_unregister (struct event_pool *event_pool, int fd, int idx); int event_unregister_close (struct event_pool *event_pool, int fd, int idx); int event_dispatch (struct event_pool *event_pool); +int event_reconfigure_threads (struct event_pool *event_pool, int value); #endif /* _EVENT_H_ */ diff --git a/libglusterfs/src/glusterfs.h b/libglusterfs/src/glusterfs.h index 8059c976368..9c078e1d5f9 100644 --- a/libglusterfs/src/glusterfs.h +++ b/libglusterfs/src/glusterfs.h @@ -157,6 +157,8 @@ #define GLUSTERFS_RPC_REPLY_SIZE 24 +#define STARTING_EVENT_THREADS 1 + #define ZR_FILE_CONTENT_REQUEST(key) (!strncmp(key, ZR_FILE_CONTENT_STR, \ ZR_FILE_CONTENT_STRLEN)) |