/src/h2o/lib/common/memcached.c

Source
/*
 * Copyright (c) 2015 DeNA Co., Ltd., Kazuho Oku
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to
 * deal in the Software without restriction, including without limitation the
 * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
 * sell copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in
 * all copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
 * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
 * IN THE SOFTWARE.
 */
#include <errno.h>
#include <inttypes.h>
#include <unistd.h>
#include "yrmcds.h"
#include "h2o/linklist.h"
#include "h2o/memcached.h"
#include "h2o/rand.h"
#include "h2o/string_.h"

struct st_h2o_memcached_context_t {
    pthread_mutex_t mutex;
    pthread_cond_t cond;
    h2o_linklist_t pending;
    size_t num_threads_connected;
    char *host;
    uint16_t port;
    int text_protocol;
    h2o_iovec_t prefix;
};

struct st_h2o_memcached_conn_t {
    h2o_memcached_context_t *ctx;
    yrmcds yrmcds;
    pthread_mutex_t mutex;
    h2o_linklist_t inflight;
    int writer_exit_requested;
};

enum en_h2o_memcached_req_type_t { REQ_TYPE_GET, REQ_TYPE_SET, REQ_TYPE_DELETE };

struct st_h2o_memcached_req_t {
    enum en_h2o_memcached_req_type_t type;
    int refcnt;
    h2o_linklist_t pending;
    h2o_linklist_t inflight;
    union {
        struct {
            h2o_multithread_receiver_t *receiver;
            h2o_multithread_message_t message;
            h2o_memcached_get_cb cb;
            void *cb_data;
            int value_is_encoded;
            h2o_iovec_t value;
            uint32_t serial;
        } get;
        struct {
            h2o_iovec_t value;
            uint32_t expiration;
        } set;
    } data;
    struct {
        size_t len;
        char base[1];
    } key;
};

static h2o_memcached_req_t *create_req(h2o_memcached_context_t *ctx, enum en_h2o_memcached_req_type_t type, h2o_iovec_t key,
                                       int encode_key)
{
    h2o_memcached_req_t *req = h2o_mem_alloc(offsetof(h2o_memcached_req_t, key.base) + ctx->prefix.len +
                                             (encode_key ? (key.len + 2) / 3 * 4 + 1 : key.len));
    req->type = type;
    req->refcnt = 1;
    req->pending = (h2o_linklist_t){NULL};
    req->inflight = (h2o_linklist_t){NULL};
    memset(&req->data, 0, sizeof(req->data));
    if (ctx->prefix.len != 0)
        memcpy(req->key.base, ctx->prefix.base, ctx->prefix.len);
    req->key.len = ctx->prefix.len;
    if (encode_key) {
        req->key.len += h2o_base64_encode(req->key.base + req->key.len, key.base, key.len, 1);
    } else {
        memcpy(req->key.base + req->key.len, key.base, key.len);
        req->key.len += key.len;
    }
    return req;
}

static void free_req(h2o_memcached_req_t *req)
{
    if (__sync_sub_and_fetch(&req->refcnt, 1) != 0)
        return;

    assert(!h2o_linklist_is_linked(&req->pending));
    switch (req->type) {
    case REQ_TYPE_GET:
        assert(!h2o_linklist_is_linked(&req->data.get.message.link));
        h2o_mem_set_secure(req->data.get.value.base, 0, req->data.get.value.len);
        free(req->data.get.value.base);
        break;
    case REQ_TYPE_SET:
        h2o_mem_set_secure(req->data.set.value.base, 0, req->data.set.value.len);
        free(req->data.set.value.base);
        break;
    case REQ_TYPE_DELETE:
        break;
    default:
        assert(!"FIXME");
        break;
    }
    free(req);
}

static void retain_req(h2o_memcached_req_t *req)
{
    __sync_add_and_fetch(&req->refcnt, 1);
}

static void discard_req(h2o_memcached_req_t *req)
{
    switch (req->type) {
    case REQ_TYPE_GET:
        h2o_multithread_send_message(req->data.get.receiver, &req->data.get.message);
        break;
    default:
        free_req(req);
        break;
    }
}

static h2o_memcached_req_t *pop_inflight(struct st_h2o_memcached_conn_t *conn, uint32_t serial)
{
    h2o_memcached_req_t *req;

    pthread_mutex_lock(&conn->mutex);

    if (conn->yrmcds.text_mode) {
        /* in text mode, responses are returned in order (and we may receive responses for commands other than GET) */
        if (!h2o_linklist_is_empty(&conn->inflight)) {
            req = H2O_STRUCT_FROM_MEMBER(h2o_memcached_req_t, inflight, conn->inflight.next);
            assert(req->type == REQ_TYPE_GET);
            if (req->data.get.serial == serial)
                goto Found;
        }
    } else {
        /* in binary mode, responses are received out-of-order (and we would only recieve responses for GET) */
        h2o_linklist_t *node;
        for (node = conn->inflight.next; node != &conn->inflight; node = node->next) {
            req = H2O_STRUCT_FROM_MEMBER(h2o_memcached_req_t, inflight, node);
            assert(req->type == REQ_TYPE_GET);
            if (req->data.get.serial == serial)
                goto Found;
        }
    }

    /* not found */
    pthread_mutex_unlock(&conn->mutex);
    return NULL;

Found:
    h2o_linklist_unlink(&req->inflight);
    pthread_mutex_unlock(&conn->mutex);
    return req;
}

static void *writer_main(void *_conn)
{
    struct st_h2o_memcached_conn_t *conn = _conn;
    yrmcds_error err;

    pthread_mutex_lock(&conn->ctx->mutex);

    while (!__sync_add_and_fetch(&conn->writer_exit_requested, 0)) {
        while (!h2o_linklist_is_empty(&conn->ctx->pending)) {
            h2o_memcached_req_t *req = H2O_STRUCT_FROM_MEMBER(h2o_memcached_req_t, pending, conn->ctx->pending.next);
            h2o_linklist_unlink(&req->pending);
            pthread_mutex_unlock(&conn->ctx->mutex);

            switch (req->type) {
            case REQ_TYPE_GET:
                retain_req(req);
                pthread_mutex_lock(&conn->mutex);
                h2o_linklist_insert(&conn->inflight, &req->inflight);
                pthread_mutex_unlock(&conn->mutex);
                err = yrmcds_get(&conn->yrmcds, req->key.base, req->key.len, 0, &req->data.get.serial);
                break;
            case REQ_TYPE_SET:
                err = yrmcds_set(&conn->yrmcds, req->key.base, req->key.len, req->data.set.value.base, req->data.set.value.len, 0,
                                 req->data.set.expiration, 0, !conn->yrmcds.text_mode, NULL);
                break;
            case REQ_TYPE_DELETE:
                err = yrmcds_remove(&conn->yrmcds, req->key.base, req->key.len, !conn->yrmcds.text_mode, NULL);
                break;
            default:
                h2o_error_printf("[lib/common/memcached.c] unknown type:%d\n", (int)req->type);
                err = YRMCDS_NOT_IMPLEMENTED;
                break;
            }
            free_req(req);
            if (err != YRMCDS_OK)
                goto Error;

            pthread_mutex_lock(&conn->ctx->mutex);
        }
        pthread_cond_wait(&conn->ctx->cond, &conn->ctx->mutex);
    }

    pthread_mutex_unlock(&conn->ctx->mutex);
    return NULL;

Error:
    h2o_error_printf("[lib/common/memcached.c] failed to send request; %s\n", yrmcds_strerror(err));
    /* doc says the call can be used to interrupt yrmcds_recv */
    yrmcds_shutdown(&conn->yrmcds);

    return NULL;
}

static void connect_to_server(h2o_memcached_context_t *ctx, yrmcds *yrmcds)
{
    size_t failcnt;
    yrmcds_error err;

    for (failcnt = 0; (err = yrmcds_connect(yrmcds, ctx->host, ctx->port)) != YRMCDS_OK; ++failcnt) {
        if (failcnt == 0) {
            h2o_error_printf("[lib/common/memcached.c] failed to connect to memcached at %s:%" PRIu16 ", %s\n", ctx->host,
                             ctx->port, yrmcds_strerror(err));
        }
        ++failcnt;
        usleep(2000000 + h2o_rand() % 3000000); /* sleep 2 to 5 seconds */
    }
    /* connected */
    if (ctx->text_protocol)
        yrmcds_text_mode(yrmcds);
    h2o_error_printf("[lib/common/memcached.c] connected to memcached at %s:%" PRIu16 "\n", ctx->host, ctx->port);
}

static void reader_main(h2o_memcached_context_t *ctx)
{
    struct st_h2o_memcached_conn_t conn = {ctx, {0}, PTHREAD_MUTEX_INITIALIZER, {&conn.inflight, &conn.inflight}, 0};
    pthread_t writer_thread;
    yrmcds_response resp;
    yrmcds_error err;
    int ret;

    /* connect to server and start the writer thread */
    connect_to_server(conn.ctx, &conn.yrmcds);
    if ((ret = pthread_create(&writer_thread, NULL, writer_main, &conn)) != 0) {
        char buf[128];
        h2o_fatal("pthread_create: %s", h2o_strerror_r(ret, buf, sizeof(buf)));
    }

    pthread_mutex_lock(&conn.ctx->mutex);
    ++conn.ctx->num_threads_connected;
    pthread_mutex_unlock(&conn.ctx->mutex);

    /* receive data until an error occurs */
    while (1) {
        if ((err = yrmcds_recv(&conn.yrmcds, &resp)) != YRMCDS_OK) {
            h2o_error_printf("[lib/common/memcached.c] yrmcds_recv:%s\n", yrmcds_strerror(err));
            break;
        }
        h2o_memcached_req_t *req = pop_inflight(&conn, resp.serial);
        if (req == NULL) {
            if (conn.yrmcds.text_mode)
                continue;
            h2o_error_printf("[lib/common/memcached.c] received unexpected serial\n");
            break;
        }
        if (resp.status == YRMCDS_STATUS_OK) {
            req->data.get.value = h2o_iovec_init(h2o_mem_alloc(resp.data_len), resp.data_len);
            memcpy(req->data.get.value.base, resp.data, resp.data_len);
            h2o_mem_set_secure((void *)resp.data, 0, resp.data_len);
        }
        h2o_multithread_send_message(req->data.get.receiver, &req->data.get.message);
    }

    /* send error to all the reqs in-flight */
    pthread_mutex_lock(&conn.mutex);
    while (!h2o_linklist_is_empty(&conn.inflight)) {
        h2o_memcached_req_t *req = H2O_STRUCT_FROM_MEMBER(h2o_memcached_req_t, inflight, conn.inflight.next);
        h2o_linklist_unlink(&req->inflight);
        assert(req->type == REQ_TYPE_GET);
        h2o_multithread_send_message(req->data.get.receiver, &req->data.get.message);
    }
    pthread_mutex_unlock(&conn.mutex);

    /* stop the writer thread */
    __sync_add_and_fetch(&conn.writer_exit_requested, 1);
    pthread_mutex_lock(&conn.ctx->mutex);
    pthread_cond_broadcast(&conn.ctx->cond);
    pthread_mutex_unlock(&conn.ctx->mutex);
    pthread_join(writer_thread, NULL);

    /* decrement num_threads_connected, and discard all the pending requests if no connections are alive */
    pthread_mutex_lock(&conn.ctx->mutex);
    if (--conn.ctx->num_threads_connected == 0) {
        while (!h2o_linklist_is_empty(&conn.ctx->pending)) {
            h2o_memcached_req_t *req = H2O_STRUCT_FROM_MEMBER(h2o_memcached_req_t, pending, conn.ctx->pending.next);
            h2o_linklist_unlink(&req->pending);
            discard_req(req);
        }
    }
    pthread_mutex_unlock(&conn.ctx->mutex);

    /* close the connection */
    yrmcds_close(&conn.yrmcds);
}

static void *thread_main(void *_ctx)
{
    h2o_memcached_context_t *ctx = _ctx;

    while (1)
        reader_main(ctx);
    return NULL;
}

static void dispatch(h2o_memcached_context_t *ctx, h2o_memcached_req_t *req)
{
    pthread_mutex_lock(&ctx->mutex);

    if (ctx->num_threads_connected != 0) {
        h2o_linklist_insert(&ctx->pending, &req->pending);
        pthread_cond_signal(&ctx->cond);
    } else {
        discard_req(req);
    }

    pthread_mutex_unlock(&ctx->mutex);
}

void h2o_memcached_receiver(h2o_multithread_receiver_t *receiver, h2o_linklist_t *messages)
{
    while (!h2o_linklist_is_empty(messages)) {
        h2o_memcached_req_t *req = H2O_STRUCT_FROM_MEMBER(h2o_memcached_req_t, data.get.message.link, messages->next);
        h2o_linklist_unlink(&req->data.get.message.link);
        assert(req->type == REQ_TYPE_GET);
        if (req->data.get.cb != NULL) {
            if (req->data.get.value_is_encoded && req->data.get.value.len != 0) {
                h2o_iovec_t decoded = h2o_decode_base64url(NULL, req->data.get.value.base, req->data.get.value.len);
                h2o_mem_set_secure(req->data.get.value.base, 0, req->data.get.value.len);
                free(req->data.get.value.base);
                req->data.get.value = decoded;
            }
            req->data.get.cb(req->data.get.value, req->data.get.cb_data);
        }
        free_req(req);
    }
}

h2o_memcached_req_t *h2o_memcached_get(h2o_memcached_context_t *ctx, h2o_multithread_receiver_t *receiver, h2o_iovec_t key,
                                       h2o_memcached_get_cb cb, void *cb_data, int flags)
{
    h2o_memcached_req_t *req = create_req(ctx, REQ_TYPE_GET, key, (flags & H2O_MEMCACHED_ENCODE_KEY) != 0);
    req->data.get.receiver = receiver;
    req->data.get.cb = cb;
    req->data.get.cb_data = cb_data;
    req->data.get.value_is_encoded = (flags & H2O_MEMCACHED_ENCODE_VALUE) != 0;
    dispatch(ctx, req);
    return req;
}

void h2o_memcached_cancel_get(h2o_memcached_context_t *ctx, h2o_memcached_req_t *req)
{
    int do_free = 0;

    pthread_mutex_lock(&ctx->mutex);
    req->data.get.cb = NULL;
    if (h2o_linklist_is_linked(&req->pending)) {
        h2o_linklist_unlink(&req->pending);
        do_free = 1;
    }
    pthread_mutex_unlock(&ctx->mutex);

    if (do_free)
        free_req(req);
}

void h2o_memcached_set(h2o_memcached_context_t *ctx, h2o_iovec_t key, h2o_iovec_t value, uint32_t expiration, int flags)
{
    h2o_memcached_req_t *req = create_req(ctx, REQ_TYPE_SET, key, (flags & H2O_MEMCACHED_ENCODE_KEY) != 0);
    if ((flags & H2O_MEMCACHED_ENCODE_VALUE) != 0) {
        req->data.set.value.base = h2o_mem_alloc((value.len + 2) / 3 * 4 + 1);
        req->data.set.value.len = h2o_base64_encode(req->data.set.value.base, value.base, value.len, 1);
    } else {
        req->data.set.value = h2o_iovec_init(h2o_mem_alloc(value.len), value.len);
        memcpy(req->data.set.value.base, value.base, value.len);
    }
    req->data.set.expiration = expiration;
    dispatch(ctx, req);
}

void h2o_memcached_delete(h2o_memcached_context_t *ctx, h2o_iovec_t key, int flags)
{
    h2o_memcached_req_t *req = create_req(ctx, REQ_TYPE_DELETE, key, (flags & H2O_MEMCACHED_ENCODE_KEY) != 0);
    dispatch(ctx, req);
}

h2o_memcached_context_t *h2o_memcached_create_context(const char *host, uint16_t port, int text_protocol, size_t num_threads,
                                                      const char *prefix)
{
    h2o_memcached_context_t *ctx = h2o_mem_alloc(sizeof(*ctx));

    pthread_mutex_init(&ctx->mutex, NULL);
    pthread_cond_init(&ctx->cond, NULL);
    h2o_linklist_init_anchor(&ctx->pending);
    ctx->num_threads_connected = 0;
    ctx->host = h2o_strdup(NULL, host, SIZE_MAX).base;
    ctx->port = port;
    ctx->text_protocol = text_protocol;
    ctx->prefix = h2o_strdup(NULL, prefix, SIZE_MAX);

    { /* start the threads */
        pthread_t tid;
        pthread_attr_t attr;
        size_t i;
        pthread_attr_init(&attr);
        pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
        for (i = 0; i != num_threads; ++i)
            h2o_multithread_create_thread(&tid, &attr, thread_main, ctx);
        pthread_attr_destroy(&attr);
    }

    return ctx;
}

Coverage Report

Created: 2026-10-17 07:02

Line	Count	Source
1		/*
2		* Copyright (c) 2015 DeNA Co., Ltd., Kazuho Oku
3		*
4		* Permission is hereby granted, free of charge, to any person obtaining a copy
5		* of this software and associated documentation files (the "Software"), to
6		* deal in the Software without restriction, including without limitation the
7		* rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
8		* sell copies of the Software, and to permit persons to whom the Software is
9		* furnished to do so, subject to the following conditions:
10		*
11		* The above copyright notice and this permission notice shall be included in
12		* all copies or substantial portions of the Software.
13		*
14		* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
15		* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
16		* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
17		* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
18		* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
19		* FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
20		* IN THE SOFTWARE.
21		*/
22		#include <errno.h>
23		#include <inttypes.h>
24		#include <unistd.h>
25		#include "yrmcds.h"
26		#include "h2o/linklist.h"
27		#include "h2o/memcached.h"
28		#include "h2o/rand.h"
29		#include "h2o/string_.h"
30
31		struct st_h2o_memcached_context_t {
32		pthread_mutex_t mutex;
33		pthread_cond_t cond;
34		h2o_linklist_t pending;
35		size_t num_threads_connected;
36		char *host;
37		uint16_t port;
38		int text_protocol;
39		h2o_iovec_t prefix;
40		};
41
42		struct st_h2o_memcached_conn_t {
43		h2o_memcached_context_t *ctx;
44		yrmcds yrmcds;
45		pthread_mutex_t mutex;
46		h2o_linklist_t inflight;
47		int writer_exit_requested;
48		};
49
50		enum en_h2o_memcached_req_type_t { REQ_TYPE_GET, REQ_TYPE_SET, REQ_TYPE_DELETE };
51
52		struct st_h2o_memcached_req_t {
53		enum en_h2o_memcached_req_type_t type;
54		int refcnt;
55		h2o_linklist_t pending;
56		h2o_linklist_t inflight;
57		union {
58		struct {
59		h2o_multithread_receiver_t *receiver;
60		h2o_multithread_message_t message;
61		h2o_memcached_get_cb cb;
62		void *cb_data;
63		int value_is_encoded;
64		h2o_iovec_t value;
65		uint32_t serial;
66		} get;
67		struct {
68		h2o_iovec_t value;
69		uint32_t expiration;
70		} set;
71		} data;
72		struct {
73		size_t len;
74		char base[1];
75		} key;
76		};
77
78		static h2o_memcached_req_t create_req(h2o_memcached_context_t ctx, enum en_h2o_memcached_req_type_t type, h2o_iovec_t key,
79		int encode_key)
80	0	{
81	0	h2o_memcached_req_t *req = h2o_mem_alloc(offsetof(h2o_memcached_req_t, key.base) + ctx->prefix.len +
82	0	(encode_key ? (key.len + 2) / 3 * 4 + 1 : key.len));
83	0	req->type = type;
84	0	req->refcnt = 1;
85	0	req->pending = (h2o_linklist_t){NULL};
86	0	req->inflight = (h2o_linklist_t){NULL};
87	0	memset(&req->data, 0, sizeof(req->data));
88	0	if (ctx->prefix.len != 0)
89	0	memcpy(req->key.base, ctx->prefix.base, ctx->prefix.len);
90	0	req->key.len = ctx->prefix.len;
91	0	if (encode_key) {
92	0	req->key.len += h2o_base64_encode(req->key.base + req->key.len, key.base, key.len, 1);
93	0	} else {
94	0	memcpy(req->key.base + req->key.len, key.base, key.len);
95	0	req->key.len += key.len;
96	0	}
97	0	return req;
98	0	}
99
100		static void free_req(h2o_memcached_req_t *req)
101	0	{
102	0	if (__sync_sub_and_fetch(&req->refcnt, 1) != 0)
103	0	return;
104
105	0	assert(!h2o_linklist_is_linked(&req->pending));
106	0	switch (req->type) {
107	0	case REQ_TYPE_GET:
108	0	assert(!h2o_linklist_is_linked(&req->data.get.message.link));
109	0	h2o_mem_set_secure(req->data.get.value.base, 0, req->data.get.value.len);
110	0	free(req->data.get.value.base);
111	0	break;
112	0	case REQ_TYPE_SET:
113	0	h2o_mem_set_secure(req->data.set.value.base, 0, req->data.set.value.len);
114	0	free(req->data.set.value.base);
115	0	break;
116	0	case REQ_TYPE_DELETE:
117	0	break;
118	0	default:
119	0	assert(!"FIXME");
120	0	break;
121	0	}
122	0	free(req);
123	0	}
124
125		static void retain_req(h2o_memcached_req_t *req)
126	0	{
127	0	__sync_add_and_fetch(&req->refcnt, 1);
128	0	}
129
130		static void discard_req(h2o_memcached_req_t *req)
131	0	{
132	0	switch (req->type) {
133	0	case REQ_TYPE_GET:
134	0	h2o_multithread_send_message(req->data.get.receiver, &req->data.get.message);
135	0	break;
136	0	default:
137	0	free_req(req);
138	0	break;
139	0	}
140	0	}
141
142		static h2o_memcached_req_t pop_inflight(struct st_h2o_memcached_conn_t conn, uint32_t serial)
143	0	{
144	0	h2o_memcached_req_t *req;
145
146	0	pthread_mutex_lock(&conn->mutex);
147
148	0	if (conn->yrmcds.text_mode) {
149		/* in text mode, responses are returned in order (and we may receive responses for commands other than GET) */
150	0	if (!h2o_linklist_is_empty(&conn->inflight)) {
151	0	req = H2O_STRUCT_FROM_MEMBER(h2o_memcached_req_t, inflight, conn->inflight.next);
152	0	assert(req->type == REQ_TYPE_GET);
153	0	if (req->data.get.serial == serial)
154	0	goto Found;
155	0	}
156	0	} else {
157		/* in binary mode, responses are received out-of-order (and we would only recieve responses for GET) */
158	0	h2o_linklist_t *node;
159	0	for (node = conn->inflight.next; node != &conn->inflight; node = node->next) {
160	0	req = H2O_STRUCT_FROM_MEMBER(h2o_memcached_req_t, inflight, node);
161	0	assert(req->type == REQ_TYPE_GET);
162	0	if (req->data.get.serial == serial)
163	0	goto Found;
164	0	}
165	0	}
166
167		/* not found */
168	0	pthread_mutex_unlock(&conn->mutex);
169	0	return NULL;
170
171	0	Found:
172	0	h2o_linklist_unlink(&req->inflight);
173	0	pthread_mutex_unlock(&conn->mutex);
174	0	return req;
175	0	}
176
177		static void writer_main(void _conn)
178	0	{
179	0	struct st_h2o_memcached_conn_t *conn = _conn;
180	0	yrmcds_error err;
181
182	0	pthread_mutex_lock(&conn->ctx->mutex);
183
184	0	while (!__sync_add_and_fetch(&conn->writer_exit_requested, 0)) {
185	0	while (!h2o_linklist_is_empty(&conn->ctx->pending)) {
186	0	h2o_memcached_req_t *req = H2O_STRUCT_FROM_MEMBER(h2o_memcached_req_t, pending, conn->ctx->pending.next);
187	0	h2o_linklist_unlink(&req->pending);
188	0	pthread_mutex_unlock(&conn->ctx->mutex);
189
190	0	switch (req->type) {
191	0	case REQ_TYPE_GET:
192	0	retain_req(req);
193	0	pthread_mutex_lock(&conn->mutex);
194	0	h2o_linklist_insert(&conn->inflight, &req->inflight);
195	0	pthread_mutex_unlock(&conn->mutex);
196	0	err = yrmcds_get(&conn->yrmcds, req->key.base, req->key.len, 0, &req->data.get.serial);
197	0	break;
198	0	case REQ_TYPE_SET:
199	0	err = yrmcds_set(&conn->yrmcds, req->key.base, req->key.len, req->data.set.value.base, req->data.set.value.len, 0,
200	0	req->data.set.expiration, 0, !conn->yrmcds.text_mode, NULL);
201	0	break;
202	0	case REQ_TYPE_DELETE:
203	0	err = yrmcds_remove(&conn->yrmcds, req->key.base, req->key.len, !conn->yrmcds.text_mode, NULL);
204	0	break;
205	0	default:
206	0	h2o_error_printf("[lib/common/memcached.c] unknown type:%d\n", (int)req->type);
207	0	err = YRMCDS_NOT_IMPLEMENTED;
208	0	break;
209	0	}
210	0	free_req(req);
211	0	if (err != YRMCDS_OK)
212	0	goto Error;
213
214	0	pthread_mutex_lock(&conn->ctx->mutex);
215	0	}
216	0	pthread_cond_wait(&conn->ctx->cond, &conn->ctx->mutex);
217	0	}
218
219	0	pthread_mutex_unlock(&conn->ctx->mutex);
220	0	return NULL;
221
222	0	Error:
223	0	h2o_error_printf("[lib/common/memcached.c] failed to send request; %s\n", yrmcds_strerror(err));
224		/* doc says the call can be used to interrupt yrmcds_recv */
225	0	yrmcds_shutdown(&conn->yrmcds);
226
227	0	return NULL;
228	0	}
229
230		static void connect_to_server(h2o_memcached_context_t ctx, yrmcds yrmcds)
231	0	{
232	0	size_t failcnt;
233	0	yrmcds_error err;
234
235	0	for (failcnt = 0; (err = yrmcds_connect(yrmcds, ctx->host, ctx->port)) != YRMCDS_OK; ++failcnt) {
236	0	if (failcnt == 0) {
237	0	h2o_error_printf("[lib/common/memcached.c] failed to connect to memcached at %s:%" PRIu16 ", %s\n", ctx->host,
238	0	ctx->port, yrmcds_strerror(err));
239	0	}
240	0	++failcnt;
241	0	usleep(2000000 + h2o_rand() % 3000000); /* sleep 2 to 5 seconds */
242	0	}
243		/* connected */
244	0	if (ctx->text_protocol)
245	0	yrmcds_text_mode(yrmcds);
246	0	h2o_error_printf("[lib/common/memcached.c] connected to memcached at %s:%" PRIu16 "\n", ctx->host, ctx->port);
247	0	}
248
249		static void reader_main(h2o_memcached_context_t *ctx)
250	0	{
251	0	struct st_h2o_memcached_conn_t conn = {ctx, {0}, PTHREAD_MUTEX_INITIALIZER, {&conn.inflight, &conn.inflight}, 0};
252	0	pthread_t writer_thread;
253	0	yrmcds_response resp;
254	0	yrmcds_error err;
255	0	int ret;
256
257		/* connect to server and start the writer thread */
258	0	connect_to_server(conn.ctx, &conn.yrmcds);
259	0	if ((ret = pthread_create(&writer_thread, NULL, writer_main, &conn)) != 0) {
260	0	char buf[128];
261	0	h2o_fatal("pthread_create: %s", h2o_strerror_r(ret, buf, sizeof(buf)));
262	0	}
263
264	0	pthread_mutex_lock(&conn.ctx->mutex);
265	0	++conn.ctx->num_threads_connected;
266	0	pthread_mutex_unlock(&conn.ctx->mutex);
267
268		/* receive data until an error occurs */
269	0	while (1) {
270	0	if ((err = yrmcds_recv(&conn.yrmcds, &resp)) != YRMCDS_OK) {
271	0	h2o_error_printf("[lib/common/memcached.c] yrmcds_recv:%s\n", yrmcds_strerror(err));
272	0	break;
273	0	}
274	0	h2o_memcached_req_t *req = pop_inflight(&conn, resp.serial);
275	0	if (req == NULL) {
276	0	if (conn.yrmcds.text_mode)
277	0	continue;
278	0	h2o_error_printf("[lib/common/memcached.c] received unexpected serial\n");
279	0	break;
280	0	}
281	0	if (resp.status == YRMCDS_STATUS_OK) {
282	0	req->data.get.value = h2o_iovec_init(h2o_mem_alloc(resp.data_len), resp.data_len);
283	0	memcpy(req->data.get.value.base, resp.data, resp.data_len);
284	0	h2o_mem_set_secure((void *)resp.data, 0, resp.data_len);
285	0	}
286	0	h2o_multithread_send_message(req->data.get.receiver, &req->data.get.message);
287	0	}
288
289		/* send error to all the reqs in-flight */
290	0	pthread_mutex_lock(&conn.mutex);
291	0	while (!h2o_linklist_is_empty(&conn.inflight)) {
292	0	h2o_memcached_req_t *req = H2O_STRUCT_FROM_MEMBER(h2o_memcached_req_t, inflight, conn.inflight.next);
293	0	h2o_linklist_unlink(&req->inflight);
294	0	assert(req->type == REQ_TYPE_GET);
295	0	h2o_multithread_send_message(req->data.get.receiver, &req->data.get.message);
296	0	}
297	0	pthread_mutex_unlock(&conn.mutex);
298
299		/* stop the writer thread */
300	0	__sync_add_and_fetch(&conn.writer_exit_requested, 1);
301	0	pthread_mutex_lock(&conn.ctx->mutex);
302	0	pthread_cond_broadcast(&conn.ctx->cond);
303	0	pthread_mutex_unlock(&conn.ctx->mutex);
304	0	pthread_join(writer_thread, NULL);
305
306		/* decrement num_threads_connected, and discard all the pending requests if no connections are alive */
307	0	pthread_mutex_lock(&conn.ctx->mutex);
308	0	if (--conn.ctx->num_threads_connected == 0) {
309	0	while (!h2o_linklist_is_empty(&conn.ctx->pending)) {
310	0	h2o_memcached_req_t *req = H2O_STRUCT_FROM_MEMBER(h2o_memcached_req_t, pending, conn.ctx->pending.next);
311	0	h2o_linklist_unlink(&req->pending);
312	0	discard_req(req);
313	0	}
314	0	}
315	0	pthread_mutex_unlock(&conn.ctx->mutex);
316
317		/* close the connection */
318	0	yrmcds_close(&conn.yrmcds);
319	0	}
320
321		static void thread_main(void _ctx)
322	0	{
323	0	h2o_memcached_context_t *ctx = _ctx;
324
325	0	while (1)
326	0	reader_main(ctx);
327	0	return NULL;
328	0	}
329
330		static void dispatch(h2o_memcached_context_t ctx, h2o_memcached_req_t req)
331	0	{
332	0	pthread_mutex_lock(&ctx->mutex);
333
334	0	if (ctx->num_threads_connected != 0) {
335	0	h2o_linklist_insert(&ctx->pending, &req->pending);
336	0	pthread_cond_signal(&ctx->cond);
337	0	} else {
338	0	discard_req(req);
339	0	}
340
341	0	pthread_mutex_unlock(&ctx->mutex);
342	0	}
343
344		void h2o_memcached_receiver(h2o_multithread_receiver_t receiver, h2o_linklist_t messages)
345	0	{
346	0	while (!h2o_linklist_is_empty(messages)) {
347	0	h2o_memcached_req_t *req = H2O_STRUCT_FROM_MEMBER(h2o_memcached_req_t, data.get.message.link, messages->next);
348	0	h2o_linklist_unlink(&req->data.get.message.link);
349	0	assert(req->type == REQ_TYPE_GET);
350	0	if (req->data.get.cb != NULL) {
351	0	if (req->data.get.value_is_encoded && req->data.get.value.len != 0) {
352	0	h2o_iovec_t decoded = h2o_decode_base64url(NULL, req->data.get.value.base, req->data.get.value.len);
353	0	h2o_mem_set_secure(req->data.get.value.base, 0, req->data.get.value.len);
354	0	free(req->data.get.value.base);
355	0	req->data.get.value = decoded;
356	0	}
357	0	req->data.get.cb(req->data.get.value, req->data.get.cb_data);
358	0	}
359	0	free_req(req);
360	0	}
361	0	}
362
363		h2o_memcached_req_t h2o_memcached_get(h2o_memcached_context_t ctx, h2o_multithread_receiver_t *receiver, h2o_iovec_t key,
364		h2o_memcached_get_cb cb, void *cb_data, int flags)
365	0	{
366	0	h2o_memcached_req_t *req = create_req(ctx, REQ_TYPE_GET, key, (flags & H2O_MEMCACHED_ENCODE_KEY) != 0);
367	0	req->data.get.receiver = receiver;
368	0	req->data.get.cb = cb;
369	0	req->data.get.cb_data = cb_data;
370	0	req->data.get.value_is_encoded = (flags & H2O_MEMCACHED_ENCODE_VALUE) != 0;
371	0	dispatch(ctx, req);
372	0	return req;
373	0	}
374
375		void h2o_memcached_cancel_get(h2o_memcached_context_t ctx, h2o_memcached_req_t req)
376	0	{
377	0	int do_free = 0;
378
379	0	pthread_mutex_lock(&ctx->mutex);
380	0	req->data.get.cb = NULL;
381	0	if (h2o_linklist_is_linked(&req->pending)) {
382	0	h2o_linklist_unlink(&req->pending);
383	0	do_free = 1;
384	0	}
385	0	pthread_mutex_unlock(&ctx->mutex);
386
387	0	if (do_free)
388	0	free_req(req);
389	0	}
390
391		void h2o_memcached_set(h2o_memcached_context_t *ctx, h2o_iovec_t key, h2o_iovec_t value, uint32_t expiration, int flags)
392	0	{
393	0	h2o_memcached_req_t *req = create_req(ctx, REQ_TYPE_SET, key, (flags & H2O_MEMCACHED_ENCODE_KEY) != 0);
394	0	if ((flags & H2O_MEMCACHED_ENCODE_VALUE) != 0) {
395	0	req->data.set.value.base = h2o_mem_alloc((value.len + 2) / 3 * 4 + 1);
396	0	req->data.set.value.len = h2o_base64_encode(req->data.set.value.base, value.base, value.len, 1);
397	0	} else {
398	0	req->data.set.value = h2o_iovec_init(h2o_mem_alloc(value.len), value.len);
399	0	memcpy(req->data.set.value.base, value.base, value.len);
400	0	}
401	0	req->data.set.expiration = expiration;
402	0	dispatch(ctx, req);
403	0	}
404
405		void h2o_memcached_delete(h2o_memcached_context_t *ctx, h2o_iovec_t key, int flags)
406	0	{
407	0	h2o_memcached_req_t *req = create_req(ctx, REQ_TYPE_DELETE, key, (flags & H2O_MEMCACHED_ENCODE_KEY) != 0);
408	0	dispatch(ctx, req);
409	0	}
410
411		h2o_memcached_context_t h2o_memcached_create_context(const char host, uint16_t port, int text_protocol, size_t num_threads,
412		const char *prefix)
413	0	{
414	0	h2o_memcached_context_t ctx = h2o_mem_alloc(sizeof(ctx));
415
416	0	pthread_mutex_init(&ctx->mutex, NULL);
417	0	pthread_cond_init(&ctx->cond, NULL);
418	0	h2o_linklist_init_anchor(&ctx->pending);
419	0	ctx->num_threads_connected = 0;
420	0	ctx->host = h2o_strdup(NULL, host, SIZE_MAX).base;
421	0	ctx->port = port;
422	0	ctx->text_protocol = text_protocol;
423	0	ctx->prefix = h2o_strdup(NULL, prefix, SIZE_MAX);
424
425	0	{ /* start the threads */
426	0	pthread_t tid;
427	0	pthread_attr_t attr;
428	0	size_t i;
429	0	pthread_attr_init(&attr);
430	0	pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
431	0	for (i = 0; i != num_threads; ++i)
432	0	h2o_multithread_create_thread(&tid, &attr, thread_main, ctx);
433	0	pthread_attr_destroy(&attr);
434	0	}
435
436	0	return ctx;
437	0	}