contrib/apr-util/memcache/apr_memcache.c

   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
   7  *
   8  *     http://www.apache.org/licenses/LICENSE-2.0
   9  *
  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.
  15  */
  16
  17 #include "apr_memcache.h"
  18 #include "apr_poll.h"
  19 #include "apr_version.h"
  20 #include <stdlib.h>
  21
  22 #define BUFFER_SIZE 512
  23 struct apr_memcache_conn_t
  24 {
  25     char *buffer;
  26     apr_size_t blen;
  27     apr_pool_t *p;
  28     apr_pool_t *tp;
  29     apr_socket_t *sock;
  30     apr_bucket_brigade *bb;
  31     apr_bucket_brigade *tb;
  32     apr_memcache_server_t *ms;
  33 };
  34
  35 /* Strings for Client Commands */
  36
  37 #define MC_EOL "\r\n"
  38 #define MC_EOL_LEN (sizeof(MC_EOL)-1)
  39
  40 #define MC_WS " "
  41 #define MC_WS_LEN (sizeof(MC_WS)-1)
  42
  43 #define MC_GET "get "
  44 #define MC_GET_LEN (sizeof(MC_GET)-1)
  45
  46 #define MC_SET "set "
  47 #define MC_SET_LEN (sizeof(MC_SET)-1)
  48
  49 #define MC_ADD "add "
  50 #define MC_ADD_LEN (sizeof(MC_ADD)-1)
  51
  52 #define MC_REPLACE "replace "
  53 #define MC_REPLACE_LEN (sizeof(MC_REPLACE)-1)
  54
  55 #define MC_DELETE "delete "
  56 #define MC_DELETE_LEN (sizeof(MC_DELETE)-1)
  57
  58 #define MC_INCR "incr "
  59 #define MC_INCR_LEN (sizeof(MC_INCR)-1)
  60
  61 #define MC_DECR "decr "
  62 #define MC_DECR_LEN (sizeof(MC_DECR)-1)
  63
  64 #define MC_VERSION "version"
  65 #define MC_VERSION_LEN (sizeof(MC_VERSION)-1)
  66
  67 #define MC_STATS "stats"
  68 #define MC_STATS_LEN (sizeof(MC_STATS)-1)
  69
  70 #define MC_QUIT "quit"
  71 #define MC_QUIT_LEN (sizeof(MC_QUIT)-1)
  72
  73 /* Strings for Server Replies */
  74
  75 #define MS_STORED "STORED"
  76 #define MS_STORED_LEN (sizeof(MS_STORED)-1)
  77
  78 #define MS_NOT_STORED "NOT_STORED"
  79 #define MS_NOT_STORED_LEN (sizeof(MS_NOT_STORED)-1)
  80
  81 #define MS_DELETED "DELETED"
  82 #define MS_DELETED_LEN (sizeof(MS_DELETED)-1)
  83
  84 #define MS_NOT_FOUND "NOT_FOUND"
  85 #define MS_NOT_FOUND_LEN (sizeof(MS_NOT_FOUND)-1)
  86
  87 #define MS_VALUE "VALUE"
  88 #define MS_VALUE_LEN (sizeof(MS_VALUE)-1)
  89
  90 #define MS_ERROR "ERROR"
  91 #define MS_ERROR_LEN (sizeof(MS_ERROR)-1)
  92
  93 #define MS_VERSION "VERSION"
  94 #define MS_VERSION_LEN (sizeof(MS_VERSION)-1)
  95
  96 #define MS_STAT "STAT"
  97 #define MS_STAT_LEN (sizeof(MS_STAT)-1)
  98
  99 #define MS_END "END"
 100 #define MS_END_LEN (sizeof(MS_END)-1)
 101
 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;
 108 };
 109
 110 #define MULT_GET_TIMEOUT 50000
 111
 112 static apr_status_t make_server_dead(apr_memcache_t *mc, apr_memcache_server_t *ms)
 113 {
 114 #if APR_HAS_THREADS
 115     apr_thread_mutex_lock(ms->lock);
 116 #endif
 117     ms->status = APR_MC_SERVER_DEAD;
 118     ms->btime = apr_time_now();
 119 #if APR_HAS_THREADS
 120     apr_thread_mutex_unlock(ms->lock);
 121 #endif
 122     return APR_SUCCESS;
 123 }
 124
 125 static apr_status_t make_server_live(apr_memcache_t *mc, apr_memcache_server_t *ms)
 126 {
 127     ms->status = APR_MC_SERVER_LIVE;
 128     return APR_SUCCESS;
 129 }
 130
 131
 132 APU_DECLARE(apr_status_t) apr_memcache_add_server(apr_memcache_t *mc, apr_memcache_server_t *ms)
 133 {
 134     apr_status_t rv = APR_SUCCESS;
 135
 136     if(mc->ntotal >= mc->nalloc) {
 137         return APR_ENOMEM;
 138     }
 139
 140     mc->live_servers[mc->ntotal] = ms;
 141     mc->ntotal++;
 142     make_server_live(mc, ms);
 143     return rv;
 144 }
 145
 146 static apr_status_t mc_version_ping(apr_memcache_server_t *ms);
 147
 148 APU_DECLARE(apr_memcache_server_t *)
 149 apr_memcache_find_server_hash(apr_memcache_t *mc, const apr_uint32_t hash)
 150 {
 151     if (mc->server_func) {
 152         return mc->server_func(mc->server_baton, mc, hash);
 153     }
 154     else {
 155         return apr_memcache_find_server_hash_default(NULL, mc, hash);
 156     }
 157 }
 158
 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)
 162 {
 163     apr_memcache_server_t *ms = NULL;
 164     apr_uint32_t h = hash ? hash : 1;
 165     apr_uint32_t i = 0;
 166     apr_time_t curtime = 0;
 167
 168     if(mc->ntotal == 0) {
 169         return NULL;
 170     }
 171
 172     do {
 173         ms = mc->live_servers[h % mc->ntotal];
 174         if(ms->status == APR_MC_SERVER_LIVE) {
 175             break;
 176         }
 177         else {
 178             if (curtime == 0) {
 179                 curtime = apr_time_now();
 180             }
 181 #if APR_HAS_THREADS
 182             apr_thread_mutex_lock(ms->lock);
 183 #endif
 184             /* Try the dead server, every 5 seconds */
 185             if (curtime - ms->btime >  apr_time_from_sec(5)) {
 186                 ms->btime = curtime;
 187                 if (mc_version_ping(ms) == APR_SUCCESS) {
 188                     make_server_live(mc, ms);
 189 #if APR_HAS_THREADS
 190                     apr_thread_mutex_unlock(ms->lock);
 191 #endif
 192                     break;
 193                 }
 194             }
 195 #if APR_HAS_THREADS
 196             apr_thread_mutex_unlock(ms->lock);
 197 #endif
 198         }
 199         h++;
 200         i++;
 201     } while(i < mc->ntotal);
 202
 203     if (i == mc->ntotal) {
 204         ms = NULL;
 205     }
 206
 207     return ms;
 208 }
 209
 210 APU_DECLARE(apr_memcache_server_t *) apr_memcache_find_server(apr_memcache_t *mc, const char *host, apr_port_t port)
 211 {
 212     int i;
 213
 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) {
 217
 218             return mc->live_servers[i];
 219         }
 220     }
 221
 222     return NULL;
 223 }
 224
 225 static apr_status_t ms_find_conn(apr_memcache_server_t *ms, apr_memcache_conn_t **conn)
 226 {
 227     apr_status_t rv;
 228     apr_bucket_alloc_t *balloc;
 229     apr_bucket *e;
 230
 231 #if APR_HAS_THREADS
 232     rv = apr_reslist_acquire(ms->conns, (void **)conn);
 233 #else
 234     *conn = ms->conn;
 235     rv = APR_SUCCESS;
 236 #endif
 237
 238     if (rv != APR_SUCCESS) {
 239         return rv;
 240     }
 241
 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);
 245
 246     e = apr_bucket_socket_create((*conn)->sock, balloc);
 247     APR_BRIGADE_INSERT_TAIL((*conn)->bb, e);
 248
 249     return rv;
 250 }
 251
 252 static apr_status_t ms_bad_conn(apr_memcache_server_t *ms, apr_memcache_conn_t *conn)
 253 {
 254 #if APR_HAS_THREADS
 255     return apr_reslist_invalidate(ms->conns, conn);
 256 #else
 257     return APR_SUCCESS;
 258 #endif
 259 }
 260
 261 static apr_status_t ms_release_conn(apr_memcache_server_t *ms, apr_memcache_conn_t *conn)
 262 {
 263     apr_pool_clear(conn->tp);
 264 #if APR_HAS_THREADS
 265     return apr_reslist_release(ms->conns, conn);
 266 #else
 267     return APR_SUCCESS;
 268 #endif
 269 }
 270
 271 APU_DECLARE(apr_status_t) apr_memcache_enable_server(apr_memcache_t *mc, apr_memcache_server_t *ms)
 272 {
 273     apr_status_t rv = APR_SUCCESS;
 274
 275     if (ms->status == APR_MC_SERVER_LIVE) {
 276         return rv;
 277     }
 278
 279     rv = make_server_live(mc, ms);
 280     return rv;
 281 }
 282
 283 APU_DECLARE(apr_status_t) apr_memcache_disable_server(apr_memcache_t *mc, apr_memcache_server_t *ms)
 284 {
 285     return make_server_dead(mc, ms);
 286 }
 287
 288 static apr_status_t conn_connect(apr_memcache_conn_t *conn)
 289 {
 290     apr_status_t rv = APR_SUCCESS;
 291     apr_sockaddr_t *sa;
 292 #if APR_HAVE_SOCKADDR_UN
 293     apr_int32_t family = conn->ms->host[0] != '/' ? APR_INET : APR_UNIX;
 294 #else
 295     apr_int32_t family = APR_INET;
 296 #endif
 297
 298     rv = apr_sockaddr_info_get(&sa, conn->ms->host, family, conn->ms->port, 0, conn->p);
 299     if (rv != APR_SUCCESS) {
 300         return rv;
 301     }
 302
 303     rv = apr_socket_timeout_set(conn->sock, 1 * APR_USEC_PER_SEC);
 304     if (rv != APR_SUCCESS) {
 305         return rv;
 306     }
 307
 308     rv = apr_socket_connect(conn->sock, sa);
 309     if (rv != APR_SUCCESS) {
 310         return rv;
 311     }
 312
 313     rv = apr_socket_timeout_set(conn->sock, -1);
 314     if (rv != APR_SUCCESS) {
 315         return rv;
 316     }
 317
 318     return rv;
 319 }
 320
 321
 322 static apr_status_t
 323 mc_conn_construct(void **conn_, void *params, apr_pool_t *pool)
 324 {
 325     apr_status_t rv = APR_SUCCESS;
 326     apr_memcache_conn_t *conn;
 327     apr_pool_t *np;
 328     apr_pool_t *tp;
 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;
 332 #else
 333     apr_int32_t family = APR_INET;
 334 #endif
 335
 336     rv = apr_pool_create(&np, pool);
 337     if (rv != APR_SUCCESS) {
 338         return rv;
 339     }
 340
 341     rv = apr_pool_create(&tp, np);
 342     if (rv != APR_SUCCESS) {
 343         apr_pool_destroy(np);
 344         return rv;
 345     }
 346
 347     conn = apr_palloc(np, sizeof( apr_memcache_conn_t ));
 348
 349     conn->p = np;
 350     conn->tp = tp;
 351
 352     rv = apr_socket_create(&conn->sock, family, SOCK_STREAM, 0, np);
 353
 354     if (rv != APR_SUCCESS) {
 355         apr_pool_destroy(np);
 356         return rv;
 357     }
 358
 359     conn->buffer = apr_palloc(conn->p, BUFFER_SIZE + 1);
 360     conn->blen = 0;
 361     conn->ms = ms;
 362
 363     rv = conn_connect(conn);
 364     if (rv != APR_SUCCESS) {
 365         apr_pool_destroy(np);
 366     }
 367     else {
 368         *conn_ = conn;
 369     }
 370
 371     return rv;
 372 }
 373
 374 #if APR_HAS_THREADS
 375 static apr_status_t
 376 mc_conn_destruct(void *conn_, void *params, apr_pool_t *pool)
 377 {
 378     apr_memcache_conn_t *conn = (apr_memcache_conn_t*)conn_;
 379     struct iovec vec[2];
 380     apr_size_t written;
 381
 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;
 385
 386     vec[1].iov_base = MC_EOL;
 387     vec[1].iov_len = MC_EOL_LEN;
 388
 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);
 392
 393     apr_pool_destroy(conn->p);
 394
 395     return APR_SUCCESS;
 396 }
 397 #endif
 398
 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)
 404 {
 405     apr_status_t rv = APR_SUCCESS;
 406     apr_memcache_server_t *server;
 407     apr_pool_t *np;
 408
 409     rv = apr_pool_create(&np, p);
 410
 411     server = apr_palloc(np, sizeof(apr_memcache_server_t));
 412
 413     server->p = np;
 414     server->host = apr_pstrdup(np, host);
 415     server->port = port;
 416     server->status = APR_MC_SERVER_DEAD;
 417 #if APR_HAS_THREADS
 418     rv = apr_thread_mutex_create(&server->lock, APR_THREAD_MUTEX_DEFAULT, np);
 419     if (rv != APR_SUCCESS) {
 420         return rv;
 421     }
 422
 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 */
 430                                server, np);
 431     if (rv != APR_SUCCESS) {
 432         return rv;
 433     }
 434
 435     apr_reslist_cleanup_order_set(server->conns, APR_RESLIST_CLEANUP_FIRST);
 436 #else
 437     rv = mc_conn_construct((void**)&(server->conn), server, np);
 438     if (rv != APR_SUCCESS) {
 439         return rv;
 440     }
 441 #endif
 442
 443     *ms = server;
 444
 445     return rv;
 446 }
 447
 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)
 451 {
 452     apr_status_t rv = APR_SUCCESS;
 453     apr_memcache_t *mc;
 454
 455     mc = apr_palloc(p, sizeof(apr_memcache_t));
 456     mc->p = p;
 457     mc->nalloc = max_servers;
 458     mc->ntotal = 0;
 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;
 464     *memcache = mc;
 465     return rv;
 466 }
 467
 468
 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.
 473  */
 474
 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,
 540 };
 541
 542 APU_DECLARE(apr_uint32_t) apr_memcache_hash_crc32(void *baton,
 543                                                   const char *data,
 544                                                   const apr_size_t data_len)
 545 {
 546     apr_uint32_t i;
 547     apr_uint32_t crc;
 548     crc = ~0;
 549
 550     for (i = 0; i < data_len; i++)
 551         crc = (crc >> 8) ^ crc32tab[(crc ^ (data[i])) & 0xff];
 552
 553     return ~crc;
 554 }
 555
 556 APU_DECLARE(apr_uint32_t) apr_memcache_hash_default(void *baton,
 557                                                     const char *data,
 558                                                     const apr_size_t data_len)
 559 {
 560     /* The default Perl Client doesn't actually use just crc32 -- it shifts it again
 561      * like this....
 562      */
 563     return ((apr_memcache_hash_crc32(baton, data, data_len) >> 16) & 0x7fff);
 564 }
 565
 566 APU_DECLARE(apr_uint32_t) apr_memcache_hash(apr_memcache_t *mc,
 567                                             const char *data,
 568                                             const apr_size_t data_len)
 569 {
 570     if (mc->hash_func) {
 571         return mc->hash_func(mc->hash_baton, data, data_len);
 572     }
 573     else {
 574         return apr_memcache_hash_default(NULL, data, data_len);
 575     }
 576 }
 577
 578 static apr_status_t get_server_line(apr_memcache_conn_t *conn)
 579 {
 580     apr_size_t bsize = BUFFER_SIZE;
 581     apr_status_t rv = APR_SUCCESS;
 582
 583     rv = apr_brigade_split_line(conn->tb, conn->bb, APR_BLOCK_READ, BUFFER_SIZE);
 584
 585     if (rv != APR_SUCCESS) {
 586         return rv;
 587     }
 588
 589     rv = apr_brigade_flatten(conn->tb, conn->buffer, &bsize);
 590
 591     if (rv != APR_SUCCESS) {
 592         return rv;
 593     }
 594
 595     conn->blen = bsize;
 596     conn->buffer[bsize] = '\0';
 597
 598     return apr_brigade_cleanup(conn->tb);
 599 }
 600
 601 static apr_status_t storage_cmd_write(apr_memcache_t *mc,
 602                                       char *cmd,
 603                                       const apr_size_t cmd_size,
 604                                       const char *key,
 605                                       char *data,
 606                                       const apr_size_t data_size,
 607                                       apr_uint32_t timeout,
 608                                       apr_uint16_t flags)
 609 {
 610     apr_uint32_t hash;
 611     apr_memcache_server_t *ms;
 612     apr_memcache_conn_t *conn;
 613     apr_status_t rv;
 614     apr_size_t written;
 615     struct iovec vec[5];
 616     apr_size_t klen;
 617
 618     apr_size_t key_size = strlen(key);
 619
 620     hash = apr_memcache_hash(mc, key, key_size);
 621
 622     ms = apr_memcache_find_server_hash(mc, hash);
 623
 624     if (ms == NULL)
 625         return APR_NOTFOUND;
 626
 627     rv = ms_find_conn(ms, &conn);
 628
 629     if (rv != APR_SUCCESS) {
 630         apr_memcache_disable_server(mc, ms);
 631         return rv;
 632     }
 633
 634     /* <command name> <key> <flags> <exptime> <bytes>\r\n<data>\r\n */
 635
 636     vec[0].iov_base = cmd;
 637     vec[0].iov_len  = cmd_size;
 638
 639     vec[1].iov_base = (void*)key;
 640     vec[1].iov_len  = key_size;
 641
 642     klen = apr_snprintf(conn->buffer, BUFFER_SIZE, " %u %u %" APR_SIZE_T_FMT " " MC_EOL,
 643                         flags, timeout, data_size);
 644
 645     vec[2].iov_base = conn->buffer;
 646     vec[2].iov_len  = klen;
 647
 648     vec[3].iov_base = data;
 649     vec[3].iov_len  = data_size;
 650
 651     vec[4].iov_base = MC_EOL;
 652     vec[4].iov_len  = MC_EOL_LEN;
 653
 654     rv = apr_socket_sendv(conn->sock, vec, 5, &written);
 655
 656     if (rv != APR_SUCCESS) {
 657         ms_bad_conn(ms, conn);
 658         apr_memcache_disable_server(mc, ms);
 659         return rv;
 660     }
 661
 662     rv = get_server_line(conn);
 663
 664     if (rv != APR_SUCCESS) {
 665         ms_bad_conn(ms, conn);
 666         apr_memcache_disable_server(mc, ms);
 667         return rv;
 668     }
 669
 670     if (strcmp(conn->buffer, MS_STORED MC_EOL) == 0) {
 671         rv = APR_SUCCESS;
 672     }
 673     else if (strcmp(conn->buffer, MS_NOT_STORED MC_EOL) == 0) {
 674         rv = APR_EEXIST;
 675     }
 676     else {
 677         rv = APR_EGENERAL;
 678     }
 679
 680     ms_release_conn(ms, conn);
 681
 682     return rv;
 683 }
 684
 685 APU_DECLARE(apr_status_t)
 686 apr_memcache_set(apr_memcache_t *mc,
 687                  const char *key,
 688                  char *data,
 689                  const apr_size_t data_size,
 690                  apr_uint32_t timeout,
 691                  apr_uint16_t flags)
 692 {
 693     return storage_cmd_write(mc,
 694                            MC_SET, MC_SET_LEN,
 695                            key,
 696                            data, data_size,
 697                            timeout, flags);
 698 }
 699
 700 APU_DECLARE(apr_status_t)
 701 apr_memcache_add(apr_memcache_t *mc,
 702                  const char *key,
 703                  char *data,
 704                  const apr_size_t data_size,
 705                  apr_uint32_t timeout,
 706                  apr_uint16_t flags)
 707 {
 708     return storage_cmd_write(mc,
 709                            MC_ADD, MC_ADD_LEN,
 710                            key,
 711                            data, data_size,
 712                            timeout, flags);
 713 }
 714
 715 APU_DECLARE(apr_status_t)
 716 apr_memcache_replace(apr_memcache_t *mc,
 717                  const char *key,
 718                  char *data,
 719                  const apr_size_t data_size,
 720                  apr_uint32_t timeout,
 721                  apr_uint16_t flags)
 722 {
 723     return storage_cmd_write(mc,
 724                            MC_REPLACE, MC_REPLACE_LEN,
 725                            key,
 726                            data, data_size,
 727                            timeout, flags);
 728
 729 }
 730
 731 /*
 732  * Parses a decimal size from size_str, returning the value in *size.
 733  * Returns 1 if parsing was successful, 0 if parsing failed.
 734  */
 735 static int parse_size(const char *size_str, apr_size_t *size)
 736 {
 737     char *endptr;
 738     long size_as_long;
 739
 740     errno = 0;
 741     size_as_long = strtol(size_str, &endptr, 10);
 742     if ((size_as_long < 0) || (errno != 0) || (endptr == size_str) ||
 743         (endptr[0] != ' ' && (endptr[0] != '\r' || endptr[1] != '\n'))) {
 744         return 0;
 745     }
 746
 747     *size = (unsigned long)size_as_long;
 748     return 1;
 749 }
 750
 751 APU_DECLARE(apr_status_t)
 752 apr_memcache_getp(apr_memcache_t *mc,
 753                   apr_pool_t *p,
 754                   const char *key,
 755                   char **baton,
 756                   apr_size_t *new_length,
 757                   apr_uint16_t *flags_)
 758 {
 759     apr_status_t rv;
 760     apr_memcache_server_t *ms;
 761     apr_memcache_conn_t *conn;
 762     apr_uint32_t hash;
 763     apr_size_t written;
 764     apr_size_t klen = strlen(key);
 765     struct iovec vec[3];
 766
 767     hash = apr_memcache_hash(mc, key, klen);
 768     ms = apr_memcache_find_server_hash(mc, hash);
 769     if (ms == NULL)
 770         return APR_NOTFOUND;
 771
 772     rv = ms_find_conn(ms, &conn);
 773
 774     if (rv != APR_SUCCESS) {
 775         apr_memcache_disable_server(mc, ms);
 776         return rv;
 777     }
 778
 779     /* get <key>[ <key>[...]]\r\n */
 780     vec[0].iov_base = MC_GET;
 781     vec[0].iov_len  = MC_GET_LEN;
 782
 783     vec[1].iov_base = (void*)key;
 784     vec[1].iov_len  = klen;
 785
 786     vec[2].iov_base = MC_EOL;
 787     vec[2].iov_len  = MC_EOL_LEN;
 788
 789     rv = apr_socket_sendv(conn->sock, vec, 3, &written);
 790
 791     if (rv != APR_SUCCESS) {
 792         ms_bad_conn(ms, conn);
 793         apr_memcache_disable_server(mc, ms);
 794         return rv;
 795     }
 796
 797     rv = get_server_line(conn);
 798     if (rv != APR_SUCCESS) {
 799         ms_bad_conn(ms, conn);
 800         apr_memcache_disable_server(mc, ms);
 801         return rv;
 802     }
 803
 804     if (strncmp(MS_VALUE, conn->buffer, MS_VALUE_LEN) == 0) {
 805         char *flags;
 806         char *length;
 807         char *last;
 808         apr_size_t len = 0;
 809
 810         flags = apr_strtok(conn->buffer, " ", &last);
 811         flags = apr_strtok(NULL, " ", &last);
 812         flags = apr_strtok(NULL, " ", &last);
 813
 814         if (flags_) {
 815             *flags_ = atoi(flags);
 816         }
 817
 818         length = apr_strtok(NULL, " ", &last);
 819         if (!length || !parse_size(length, &len)) {
 820             ms_bad_conn(ms, conn);
 821             apr_memcache_disable_server(mc, ms);
 822             return APR_EGENERAL;
 823         }
 824         else {
 825             apr_bucket_brigade *bbb;
 826             apr_bucket *e;
 827
 828             /* eat the trailing \r\n */
 829             rv = apr_brigade_partition(conn->bb, len+2, &e);
 830             if (rv != APR_SUCCESS) {
 831                 ms_bad_conn(ms, conn);
 832                 apr_memcache_disable_server(mc, ms);
 833                 return rv;
 834             }
 835
 836             bbb = apr_brigade_split(conn->bb, e);
 837
 838             rv = apr_brigade_pflatten(conn->bb, baton, &len, p);
 839             if (rv != APR_SUCCESS) {
 840                 ms_bad_conn(ms, conn);
 841                 return rv;
 842             }
 843
 844             rv = apr_brigade_destroy(conn->bb);
 845             if (rv != APR_SUCCESS) {
 846                 ms_bad_conn(ms, conn);
 847                 return rv;
 848             }
 849
 850             conn->bb = bbb;
 851
 852             *new_length = len - 2;
 853             (*baton)[*new_length] = '\0';
 854         }
 855
 856         rv = get_server_line(conn);
 857         if (rv != APR_SUCCESS) {
 858             ms_bad_conn(ms, conn);
 859             apr_memcache_disable_server(mc, ms);
 860             return rv;
 861         }
 862
 863         if (strncmp(MS_END, conn->buffer, MS_END_LEN) != 0) {
 864             ms_bad_conn(ms, conn);
 865             apr_memcache_disable_server(mc, ms);
 866             return APR_EGENERAL;
 867         }
 868     }
 869     else if (strncmp(MS_END, conn->buffer, MS_END_LEN) == 0) {
 870         rv = APR_NOTFOUND;
 871     }
 872     else {
 873         ms_bad_conn(ms, conn);
 874         apr_memcache_disable_server(mc, ms);
 875         return APR_EGENERAL;
 876     }
 877
 878     ms_release_conn(ms, conn);
 879
 880     return rv;
 881 }
 882
 883 APU_DECLARE(apr_status_t)
 884 apr_memcache_delete(apr_memcache_t *mc,
 885                     const char *key,
 886                     apr_uint32_t timeout)
 887 {
 888     apr_status_t rv;
 889     apr_memcache_server_t *ms;
 890     apr_memcache_conn_t *conn;
 891     apr_uint32_t hash;
 892     apr_size_t written;
 893     struct iovec vec[3];
 894     apr_size_t klen = strlen(key);
 895
 896     hash = apr_memcache_hash(mc, key, klen);
 897     ms = apr_memcache_find_server_hash(mc, hash);
 898     if (ms == NULL)
 899         return APR_NOTFOUND;
 900
 901     rv = ms_find_conn(ms, &conn);
 902
 903     if (rv != APR_SUCCESS) {
 904         apr_memcache_disable_server(mc, ms);
 905         return rv;
 906     }
 907
 908     /* delete <key> <time>\r\n */
 909     vec[0].iov_base = MC_DELETE;
 910     vec[0].iov_len  = MC_DELETE_LEN;
 911
 912     vec[1].iov_base = (void*)key;
 913     vec[1].iov_len  = klen;
 914
 915     klen = apr_snprintf(conn->buffer, BUFFER_SIZE, " %u" MC_EOL, timeout);
 916
 917     vec[2].iov_base = conn->buffer;
 918     vec[2].iov_len  = klen;
 919
 920     rv = apr_socket_sendv(conn->sock, vec, 3, &written);
 921
 922     if (rv != APR_SUCCESS) {
 923         ms_bad_conn(ms, conn);
 924         apr_memcache_disable_server(mc, ms);
 925         return rv;
 926     }
 927
 928     rv = get_server_line(conn);
 929     if (rv != APR_SUCCESS) {
 930         ms_bad_conn(ms, conn);
 931         apr_memcache_disable_server(mc, ms);
 932         return rv;
 933     }
 934
 935     if (strncmp(MS_DELETED, conn->buffer, MS_DELETED_LEN) == 0) {
 936         rv = APR_SUCCESS;
 937     }
 938     else if (strncmp(MS_NOT_FOUND, conn->buffer, MS_NOT_FOUND_LEN) == 0) {
 939         rv = APR_NOTFOUND;
 940     }
 941     else {
 942         rv = APR_EGENERAL;
 943     }
 944
 945     ms_release_conn(ms, conn);
 946
 947     return rv;
 948 }
 949
 950 static apr_status_t num_cmd_write(apr_memcache_t *mc,
 951                                       char *cmd,
 952                                       const apr_uint32_t cmd_size,
 953                                       const char *key,
 954                                       const apr_int32_t inc,
 955                                       apr_uint32_t *new_value)
 956 {
 957     apr_status_t rv;
 958     apr_memcache_server_t *ms;
 959     apr_memcache_conn_t *conn;
 960     apr_uint32_t hash;
 961     apr_size_t written;
 962     struct iovec vec[3];
 963     apr_size_t klen = strlen(key);
 964
 965     hash = apr_memcache_hash(mc, key, klen);
 966     ms = apr_memcache_find_server_hash(mc, hash);
 967     if (ms == NULL)
 968         return APR_NOTFOUND;
 969
 970     rv = ms_find_conn(ms, &conn);
 971
 972     if (rv != APR_SUCCESS) {
 973         apr_memcache_disable_server(mc, ms);
 974         return rv;
 975     }
 976
 977     /* <cmd> <key> <value>\r\n */
 978     vec[0].iov_base = cmd;
 979     vec[0].iov_len  = cmd_size;
 980
 981     vec[1].iov_base = (void*)key;
 982     vec[1].iov_len  = klen;
 983
 984     klen = apr_snprintf(conn->buffer, BUFFER_SIZE, " %u" MC_EOL, inc);
 985
 986     vec[2].iov_base = conn->buffer;
 987     vec[2].iov_len  = klen;
 988
 989     rv = apr_socket_sendv(conn->sock, vec, 3, &written);
 990
 991     if (rv != APR_SUCCESS) {
 992         ms_bad_conn(ms, conn);
 993         apr_memcache_disable_server(mc, ms);
 994         return rv;
 995     }
 996
 997     rv = get_server_line(conn);
 998     if (rv != APR_SUCCESS) {
 999         ms_bad_conn(ms, conn);
1000         apr_memcache_disable_server(mc, ms);
1001         return rv;
1002     }
1003
1004     if (strncmp(MS_ERROR, conn->buffer, MS_ERROR_LEN) == 0) {
1005         rv = APR_EGENERAL;
1006     }
1007     else if (strncmp(MS_NOT_FOUND, conn->buffer, MS_NOT_FOUND_LEN) == 0) {
1008         rv = APR_NOTFOUND;
1009     }
1010     else {
1011         if (new_value) {
1012             *new_value = atoi(conn->buffer);
1013         }
1014         rv = APR_SUCCESS;
1015     }
1016
1017     ms_release_conn(ms, conn);
1018
1019     return rv;
1020 }
1021
1022 APU_DECLARE(apr_status_t)
1023 apr_memcache_incr(apr_memcache_t *mc,
1024                     const char *key,
1025                     apr_int32_t inc,
1026                     apr_uint32_t *new_value)
1027 {
1028     return num_cmd_write(mc,
1029                          MC_INCR,
1030                          MC_INCR_LEN,
1031                          key,
1032                          inc,
1033                          new_value);
1034 }
1035
1036
1037 APU_DECLARE(apr_status_t)
1038 apr_memcache_decr(apr_memcache_t *mc,
1039                     const char *key,
1040                     apr_int32_t inc,
1041                     apr_uint32_t *new_value)
1042 {
1043     return num_cmd_write(mc,
1044                          MC_DECR,
1045                          MC_DECR_LEN,
1046                          key,
1047                          inc,
1048                          new_value);
1049 }
1050
1051
1052
1053 APU_DECLARE(apr_status_t)
1054 apr_memcache_version(apr_memcache_server_t *ms,
1055                   apr_pool_t *p,
1056                   char **baton)
1057 {
1058     apr_status_t rv;
1059     apr_memcache_conn_t *conn;
1060     apr_size_t written;
1061     struct iovec vec[2];
1062
1063     rv = ms_find_conn(ms, &conn);
1064
1065     if (rv != APR_SUCCESS) {
1066         return rv;
1067     }
1068
1069     /* version\r\n */
1070     vec[0].iov_base = MC_VERSION;
1071     vec[0].iov_len  = MC_VERSION_LEN;
1072
1073     vec[1].iov_base = MC_EOL;
1074     vec[1].iov_len  = MC_EOL_LEN;
1075
1076     rv = apr_socket_sendv(conn->sock, vec, 2, &written);
1077
1078     if (rv != APR_SUCCESS) {
1079         ms_bad_conn(ms, conn);
1080         return rv;
1081     }
1082
1083     rv = get_server_line(conn);
1084     if (rv != APR_SUCCESS) {
1085         ms_bad_conn(ms, conn);
1086         return rv;
1087     }
1088
1089     if (strncmp(MS_VERSION, conn->buffer, MS_VERSION_LEN) == 0) {
1090         *baton = apr_pstrmemdup(p, conn->buffer+MS_VERSION_LEN+1,
1091                                 conn->blen - MS_VERSION_LEN - 2);
1092         rv = APR_SUCCESS;
1093     }
1094     else {
1095         rv = APR_EGENERAL;
1096     }
1097
1098     ms_release_conn(ms, conn);
1099
1100     return rv;
1101 }
1102
1103 apr_status_t mc_version_ping(apr_memcache_server_t *ms)
1104 {
1105     apr_status_t rv;
1106     apr_size_t written;
1107     struct iovec vec[2];
1108     apr_memcache_conn_t *conn;
1109
1110     rv = ms_find_conn(ms, &conn);
1111
1112     if (rv != APR_SUCCESS) {
1113         return rv;
1114     }
1115
1116     /* version\r\n */
1117     vec[0].iov_base = MC_VERSION;
1118     vec[0].iov_len  = MC_VERSION_LEN;
1119
1120     vec[1].iov_base = MC_EOL;
1121     vec[1].iov_len  = MC_EOL_LEN;
1122
1123     rv = apr_socket_sendv(conn->sock, vec, 2, &written);
1124
1125     if (rv != APR_SUCCESS) {
1126         ms_bad_conn(ms, conn);
1127         return rv;
1128     }
1129
1130     rv = get_server_line(conn);
1131     ms_release_conn(ms, conn);
1132     return rv;
1133 }
1134
1135
1136 APU_DECLARE(void)
1137 apr_memcache_add_multget_key(apr_pool_t *data_pool,
1138                              const char* key,
1139                              apr_hash_t **values)
1140 {
1141     apr_memcache_value_t* value;
1142     apr_size_t klen = strlen(key);
1143
1144     /* create the value hash if need be */
1145     if (!*values) {
1146         *values = apr_hash_make(data_pool);
1147     }
1148
1149     /* init key and add it to the value hash */
1150     value = apr_pcalloc(data_pool, sizeof(apr_memcache_value_t));
1151
1152     value->status = APR_NOTFOUND;
1153     value->key = apr_pstrdup(data_pool, key);
1154
1155     apr_hash_set(*values, value->key, klen, value);
1156 }
1157
1158 static void mget_conn_result(int serverup,
1159                              int connup,
1160                              apr_status_t rv,
1161                              apr_memcache_t *mc,
1162                              apr_memcache_server_t *ms,
1163                              apr_memcache_conn_t *conn,
1164                              struct cache_server_query_t *server_query,
1165                              apr_hash_t *values,
1166                              apr_hash_t *server_queries)
1167 {
1168     apr_int32_t j;
1169     apr_memcache_value_t* value;
1170
1171     apr_hash_set(server_queries, &ms, sizeof(ms), NULL);
1172
1173     if (connup) {
1174         ms_release_conn(ms, conn);
1175     } else {
1176         ms_bad_conn(ms, conn);
1177
1178         if (!serverup) {
1179             apr_memcache_disable_server(mc, ms);
1180         }
1181     }
1182
1183     for (j = 1; j < server_query->query_vec_count ; j+=2) {
1184         if (server_query->query_vec[j].iov_base) {
1185             value = apr_hash_get(values, server_query->query_vec[j].iov_base,
1186                                  strlen(server_query->query_vec[j].iov_base));
1187
1188             if (value->status == APR_NOTFOUND) {
1189                 value->status = rv;
1190             }
1191         }
1192     }
1193 }
1194
1195 APU_DECLARE(apr_status_t)
1196 apr_memcache_multgetp(apr_memcache_t *mc,
1197                       apr_pool_t *temp_pool,
1198                       apr_pool_t *data_pool,
1199                       apr_hash_t *values)
1200 {
1201     apr_status_t rv;
1202     apr_memcache_server_t* ms;
1203     apr_memcache_conn_t* conn;
1204     apr_uint32_t hash;
1205     apr_size_t written;
1206     apr_size_t klen;
1207
1208     apr_memcache_value_t* value;
1209     apr_hash_index_t* value_hash_index;
1210
1211     /* this is a little over aggresive, but beats multiple loops
1212      * to figure out how long each vector needs to be per-server.
1213      */
1214     apr_int32_t veclen = 2 + 2 * apr_hash_count(values) - 1; /* get <key>[<space><key>...]\r\n */
1215     apr_int32_t i, j;
1216     apr_int32_t queries_sent;
1217     apr_int32_t queries_recvd;
1218
1219     apr_hash_t * server_queries = apr_hash_make(temp_pool);
1220     struct cache_server_query_t* server_query;
1221     apr_hash_index_t * query_hash_index;
1222
1223     apr_pollset_t* pollset;
1224     const apr_pollfd_t* activefds;
1225     apr_pollfd_t* pollfds;
1226
1227
1228     /* build all the queries */
1229     value_hash_index = apr_hash_first(temp_pool, values);
1230     while (value_hash_index) {
1231         void *v;
1232         apr_hash_this(value_hash_index, NULL, NULL, &v);
1233         value = v;
1234         value_hash_index = apr_hash_next(value_hash_index);
1235         klen = strlen(value->key);
1236
1237         hash = apr_memcache_hash(mc, value->key, klen);
1238         ms = apr_memcache_find_server_hash(mc, hash);
1239         if (ms == NULL) {
1240             continue;
1241         }
1242
1243         server_query = apr_hash_get(server_queries, &ms, sizeof(ms));
1244
1245         if (!server_query) {
1246             rv = ms_find_conn(ms, &conn);
1247
1248             if (rv != APR_SUCCESS) {
1249                 apr_memcache_disable_server(mc, ms);
1250                 value->status = rv;
1251                 continue;
1252             }
1253
1254             server_query = apr_pcalloc(temp_pool,sizeof(struct cache_server_query_t));
1255
1256             apr_hash_set(server_queries, &ms, sizeof(ms), server_query);
1257
1258             server_query->ms = ms;
1259             server_query->conn = conn;
1260             server_query->query_vec = apr_pcalloc(temp_pool, sizeof(struct iovec)*veclen);
1261
1262             /* set up the first key */
1263             server_query->query_vec[0].iov_base = MC_GET;
1264             server_query->query_vec[0].iov_len  = MC_GET_LEN;
1265
1266             server_query->query_vec[1].iov_base = (void*)(value->key);
1267             server_query->query_vec[1].iov_len  = klen;
1268
1269             server_query->query_vec[2].iov_base = MC_EOL;
1270             server_query->query_vec[2].iov_len  = MC_EOL_LEN;
1271
1272             server_query->query_vec_count = 3;
1273         }
1274         else {
1275             j = server_query->query_vec_count - 1;
1276
1277             server_query->query_vec[j].iov_base = MC_WS;
1278             server_query->query_vec[j].iov_len  = MC_WS_LEN;
1279             j++;
1280
1281             server_query->query_vec[j].iov_base = (void*)(value->key);
1282             server_query->query_vec[j].iov_len  = klen;
1283             j++;
1284
1285             server_query->query_vec[j].iov_base = MC_EOL;
1286             server_query->query_vec[j].iov_len  = MC_EOL_LEN;
1287             j++;
1288
1289            server_query->query_vec_count = j;
1290         }
1291     }
1292
1293     /* create polling structures */
1294     pollfds = apr_pcalloc(temp_pool, apr_hash_count(server_queries) * sizeof(apr_pollfd_t));
1295
1296     rv = apr_pollset_create(&pollset, apr_hash_count(server_queries), temp_pool, 0);
1297
1298     if (rv != APR_SUCCESS) {
1299         query_hash_index = apr_hash_first(temp_pool, server_queries);
1300
1301         while (query_hash_index) {
1302             void *v;
1303             apr_hash_this(query_hash_index, NULL, NULL, &v);
1304             server_query = v;
1305             query_hash_index = apr_hash_next(query_hash_index);
1306
1307             mget_conn_result(TRUE, TRUE, rv, mc, server_query->ms, server_query->conn,
1308                              server_query, values, server_queries);
1309         }
1310
1311         return rv;
1312     }
1313
1314     /* send all the queries */
1315     queries_sent = 0;
1316     query_hash_index = apr_hash_first(temp_pool, server_queries);
1317
1318     while (query_hash_index) {
1319         void *v;
1320         apr_hash_this(query_hash_index, NULL, NULL, &v);
1321         server_query = v;
1322         query_hash_index = apr_hash_next(query_hash_index);
1323
1324         conn = server_query->conn;
1325         ms = server_query->ms;
1326
1327         for (i = 0, rv = APR_SUCCESS; i < veclen && rv == APR_SUCCESS; i += APR_MAX_IOVEC_SIZE) {
1328             rv = apr_socket_sendv(conn->sock, &(server_query->query_vec[i]),
1329                                   veclen-i>APR_MAX_IOVEC_SIZE ? APR_MAX_IOVEC_SIZE : veclen-i , &written);
1330         }
1331
1332         if (rv != APR_SUCCESS) {
1333             mget_conn_result(FALSE, FALSE, rv, mc, ms, conn,
1334                              server_query, values, server_queries);
1335             continue;
1336         }
1337
1338         pollfds[queries_sent].desc_type = APR_POLL_SOCKET;
1339         pollfds[queries_sent].reqevents = APR_POLLIN;
1340         pollfds[queries_sent].p = temp_pool;
1341         pollfds[queries_sent].desc.s = conn->sock;
1342         pollfds[queries_sent].client_data = (void *)server_query;
1343         apr_pollset_add (pollset, &pollfds[queries_sent]);
1344
1345         queries_sent++;
1346     }
1347
1348     while (queries_sent) {
1349         rv = apr_pollset_poll(pollset, MULT_GET_TIMEOUT, &queries_recvd, &activefds);
1350
1351         if (rv != APR_SUCCESS) {
1352             /* timeout */
1353             queries_sent = 0;
1354             continue;
1355         }
1356         for (i = 0; i < queries_recvd; i++) {
1357             server_query = activefds[i].client_data;
1358             conn = server_query->conn;
1359             ms = server_query->ms;
1360
1361            rv = get_server_line(conn);
1362
1363            if (rv != APR_SUCCESS) {
1364                apr_pollset_remove (pollset, &activefds[i]);
1365                mget_conn_result(FALSE, FALSE, rv, mc, ms, conn,
1366                                 server_query, values, server_queries);
1367                queries_sent--;
1368                continue;
1369            }
1370
1371            if (strncmp(MS_VALUE, conn->buffer, MS_VALUE_LEN) == 0) {
1372                char *key;
1373                char *flags;
1374                char *length;
1375                char *last;
1376                char *data;
1377                apr_size_t len = 0;
1378                apr_bucket *e = NULL;
1379
1380                key = apr_strtok(conn->buffer, " ", &last); /* just the VALUE, ignore */
1381                key = apr_strtok(NULL, " ", &last);
1382                flags = apr_strtok(NULL, " ", &last);
1383                length = apr_strtok(NULL, " ", &last);
1384
1385                if (!length || !parse_size(length, &len)) {
1386                    rv = APR_EGENERAL;
1387                }
1388                else {
1389                    /* eat the trailing \r\n */
1390                    rv = apr_brigade_partition(conn->bb, len+2, &e);
1391                }
1392                if (rv != APR_SUCCESS) {
1393                    apr_pollset_remove (pollset, &activefds[i]);
1394                    mget_conn_result(TRUE, FALSE, rv, mc, ms, conn,
1395                                     server_query, values, server_queries);
1396                    queries_sent--;
1397                    continue;
1398                }
1399
1400                value = apr_hash_get(values, key, strlen(key));
1401                if (value) {
1402                    apr_bucket_brigade *bbb;
1403
1404                    bbb = apr_brigade_split(conn->bb, e);
1405
1406                    rv = apr_brigade_pflatten(conn->bb, &data, &len, data_pool);
1407                    if (rv != APR_SUCCESS) {
1408                        apr_pollset_remove (pollset, &activefds[i]);
1409                        mget_conn_result(TRUE, FALSE, rv, mc, ms, conn,
1410                                         server_query, values, server_queries);
1411                        queries_sent--;
1412                        continue;
1413                    }
1414
1415                    rv = apr_brigade_destroy(conn->bb);
1416                    if (rv != APR_SUCCESS) {
1417                        apr_pollset_remove (pollset, &activefds[i]);
1418                        mget_conn_result(TRUE, FALSE, rv, mc, ms, conn,
1419                                         server_query, values, server_queries);
1420                        queries_sent--;
1421                        continue;
1422                    }
1423
1424                    conn->bb = bbb;
1425
1426                    value->len = len - 2;
1427                    data[value->len] = '\0';
1428                    value->data = data;
1429
1430                    value->status = rv;
1431                    value->flags = atoi(flags);
1432
1433                    /* stay on the server */
1434                    i--;
1435                }
1436                else {
1437                    /* Server Sent back a key I didn't ask for or my
1438                     * hash is corrupt */
1439                    rv = APR_EGENERAL;
1440                }
1441            }
1442            else if (strncmp(MS_END, conn->buffer, MS_END_LEN) == 0) {
1443                /* this connection is done */
1444                apr_pollset_remove (pollset, &activefds[i]);
1445                ms_release_conn(ms, conn);
1446                apr_hash_set(server_queries, &ms, sizeof(ms), NULL);
1447                queries_sent--;
1448            }
1449            else {
1450                /* unknown reply? */
1451                rv = APR_EGENERAL;
1452            }
1453            if (rv != APR_SUCCESS) {
1454                apr_pollset_remove (pollset, &activefds[i]);
1455                mget_conn_result(TRUE, FALSE, rv, mc, ms, conn,
1456                                 server_query, values, server_queries);
1457                queries_sent--;
1458            }
1459         } /* /for */
1460     } /* /while */
1461
1462     query_hash_index = apr_hash_first(temp_pool, server_queries);
1463     while (query_hash_index) {
1464         void *v;
1465         apr_hash_this(query_hash_index, NULL, NULL, &v);
1466         server_query = v;
1467         query_hash_index = apr_hash_next(query_hash_index);
1468
1469         conn = server_query->conn;
1470         ms = server_query->ms;
1471
1472         mget_conn_result(TRUE, (rv == APR_SUCCESS), rv, mc, ms, conn,
1473                          server_query, values, server_queries);
1474         continue;
1475     }
1476
1477     apr_pollset_destroy(pollset);
1478     apr_pool_clear(temp_pool);
1479     return APR_SUCCESS;
1480
1481 }
1482
1483
1484
1485 /**
1486  * Define all of the strings for stats
1487  */
1488
1489 #define STAT_pid MS_STAT " pid "
1490 #define STAT_pid_LEN (sizeof(STAT_pid)-1)
1491
1492 #define STAT_uptime MS_STAT " uptime "
1493 #define STAT_uptime_LEN (sizeof(STAT_uptime)-1)
1494
1495 #define STAT_time MS_STAT " time "
1496 #define STAT_time_LEN (sizeof(STAT_time)-1)
1497
1498 #define STAT_version MS_STAT " version "
1499 #define STAT_version_LEN (sizeof(STAT_version)-1)
1500
1501 #define STAT_pointer_size MS_STAT " pointer_size "
1502 #define STAT_pointer_size_LEN (sizeof(STAT_pointer_size)-1)
1503
1504 #define STAT_rusage_user MS_STAT " rusage_user "
1505 #define STAT_rusage_user_LEN (sizeof(STAT_rusage_user)-1)
1506
1507 #define STAT_rusage_system MS_STAT " rusage_system "
1508 #define STAT_rusage_system_LEN (sizeof(STAT_rusage_system)-1)
1509
1510 #define STAT_curr_items MS_STAT " curr_items "
1511 #define STAT_curr_items_LEN (sizeof(STAT_curr_items)-1)
1512
1513 #define STAT_total_items MS_STAT " total_items "
1514 #define STAT_total_items_LEN (sizeof(STAT_total_items)-1)
1515
1516 #define STAT_bytes MS_STAT " bytes "
1517 #define STAT_bytes_LEN (sizeof(STAT_bytes)-1)
1518
1519 #define STAT_curr_connections MS_STAT " curr_connections "
1520 #define STAT_curr_connections_LEN (sizeof(STAT_curr_connections)-1)
1521
1522 #define STAT_total_connections MS_STAT " total_connections "
1523 #define STAT_total_connections_LEN (sizeof(STAT_total_connections)-1)
1524
1525 #define STAT_connection_structures MS_STAT " connection_structures "
1526 #define STAT_connection_structures_LEN (sizeof(STAT_connection_structures)-1)
1527
1528 #define STAT_cmd_get MS_STAT " cmd_get "
1529 #define STAT_cmd_get_LEN (sizeof(STAT_cmd_get)-1)
1530
1531 #define STAT_cmd_set MS_STAT " cmd_set "
1532 #define STAT_cmd_set_LEN (sizeof(STAT_cmd_set)-1)
1533
1534 #define STAT_get_hits MS_STAT " get_hits "
1535 #define STAT_get_hits_LEN (sizeof(STAT_get_hits)-1)
1536
1537 #define STAT_get_misses MS_STAT " get_misses "
1538 #define STAT_get_misses_LEN (sizeof(STAT_get_misses)-1)
1539
1540 #define STAT_evictions MS_STAT " evictions "
1541 #define STAT_evictions_LEN (sizeof(STAT_evictions)-1)
1542
1543 #define STAT_bytes_read MS_STAT " bytes_read "
1544 #define STAT_bytes_read_LEN (sizeof(STAT_bytes_read)-1)
1545
1546 #define STAT_bytes_written MS_STAT " bytes_written "
1547 #define STAT_bytes_written_LEN (sizeof(STAT_bytes_written)-1)
1548
1549 #define STAT_limit_maxbytes MS_STAT " limit_maxbytes "
1550 #define STAT_limit_maxbytes_LEN (sizeof(STAT_limit_maxbytes)-1)
1551
1552 #define STAT_threads MS_STAT " threads "
1553 #define STAT_threads_LEN (sizeof(STAT_threads)-1)
1554
1555 static const char *stat_read_string(apr_pool_t *p, char *buf, apr_size_t len)
1556 {
1557     /* remove trailing \r\n and null char */
1558     return apr_pstrmemdup(p, buf, len-2);
1559 }
1560
1561 static apr_uint32_t stat_read_uint32(apr_pool_t *p, char *buf, apr_size_t  len)
1562 {
1563     buf[len-2] = '\0';
1564     return atoi(buf);
1565 }
1566
1567 static apr_uint64_t stat_read_uint64(apr_pool_t *p, char *buf, apr_size_t  len)
1568 {
1569     buf[len-2] = '\0';
1570     return apr_atoi64(buf);
1571 }
1572
1573 static apr_time_t stat_read_time(apr_pool_t *p, char *buf, apr_size_t  len)
1574 {
1575     buf[len-2] = '\0';
1576     return apr_time_from_sec(atoi(buf));
1577 }
1578
1579 static apr_time_t stat_read_rtime(apr_pool_t *p, char *buf, apr_size_t  len)
1580 {
1581     char *tok;
1582     char *secs;
1583     char *usecs;
1584     const char *sep = ":.";
1585
1586     buf[len-2] = '\0';
1587
1588     secs = apr_strtok(buf, sep, &tok);
1589     usecs = apr_strtok(NULL, sep, &tok);
1590     if (secs && usecs) {
1591         return apr_time_make(atoi(secs), atoi(usecs));
1592     }
1593     else {
1594         return apr_time_make(0, 0);
1595     }
1596 }
1597
1598 /**
1599  * I got tired of Typing. Meh.
1600  *
1601  * TODO: Convert it to static tables to make it cooler.
1602  */
1603
1604 #define mc_stat_cmp(name) \
1605     strncmp(STAT_ ## name, conn->buffer, STAT_ ## name ## _LEN) == 0
1606
1607 #define mc_stat_str(name) \
1608     stat_read_string(p, conn->buffer + name, \
1609                      conn->blen - name)
1610
1611 #define mc_stat_uint32(name) \
1612     stat_read_uint32(p, conn->buffer + name, \
1613                      conn->blen - name)
1614
1615 #define mc_stat_uint64(name) \
1616     stat_read_uint64(p, conn->buffer + name, \
1617                      conn->blen - name)
1618
1619 #define mc_stat_time(name) \
1620     stat_read_time(p, conn->buffer + name, \
1621                      conn->blen - name)
1622
1623 #define mc_stat_rtime(name) \
1624     stat_read_rtime(p, conn->buffer + name, \
1625                      conn->blen - name)
1626
1627
1628 #define mc_do_stat(name, type) \
1629     if (mc_stat_cmp(name)) { \
1630         stats-> name = mc_stat_ ## type ((STAT_ ## name ## _LEN)); \
1631     }
1632
1633 static void update_stats(apr_pool_t *p, apr_memcache_conn_t *conn,
1634                          apr_memcache_stats_t *stats)
1635 {
1636
1637     mc_do_stat(version, str)
1638     else mc_do_stat(pid, uint32)
1639     else mc_do_stat(uptime, uint32)
1640     else mc_do_stat(pointer_size, uint32)
1641     else mc_do_stat(time, time)
1642     else mc_do_stat(rusage_user, rtime)
1643     else mc_do_stat(rusage_system, rtime)
1644     else mc_do_stat(curr_items, uint32)
1645     else mc_do_stat(total_items, uint32)
1646     else mc_do_stat(bytes, uint64)
1647     else mc_do_stat(curr_connections, uint32)
1648     else mc_do_stat(total_connections, uint32)
1649     else mc_do_stat(connection_structures, uint32)
1650     else mc_do_stat(cmd_get, uint32)
1651     else mc_do_stat(cmd_set, uint32)
1652     else mc_do_stat(get_hits, uint32)
1653     else mc_do_stat(get_misses, uint32)
1654     else mc_do_stat(evictions, uint64)
1655     else mc_do_stat(bytes_read, uint64)
1656     else mc_do_stat(bytes_written, uint64)
1657     else mc_do_stat(limit_maxbytes, uint32)
1658     else mc_do_stat(threads, uint32)
1659 }
1660
1661 APU_DECLARE(apr_status_t)
1662 apr_memcache_stats(apr_memcache_server_t *ms,
1663                   apr_pool_t *p,
1664                   apr_memcache_stats_t **stats)
1665 {
1666     apr_memcache_stats_t *ret;
1667     apr_status_t rv;
1668     apr_memcache_conn_t *conn;
1669     apr_size_t written;
1670     struct iovec vec[2];
1671
1672     rv = ms_find_conn(ms, &conn);
1673
1674     if (rv != APR_SUCCESS) {
1675         return rv;
1676     }
1677
1678     /* version\r\n */
1679     vec[0].iov_base = MC_STATS;
1680     vec[0].iov_len  = MC_STATS_LEN;
1681
1682     vec[1].iov_base = MC_EOL;
1683     vec[1].iov_len  = MC_EOL_LEN;
1684
1685     rv = apr_socket_sendv(conn->sock, vec, 2, &written);
1686
1687     if (rv != APR_SUCCESS) {
1688         ms_bad_conn(ms, conn);
1689         return rv;
1690     }
1691
1692     ret = apr_pcalloc(p, sizeof(apr_memcache_stats_t));
1693
1694     do {
1695         rv = get_server_line(conn);
1696         if (rv != APR_SUCCESS) {
1697             ms_bad_conn(ms, conn);
1698             return rv;
1699         }
1700
1701         if (strncmp(MS_END, conn->buffer, MS_END_LEN) == 0) {
1702             rv = APR_SUCCESS;
1703             break;
1704         }
1705         else if (strncmp(MS_STAT, conn->buffer, MS_STAT_LEN) == 0) {
1706             update_stats(p, conn, ret);
1707             continue;
1708         }
1709         else {
1710             rv = APR_EGENERAL;
1711             break;
1712         }
1713
1714     } while(1);
1715
1716     ms_release_conn(ms, conn);
1717
1718     if (stats) {
1719         *stats = ret;
1720     }
1721
1722     return rv;
1723 }
1724