[memcached] sgrimm,
r447: Forgot to svn add the new source file wi...
commits at code.sixapart.com
commits at code.sixapart.com
Fri Nov 24 08:15:45 UTC 2006
Forgot to svn add the new source file with all the thread functions!
A branches/multithreaded/server/thread.c
Added: branches/multithreaded/server/thread.c
===================================================================
--- branches/multithreaded/server/thread.c 2006-11-23 22:21:26 UTC (rev 446)
+++ branches/multithreaded/server/thread.c 2006-11-24 08:15:44 UTC (rev 447)
@@ -0,0 +1,614 @@
+/* -*- Mode: C; tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*- */
+/*
+ * Thread management for memcached.
+ *
+ * $Id$
+ */
+#include "memcached.h"
+#include <stdio.h>
+#include <errno.h>
+#include <stdlib.h>
+#include <errno.h>
+
+#ifdef HAVE_MALLOC_H
+#include <malloc.h>
+#endif
+
+#ifdef USE_THREADS
+
+#include <pthread.h>
+
+#define ITEMS_PER_ALLOC 64
+
+/* An item in the connection queue. */
+typedef struct conn_queue_item CQ_ITEM;
+struct conn_queue_item {
+ int sfd;
+ int init_state;
+ int event_flags;
+ int read_buffer_size;
+ int is_udp;
+ CQ_ITEM *next;
+};
+
+/* A connection queue. */
+typedef struct conn_queue CQ;
+struct conn_queue {
+ CQ_ITEM *head;
+ CQ_ITEM *tail;
+ pthread_mutex_t lock;
+ pthread_cond_t cond;
+};
+
+/* Lock for connection freelist */
+static pthread_mutex_t conn_lock;
+
+/* Lock for cache operations (item_*, assoc_*) */
+static pthread_mutex_t cache_lock;
+
+/* Lock for slab allocator operations */
+static pthread_mutex_t slabs_lock;
+
+/* Lock for global stats */
+static pthread_mutex_t stats_lock;
+
+/* Free list of CQ_ITEM structs */
+static CQ_ITEM *cqi_freelist;
+static pthread_mutex_t cqi_freelist_lock;
+
+/*
+ * Each libevent instance has a wakeup pipe, which other threads
+ * can use to signal that they've put a new connection on its queue.
+ */
+typedef struct {
+ pthread_t thread_id; /* unique ID of this thread */
+ struct event_base *base; /* libevent handle this thread uses */
+ struct event notify_event; /* listen event for notify pipe */
+ int notify_receive_fd; /* receiving end of notify pipe */
+ int notify_send_fd; /* sending end of notify pipe */
+ CQ new_conn_queue; /* queue of new connections to handle */
+} LIBEVENT_THREAD;
+
+static LIBEVENT_THREAD *threads;
+
+/*
+ * Number of threads that have finished setting themselves up.
+ */
+static int init_count = 0;
+static pthread_mutex_t init_lock;
+static pthread_cond_t init_cond;
+
+
+static void thread_libevent_process(int fd, short which, void *arg);
+
+/*
+ * Initializes a connection queue.
+ */
+static void cq_init(CQ *cq) {
+ pthread_mutex_init(&cq->lock, NULL);
+ pthread_cond_init(&cq->cond, NULL);
+ cq->head = NULL;
+ cq->tail = NULL;
+}
+
+/*
+ * Waits for work on a connection queue.
+ */
+static CQ_ITEM *cq_pop(CQ *cq) {
+ CQ_ITEM *item;
+
+ pthread_mutex_lock(&cq->lock);
+ while (NULL == cq->head)
+ pthread_cond_wait(&cq->cond, &cq->lock);
+ item = cq->head;
+ cq->head = item->next;
+ if (NULL == cq->head)
+ cq->tail = NULL;
+ pthread_mutex_unlock(&cq->lock);
+
+ return item;
+}
+
+/*
+ * Looks for an item on a connection queue, but doesn't block if there isn't
+ * one.
+ */
+static CQ_ITEM *cq_peek(CQ *cq) {
+ CQ_ITEM *item;
+
+ pthread_mutex_lock(&cq->lock);
+ item = cq->head;
+ if (NULL != item) {
+ cq->head = item->next;
+ if (NULL == cq->head)
+ cq->tail = NULL;
+ }
+ pthread_mutex_unlock(&cq->lock);
+
+ return item;
+}
+
+/*
+ * Adds an item to a connection queue.
+ */
+static void cq_push(CQ *cq, CQ_ITEM *item) {
+ item->next = NULL;
+
+ pthread_mutex_lock(&cq->lock);
+ if (NULL == cq->tail)
+ cq->head = item;
+ else
+ cq->tail->next = item;
+ cq->tail = item;
+ pthread_cond_signal(&cq->cond);
+ pthread_mutex_unlock(&cq->lock);
+}
+
+/*
+ * Returns a fresh connection queue item.
+ */
+static CQ_ITEM *cqi_new() {
+ CQ_ITEM *item = NULL;
+ pthread_mutex_lock(&cqi_freelist_lock);
+ if (cqi_freelist) {
+ item = cqi_freelist;
+ cqi_freelist = item->next;
+ }
+ pthread_mutex_unlock(&cqi_freelist_lock);
+
+ if (NULL == item) {
+ int i;
+
+ /* Allocate a bunch of items at once to reduce fragmentation */
+ item = malloc(sizeof(CQ_ITEM) * ITEMS_PER_ALLOC);
+ if (NULL == item)
+ return NULL;
+
+ /*
+ * Link together all the new items except the first one
+ * (which we'll return to the caller) for placement on
+ * the freelist.
+ */
+ for (i = 2; i < ITEMS_PER_ALLOC - 1; i++)
+ item[i - 1].next = &item[i];
+
+ pthread_mutex_lock(&cqi_freelist_lock);
+ item[ITEMS_PER_ALLOC - 1].next = cqi_freelist;
+ cqi_freelist = &item[1];
+ pthread_mutex_unlock(&cqi_freelist_lock);
+ }
+
+ return item;
+}
+
+
+/*
+ * Frees a connection queue item (adds it to the freelist.)
+ */
+static void cqi_free(CQ_ITEM *item) {
+ pthread_mutex_lock(&cqi_freelist_lock);
+ item->next = cqi_freelist;
+ cqi_freelist = item;
+ pthread_mutex_unlock(&cqi_freelist_lock);
+}
+
+
+/*
+ * Creates a worker thread.
+ */
+static void create_worker(void *(*func)(void *), void *arg) {
+ pthread_t thread;
+ pthread_attr_t attr;
+ int ret;
+
+ pthread_attr_init(&attr);
+
+ if (ret = pthread_create(&thread, &attr, func, arg)) {
+ fprintf(stderr, "Can't create thread: %s\n",
+ strerror(ret));
+ exit(1);
+ }
+}
+
+
+/*
+ * Pulls a conn structure from the freelist, if one is available.
+ */
+conn *mt_conn_from_freelist() {
+ conn *c;
+
+ pthread_mutex_lock(&conn_lock);
+ c = do_conn_from_freelist();
+ pthread_mutex_unlock(&conn_lock);
+
+ return c;
+}
+
+
+/*
+ * Adds a conn structure to the freelist.
+ *
+ * Returns 0 on success, 1 if the structure couldn't be added.
+ */
+int mt_conn_add_to_freelist(conn *c) {
+ int result;
+
+ pthread_mutex_lock(&conn_lock);
+ result = do_conn_add_to_freelist(c);
+ pthread_mutex_unlock(&conn_lock);
+
+ return result;
+}
+
+/****************************** LIBEVENT THREADS *****************************/
+
+/*
+ * Set up a thread's information.
+ */
+static void setup_thread(LIBEVENT_THREAD *me) {
+ if (! me->base) {
+ me->base = event_init();
+ if (! me->base) {
+ fprintf(stderr, "Can't allocate event base\n");
+ exit(1);
+ }
+ }
+
+ /* Listen for notifications from other threads */
+ event_set(&me->notify_event, me->notify_receive_fd,
+ EV_READ | EV_PERSIST, thread_libevent_process, me);
+ event_base_set(me->base, &me->notify_event);
+
+ if (event_add(&me->notify_event, 0) == -1) {
+ fprintf(stderr, "Can't monitor libevent notify pipe\n");
+ exit(1);
+ }
+
+ cq_init(&me->new_conn_queue);
+}
+
+
+/*
+ * Worker thread: main event loop
+ */
+static void *worker_libevent(void *arg) {
+ LIBEVENT_THREAD *me = arg;
+
+ /* Any per-thread setup can happen here; thread_init() will block until
+ * all threads have finished initializing.
+ */
+
+ pthread_mutex_lock(&init_lock);
+ init_count++;
+ pthread_cond_signal(&init_cond);
+ pthread_mutex_unlock(&init_lock);
+
+ event_base_loop(me->base, 0);
+}
+
+
+/*
+ * Processes an incoming "handle a new connection" item. This is called when
+ * input arrives on the libevent wakeup pipe.
+ */
+static void thread_libevent_process(int fd, short which, void *arg) {
+ LIBEVENT_THREAD *me = arg;
+ CQ_ITEM *item;
+ char buf[1];
+
+ if (read(fd, buf, 1) != 1)
+ if (settings.verbose > 0)
+ fprintf(stderr, "Can't read from libevent pipe\n");
+
+ if (item = cq_peek(&me->new_conn_queue)) {
+ conn *c = conn_new(item->sfd, item->init_state, item->event_flags,
+ item->read_buffer_size, item->is_udp, me->base);
+ if (!c) {
+ if (item->is_udp) {
+ fprintf(stderr, "Can't listen for events on UDP socket\n");
+ exit(1);
+ }
+ else {
+ if (settings.verbose > 0) {
+ fprintf(stderr, "Can't listen for events on fd %d\n",
+ item->sfd);
+ }
+ close(item->sfd);
+ }
+ }
+ cqi_free(item);
+ }
+}
+
+/* Which thread we assigned a connection to most recently. */
+static int last_thread = -1;
+
+/*
+ * Dispatches a new connection to another thread. This is only ever called
+ * from the main thread, either during initialization (for UDP) or because
+ * of an incoming connection.
+ */
+void dispatch_conn_new(int sfd, int init_state, int event_flags,
+ int read_buffer_size, int is_udp) {
+ CQ_ITEM *item = cqi_new();
+ int thread = (last_thread + 1) % settings.num_threads;
+
+ last_thread = thread;
+
+ item->sfd = sfd;
+ item->init_state = init_state;
+ item->event_flags = event_flags;
+ item->read_buffer_size = read_buffer_size;
+ item->is_udp = is_udp;
+
+ cq_push(&threads[thread].new_conn_queue, item);
+ if (write(threads[thread].notify_send_fd, "", 1) != 1) {
+ perror("Writing to thread notify pipe");
+ }
+}
+
+/*
+ * Returns true if this is the thread that listens for new TCP connections.
+ */
+int mt_is_listen_thread() {
+ return pthread_self() == threads[0].thread_id;
+}
+
+/********************************* ITEM ACCESS *******************************/
+
+/*
+ * Walks through the list of deletes that have been deferred because the items
+ * were locked down at the tmie.
+ */
+void mt_run_deferred_deletes() {
+ pthread_mutex_lock(&cache_lock);
+ do_run_deferred_deletes();
+ pthread_mutex_unlock(&cache_lock);
+}
+
+/*
+ * Allocates a new item.
+ */
+item *mt_item_alloc(char *key, size_t nkey, int flags, rel_time_t exptime, int nbytes) {
+ item *it;
+ pthread_mutex_lock(&cache_lock);
+ it = do_item_alloc(key, nkey, flags, exptime, nbytes);
+ pthread_mutex_unlock(&cache_lock);
+ return it;
+}
+
+/*
+ * Returns an item if it hasn't been marked as expired or deleted,
+ * lazy-expiring as needed.
+ */
+item *mt_item_get_notedeleted(char *key, size_t nkey, int *delete_locked) {
+ item *it;
+ pthread_mutex_lock(&cache_lock);
+ it = do_item_get_notedeleted(key, nkey, delete_locked);
+ pthread_mutex_unlock(&cache_lock);
+ return it;
+}
+
+/*
+ * Returns an item whether or not it's been marked as expired or deleted.
+ */
+item *mt_item_get_nocheck(char *key, size_t nkey) {
+ item *it;
+
+ pthread_mutex_lock(&cache_lock);
+ it = assoc_find(key, nkey);
+ it->refcount++;
+ pthread_mutex_unlock(&cache_lock);
+ return it;
+}
+
+/*
+ * Links an item into the LRU and hashtable.
+ */
+int mt_item_link(item *item) {
+ int ret;
+
+ pthread_mutex_lock(&cache_lock);
+ ret = do_item_link(item);
+ pthread_mutex_unlock(&cache_lock);
+ return ret;
+}
+
+/*
+ * Decrements the reference count on an item and adds it to the freelist if
+ * needed.
+ */
+void mt_item_remove(item *item) {
+ pthread_mutex_lock(&cache_lock);
+ do_item_remove(item);
+ pthread_mutex_unlock(&cache_lock);
+}
+
+/*
+ * Replaces one item with another in the hashtable.
+ */
+int mt_item_replace(item *old, item *new) {
+ int ret;
+
+ pthread_mutex_lock(&cache_lock);
+ ret = do_item_replace(old, new);
+ pthread_mutex_unlock(&cache_lock);
+ return ret;
+}
+
+/*
+ * Unlinks an item from the LRU and hashtable.
+ */
+void mt_item_unlink(item *item) {
+ pthread_mutex_lock(&cache_lock);
+ do_item_unlink(item);
+ pthread_mutex_unlock(&cache_lock);
+}
+
+/*
+ * Moves an item to the back of the LRU queue.
+ */
+void mt_item_update(item *item) {
+ pthread_mutex_lock(&cache_lock);
+ do_item_update(item);
+ pthread_mutex_unlock(&cache_lock);
+}
+
+/*
+ * Adds an item to the deferred-delete list so it can be reaped later.
+ */
+char *mt_defer_delete(item *item, time_t exptime) {
+ char *ret;
+
+ pthread_mutex_lock(&cache_lock);
+ ret = do_defer_delete(item, exptime);
+ pthread_mutex_unlock(&cache_lock);
+ return ret;
+}
+
+/*
+ * Does arithmetic on a numeric item value.
+ */
+char *mt_add_delta(item *item, int incr, unsigned int delta, char *buf) {
+ char *ret;
+
+ pthread_mutex_lock(&cache_lock);
+ ret = do_add_delta(item, incr, delta, buf);
+ pthread_mutex_unlock(&cache_lock);
+ return ret;
+}
+
+/*
+ * Stores an item in the cache (high level, obeys set/add/replace semantics)
+ */
+int mt_store_item(item *item, int comm) {
+ int ret;
+
+ pthread_mutex_lock(&cache_lock);
+ ret = do_store_item(item, comm);
+ pthread_mutex_unlock(&cache_lock);
+ return ret;
+}
+
+/*
+ * Flushes expired items after a flush_all call
+ */
+void mt_item_flush_expired() {
+ pthread_mutex_lock(&cache_lock);
+ do_item_flush_expired();
+ pthread_mutex_unlock(&cache_lock);
+}
+
+/****************************** HASHTABLE MODULE *****************************/
+
+void mt_assoc_move_next_bucket() {
+ pthread_mutex_lock(&cache_lock);
+ do_assoc_move_next_bucket();
+ pthread_mutex_unlock(&cache_lock);
+}
+
+/******************************* SLAB ALLOCATOR ******************************/
+
+void *mt_slabs_alloc(size_t size) {
+ void *ret;
+
+ pthread_mutex_lock(&slabs_lock);
+ ret = do_slabs_alloc(size);
+ pthread_mutex_unlock(&slabs_lock);
+ return ret;
+}
+
+void mt_slabs_free(void *ptr, size_t size) {
+ pthread_mutex_lock(&slabs_lock);
+ do_slabs_free(ptr, size);
+ pthread_mutex_unlock(&slabs_lock);
+}
+
+char *mt_slabs_stats(int *buflen) {
+ char *ret;
+
+ pthread_mutex_lock(&slabs_lock);
+ ret = do_slabs_stats(buflen);
+ pthread_mutex_unlock(&slabs_lock);
+ return ret;
+}
+
+#ifdef ALLOW_SLABS_REASSIGN
+int mt_slabs_reassign(unsigned char srcid, unsigned char dstid) {
+ int ret;
+
+ pthread_mutex_lock(&slabs_lock);
+ ret = do_slabs_reassign(srcid, dstid);
+ pthread_mutex_unlock(&slabs_lock);
+ return ret;
+}
+#endif
+
+/******************************* GLOBAL STATS ******************************/
+
+void mt_stats_lock() {
+ pthread_mutex_lock(&stats_lock);
+}
+
+void mt_stats_unlock() {
+ pthread_mutex_unlock(&stats_lock);
+}
+
+/*
+ * Initializes the thread subsystem, creating various worker threads.
+ *
+ * nthreads Number of event handler threads to spawn
+ * main_base Event base for main thread
+ */
+void thread_init(int nthreads, struct event_base *main_base) {
+ int i;
+ pthread_t *thread;
+
+ pthread_mutex_init(&cache_lock, NULL);
+ pthread_mutex_init(&conn_lock, NULL);
+ pthread_mutex_init(&slabs_lock, NULL);
+ pthread_mutex_init(&stats_lock, NULL);
+
+ pthread_mutex_init(&init_lock, NULL);
+ pthread_cond_init(&init_cond, NULL);
+
+ pthread_mutex_init(&cqi_freelist_lock, NULL);
+ cqi_freelist = NULL;
+
+ threads = malloc(sizeof(LIBEVENT_THREAD) * nthreads);
+ if (! threads) {
+ perror("Can't allocate thread descriptors");
+ exit(1);
+ }
+
+ threads[0].base = main_base;
+ threads[0].thread_id = pthread_self();
+
+ for (i = 0; i < nthreads; i++) {
+ int fds[2];
+ if (pipe(fds)) {
+ perror("Can't create notify pipe");
+ exit(1);
+ }
+
+ threads[i].notify_receive_fd = fds[0];
+ threads[i].notify_send_fd = fds[1];
+
+ setup_thread(&threads[i]);
+ }
+
+ /* Create threads after we've done all the libevent setup. */
+ for (i = 1; i < nthreads; i++) {
+ create_worker(worker_libevent, &threads[i]);
+ }
+
+ /* Wait for all the threads to set themselves up before returning. */
+ pthread_mutex_lock(&init_lock);
+ init_count++; // main thread
+ while (init_count < nthreads) {
+ pthread_cond_wait(&init_cond, &init_lock);
+ }
+ pthread_mutex_unlock(&init_lock);
+}
+
+#endif
More information about the memcached-commits
mailing list