1 /* Licensed to the Apache Software Foundation (ASF) under one or more
2 * contributor license agreements. See the NOTICE file distributed with
3 * this work for additional information regarding copyright ownership.
4 * The ASF licenses this file to You under the Apache License, Version 2.0
5 * (the "License"); you may not use this file except in compliance with
6 * the License. You may obtain a copy of the License at
8 * http://www.apache.org/licenses/LICENSE-2.0
10 * Unless required by applicable law or agreed to in writing, software
11 * distributed under the License is distributed on an "AS IS" BASIS,
12 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 * See the License for the specific language governing permissions and
14 * limitations under the License.
17 #include "apr_memcache.h"
19 #include "apr_version.h"
22 #define BUFFER_SIZE 512
23 struct apr_memcache_conn_t
30 apr_bucket_brigade *bb;
31 apr_bucket_brigade *tb;
32 apr_memcache_server_t *ms;
35 /* Strings for Client Commands */
38 #define MC_EOL_LEN (sizeof(MC_EOL)-1)
41 #define MC_WS_LEN (sizeof(MC_WS)-1)
44 #define MC_GET_LEN (sizeof(MC_GET)-1)
47 #define MC_SET_LEN (sizeof(MC_SET)-1)
50 #define MC_ADD_LEN (sizeof(MC_ADD)-1)
52 #define MC_REPLACE "replace "
53 #define MC_REPLACE_LEN (sizeof(MC_REPLACE)-1)
55 #define MC_DELETE "delete "
56 #define MC_DELETE_LEN (sizeof(MC_DELETE)-1)
58 #define MC_INCR "incr "
59 #define MC_INCR_LEN (sizeof(MC_INCR)-1)
61 #define MC_DECR "decr "
62 #define MC_DECR_LEN (sizeof(MC_DECR)-1)
64 #define MC_VERSION "version"
65 #define MC_VERSION_LEN (sizeof(MC_VERSION)-1)
67 #define MC_STATS "stats"
68 #define MC_STATS_LEN (sizeof(MC_STATS)-1)
70 #define MC_QUIT "quit"
71 #define MC_QUIT_LEN (sizeof(MC_QUIT)-1)
73 /* Strings for Server Replies */
75 #define MS_STORED "STORED"
76 #define MS_STORED_LEN (sizeof(MS_STORED)-1)
78 #define MS_NOT_STORED "NOT_STORED"
79 #define MS_NOT_STORED_LEN (sizeof(MS_NOT_STORED)-1)
81 #define MS_DELETED "DELETED"
82 #define MS_DELETED_LEN (sizeof(MS_DELETED)-1)
84 #define MS_NOT_FOUND "NOT_FOUND"
85 #define MS_NOT_FOUND_LEN (sizeof(MS_NOT_FOUND)-1)
87 #define MS_VALUE "VALUE"
88 #define MS_VALUE_LEN (sizeof(MS_VALUE)-1)
90 #define MS_ERROR "ERROR"
91 #define MS_ERROR_LEN (sizeof(MS_ERROR)-1)
93 #define MS_VERSION "VERSION"
94 #define MS_VERSION_LEN (sizeof(MS_VERSION)-1)
96 #define MS_STAT "STAT"
97 #define MS_STAT_LEN (sizeof(MS_STAT)-1)
100 #define MS_END_LEN (sizeof(MS_END)-1)
102 /** Server and Query Structure for a multiple get */
103 struct cache_server_query_t {
104 apr_memcache_server_t* ms;
105 apr_memcache_conn_t* conn;
106 struct iovec* query_vec;
107 apr_int32_t query_vec_count;
110 #define MULT_GET_TIMEOUT 50000
112 static apr_status_t make_server_dead(apr_memcache_t *mc, apr_memcache_server_t *ms)
115 apr_thread_mutex_lock(ms->lock);
117 ms->status = APR_MC_SERVER_DEAD;
118 ms->btime = apr_time_now();
120 apr_thread_mutex_unlock(ms->lock);
125 static apr_status_t make_server_live(apr_memcache_t *mc, apr_memcache_server_t *ms)
127 ms->status = APR_MC_SERVER_LIVE;
132 APU_DECLARE(apr_status_t) apr_memcache_add_server(apr_memcache_t *mc, apr_memcache_server_t *ms)
134 apr_status_t rv = APR_SUCCESS;
136 if(mc->ntotal >= mc->nalloc) {
140 mc->live_servers[mc->ntotal] = ms;
142 make_server_live(mc, ms);
146 static apr_status_t mc_version_ping(apr_memcache_server_t *ms);
148 APU_DECLARE(apr_memcache_server_t *)
149 apr_memcache_find_server_hash(apr_memcache_t *mc, const apr_uint32_t hash)
151 if (mc->server_func) {
152 return mc->server_func(mc->server_baton, mc, hash);
155 return apr_memcache_find_server_hash_default(NULL, mc, hash);
159 APU_DECLARE(apr_memcache_server_t *)
160 apr_memcache_find_server_hash_default(void *baton, apr_memcache_t *mc,
161 const apr_uint32_t hash)
163 apr_memcache_server_t *ms = NULL;
164 apr_uint32_t h = hash ? hash : 1;
166 apr_time_t curtime = 0;
168 if(mc->ntotal == 0) {
173 ms = mc->live_servers[h % mc->ntotal];
174 if(ms->status == APR_MC_SERVER_LIVE) {
179 curtime = apr_time_now();
182 apr_thread_mutex_lock(ms->lock);
184 /* Try the dead server, every 5 seconds */
185 if (curtime - ms->btime > apr_time_from_sec(5)) {
187 if (mc_version_ping(ms) == APR_SUCCESS) {
188 make_server_live(mc, ms);
190 apr_thread_mutex_unlock(ms->lock);
196 apr_thread_mutex_unlock(ms->lock);
201 } while(i < mc->ntotal);
203 if (i == mc->ntotal) {
210 APU_DECLARE(apr_memcache_server_t *) apr_memcache_find_server(apr_memcache_t *mc, const char *host, apr_port_t port)
214 for (i = 0; i < mc->ntotal; i++) {
215 if (strcmp(mc->live_servers[i]->host, host) == 0
216 && mc->live_servers[i]->port == port) {
218 return mc->live_servers[i];
225 static apr_status_t ms_find_conn(apr_memcache_server_t *ms, apr_memcache_conn_t **conn)
228 apr_bucket_alloc_t *balloc;
232 rv = apr_reslist_acquire(ms->conns, (void **)conn);
238 if (rv != APR_SUCCESS) {
242 balloc = apr_bucket_alloc_create((*conn)->tp);
243 (*conn)->bb = apr_brigade_create((*conn)->tp, balloc);
244 (*conn)->tb = apr_brigade_create((*conn)->tp, balloc);
246 e = apr_bucket_socket_create((*conn)->sock, balloc);
247 APR_BRIGADE_INSERT_TAIL((*conn)->bb, e);
252 static apr_status_t ms_bad_conn(apr_memcache_server_t *ms, apr_memcache_conn_t *conn)
255 return apr_reslist_invalidate(ms->conns, conn);
261 static apr_status_t ms_release_conn(apr_memcache_server_t *ms, apr_memcache_conn_t *conn)
263 apr_pool_clear(conn->tp);
265 return apr_reslist_release(ms->conns, conn);
271 APU_DECLARE(apr_status_t) apr_memcache_enable_server(apr_memcache_t *mc, apr_memcache_server_t *ms)
273 apr_status_t rv = APR_SUCCESS;
275 if (ms->status == APR_MC_SERVER_LIVE) {
279 rv = make_server_live(mc, ms);
283 APU_DECLARE(apr_status_t) apr_memcache_disable_server(apr_memcache_t *mc, apr_memcache_server_t *ms)
285 return make_server_dead(mc, ms);
288 static apr_status_t conn_connect(apr_memcache_conn_t *conn)
290 apr_status_t rv = APR_SUCCESS;
292 #if APR_HAVE_SOCKADDR_UN
293 apr_int32_t family = conn->ms->host[0] != '/' ? APR_INET : APR_UNIX;
295 apr_int32_t family = APR_INET;
298 rv = apr_sockaddr_info_get(&sa, conn->ms->host, family, conn->ms->port, 0, conn->p);
299 if (rv != APR_SUCCESS) {
303 rv = apr_socket_timeout_set(conn->sock, 1 * APR_USEC_PER_SEC);
304 if (rv != APR_SUCCESS) {
308 rv = apr_socket_connect(conn->sock, sa);
309 if (rv != APR_SUCCESS) {
313 rv = apr_socket_timeout_set(conn->sock, -1);
314 if (rv != APR_SUCCESS) {
323 mc_conn_construct(void **conn_, void *params, apr_pool_t *pool)
325 apr_status_t rv = APR_SUCCESS;
326 apr_memcache_conn_t *conn;
329 apr_memcache_server_t *ms = params;
330 #if APR_HAVE_SOCKADDR_UN
331 apr_int32_t family = ms->host[0] != '/' ? APR_INET : APR_UNIX;
333 apr_int32_t family = APR_INET;
336 rv = apr_pool_create(&np, pool);
337 if (rv != APR_SUCCESS) {
341 rv = apr_pool_create(&tp, np);
342 if (rv != APR_SUCCESS) {
343 apr_pool_destroy(np);
347 conn = apr_palloc(np, sizeof( apr_memcache_conn_t ));
352 rv = apr_socket_create(&conn->sock, family, SOCK_STREAM, 0, np);
354 if (rv != APR_SUCCESS) {
355 apr_pool_destroy(np);
359 conn->buffer = apr_palloc(conn->p, BUFFER_SIZE);
363 rv = conn_connect(conn);
364 if (rv != APR_SUCCESS) {
365 apr_pool_destroy(np);
376 mc_conn_destruct(void *conn_, void *params, apr_pool_t *pool)
378 apr_memcache_conn_t *conn = (apr_memcache_conn_t*)conn_;
382 /* send a quit message to the memcached server to be nice about it. */
383 vec[0].iov_base = MC_QUIT;
384 vec[0].iov_len = MC_QUIT_LEN;
386 vec[1].iov_base = MC_EOL;
387 vec[1].iov_len = MC_EOL_LEN;
389 /* Return values not checked, since we just want to make it go away. */
390 apr_socket_sendv(conn->sock, vec, 2, &written);
391 apr_socket_close(conn->sock);
393 apr_pool_destroy(conn->p);
399 APU_DECLARE(apr_status_t) apr_memcache_server_create(apr_pool_t *p,
400 const char *host, apr_port_t port,
401 apr_uint32_t min, apr_uint32_t smax,
402 apr_uint32_t max, apr_uint32_t ttl,
403 apr_memcache_server_t **ms)
405 apr_status_t rv = APR_SUCCESS;
406 apr_memcache_server_t *server;
409 rv = apr_pool_create(&np, p);
411 server = apr_palloc(np, sizeof(apr_memcache_server_t));
414 server->host = apr_pstrdup(np, host);
416 server->status = APR_MC_SERVER_DEAD;
418 rv = apr_thread_mutex_create(&server->lock, APR_THREAD_MUTEX_DEFAULT, np);
419 if (rv != APR_SUCCESS) {
423 rv = apr_reslist_create(&server->conns,
424 min, /* hard minimum */
425 smax, /* soft maximum */
426 max, /* hard maximum */
427 ttl, /* Time to live */
428 mc_conn_construct, /* Make a New Connection */
429 mc_conn_destruct, /* Kill Old Connection */
431 if (rv != APR_SUCCESS) {
435 apr_reslist_cleanup_order_set(server->conns, APR_RESLIST_CLEANUP_FIRST);
437 rv = mc_conn_construct((void**)&(server->conn), server, np);
438 if (rv != APR_SUCCESS) {
448 APU_DECLARE(apr_status_t) apr_memcache_create(apr_pool_t *p,
449 apr_uint16_t max_servers, apr_uint32_t flags,
450 apr_memcache_t **memcache)
452 apr_status_t rv = APR_SUCCESS;
455 mc = apr_palloc(p, sizeof(apr_memcache_t));
457 mc->nalloc = max_servers;
459 mc->live_servers = apr_palloc(p, mc->nalloc * sizeof(struct apr_memcache_server_t *));
460 mc->hash_func = NULL;
461 mc->hash_baton = NULL;
462 mc->server_func = NULL;
463 mc->server_baton = NULL;
469 /* The crc32 functions and data was originally written by Spencer
470 * Garrett <srg@quick.com> and was gleaned from the PostgreSQL source
471 * tree via the files contrib/ltree/crc32.[ch] and from FreeBSD at
472 * src/usr.bin/cksum/crc32.c.
475 static const apr_uint32_t crc32tab[256] = {
476 0x00000000, 0x77073096, 0xee0e612c, 0x990951ba,
477 0x076dc419, 0x706af48f, 0xe963a535, 0x9e6495a3,
478 0x0edb8832, 0x79dcb8a4, 0xe0d5e91e, 0x97d2d988,
479 0x09b64c2b, 0x7eb17cbd, 0xe7b82d07, 0x90bf1d91,
480 0x1db71064, 0x6ab020f2, 0xf3b97148, 0x84be41de,
481 0x1adad47d, 0x6ddde4eb, 0xf4d4b551, 0x83d385c7,
482 0x136c9856, 0x646ba8c0, 0xfd62f97a, 0x8a65c9ec,
483 0x14015c4f, 0x63066cd9, 0xfa0f3d63, 0x8d080df5,
484 0x3b6e20c8, 0x4c69105e, 0xd56041e4, 0xa2677172,
485 0x3c03e4d1, 0x4b04d447, 0xd20d85fd, 0xa50ab56b,
486 0x35b5a8fa, 0x42b2986c, 0xdbbbc9d6, 0xacbcf940,
487 0x32d86ce3, 0x45df5c75, 0xdcd60dcf, 0xabd13d59,
488 0x26d930ac, 0x51de003a, 0xc8d75180, 0xbfd06116,
489 0x21b4f4b5, 0x56b3c423, 0xcfba9599, 0xb8bda50f,
490 0x2802b89e, 0x5f058808, 0xc60cd9b2, 0xb10be924,
491 0x2f6f7c87, 0x58684c11, 0xc1611dab, 0xb6662d3d,
492 0x76dc4190, 0x01db7106, 0x98d220bc, 0xefd5102a,
493 0x71b18589, 0x06b6b51f, 0x9fbfe4a5, 0xe8b8d433,
494 0x7807c9a2, 0x0f00f934, 0x9609a88e, 0xe10e9818,
495 0x7f6a0dbb, 0x086d3d2d, 0x91646c97, 0xe6635c01,
496 0x6b6b51f4, 0x1c6c6162, 0x856530d8, 0xf262004e,
497 0x6c0695ed, 0x1b01a57b, 0x8208f4c1, 0xf50fc457,
498 0x65b0d9c6, 0x12b7e950, 0x8bbeb8ea, 0xfcb9887c,
499 0x62dd1ddf, 0x15da2d49, 0x8cd37cf3, 0xfbd44c65,
500 0x4db26158, 0x3ab551ce, 0xa3bc0074, 0xd4bb30e2,
501 0x4adfa541, 0x3dd895d7, 0xa4d1c46d, 0xd3d6f4fb,
502 0x4369e96a, 0x346ed9fc, 0xad678846, 0xda60b8d0,
503 0x44042d73, 0x33031de5, 0xaa0a4c5f, 0xdd0d7cc9,
504 0x5005713c, 0x270241aa, 0xbe0b1010, 0xc90c2086,
505 0x5768b525, 0x206f85b3, 0xb966d409, 0xce61e49f,
506 0x5edef90e, 0x29d9c998, 0xb0d09822, 0xc7d7a8b4,
507 0x59b33d17, 0x2eb40d81, 0xb7bd5c3b, 0xc0ba6cad,
508 0xedb88320, 0x9abfb3b6, 0x03b6e20c, 0x74b1d29a,
509 0xead54739, 0x9dd277af, 0x04db2615, 0x73dc1683,
510 0xe3630b12, 0x94643b84, 0x0d6d6a3e, 0x7a6a5aa8,
511 0xe40ecf0b, 0x9309ff9d, 0x0a00ae27, 0x7d079eb1,
512 0xf00f9344, 0x8708a3d2, 0x1e01f268, 0x6906c2fe,
513 0xf762575d, 0x806567cb, 0x196c3671, 0x6e6b06e7,
514 0xfed41b76, 0x89d32be0, 0x10da7a5a, 0x67dd4acc,
515 0xf9b9df6f, 0x8ebeeff9, 0x17b7be43, 0x60b08ed5,
516 0xd6d6a3e8, 0xa1d1937e, 0x38d8c2c4, 0x4fdff252,
517 0xd1bb67f1, 0xa6bc5767, 0x3fb506dd, 0x48b2364b,
518 0xd80d2bda, 0xaf0a1b4c, 0x36034af6, 0x41047a60,
519 0xdf60efc3, 0xa867df55, 0x316e8eef, 0x4669be79,
520 0xcb61b38c, 0xbc66831a, 0x256fd2a0, 0x5268e236,
521 0xcc0c7795, 0xbb0b4703, 0x220216b9, 0x5505262f,
522 0xc5ba3bbe, 0xb2bd0b28, 0x2bb45a92, 0x5cb36a04,
523 0xc2d7ffa7, 0xb5d0cf31, 0x2cd99e8b, 0x5bdeae1d,
524 0x9b64c2b0, 0xec63f226, 0x756aa39c, 0x026d930a,
525 0x9c0906a9, 0xeb0e363f, 0x72076785, 0x05005713,
526 0x95bf4a82, 0xe2b87a14, 0x7bb12bae, 0x0cb61b38,
527 0x92d28e9b, 0xe5d5be0d, 0x7cdcefb7, 0x0bdbdf21,
528 0x86d3d2d4, 0xf1d4e242, 0x68ddb3f8, 0x1fda836e,
529 0x81be16cd, 0xf6b9265b, 0x6fb077e1, 0x18b74777,
530 0x88085ae6, 0xff0f6a70, 0x66063bca, 0x11010b5c,
531 0x8f659eff, 0xf862ae69, 0x616bffd3, 0x166ccf45,
532 0xa00ae278, 0xd70dd2ee, 0x4e048354, 0x3903b3c2,
533 0xa7672661, 0xd06016f7, 0x4969474d, 0x3e6e77db,
534 0xaed16a4a, 0xd9d65adc, 0x40df0b66, 0x37d83bf0,
535 0xa9bcae53, 0xdebb9ec5, 0x47b2cf7f, 0x30b5ffe9,
536 0xbdbdf21c, 0xcabac28a, 0x53b39330, 0x24b4a3a6,
537 0xbad03605, 0xcdd70693, 0x54de5729, 0x23d967bf,
538 0xb3667a2e, 0xc4614ab8, 0x5d681b02, 0x2a6f2b94,
539 0xb40bbe37, 0xc30c8ea1, 0x5a05df1b, 0x2d02ef8d,
542 APU_DECLARE(apr_uint32_t) apr_memcache_hash_crc32(void *baton,
544 const apr_size_t data_len)
550 for (i = 0; i < data_len; i++)
551 crc = (crc >> 8) ^ crc32tab[(crc ^ (data[i])) & 0xff];
556 APU_DECLARE(apr_uint32_t) apr_memcache_hash_default(void *baton,
558 const apr_size_t data_len)
560 /* The default Perl Client doesn't actually use just crc32 -- it shifts it again
563 return ((apr_memcache_hash_crc32(baton, data, data_len) >> 16) & 0x7fff);
566 APU_DECLARE(apr_uint32_t) apr_memcache_hash(apr_memcache_t *mc,
568 const apr_size_t data_len)
571 return mc->hash_func(mc->hash_baton, data, data_len);
574 return apr_memcache_hash_default(NULL, data, data_len);
578 static apr_status_t get_server_line(apr_memcache_conn_t *conn)
580 apr_size_t bsize = BUFFER_SIZE;
581 apr_status_t rv = APR_SUCCESS;
583 rv = apr_brigade_split_line(conn->tb, conn->bb, APR_BLOCK_READ, BUFFER_SIZE);
585 if (rv != APR_SUCCESS) {
589 rv = apr_brigade_flatten(conn->tb, conn->buffer, &bsize);
591 if (rv != APR_SUCCESS) {
596 conn->buffer[bsize] = '\0';
598 return apr_brigade_cleanup(conn->tb);
601 static apr_status_t storage_cmd_write(apr_memcache_t *mc,
603 const apr_size_t cmd_size,
606 const apr_size_t data_size,
607 apr_uint32_t timeout,
611 apr_memcache_server_t *ms;
612 apr_memcache_conn_t *conn;
618 apr_size_t key_size = strlen(key);
620 hash = apr_memcache_hash(mc, key, key_size);
622 ms = apr_memcache_find_server_hash(mc, hash);
627 rv = ms_find_conn(ms, &conn);
629 if (rv != APR_SUCCESS) {
630 apr_memcache_disable_server(mc, ms);
634 /* <command name> <key> <flags> <exptime> <bytes>\r\n<data>\r\n */
636 vec[0].iov_base = cmd;
637 vec[0].iov_len = cmd_size;
639 vec[1].iov_base = (void*)key;
640 vec[1].iov_len = key_size;
642 klen = apr_snprintf(conn->buffer, BUFFER_SIZE, " %u %u %" APR_SIZE_T_FMT " " MC_EOL,
643 flags, timeout, data_size);
645 vec[2].iov_base = conn->buffer;
646 vec[2].iov_len = klen;
648 vec[3].iov_base = data;
649 vec[3].iov_len = data_size;
651 vec[4].iov_base = MC_EOL;
652 vec[4].iov_len = MC_EOL_LEN;
654 rv = apr_socket_sendv(conn->sock, vec, 5, &written);
656 if (rv != APR_SUCCESS) {
657 ms_bad_conn(ms, conn);
658 apr_memcache_disable_server(mc, ms);
662 rv = get_server_line(conn);
664 if (rv != APR_SUCCESS) {
665 ms_bad_conn(ms, conn);
666 apr_memcache_disable_server(mc, ms);
670 if (strcmp(conn->buffer, MS_STORED MC_EOL) == 0) {
673 else if (strcmp(conn->buffer, MS_NOT_STORED MC_EOL) == 0) {
680 ms_release_conn(ms, conn);
685 APU_DECLARE(apr_status_t)
686 apr_memcache_set(apr_memcache_t *mc,
689 const apr_size_t data_size,
690 apr_uint32_t timeout,
693 return storage_cmd_write(mc,
700 APU_DECLARE(apr_status_t)
701 apr_memcache_add(apr_memcache_t *mc,
704 const apr_size_t data_size,
705 apr_uint32_t timeout,
708 return storage_cmd_write(mc,
715 APU_DECLARE(apr_status_t)
716 apr_memcache_replace(apr_memcache_t *mc,
719 const apr_size_t data_size,
720 apr_uint32_t timeout,
723 return storage_cmd_write(mc,
724 MC_REPLACE, MC_REPLACE_LEN,
731 APU_DECLARE(apr_status_t)
732 apr_memcache_getp(apr_memcache_t *mc,
736 apr_size_t *new_length,
737 apr_uint16_t *flags_)
740 apr_memcache_server_t *ms;
741 apr_memcache_conn_t *conn;
744 apr_size_t klen = strlen(key);
747 hash = apr_memcache_hash(mc, key, klen);
748 ms = apr_memcache_find_server_hash(mc, hash);
752 rv = ms_find_conn(ms, &conn);
754 if (rv != APR_SUCCESS) {
755 apr_memcache_disable_server(mc, ms);
759 /* get <key>[ <key>[...]]\r\n */
760 vec[0].iov_base = MC_GET;
761 vec[0].iov_len = MC_GET_LEN;
763 vec[1].iov_base = (void*)key;
764 vec[1].iov_len = klen;
766 vec[2].iov_base = MC_EOL;
767 vec[2].iov_len = MC_EOL_LEN;
769 rv = apr_socket_sendv(conn->sock, vec, 3, &written);
771 if (rv != APR_SUCCESS) {
772 ms_bad_conn(ms, conn);
773 apr_memcache_disable_server(mc, ms);
777 rv = get_server_line(conn);
778 if (rv != APR_SUCCESS) {
779 ms_bad_conn(ms, conn);
780 apr_memcache_disable_server(mc, ms);
784 if (strncmp(MS_VALUE, conn->buffer, MS_VALUE_LEN) == 0) {
790 flags = apr_strtok(conn->buffer, " ", &last);
791 flags = apr_strtok(NULL, " ", &last);
792 flags = apr_strtok(NULL, " ", &last);
795 *flags_ = atoi(flags);
798 length = apr_strtok(NULL, " ", &last);
800 len = strtol(length, (char **)NULL, 10);
808 apr_bucket_brigade *bbb;
811 /* eat the trailing \r\n */
812 rv = apr_brigade_partition(conn->bb, len+2, &e);
814 if (rv != APR_SUCCESS) {
815 ms_bad_conn(ms, conn);
816 apr_memcache_disable_server(mc, ms);
820 bbb = apr_brigade_split(conn->bb, e);
822 rv = apr_brigade_pflatten(conn->bb, baton, &len, p);
824 if (rv != APR_SUCCESS) {
825 ms_bad_conn(ms, conn);
826 apr_memcache_disable_server(mc, ms);
830 rv = apr_brigade_destroy(conn->bb);
831 if (rv != APR_SUCCESS) {
832 ms_bad_conn(ms, conn);
833 apr_memcache_disable_server(mc, ms);
839 *new_length = len - 2;
840 (*baton)[*new_length] = '\0';
843 rv = get_server_line(conn);
844 if (rv != APR_SUCCESS) {
845 ms_bad_conn(ms, conn);
846 apr_memcache_disable_server(mc, ms);
850 if (strncmp(MS_END, conn->buffer, MS_END_LEN) != 0) {
854 else if (strncmp(MS_END, conn->buffer, MS_END_LEN) == 0) {
861 ms_release_conn(ms, conn);
866 APU_DECLARE(apr_status_t)
867 apr_memcache_delete(apr_memcache_t *mc,
869 apr_uint32_t timeout)
872 apr_memcache_server_t *ms;
873 apr_memcache_conn_t *conn;
877 apr_size_t klen = strlen(key);
879 hash = apr_memcache_hash(mc, key, klen);
880 ms = apr_memcache_find_server_hash(mc, hash);
884 rv = ms_find_conn(ms, &conn);
886 if (rv != APR_SUCCESS) {
887 apr_memcache_disable_server(mc, ms);
891 /* delete <key> <time>\r\n */
892 vec[0].iov_base = MC_DELETE;
893 vec[0].iov_len = MC_DELETE_LEN;
895 vec[1].iov_base = (void*)key;
896 vec[1].iov_len = klen;
898 klen = apr_snprintf(conn->buffer, BUFFER_SIZE, " %u" MC_EOL, timeout);
900 vec[2].iov_base = conn->buffer;
901 vec[2].iov_len = klen;
903 rv = apr_socket_sendv(conn->sock, vec, 3, &written);
905 if (rv != APR_SUCCESS) {
906 ms_bad_conn(ms, conn);
907 apr_memcache_disable_server(mc, ms);
911 rv = get_server_line(conn);
912 if (rv != APR_SUCCESS) {
913 ms_bad_conn(ms, conn);
914 apr_memcache_disable_server(mc, ms);
918 if (strncmp(MS_DELETED, conn->buffer, MS_DELETED_LEN) == 0) {
921 else if (strncmp(MS_NOT_FOUND, conn->buffer, MS_NOT_FOUND_LEN) == 0) {
928 ms_release_conn(ms, conn);
933 static apr_status_t num_cmd_write(apr_memcache_t *mc,
935 const apr_uint32_t cmd_size,
937 const apr_int32_t inc,
938 apr_uint32_t *new_value)
941 apr_memcache_server_t *ms;
942 apr_memcache_conn_t *conn;
946 apr_size_t klen = strlen(key);
948 hash = apr_memcache_hash(mc, key, klen);
949 ms = apr_memcache_find_server_hash(mc, hash);
953 rv = ms_find_conn(ms, &conn);
955 if (rv != APR_SUCCESS) {
956 apr_memcache_disable_server(mc, ms);
960 /* <cmd> <key> <value>\r\n */
961 vec[0].iov_base = cmd;
962 vec[0].iov_len = cmd_size;
964 vec[1].iov_base = (void*)key;
965 vec[1].iov_len = klen;
967 klen = apr_snprintf(conn->buffer, BUFFER_SIZE, " %u" MC_EOL, inc);
969 vec[2].iov_base = conn->buffer;
970 vec[2].iov_len = klen;
972 rv = apr_socket_sendv(conn->sock, vec, 3, &written);
974 if (rv != APR_SUCCESS) {
975 ms_bad_conn(ms, conn);
976 apr_memcache_disable_server(mc, ms);
980 rv = get_server_line(conn);
981 if (rv != APR_SUCCESS) {
982 ms_bad_conn(ms, conn);
983 apr_memcache_disable_server(mc, ms);
987 if (strncmp(MS_ERROR, conn->buffer, MS_ERROR_LEN) == 0) {
990 else if (strncmp(MS_NOT_FOUND, conn->buffer, MS_NOT_FOUND_LEN) == 0) {
995 *new_value = atoi(conn->buffer);
1000 ms_release_conn(ms, conn);
1005 APU_DECLARE(apr_status_t)
1006 apr_memcache_incr(apr_memcache_t *mc,
1009 apr_uint32_t *new_value)
1011 return num_cmd_write(mc,
1020 APU_DECLARE(apr_status_t)
1021 apr_memcache_decr(apr_memcache_t *mc,
1024 apr_uint32_t *new_value)
1026 return num_cmd_write(mc,
1036 APU_DECLARE(apr_status_t)
1037 apr_memcache_version(apr_memcache_server_t *ms,
1042 apr_memcache_conn_t *conn;
1044 struct iovec vec[2];
1046 rv = ms_find_conn(ms, &conn);
1048 if (rv != APR_SUCCESS) {
1053 vec[0].iov_base = MC_VERSION;
1054 vec[0].iov_len = MC_VERSION_LEN;
1056 vec[1].iov_base = MC_EOL;
1057 vec[1].iov_len = MC_EOL_LEN;
1059 rv = apr_socket_sendv(conn->sock, vec, 2, &written);
1061 if (rv != APR_SUCCESS) {
1062 ms_bad_conn(ms, conn);
1066 rv = get_server_line(conn);
1067 if (rv != APR_SUCCESS) {
1068 ms_bad_conn(ms, conn);
1072 if (strncmp(MS_VERSION, conn->buffer, MS_VERSION_LEN) == 0) {
1073 *baton = apr_pstrmemdup(p, conn->buffer+MS_VERSION_LEN+1,
1074 conn->blen - MS_VERSION_LEN - 2);
1081 ms_release_conn(ms, conn);
1086 apr_status_t mc_version_ping(apr_memcache_server_t *ms)
1090 struct iovec vec[2];
1091 apr_memcache_conn_t *conn;
1093 rv = ms_find_conn(ms, &conn);
1095 if (rv != APR_SUCCESS) {
1100 vec[0].iov_base = MC_VERSION;
1101 vec[0].iov_len = MC_VERSION_LEN;
1103 vec[1].iov_base = MC_EOL;
1104 vec[1].iov_len = MC_EOL_LEN;
1106 rv = apr_socket_sendv(conn->sock, vec, 2, &written);
1108 if (rv != APR_SUCCESS) {
1109 ms_bad_conn(ms, conn);
1113 rv = get_server_line(conn);
1114 ms_release_conn(ms, conn);
1120 apr_memcache_add_multget_key(apr_pool_t *data_pool,
1122 apr_hash_t **values)
1124 apr_memcache_value_t* value;
1125 apr_size_t klen = strlen(key);
1127 /* create the value hash if need be */
1129 *values = apr_hash_make(data_pool);
1132 /* init key and add it to the value hash */
1133 value = apr_pcalloc(data_pool, sizeof(apr_memcache_value_t));
1135 value->status = APR_NOTFOUND;
1136 value->key = apr_pstrdup(data_pool, key);
1138 apr_hash_set(*values, value->key, klen, value);
1141 static void mget_conn_result(int serverup,
1145 apr_memcache_server_t *ms,
1146 apr_memcache_conn_t *conn,
1147 struct cache_server_query_t *server_query,
1149 apr_hash_t *server_queries)
1152 apr_memcache_value_t* value;
1154 apr_hash_set(server_queries, &ms, sizeof(ms), NULL);
1157 ms_release_conn(ms, conn);
1159 ms_bad_conn(ms, conn);
1162 apr_memcache_disable_server(mc, ms);
1166 for (j = 1; j < server_query->query_vec_count ; j+=2) {
1167 if (server_query->query_vec[j].iov_base) {
1168 value = apr_hash_get(values, server_query->query_vec[j].iov_base,
1169 strlen(server_query->query_vec[j].iov_base));
1171 if (value->status == APR_NOTFOUND) {
1178 APU_DECLARE(apr_status_t)
1179 apr_memcache_multgetp(apr_memcache_t *mc,
1180 apr_pool_t *temp_pool,
1181 apr_pool_t *data_pool,
1185 apr_memcache_server_t* ms;
1186 apr_memcache_conn_t* conn;
1191 apr_memcache_value_t* value;
1192 apr_hash_index_t* value_hash_index;
1194 /* this is a little over aggresive, but beats multiple loops
1195 * to figure out how long each vector needs to be per-server.
1197 apr_int32_t veclen = 2 + 2 * apr_hash_count(values) - 1; /* get <key>[<space><key>...]\r\n */
1199 apr_int32_t queries_sent;
1200 apr_int32_t queries_recvd;
1202 apr_hash_t * server_queries = apr_hash_make(temp_pool);
1203 struct cache_server_query_t* server_query;
1204 apr_hash_index_t * query_hash_index;
1206 apr_pollset_t* pollset;
1207 const apr_pollfd_t* activefds;
1208 apr_pollfd_t* pollfds;
1211 /* build all the queries */
1212 value_hash_index = apr_hash_first(temp_pool, values);
1213 while (value_hash_index) {
1215 apr_hash_this(value_hash_index, NULL, NULL, &v);
1217 value_hash_index = apr_hash_next(value_hash_index);
1218 klen = strlen(value->key);
1220 hash = apr_memcache_hash(mc, value->key, klen);
1221 ms = apr_memcache_find_server_hash(mc, hash);
1226 server_query = apr_hash_get(server_queries, &ms, sizeof(ms));
1228 if (!server_query) {
1229 rv = ms_find_conn(ms, &conn);
1231 if (rv != APR_SUCCESS) {
1232 apr_memcache_disable_server(mc, ms);
1237 server_query = apr_pcalloc(temp_pool,sizeof(struct cache_server_query_t));
1239 apr_hash_set(server_queries, &ms, sizeof(ms), server_query);
1241 server_query->ms = ms;
1242 server_query->conn = conn;
1243 server_query->query_vec = apr_pcalloc(temp_pool, sizeof(struct iovec)*veclen);
1245 /* set up the first key */
1246 server_query->query_vec[0].iov_base = MC_GET;
1247 server_query->query_vec[0].iov_len = MC_GET_LEN;
1249 server_query->query_vec[1].iov_base = (void*)(value->key);
1250 server_query->query_vec[1].iov_len = klen;
1252 server_query->query_vec[2].iov_base = MC_EOL;
1253 server_query->query_vec[2].iov_len = MC_EOL_LEN;
1255 server_query->query_vec_count = 3;
1258 j = server_query->query_vec_count - 1;
1260 server_query->query_vec[j].iov_base = MC_WS;
1261 server_query->query_vec[j].iov_len = MC_WS_LEN;
1264 server_query->query_vec[j].iov_base = (void*)(value->key);
1265 server_query->query_vec[j].iov_len = klen;
1268 server_query->query_vec[j].iov_base = MC_EOL;
1269 server_query->query_vec[j].iov_len = MC_EOL_LEN;
1272 server_query->query_vec_count = j;
1276 /* create polling structures */
1277 pollfds = apr_pcalloc(temp_pool, apr_hash_count(server_queries) * sizeof(apr_pollfd_t));
1279 rv = apr_pollset_create(&pollset, apr_hash_count(server_queries), temp_pool, 0);
1281 if (rv != APR_SUCCESS) {
1282 query_hash_index = apr_hash_first(temp_pool, server_queries);
1284 while (query_hash_index) {
1286 apr_hash_this(query_hash_index, NULL, NULL, &v);
1288 query_hash_index = apr_hash_next(query_hash_index);
1290 mget_conn_result(TRUE, TRUE, rv, mc, server_query->ms, server_query->conn,
1291 server_query, values, server_queries);
1297 /* send all the queries */
1299 query_hash_index = apr_hash_first(temp_pool, server_queries);
1301 while (query_hash_index) {
1303 apr_hash_this(query_hash_index, NULL, NULL, &v);
1305 query_hash_index = apr_hash_next(query_hash_index);
1307 conn = server_query->conn;
1308 ms = server_query->ms;
1310 for (i = 0, rv = APR_SUCCESS; i < veclen && rv == APR_SUCCESS; i += APR_MAX_IOVEC_SIZE) {
1311 rv = apr_socket_sendv(conn->sock, &(server_query->query_vec[i]),
1312 veclen-i>APR_MAX_IOVEC_SIZE ? APR_MAX_IOVEC_SIZE : veclen-i , &written);
1315 if (rv != APR_SUCCESS) {
1316 mget_conn_result(FALSE, FALSE, rv, mc, ms, conn,
1317 server_query, values, server_queries);
1321 pollfds[queries_sent].desc_type = APR_POLL_SOCKET;
1322 pollfds[queries_sent].reqevents = APR_POLLIN;
1323 pollfds[queries_sent].p = temp_pool;
1324 pollfds[queries_sent].desc.s = conn->sock;
1325 pollfds[queries_sent].client_data = (void *)server_query;
1326 apr_pollset_add (pollset, &pollfds[queries_sent]);
1331 while (queries_sent) {
1332 rv = apr_pollset_poll(pollset, MULT_GET_TIMEOUT, &queries_recvd, &activefds);
1334 if (rv != APR_SUCCESS) {
1339 for (i = 0; i < queries_recvd; i++) {
1340 server_query = activefds[i].client_data;
1341 conn = server_query->conn;
1342 ms = server_query->ms;
1344 rv = get_server_line(conn);
1346 if (rv != APR_SUCCESS) {
1347 apr_pollset_remove (pollset, &activefds[i]);
1348 mget_conn_result(FALSE, FALSE, rv, mc, ms, conn,
1349 server_query, values, server_queries);
1354 if (strncmp(MS_VALUE, conn->buffer, MS_VALUE_LEN) == 0) {
1362 key = apr_strtok(conn->buffer, " ", &last); /* just the VALUE, ignore */
1363 key = apr_strtok(NULL, " ", &last);
1364 flags = apr_strtok(NULL, " ", &last);
1367 length = apr_strtok(NULL, " ", &last);
1369 len = strtol(length, (char **) NULL, 10);
1372 value = apr_hash_get(values, key, strlen(key));
1377 apr_bucket_brigade *bbb;
1380 /* eat the trailing \r\n */
1381 rv = apr_brigade_partition(conn->bb, len+2, &e);
1383 if (rv != APR_SUCCESS) {
1384 apr_pollset_remove (pollset, &activefds[i]);
1385 mget_conn_result(FALSE, FALSE, rv, mc, ms, conn,
1386 server_query, values, server_queries);
1391 bbb = apr_brigade_split(conn->bb, e);
1393 rv = apr_brigade_pflatten(conn->bb, &data, &len, data_pool);
1395 if (rv != APR_SUCCESS) {
1396 apr_pollset_remove (pollset, &activefds[i]);
1397 mget_conn_result(FALSE, FALSE, rv, mc, ms, conn,
1398 server_query, values, server_queries);
1403 rv = apr_brigade_destroy(conn->bb);
1404 if (rv != APR_SUCCESS) {
1405 apr_pollset_remove (pollset, &activefds[i]);
1406 mget_conn_result(FALSE, FALSE, rv, mc, ms, conn,
1407 server_query, values, server_queries);
1414 value->len = len - 2;
1415 data[value->len] = '\0';
1420 value->flags = atoi(flags);
1422 /* stay on the server */
1427 /* TODO: Server Sent back a key I didn't ask for or my
1428 * hash is corrupt */
1431 else if (strncmp(MS_END, conn->buffer, MS_END_LEN) == 0) {
1432 /* this connection is done */
1433 apr_pollset_remove (pollset, &activefds[i]);
1434 ms_release_conn(ms, conn);
1435 apr_hash_set(server_queries, &ms, sizeof(ms), NULL);
1440 /* unknown reply? */
1447 query_hash_index = apr_hash_first(temp_pool, server_queries);
1448 while (query_hash_index) {
1450 apr_hash_this(query_hash_index, NULL, NULL, &v);
1452 query_hash_index = apr_hash_next(query_hash_index);
1454 conn = server_query->conn;
1455 ms = server_query->ms;
1457 mget_conn_result(TRUE, (rv == APR_SUCCESS), rv, mc, ms, conn,
1458 server_query, values, server_queries);
1462 apr_pollset_destroy(pollset);
1463 apr_pool_clear(temp_pool);
1471 * Define all of the strings for stats
1474 #define STAT_pid MS_STAT " pid "
1475 #define STAT_pid_LEN (sizeof(STAT_pid)-1)
1477 #define STAT_uptime MS_STAT " uptime "
1478 #define STAT_uptime_LEN (sizeof(STAT_uptime)-1)
1480 #define STAT_time MS_STAT " time "
1481 #define STAT_time_LEN (sizeof(STAT_time)-1)
1483 #define STAT_version MS_STAT " version "
1484 #define STAT_version_LEN (sizeof(STAT_version)-1)
1486 #define STAT_pointer_size MS_STAT " pointer_size "
1487 #define STAT_pointer_size_LEN (sizeof(STAT_pointer_size)-1)
1489 #define STAT_rusage_user MS_STAT " rusage_user "
1490 #define STAT_rusage_user_LEN (sizeof(STAT_rusage_user)-1)
1492 #define STAT_rusage_system MS_STAT " rusage_system "
1493 #define STAT_rusage_system_LEN (sizeof(STAT_rusage_system)-1)
1495 #define STAT_curr_items MS_STAT " curr_items "
1496 #define STAT_curr_items_LEN (sizeof(STAT_curr_items)-1)
1498 #define STAT_total_items MS_STAT " total_items "
1499 #define STAT_total_items_LEN (sizeof(STAT_total_items)-1)
1501 #define STAT_bytes MS_STAT " bytes "
1502 #define STAT_bytes_LEN (sizeof(STAT_bytes)-1)
1504 #define STAT_curr_connections MS_STAT " curr_connections "
1505 #define STAT_curr_connections_LEN (sizeof(STAT_curr_connections)-1)
1507 #define STAT_total_connections MS_STAT " total_connections "
1508 #define STAT_total_connections_LEN (sizeof(STAT_total_connections)-1)
1510 #define STAT_connection_structures MS_STAT " connection_structures "
1511 #define STAT_connection_structures_LEN (sizeof(STAT_connection_structures)-1)
1513 #define STAT_cmd_get MS_STAT " cmd_get "
1514 #define STAT_cmd_get_LEN (sizeof(STAT_cmd_get)-1)
1516 #define STAT_cmd_set MS_STAT " cmd_set "
1517 #define STAT_cmd_set_LEN (sizeof(STAT_cmd_set)-1)
1519 #define STAT_get_hits MS_STAT " get_hits "
1520 #define STAT_get_hits_LEN (sizeof(STAT_get_hits)-1)
1522 #define STAT_get_misses MS_STAT " get_misses "
1523 #define STAT_get_misses_LEN (sizeof(STAT_get_misses)-1)
1525 #define STAT_evictions MS_STAT " evictions "
1526 #define STAT_evictions_LEN (sizeof(STAT_evictions)-1)
1528 #define STAT_bytes_read MS_STAT " bytes_read "
1529 #define STAT_bytes_read_LEN (sizeof(STAT_bytes_read)-1)
1531 #define STAT_bytes_written MS_STAT " bytes_written "
1532 #define STAT_bytes_written_LEN (sizeof(STAT_bytes_written)-1)
1534 #define STAT_limit_maxbytes MS_STAT " limit_maxbytes "
1535 #define STAT_limit_maxbytes_LEN (sizeof(STAT_limit_maxbytes)-1)
1537 #define STAT_threads MS_STAT " threads "
1538 #define STAT_threads_LEN (sizeof(STAT_threads)-1)
1540 static const char *stat_read_string(apr_pool_t *p, char *buf, apr_size_t len)
1542 /* remove trailing \r\n and null char */
1543 return apr_pstrmemdup(p, buf, len-2);
1546 static apr_uint32_t stat_read_uint32(apr_pool_t *p, char *buf, apr_size_t len)
1552 static apr_uint64_t stat_read_uint64(apr_pool_t *p, char *buf, apr_size_t len)
1555 return apr_atoi64(buf);
1558 static apr_time_t stat_read_time(apr_pool_t *p, char *buf, apr_size_t len)
1561 return apr_time_from_sec(atoi(buf));
1564 static apr_time_t stat_read_rtime(apr_pool_t *p, char *buf, apr_size_t len)
1569 const char *sep = ":.";
1573 secs = apr_strtok(buf, sep, &tok);
1574 usecs = apr_strtok(NULL, sep, &tok);
1575 if (secs && usecs) {
1576 return apr_time_make(atoi(secs), atoi(usecs));
1579 return apr_time_make(0, 0);
1584 * I got tired of Typing. Meh.
1586 * TODO: Convert it to static tables to make it cooler.
1589 #define mc_stat_cmp(name) \
1590 strncmp(STAT_ ## name, conn->buffer, STAT_ ## name ## _LEN) == 0
1592 #define mc_stat_str(name) \
1593 stat_read_string(p, conn->buffer + name, \
1596 #define mc_stat_uint32(name) \
1597 stat_read_uint32(p, conn->buffer + name, \
1600 #define mc_stat_uint64(name) \
1601 stat_read_uint64(p, conn->buffer + name, \
1604 #define mc_stat_time(name) \
1605 stat_read_time(p, conn->buffer + name, \
1608 #define mc_stat_rtime(name) \
1609 stat_read_rtime(p, conn->buffer + name, \
1613 #define mc_do_stat(name, type) \
1614 if (mc_stat_cmp(name)) { \
1615 stats-> name = mc_stat_ ## type ((STAT_ ## name ## _LEN)); \
1618 static void update_stats(apr_pool_t *p, apr_memcache_conn_t *conn,
1619 apr_memcache_stats_t *stats)
1622 mc_do_stat(version, str)
1623 else mc_do_stat(pid, uint32)
1624 else mc_do_stat(uptime, uint32)
1625 else mc_do_stat(pointer_size, uint32)
1626 else mc_do_stat(time, time)
1627 else mc_do_stat(rusage_user, rtime)
1628 else mc_do_stat(rusage_system, rtime)
1629 else mc_do_stat(curr_items, uint32)
1630 else mc_do_stat(total_items, uint32)
1631 else mc_do_stat(bytes, uint64)
1632 else mc_do_stat(curr_connections, uint32)
1633 else mc_do_stat(total_connections, uint32)
1634 else mc_do_stat(connection_structures, uint32)
1635 else mc_do_stat(cmd_get, uint32)
1636 else mc_do_stat(cmd_set, uint32)
1637 else mc_do_stat(get_hits, uint32)
1638 else mc_do_stat(get_misses, uint32)
1639 else mc_do_stat(evictions, uint64)
1640 else mc_do_stat(bytes_read, uint64)
1641 else mc_do_stat(bytes_written, uint64)
1642 else mc_do_stat(limit_maxbytes, uint32)
1643 else mc_do_stat(threads, uint32)
1646 APU_DECLARE(apr_status_t)
1647 apr_memcache_stats(apr_memcache_server_t *ms,
1649 apr_memcache_stats_t **stats)
1651 apr_memcache_stats_t *ret;
1653 apr_memcache_conn_t *conn;
1655 struct iovec vec[2];
1657 rv = ms_find_conn(ms, &conn);
1659 if (rv != APR_SUCCESS) {
1664 vec[0].iov_base = MC_STATS;
1665 vec[0].iov_len = MC_STATS_LEN;
1667 vec[1].iov_base = MC_EOL;
1668 vec[1].iov_len = MC_EOL_LEN;
1670 rv = apr_socket_sendv(conn->sock, vec, 2, &written);
1672 if (rv != APR_SUCCESS) {
1673 ms_bad_conn(ms, conn);
1677 ret = apr_pcalloc(p, sizeof(apr_memcache_stats_t));
1680 rv = get_server_line(conn);
1681 if (rv != APR_SUCCESS) {
1682 ms_bad_conn(ms, conn);
1686 if (strncmp(MS_END, conn->buffer, MS_END_LEN) == 0) {
1690 else if (strncmp(MS_STAT, conn->buffer, MS_STAT_LEN) == 0) {
1691 update_stats(p, conn, ret);
1701 ms_release_conn(ms, conn);