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_socket_t *sock;
  29     apr_bucket_alloc_t *balloc;
  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 the dead server, every 5 seconds */
 185             if (curtime - ms->btime >  apr_time_from_sec(5)) {
 186                 if (mc_version_ping(ms) == APR_SUCCESS) {
 187                     ms->btime = curtime;
 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 #if APR_HAS_THREADS
 228     return apr_reslist_acquire(ms->conns, (void **)conn);
 229 #else
 230     *conn = ms->conn;
 231     return APR_SUCCESS;
 232 #endif
 233 }
 234
 235 static apr_status_t ms_bad_conn(apr_memcache_server_t *ms, apr_memcache_conn_t *conn)
 236 {
 237 #if APR_HAS_THREADS
 238     return apr_reslist_invalidate(ms->conns, conn);
 239 #else
 240     return APR_SUCCESS;
 241 #endif
 242 }
 243
 244 static apr_status_t ms_release_conn(apr_memcache_server_t *ms, apr_memcache_conn_t *conn)
 245 {
 246 #if APR_HAS_THREADS
 247     return apr_reslist_release(ms->conns, conn);
 248 #else
 249     return APR_SUCCESS;
 250 #endif
 251 }
 252
 253 APU_DECLARE(apr_status_t) apr_memcache_enable_server(apr_memcache_t *mc, apr_memcache_server_t *ms)
 254 {
 255     apr_status_t rv = APR_SUCCESS;
 256
 257     if (ms->status == APR_MC_SERVER_LIVE) {
 258         return rv;
 259     }
 260
 261     rv = make_server_live(mc, ms);
 262     return rv;
 263 }
 264
 265 APU_DECLARE(apr_status_t) apr_memcache_disable_server(apr_memcache_t *mc, apr_memcache_server_t *ms)
 266 {
 267     return make_server_dead(mc, ms);
 268 }
 269
 270 static apr_status_t conn_connect(apr_memcache_conn_t *conn)
 271 {
 272     apr_status_t rv = APR_SUCCESS;
 273     apr_sockaddr_t *sa;
 274
 275     rv = apr_sockaddr_info_get(&sa, conn->ms->host, APR_INET, conn->ms->port, 0, conn->p);
 276     if (rv != APR_SUCCESS) {
 277         return rv;
 278     }
 279
 280     rv = apr_socket_timeout_set(conn->sock, 1 * APR_USEC_PER_SEC);
 281     if (rv != APR_SUCCESS) {
 282         return rv;
 283     }
 284
 285     rv = apr_socket_connect(conn->sock, sa);
 286     if (rv != APR_SUCCESS) {
 287         return rv;
 288     }
 289
 290     rv = apr_socket_timeout_set(conn->sock, -1);
 291     if (rv != APR_SUCCESS) {
 292         return rv;
 293     }
 294
 295     return rv;
 296 }
 297
 298
 299 static apr_status_t
 300 mc_conn_construct(void **conn_, void *params, apr_pool_t *pool)
 301 {
 302     apr_status_t rv = APR_SUCCESS;
 303     apr_memcache_conn_t *conn;
 304     apr_bucket *e;
 305     apr_pool_t *np;
 306     apr_memcache_server_t *ms = params;
 307
 308     rv = apr_pool_create(&np, pool);
 309     if (rv != APR_SUCCESS) {
 310         return rv;
 311     }
 312
 313     conn = apr_palloc(np, sizeof( apr_memcache_conn_t ));
 314
 315     conn->p = np;
 316
 317     rv = apr_socket_create(&conn->sock, APR_INET, SOCK_STREAM, 0, np);
 318
 319     if (rv != APR_SUCCESS) {
 320         apr_pool_destroy(np);
 321         return rv;
 322     }
 323
 324     conn->balloc = apr_bucket_alloc_create(conn->p);
 325     conn->bb = apr_brigade_create(conn->p, conn->balloc);
 326     conn->tb = apr_brigade_create(conn->p, conn->balloc);
 327     conn->buffer = apr_palloc(conn->p, BUFFER_SIZE);
 328     conn->blen = 0;
 329     conn->ms = ms;
 330
 331     e = apr_bucket_socket_create(conn->sock, conn->balloc);
 332     APR_BRIGADE_INSERT_TAIL(conn->bb, e);
 333
 334     rv = conn_connect(conn);
 335     if (rv != APR_SUCCESS) {
 336         apr_pool_destroy(np);
 337     }
 338     else {
 339         *conn_ = conn;
 340     }
 341
 342     return rv;
 343 }
 344
 345 #if APR_HAS_THREADS
 346 static apr_status_t
 347 mc_conn_destruct(void *conn_, void *params, apr_pool_t *pool)
 348 {
 349     apr_memcache_conn_t *conn = (apr_memcache_conn_t*)conn_;
 350     struct iovec vec[2];
 351     apr_size_t written;
 352
 353     /* send a quit message to the memcached server to be nice about it. */
 354     vec[0].iov_base = MC_QUIT;
 355     vec[0].iov_len = MC_QUIT_LEN;
 356
 357     vec[1].iov_base = MC_EOL;
 358     vec[1].iov_len = MC_EOL_LEN;
 359
 360     /* Return values not checked, since we just want to make it go away. */
 361     apr_socket_sendv(conn->sock, vec, 2, &written);
 362     apr_socket_close(conn->sock);
 363
 364     apr_pool_destroy(conn->p);
 365
 366     return APR_SUCCESS;
 367 }
 368 #endif
 369
 370 APU_DECLARE(apr_status_t) apr_memcache_server_create(apr_pool_t *p,
 371                                                      const char *host, apr_port_t port,
 372                                                      apr_uint32_t min, apr_uint32_t smax,
 373                                                      apr_uint32_t max, apr_uint32_t ttl,
 374                                                      apr_memcache_server_t **ms)
 375 {
 376     apr_status_t rv = APR_SUCCESS;
 377     apr_memcache_server_t *server;
 378     apr_pool_t *np;
 379
 380     rv = apr_pool_create(&np, p);
 381
 382     server = apr_palloc(np, sizeof(apr_memcache_server_t));
 383
 384     server->p = np;
 385     server->host = apr_pstrdup(np, host);
 386     server->port = port;
 387     server->status = APR_MC_SERVER_DEAD;
 388 #if APR_HAS_THREADS
 389     rv = apr_thread_mutex_create(&server->lock, APR_THREAD_MUTEX_DEFAULT, np);
 390     if (rv != APR_SUCCESS) {
 391         return rv;
 392     }
 393
 394     rv = apr_reslist_create(&server->conns,
 395                                min,                     /* hard minimum */
 396                                smax,                    /* soft maximum */
 397                                max,                     /* hard maximum */
 398                                ttl,                     /* Time to live */
 399                                mc_conn_construct,       /* Make a New Connection */
 400                                mc_conn_destruct,        /* Kill Old Connection */
 401                                server, np);
 402
 403     apr_reslist_cleanup_order_set(server->conns, APR_RESLIST_CLEANUP_FIRST);
 404 #else
 405     rv = mc_conn_construct((void**)&(server->conn), server, np);
 406 #endif
 407
 408     if (rv != APR_SUCCESS) {
 409         return rv;
 410     }
 411
 412     *ms = server;
 413
 414     return rv;
 415 }
 416
 417 APU_DECLARE(apr_status_t) apr_memcache_create(apr_pool_t *p,
 418                                               apr_uint16_t max_servers, apr_uint32_t flags,
 419                                               apr_memcache_t **memcache)
 420 {
 421     apr_status_t rv = APR_SUCCESS;
 422     apr_memcache_t *mc;
 423
 424     mc = apr_palloc(p, sizeof(apr_memcache_t));
 425     mc->p = p;
 426     mc->nalloc = max_servers;
 427     mc->ntotal = 0;
 428     mc->live_servers = apr_palloc(p, mc->nalloc * sizeof(struct apr_memcache_server_t *));
 429     mc->hash_func = NULL;
 430     mc->hash_baton = NULL;
 431     mc->server_func = NULL;
 432     mc->server_baton = NULL;
 433     *memcache = mc;
 434     return rv;
 435 }
 436
 437
 438 /* The crc32 functions and data was originally written by Spencer
 439  * Garrett <srg@quick.com> and was gleaned from the PostgreSQL source
 440  * tree via the files contrib/ltree/crc32.[ch] and from FreeBSD at
 441  * src/usr.bin/cksum/crc32.c.
 442  */
 443
 444 static const apr_uint32_t crc32tab[256] = {
 445   0x00000000, 0x77073096, 0xee0e612c, 0x990951ba,
 446   0x076dc419, 0x706af48f, 0xe963a535, 0x9e6495a3,
 447   0x0edb8832, 0x79dcb8a4, 0xe0d5e91e, 0x97d2d988,
 448   0x09b64c2b, 0x7eb17cbd, 0xe7b82d07, 0x90bf1d91,
 449   0x1db71064, 0x6ab020f2, 0xf3b97148, 0x84be41de,
 450   0x1adad47d, 0x6ddde4eb, 0xf4d4b551, 0x83d385c7,
 451   0x136c9856, 0x646ba8c0, 0xfd62f97a, 0x8a65c9ec,
 452   0x14015c4f, 0x63066cd9, 0xfa0f3d63, 0x8d080df5,
 453   0x3b6e20c8, 0x4c69105e, 0xd56041e4, 0xa2677172,
 454   0x3c03e4d1, 0x4b04d447, 0xd20d85fd, 0xa50ab56b,
 455   0x35b5a8fa, 0x42b2986c, 0xdbbbc9d6, 0xacbcf940,
 456   0x32d86ce3, 0x45df5c75, 0xdcd60dcf, 0xabd13d59,
 457   0x26d930ac, 0x51de003a, 0xc8d75180, 0xbfd06116,
 458   0x21b4f4b5, 0x56b3c423, 0xcfba9599, 0xb8bda50f,
 459   0x2802b89e, 0x5f058808, 0xc60cd9b2, 0xb10be924,
 460   0x2f6f7c87, 0x58684c11, 0xc1611dab, 0xb6662d3d,
 461   0x76dc4190, 0x01db7106, 0x98d220bc, 0xefd5102a,
 462   0x71b18589, 0x06b6b51f, 0x9fbfe4a5, 0xe8b8d433,
 463   0x7807c9a2, 0x0f00f934, 0x9609a88e, 0xe10e9818,
 464   0x7f6a0dbb, 0x086d3d2d, 0x91646c97, 0xe6635c01,
 465   0x6b6b51f4, 0x1c6c6162, 0x856530d8, 0xf262004e,
 466   0x6c0695ed, 0x1b01a57b, 0x8208f4c1, 0xf50fc457,
 467   0x65b0d9c6, 0x12b7e950, 0x8bbeb8ea, 0xfcb9887c,
 468   0x62dd1ddf, 0x15da2d49, 0x8cd37cf3, 0xfbd44c65,
 469   0x4db26158, 0x3ab551ce, 0xa3bc0074, 0xd4bb30e2,
 470   0x4adfa541, 0x3dd895d7, 0xa4d1c46d, 0xd3d6f4fb,
 471   0x4369e96a, 0x346ed9fc, 0xad678846, 0xda60b8d0,
 472   0x44042d73, 0x33031de5, 0xaa0a4c5f, 0xdd0d7cc9,
 473   0x5005713c, 0x270241aa, 0xbe0b1010, 0xc90c2086,
 474   0x5768b525, 0x206f85b3, 0xb966d409, 0xce61e49f,
 475   0x5edef90e, 0x29d9c998, 0xb0d09822, 0xc7d7a8b4,
 476   0x59b33d17, 0x2eb40d81, 0xb7bd5c3b, 0xc0ba6cad,
 477   0xedb88320, 0x9abfb3b6, 0x03b6e20c, 0x74b1d29a,
 478   0xead54739, 0x9dd277af, 0x04db2615, 0x73dc1683,
 479   0xe3630b12, 0x94643b84, 0x0d6d6a3e, 0x7a6a5aa8,
 480   0xe40ecf0b, 0x9309ff9d, 0x0a00ae27, 0x7d079eb1,
 481   0xf00f9344, 0x8708a3d2, 0x1e01f268, 0x6906c2fe,
 482   0xf762575d, 0x806567cb, 0x196c3671, 0x6e6b06e7,
 483   0xfed41b76, 0x89d32be0, 0x10da7a5a, 0x67dd4acc,
 484   0xf9b9df6f, 0x8ebeeff9, 0x17b7be43, 0x60b08ed5,
 485   0xd6d6a3e8, 0xa1d1937e, 0x38d8c2c4, 0x4fdff252,
 486   0xd1bb67f1, 0xa6bc5767, 0x3fb506dd, 0x48b2364b,
 487   0xd80d2bda, 0xaf0a1b4c, 0x36034af6, 0x41047a60,
 488   0xdf60efc3, 0xa867df55, 0x316e8eef, 0x4669be79,
 489   0xcb61b38c, 0xbc66831a, 0x256fd2a0, 0x5268e236,
 490   0xcc0c7795, 0xbb0b4703, 0x220216b9, 0x5505262f,
 491   0xc5ba3bbe, 0xb2bd0b28, 0x2bb45a92, 0x5cb36a04,
 492   0xc2d7ffa7, 0xb5d0cf31, 0x2cd99e8b, 0x5bdeae1d,
 493   0x9b64c2b0, 0xec63f226, 0x756aa39c, 0x026d930a,
 494   0x9c0906a9, 0xeb0e363f, 0x72076785, 0x05005713,
 495   0x95bf4a82, 0xe2b87a14, 0x7bb12bae, 0x0cb61b38,
 496   0x92d28e9b, 0xe5d5be0d, 0x7cdcefb7, 0x0bdbdf21,
 497   0x86d3d2d4, 0xf1d4e242, 0x68ddb3f8, 0x1fda836e,
 498   0x81be16cd, 0xf6b9265b, 0x6fb077e1, 0x18b74777,
 499   0x88085ae6, 0xff0f6a70, 0x66063bca, 0x11010b5c,
 500   0x8f659eff, 0xf862ae69, 0x616bffd3, 0x166ccf45,
 501   0xa00ae278, 0xd70dd2ee, 0x4e048354, 0x3903b3c2,
 502   0xa7672661, 0xd06016f7, 0x4969474d, 0x3e6e77db,
 503   0xaed16a4a, 0xd9d65adc, 0x40df0b66, 0x37d83bf0,
 504   0xa9bcae53, 0xdebb9ec5, 0x47b2cf7f, 0x30b5ffe9,
 505   0xbdbdf21c, 0xcabac28a, 0x53b39330, 0x24b4a3a6,
 506   0xbad03605, 0xcdd70693, 0x54de5729, 0x23d967bf,
 507   0xb3667a2e, 0xc4614ab8, 0x5d681b02, 0x2a6f2b94,
 508   0xb40bbe37, 0xc30c8ea1, 0x5a05df1b, 0x2d02ef8d,
 509 };
 510
 511 APU_DECLARE(apr_uint32_t) apr_memcache_hash_crc32(void *baton,
 512                                                   const char *data,
 513                                                   const apr_size_t data_len)
 514 {
 515     apr_uint32_t i;
 516     apr_uint32_t crc;
 517     crc = ~0;
 518
 519     for (i = 0; i < data_len; i++)
 520         crc = (crc >> 8) ^ crc32tab[(crc ^ (data[i])) & 0xff];
 521
 522     return ~crc;
 523 }
 524
 525 APU_DECLARE(apr_uint32_t) apr_memcache_hash_default(void *baton,
 526                                                     const char *data,
 527                                                     const apr_size_t data_len)
 528 {
 529     /* The default Perl Client doesn't actually use just crc32 -- it shifts it again
 530      * like this....
 531      */
 532     return ((apr_memcache_hash_crc32(baton, data, data_len) >> 16) & 0x7fff);
 533 }
 534
 535 APU_DECLARE(apr_uint32_t) apr_memcache_hash(apr_memcache_t *mc,
 536                                             const char *data,
 537                                             const apr_size_t data_len)
 538 {
 539     if (mc->hash_func) {
 540         return mc->hash_func(mc->hash_baton, data, data_len);
 541     }
 542     else {
 543         return apr_memcache_hash_default(NULL, data, data_len);
 544     }
 545 }
 546
 547 static apr_status_t get_server_line(apr_memcache_conn_t *conn)
 548 {
 549     apr_size_t bsize = BUFFER_SIZE;
 550     apr_status_t rv = APR_SUCCESS;
 551
 552     rv = apr_brigade_split_line(conn->tb, conn->bb, APR_BLOCK_READ, BUFFER_SIZE);
 553
 554     if (rv != APR_SUCCESS) {
 555         return rv;
 556     }
 557
 558     rv = apr_brigade_flatten(conn->tb, conn->buffer, &bsize);
 559
 560     if (rv != APR_SUCCESS) {
 561         return rv;
 562     }
 563
 564     conn->blen = bsize;
 565     conn->buffer[bsize] = '\0';
 566
 567     return apr_brigade_cleanup(conn->tb);
 568 }
 569
 570 static apr_status_t storage_cmd_write(apr_memcache_t *mc,
 571                                       char *cmd,
 572                                       const apr_size_t cmd_size,
 573                                       const char *key,
 574                                       char *data,
 575                                       const apr_size_t data_size,
 576                                       apr_uint32_t timeout,
 577                                       apr_uint16_t flags)
 578 {
 579     apr_uint32_t hash;
 580     apr_memcache_server_t *ms;
 581     apr_memcache_conn_t *conn;
 582     apr_status_t rv;
 583     apr_size_t written;
 584     struct iovec vec[5];
 585     apr_size_t klen;
 586
 587     apr_size_t key_size = strlen(key);
 588
 589     hash = apr_memcache_hash(mc, key, key_size);
 590
 591     ms = apr_memcache_find_server_hash(mc, hash);
 592
 593     if (ms == NULL)
 594         return APR_NOTFOUND;
 595
 596     rv = ms_find_conn(ms, &conn);
 597
 598     if (rv != APR_SUCCESS) {
 599         apr_memcache_disable_server(mc, ms);
 600         return rv;
 601     }
 602
 603     /* <command name> <key> <flags> <exptime> <bytes>\r\n<data>\r\n */
 604
 605     vec[0].iov_base = cmd;
 606     vec[0].iov_len  = cmd_size;
 607
 608     vec[1].iov_base = (void*)key;
 609     vec[1].iov_len  = key_size;
 610
 611     klen = apr_snprintf(conn->buffer, BUFFER_SIZE, " %u %u %" APR_SIZE_T_FMT " " MC_EOL,
 612                         flags, timeout, data_size);
 613
 614     vec[2].iov_base = conn->buffer;
 615     vec[2].iov_len  = klen;
 616
 617     vec[3].iov_base = data;
 618     vec[3].iov_len  = data_size;
 619
 620     vec[4].iov_base = MC_EOL;
 621     vec[4].iov_len  = MC_EOL_LEN;
 622
 623     rv = apr_socket_sendv(conn->sock, vec, 5, &written);
 624
 625     if (rv != APR_SUCCESS) {
 626         ms_bad_conn(ms, conn);
 627         apr_memcache_disable_server(mc, ms);
 628         return rv;
 629     }
 630
 631     rv = get_server_line(conn);
 632
 633     if (rv != APR_SUCCESS) {
 634         ms_bad_conn(ms, conn);
 635         apr_memcache_disable_server(mc, ms);
 636         return rv;
 637     }
 638
 639     if (strcmp(conn->buffer, MS_STORED MC_EOL) == 0) {
 640         rv = APR_SUCCESS;
 641     }
 642     else if (strcmp(conn->buffer, MS_NOT_STORED MC_EOL) == 0) {
 643         rv = APR_EEXIST;
 644     }
 645     else {
 646         rv = APR_EGENERAL;
 647     }
 648
 649     ms_release_conn(ms, conn);
 650
 651     return rv;
 652 }
 653
 654 APU_DECLARE(apr_status_t)
 655 apr_memcache_set(apr_memcache_t *mc,
 656                  const char *key,
 657                  char *data,
 658                  const apr_size_t data_size,
 659                  apr_uint32_t timeout,
 660                  apr_uint16_t flags)
 661 {
 662     return storage_cmd_write(mc,
 663                            MC_SET, MC_SET_LEN,
 664                            key,
 665                            data, data_size,
 666                            timeout, flags);
 667 }
 668
 669 APU_DECLARE(apr_status_t)
 670 apr_memcache_add(apr_memcache_t *mc,
 671                  const char *key,
 672                  char *data,
 673                  const apr_size_t data_size,
 674                  apr_uint32_t timeout,
 675                  apr_uint16_t flags)
 676 {
 677     return storage_cmd_write(mc,
 678                            MC_ADD, MC_ADD_LEN,
 679                            key,
 680                            data, data_size,
 681                            timeout, flags);
 682 }
 683
 684 APU_DECLARE(apr_status_t)
 685 apr_memcache_replace(apr_memcache_t *mc,
 686                  const char *key,
 687                  char *data,
 688                  const apr_size_t data_size,
 689                  apr_uint32_t timeout,
 690                  apr_uint16_t flags)
 691 {
 692     return storage_cmd_write(mc,
 693                            MC_REPLACE, MC_REPLACE_LEN,
 694                            key,
 695                            data, data_size,
 696                            timeout, flags);
 697
 698 }
 699
 700 APU_DECLARE(apr_status_t)
 701 apr_memcache_getp(apr_memcache_t *mc,
 702                   apr_pool_t *p,
 703                   const char *key,
 704                   char **baton,
 705                   apr_size_t *new_length,
 706                   apr_uint16_t *flags_)
 707 {
 708     apr_status_t rv;
 709     apr_memcache_server_t *ms;
 710     apr_memcache_conn_t *conn;
 711     apr_uint32_t hash;
 712     apr_size_t written;
 713     apr_size_t klen = strlen(key);
 714     struct iovec vec[3];
 715
 716     hash = apr_memcache_hash(mc, key, klen);
 717     ms = apr_memcache_find_server_hash(mc, hash);
 718     if (ms == NULL)
 719         return APR_NOTFOUND;
 720
 721     rv = ms_find_conn(ms, &conn);
 722
 723     if (rv != APR_SUCCESS) {
 724         apr_memcache_disable_server(mc, ms);
 725         return rv;
 726     }
 727
 728     /* get <key>[ <key>[...]]\r\n */
 729     vec[0].iov_base = MC_GET;
 730     vec[0].iov_len  = MC_GET_LEN;
 731
 732     vec[1].iov_base = (void*)key;
 733     vec[1].iov_len  = klen;
 734
 735     vec[2].iov_base = MC_EOL;
 736     vec[2].iov_len  = MC_EOL_LEN;
 737
 738     rv = apr_socket_sendv(conn->sock, vec, 3, &written);
 739
 740     if (rv != APR_SUCCESS) {
 741         ms_bad_conn(ms, conn);
 742         apr_memcache_disable_server(mc, ms);
 743         return rv;
 744     }
 745
 746     rv = get_server_line(conn);
 747     if (rv != APR_SUCCESS) {
 748         ms_bad_conn(ms, conn);
 749         apr_memcache_disable_server(mc, ms);
 750         return rv;
 751     }
 752
 753     if (strncmp(MS_VALUE, conn->buffer, MS_VALUE_LEN) == 0) {
 754         char *flags;
 755         char *length;
 756         char *start;
 757         char *last;
 758         apr_size_t len = 0;
 759
 760         start = conn->buffer;
 761         flags = apr_strtok(conn->buffer, " ", &last);
 762         flags = apr_strtok(NULL, " ", &last);
 763         flags = apr_strtok(NULL, " ", &last);
 764
 765         if (flags_) {
 766             *flags_ = atoi(flags);
 767         }
 768
 769         length = apr_strtok(NULL, " ", &last);
 770         if (length) {
 771             len = atoi(length);
 772         }
 773
 774         if (len < 0)  {
 775             *new_length = 0;
 776             *baton = NULL;
 777         }
 778         else {
 779             apr_bucket_brigade *bbb;
 780             apr_bucket *e;
 781
 782             /* eat the trailing \r\n */
 783             rv = apr_brigade_partition(conn->bb, len+2, &e);
 784
 785             if (rv != APR_SUCCESS) {
 786                 ms_bad_conn(ms, conn);
 787                 apr_memcache_disable_server(mc, ms);
 788                 return rv;
 789             }
 790
 791             bbb = apr_brigade_split(conn->bb, e);
 792
 793             rv = apr_brigade_pflatten(conn->bb, baton, &len, p);
 794
 795             if (rv != APR_SUCCESS) {
 796                 ms_bad_conn(ms, conn);
 797                 apr_memcache_disable_server(mc, ms);
 798                 return rv;
 799             }
 800
 801             rv = apr_brigade_destroy(conn->bb);
 802             if (rv != APR_SUCCESS) {
 803                 ms_bad_conn(ms, conn);
 804                 apr_memcache_disable_server(mc, ms);
 805                 return rv;
 806             }
 807
 808             conn->bb = bbb;
 809
 810             *new_length = len - 2;
 811             (*baton)[*new_length] = '\0';
 812         }
 813
 814         rv = get_server_line(conn);
 815         if (rv != APR_SUCCESS) {
 816             ms_bad_conn(ms, conn);
 817             apr_memcache_disable_server(mc, ms);
 818             return rv;
 819         }
 820
 821         if (strncmp(MS_END, conn->buffer, MS_END_LEN) != 0) {
 822             rv = APR_EGENERAL;
 823         }
 824     }
 825     else if (strncmp(MS_END, conn->buffer, MS_END_LEN) == 0) {
 826         rv = APR_NOTFOUND;
 827     }
 828     else {
 829         rv = APR_EGENERAL;
 830     }
 831
 832     ms_release_conn(ms, conn);
 833
 834     return rv;
 835 }
 836
 837 APU_DECLARE(apr_status_t)
 838 apr_memcache_delete(apr_memcache_t *mc,
 839                     const char *key,
 840                     apr_uint32_t timeout)
 841 {
 842     apr_status_t rv;
 843     apr_memcache_server_t *ms;
 844     apr_memcache_conn_t *conn;
 845     apr_uint32_t hash;
 846     apr_size_t written;
 847     struct iovec vec[3];
 848     apr_size_t klen = strlen(key);
 849
 850     hash = apr_memcache_hash(mc, key, klen);
 851     ms = apr_memcache_find_server_hash(mc, hash);
 852     if (ms == NULL)
 853         return APR_NOTFOUND;
 854
 855     rv = ms_find_conn(ms, &conn);
 856
 857     if (rv != APR_SUCCESS) {
 858         apr_memcache_disable_server(mc, ms);
 859         return rv;
 860     }
 861
 862     /* delete <key> <time>\r\n */
 863     vec[0].iov_base = MC_DELETE;
 864     vec[0].iov_len  = MC_DELETE_LEN;
 865
 866     vec[1].iov_base = (void*)key;
 867     vec[1].iov_len  = klen;
 868
 869     klen = apr_snprintf(conn->buffer, BUFFER_SIZE, " %u" MC_EOL, timeout);
 870
 871     vec[2].iov_base = conn->buffer;
 872     vec[2].iov_len  = klen;
 873
 874     rv = apr_socket_sendv(conn->sock, vec, 3, &written);
 875
 876     if (rv != APR_SUCCESS) {
 877         ms_bad_conn(ms, conn);
 878         apr_memcache_disable_server(mc, ms);
 879         return rv;
 880     }
 881
 882     rv = get_server_line(conn);
 883     if (rv != APR_SUCCESS) {
 884         ms_bad_conn(ms, conn);
 885         apr_memcache_disable_server(mc, ms);
 886         return rv;
 887     }
 888
 889     if (strncmp(MS_DELETED, conn->buffer, MS_DELETED_LEN) == 0) {
 890         rv = APR_SUCCESS;
 891     }
 892     else if (strncmp(MS_NOT_FOUND, conn->buffer, MS_NOT_FOUND_LEN) == 0) {
 893         rv = APR_NOTFOUND;
 894     }
 895     else {
 896         rv = APR_EGENERAL;
 897     }
 898
 899     ms_release_conn(ms, conn);
 900
 901     return rv;
 902 }
 903
 904 static apr_status_t num_cmd_write(apr_memcache_t *mc,
 905                                       char *cmd,
 906                                       const apr_uint32_t cmd_size,
 907                                       const char *key,
 908                                       const apr_int32_t inc,
 909                                       apr_uint32_t *new_value)
 910 {
 911     apr_status_t rv;
 912     apr_memcache_server_t *ms;
 913     apr_memcache_conn_t *conn;
 914     apr_uint32_t hash;
 915     apr_size_t written;
 916     struct iovec vec[3];
 917     apr_size_t klen = strlen(key);
 918
 919     hash = apr_memcache_hash(mc, key, klen);
 920     ms = apr_memcache_find_server_hash(mc, hash);
 921     if (ms == NULL)
 922         return APR_NOTFOUND;
 923
 924     rv = ms_find_conn(ms, &conn);
 925
 926     if (rv != APR_SUCCESS) {
 927         apr_memcache_disable_server(mc, ms);
 928         return rv;
 929     }
 930
 931     /* <cmd> <key> <value>\r\n */
 932     vec[0].iov_base = cmd;
 933     vec[0].iov_len  = cmd_size;
 934
 935     vec[1].iov_base = (void*)key;
 936     vec[1].iov_len  = klen;
 937
 938     klen = apr_snprintf(conn->buffer, BUFFER_SIZE, " %u" MC_EOL, inc);
 939
 940     vec[2].iov_base = conn->buffer;
 941     vec[2].iov_len  = klen;
 942
 943     rv = apr_socket_sendv(conn->sock, vec, 3, &written);
 944
 945     if (rv != APR_SUCCESS) {
 946         ms_bad_conn(ms, conn);
 947         apr_memcache_disable_server(mc, ms);
 948         return rv;
 949     }
 950
 951     rv = get_server_line(conn);
 952     if (rv != APR_SUCCESS) {
 953         ms_bad_conn(ms, conn);
 954         apr_memcache_disable_server(mc, ms);
 955         return rv;
 956     }
 957
 958     if (strncmp(MS_ERROR, conn->buffer, MS_ERROR_LEN) == 0) {
 959         rv = APR_EGENERAL;
 960     }
 961     else if (strncmp(MS_NOT_FOUND, conn->buffer, MS_NOT_FOUND_LEN) == 0) {
 962         rv = APR_NOTFOUND;
 963     }
 964     else {
 965         if (new_value) {
 966             *new_value = atoi(conn->buffer);
 967         }
 968         rv = APR_SUCCESS;
 969     }
 970
 971     ms_release_conn(ms, conn);
 972
 973     return rv;
 974 }
 975
 976 APU_DECLARE(apr_status_t)
 977 apr_memcache_incr(apr_memcache_t *mc,
 978                     const char *key,
 979                     apr_int32_t inc,
 980                     apr_uint32_t *new_value)
 981 {
 982     return num_cmd_write(mc,
 983                          MC_INCR,
 984                          MC_INCR_LEN,
 985                          key,
 986                          inc,
 987                          new_value);
 988 }
 989
 990
 991 APU_DECLARE(apr_status_t)
 992 apr_memcache_decr(apr_memcache_t *mc,
 993                     const char *key,
 994                     apr_int32_t inc,
 995                     apr_uint32_t *new_value)
 996 {
 997     return num_cmd_write(mc,
 998                          MC_DECR,
 999                          MC_DECR_LEN,
1000                          key,
1001                          inc,
1002                          new_value);
1003 }
1004
1005
1006
1007 APU_DECLARE(apr_status_t)
1008 apr_memcache_version(apr_memcache_server_t *ms,
1009                   apr_pool_t *p,
1010                   char **baton)
1011 {
1012     apr_status_t rv;
1013     apr_memcache_conn_t *conn;
1014     apr_size_t written;
1015     struct iovec vec[2];
1016
1017     rv = ms_find_conn(ms, &conn);
1018
1019     if (rv != APR_SUCCESS) {
1020         return rv;
1021     }
1022
1023     /* version\r\n */
1024     vec[0].iov_base = MC_VERSION;
1025     vec[0].iov_len  = MC_VERSION_LEN;
1026
1027     vec[1].iov_base = MC_EOL;
1028     vec[1].iov_len  = MC_EOL_LEN;
1029
1030     rv = apr_socket_sendv(conn->sock, vec, 2, &written);
1031
1032     if (rv != APR_SUCCESS) {
1033         ms_bad_conn(ms, conn);
1034         return rv;
1035     }
1036
1037     rv = get_server_line(conn);
1038     if (rv != APR_SUCCESS) {
1039         ms_bad_conn(ms, conn);
1040         return rv;
1041     }
1042
1043     if (strncmp(MS_VERSION, conn->buffer, MS_VERSION_LEN) == 0) {
1044         *baton = apr_pstrmemdup(p, conn->buffer+MS_VERSION_LEN+1,
1045                                 conn->blen - MS_VERSION_LEN - 2);
1046         rv = APR_SUCCESS;
1047     }
1048     else {
1049         rv = APR_EGENERAL;
1050     }
1051
1052     ms_release_conn(ms, conn);
1053
1054     return rv;
1055 }
1056
1057 apr_status_t mc_version_ping(apr_memcache_server_t *ms)
1058 {
1059     apr_status_t rv;
1060     apr_size_t written;
1061     struct iovec vec[2];
1062     apr_memcache_conn_t *conn;
1063
1064     rv = ms_find_conn(ms, &conn);
1065
1066     if (rv != APR_SUCCESS) {
1067         return rv;
1068     }
1069
1070     /* version\r\n */
1071     vec[0].iov_base = MC_VERSION;
1072     vec[0].iov_len  = MC_VERSION_LEN;
1073
1074     vec[1].iov_base = MC_EOL;
1075     vec[1].iov_len  = MC_EOL_LEN;
1076
1077     rv = apr_socket_sendv(conn->sock, vec, 2, &written);
1078
1079     if (rv != APR_SUCCESS) {
1080         ms_bad_conn(ms, conn);
1081         return rv;
1082     }
1083
1084     rv = get_server_line(conn);
1085     ms_release_conn(ms, conn);
1086     return rv;
1087 }
1088
1089
1090 APU_DECLARE(void)
1091 apr_memcache_add_multget_key(apr_pool_t *data_pool,
1092                              const char* key,
1093                              apr_hash_t **values)
1094 {
1095     apr_memcache_value_t* value;
1096     apr_size_t klen = strlen(key);
1097
1098     /* create the value hash if need be */
1099     if (!*values) {
1100         *values = apr_hash_make(data_pool);
1101     }
1102
1103     /* init key and add it to the value hash */
1104     value = apr_pcalloc(data_pool, sizeof(apr_memcache_value_t));
1105
1106     value->status = APR_NOTFOUND;
1107     value->key = apr_pstrdup(data_pool, key);
1108
1109     apr_hash_set(*values, value->key, klen, value);
1110 }
1111
1112 static void mget_conn_result(int up,
1113                              apr_status_t rv,
1114                              apr_memcache_t *mc,
1115                              apr_memcache_server_t *ms,
1116                              apr_memcache_conn_t *conn,
1117                              struct cache_server_query_t *server_query,
1118                              apr_hash_t *values,
1119                              apr_hash_t *server_queries)
1120 {
1121     apr_int32_t j;
1122     apr_memcache_value_t* value;
1123
1124     if (!up) {
1125         ms_bad_conn(ms, conn);
1126         apr_memcache_disable_server(mc, ms);
1127     }
1128
1129     for (j = 1; j < server_query->query_vec_count ; j+=2) {
1130         if (server_query->query_vec[j].iov_base) {
1131             value = apr_hash_get(values, server_query->query_vec[j].iov_base,
1132                                  strlen(server_query->query_vec[j].iov_base));
1133
1134             if (value->status == APR_NOTFOUND) {
1135                 value->status = rv;
1136             }
1137         }
1138     }
1139
1140     ms_release_conn(ms, conn);
1141
1142     apr_hash_set(server_queries, &ms, sizeof(ms), NULL);
1143 }
1144
1145 APU_DECLARE(apr_status_t)
1146 apr_memcache_multgetp(apr_memcache_t *mc,
1147                       apr_pool_t *temp_pool,
1148                       apr_pool_t *data_pool,
1149                       apr_hash_t *values)
1150 {
1151     apr_status_t rv;
1152     apr_memcache_server_t* ms;
1153     apr_memcache_conn_t* conn;
1154     apr_uint32_t hash;
1155     apr_size_t written;
1156     apr_size_t klen;
1157
1158     apr_memcache_value_t* value;
1159     apr_hash_index_t* value_hash_index;
1160
1161     /* this is a little over aggresive, but beats multiple loops
1162      * to figure out how long each vector needs to be per-server.
1163      */
1164     apr_int32_t veclen = 2 + 2 * apr_hash_count(values) - 1; /* get <key>[<space><key>...]\r\n */
1165     apr_int32_t i, j;
1166     apr_int32_t queries_sent;
1167     apr_int32_t queries_recvd;
1168
1169     apr_hash_t * server_queries = apr_hash_make(temp_pool);
1170     struct cache_server_query_t* server_query;
1171     apr_hash_index_t * query_hash_index;
1172
1173     apr_pollset_t* pollset;
1174     const apr_pollfd_t* activefds;
1175     apr_pollfd_t* pollfds;
1176
1177
1178     /* build all the queries */
1179     value_hash_index = apr_hash_first(temp_pool, values);
1180     while (value_hash_index) {
1181         void *v;
1182         apr_hash_this(value_hash_index, NULL, NULL, &v);
1183         value = v;
1184         value_hash_index = apr_hash_next(value_hash_index);
1185         klen = strlen(value->key);
1186
1187         hash = apr_memcache_hash(mc, value->key, klen);
1188         ms = apr_memcache_find_server_hash(mc, hash);
1189         if (ms == NULL) {
1190             continue;
1191         }
1192
1193         server_query = apr_hash_get(server_queries, &ms, sizeof(ms));
1194
1195         if (!server_query) {
1196             rv = ms_find_conn(ms, &conn);
1197
1198             if (rv != APR_SUCCESS) {
1199                 apr_memcache_disable_server(mc, ms);
1200                 value->status = rv;
1201                 continue;
1202             }
1203
1204             server_query = apr_pcalloc(temp_pool,sizeof(struct cache_server_query_t));
1205
1206             apr_hash_set(server_queries, &ms, sizeof(ms), server_query);
1207
1208             server_query->ms = ms;
1209             server_query->conn = conn;
1210             server_query->query_vec = apr_pcalloc(temp_pool, sizeof(struct iovec)*veclen);
1211
1212             /* set up the first key */
1213             server_query->query_vec[0].iov_base = MC_GET;
1214             server_query->query_vec[0].iov_len  = MC_GET_LEN;
1215
1216             server_query->query_vec[1].iov_base = (void*)(value->key);
1217             server_query->query_vec[1].iov_len  = klen;
1218
1219             server_query->query_vec[2].iov_base = MC_EOL;
1220             server_query->query_vec[2].iov_len  = MC_EOL_LEN;
1221
1222             server_query->query_vec_count = 3;
1223         }
1224         else {
1225             j = server_query->query_vec_count - 1;
1226
1227             server_query->query_vec[j].iov_base = MC_WS;
1228             server_query->query_vec[j].iov_len  = MC_WS_LEN;
1229             j++;
1230
1231             server_query->query_vec[j].iov_base = (void*)(value->key);
1232             server_query->query_vec[j].iov_len  = klen;
1233             j++;
1234
1235             server_query->query_vec[j].iov_base = MC_EOL;
1236             server_query->query_vec[j].iov_len  = MC_EOL_LEN;
1237             j++;
1238
1239            server_query->query_vec_count = j;
1240         }
1241     }
1242
1243     /* create polling structures */
1244     pollfds = apr_pcalloc(temp_pool, apr_hash_count(server_queries) * sizeof(apr_pollfd_t));
1245
1246     rv = apr_pollset_create(&pollset, apr_hash_count(server_queries), temp_pool, 0);
1247
1248     if (rv != APR_SUCCESS) {
1249         return rv;
1250     }
1251
1252     /* send all the queries */
1253     queries_sent = 0;
1254     query_hash_index = apr_hash_first(temp_pool, server_queries);
1255
1256     while (query_hash_index) {
1257         void *v;
1258         apr_hash_this(query_hash_index, NULL, NULL, &v);
1259         server_query = v;
1260         query_hash_index = apr_hash_next(query_hash_index);
1261
1262         conn = server_query->conn;
1263         ms = server_query->ms;
1264
1265         for (i = 0, rv = APR_SUCCESS; i < veclen && rv == APR_SUCCESS; i += APR_MAX_IOVEC_SIZE) {
1266             rv = apr_socket_sendv(conn->sock, &(server_query->query_vec[i]),
1267                                   veclen-i>APR_MAX_IOVEC_SIZE ? APR_MAX_IOVEC_SIZE : veclen-i , &written);
1268         }
1269
1270         if (rv != APR_SUCCESS) {
1271             mget_conn_result(FALSE, rv, mc, ms, conn,
1272                              server_query, values, server_queries);
1273             continue;
1274         }
1275
1276         pollfds[queries_sent].desc_type = APR_POLL_SOCKET;
1277         pollfds[queries_sent].reqevents = APR_POLLIN;
1278         pollfds[queries_sent].p = temp_pool;
1279         pollfds[queries_sent].desc.s = conn->sock;
1280         pollfds[queries_sent].client_data = (void *)server_query;
1281         apr_pollset_add (pollset, &pollfds[queries_sent]);
1282
1283         queries_sent++;
1284     }
1285
1286     while (queries_sent) {
1287         rv = apr_pollset_poll(pollset, MULT_GET_TIMEOUT, &queries_recvd, &activefds);
1288
1289         if (rv != APR_SUCCESS) {
1290             /* timeout */
1291             queries_sent = 0;
1292             continue;
1293         }
1294         for (i = 0; i < queries_recvd; i++) {
1295             server_query = activefds[i].client_data;
1296             conn = server_query->conn;
1297             ms = server_query->ms;
1298
1299            rv = get_server_line(conn);
1300
1301            if (rv != APR_SUCCESS) {
1302                apr_pollset_remove (pollset, &activefds[i]);
1303                mget_conn_result(FALSE, rv, mc, ms, conn,
1304                                 server_query, values, server_queries);
1305                queries_sent--;
1306                continue;
1307            }
1308
1309            if (strncmp(MS_VALUE, conn->buffer, MS_VALUE_LEN) == 0) {
1310                char *key;
1311                char *flags;
1312                char *length;
1313                char *start;
1314                char *last;
1315                char *data;
1316                apr_size_t len = 0;
1317
1318                start = conn->buffer;
1319                key = apr_strtok(conn->buffer, " ", &last); /* just the VALUE, ignore */
1320                key = apr_strtok(NULL, " ", &last);
1321                flags = apr_strtok(NULL, " ", &last);
1322
1323
1324                length = apr_strtok(NULL, " ", &last);
1325                if (length) {
1326                    len = atoi(length);
1327                }
1328
1329                value = apr_hash_get(values, key, strlen(key));
1330
1331
1332                if (value) {
1333                    if (len > 0)  {
1334                        apr_bucket_brigade *bbb;
1335                        apr_bucket *e;
1336
1337                        /* eat the trailing \r\n */
1338                        rv = apr_brigade_partition(conn->bb, len+2, &e);
1339
1340                        if (rv != APR_SUCCESS) {
1341                            apr_pollset_remove (pollset, &activefds[i]);
1342                            mget_conn_result(FALSE, rv, mc, ms, conn,
1343                                             server_query, values, server_queries);
1344                            queries_sent--;
1345                            continue;
1346                        }
1347
1348                        bbb = apr_brigade_split(conn->bb, e);
1349
1350                        rv = apr_brigade_pflatten(conn->bb, &data, &len, data_pool);
1351
1352                        if (rv != APR_SUCCESS) {
1353                            apr_pollset_remove (pollset, &activefds[i]);
1354                            mget_conn_result(FALSE, rv, mc, ms, conn,
1355                                             server_query, values, server_queries);
1356                            queries_sent--;
1357                            continue;
1358                        }
1359
1360                        rv = apr_brigade_destroy(conn->bb);
1361                        if (rv != APR_SUCCESS) {
1362                            apr_pollset_remove (pollset, &activefds[i]);
1363                            mget_conn_result(FALSE, rv, mc, ms, conn,
1364                                             server_query, values, server_queries);
1365                            queries_sent--;
1366                            continue;
1367                        }
1368
1369                        conn->bb = bbb;
1370
1371                        value->len = len - 2;
1372                        data[value->len] = '\0';
1373                        value->data = data;
1374                    }
1375
1376                    value->status = rv;
1377                    value->flags = atoi(flags);
1378
1379                    /* stay on the server */
1380                    i--;
1381
1382                }
1383                else {
1384                    /* TODO: Server Sent back a key I didn't ask for or my
1385                     *       hash is corrupt */
1386                }
1387            }
1388            else if (strncmp(MS_END, conn->buffer, MS_END_LEN) == 0) {
1389                /* this connection is done */
1390                apr_pollset_remove (pollset, &activefds[i]);
1391                ms_release_conn(ms, conn);
1392                apr_hash_set(server_queries, &ms, sizeof(ms), NULL);
1393
1394                queries_sent--;
1395            }
1396            else {
1397                /* unknown reply? */
1398                rv = APR_EGENERAL;
1399            }
1400
1401         } /* /for */
1402     } /* /while */
1403
1404     query_hash_index = apr_hash_first(temp_pool, server_queries);
1405     while (query_hash_index) {
1406         void *v;
1407         apr_hash_this(query_hash_index, NULL, NULL, &v);
1408         server_query = v;
1409         query_hash_index = apr_hash_next(query_hash_index);
1410
1411         conn = server_query->conn;
1412         ms = server_query->ms;
1413
1414         mget_conn_result(TRUE, rv, mc, ms, conn,
1415                          server_query, values, server_queries);
1416         continue;
1417     }
1418
1419     apr_pool_clear(temp_pool);
1420     apr_pollset_destroy(pollset);
1421     return APR_SUCCESS;
1422
1423 }
1424
1425
1426
1427 /**
1428  * Define all of the strings for stats
1429  */
1430
1431 #define STAT_pid MS_STAT " pid "
1432 #define STAT_pid_LEN (sizeof(STAT_pid)-1)
1433
1434 #define STAT_uptime MS_STAT " uptime "
1435 #define STAT_uptime_LEN (sizeof(STAT_uptime)-1)
1436
1437 #define STAT_time MS_STAT " time "
1438 #define STAT_time_LEN (sizeof(STAT_time)-1)
1439
1440 #define STAT_version MS_STAT " version "
1441 #define STAT_version_LEN (sizeof(STAT_version)-1)
1442
1443 #define STAT_pointer_size MS_STAT " pointer_size "
1444 #define STAT_pointer_size_LEN (sizeof(STAT_pointer_size)-1)
1445
1446 #define STAT_rusage_user MS_STAT " rusage_user "
1447 #define STAT_rusage_user_LEN (sizeof(STAT_rusage_user)-1)
1448
1449 #define STAT_rusage_system MS_STAT " rusage_system "
1450 #define STAT_rusage_system_LEN (sizeof(STAT_rusage_system)-1)
1451
1452 #define STAT_curr_items MS_STAT " curr_items "
1453 #define STAT_curr_items_LEN (sizeof(STAT_curr_items)-1)
1454
1455 #define STAT_total_items MS_STAT " total_items "
1456 #define STAT_total_items_LEN (sizeof(STAT_total_items)-1)
1457
1458 #define STAT_bytes MS_STAT " bytes "
1459 #define STAT_bytes_LEN (sizeof(STAT_bytes)-1)
1460
1461 #define STAT_curr_connections MS_STAT " curr_connections "
1462 #define STAT_curr_connections_LEN (sizeof(STAT_curr_connections)-1)
1463
1464 #define STAT_total_connections MS_STAT " total_connections "
1465 #define STAT_total_connections_LEN (sizeof(STAT_total_connections)-1)
1466
1467 #define STAT_connection_structures MS_STAT " connection_structures "
1468 #define STAT_connection_structures_LEN (sizeof(STAT_connection_structures)-1)
1469
1470 #define STAT_cmd_get MS_STAT " cmd_get "
1471 #define STAT_cmd_get_LEN (sizeof(STAT_cmd_get)-1)
1472
1473 #define STAT_cmd_set MS_STAT " cmd_set "
1474 #define STAT_cmd_set_LEN (sizeof(STAT_cmd_set)-1)
1475
1476 #define STAT_get_hits MS_STAT " get_hits "
1477 #define STAT_get_hits_LEN (sizeof(STAT_get_hits)-1)
1478
1479 #define STAT_get_misses MS_STAT " get_misses "
1480 #define STAT_get_misses_LEN (sizeof(STAT_get_misses)-1)
1481
1482 #define STAT_evictions MS_STAT " evictions "
1483 #define STAT_evictions_LEN (sizeof(STAT_evictions)-1)
1484
1485 #define STAT_bytes_read MS_STAT " bytes_read "
1486 #define STAT_bytes_read_LEN (sizeof(STAT_bytes_read)-1)
1487
1488 #define STAT_bytes_written MS_STAT " bytes_written "
1489 #define STAT_bytes_written_LEN (sizeof(STAT_bytes_written)-1)
1490
1491 #define STAT_limit_maxbytes MS_STAT " limit_maxbytes "
1492 #define STAT_limit_maxbytes_LEN (sizeof(STAT_limit_maxbytes)-1)
1493
1494 #define STAT_threads MS_STAT " threads "
1495 #define STAT_threads_LEN (sizeof(STAT_threads)-1)
1496
1497 static const char *stat_read_string(apr_pool_t *p, char *buf, apr_size_t len)
1498 {
1499     /* remove trailing \r\n and null char */
1500     return apr_pstrmemdup(p, buf, len-2);
1501 }
1502
1503 static apr_uint32_t stat_read_uint32(apr_pool_t *p, char *buf, apr_size_t  len)
1504 {
1505     buf[len-2] = '\0';
1506     return atoi(buf);
1507 }
1508
1509 static apr_uint64_t stat_read_uint64(apr_pool_t *p, char *buf, apr_size_t  len)
1510 {
1511     buf[len-2] = '\0';
1512     return apr_atoi64(buf);
1513 }
1514
1515 static apr_time_t stat_read_time(apr_pool_t *p, char *buf, apr_size_t  len)
1516 {
1517     buf[len-2] = '\0';
1518     return apr_time_from_sec(atoi(buf));
1519 }
1520
1521 static apr_time_t stat_read_rtime(apr_pool_t *p, char *buf, apr_size_t  len)
1522 {
1523     char *tok;
1524     char *secs;
1525     char *usecs;
1526     const char *sep = ":.";
1527
1528     buf[len-2] = '\0';
1529
1530     secs = apr_strtok(buf, sep, &tok);
1531     usecs = apr_strtok(NULL, sep, &tok);
1532     if (secs && usecs) {
1533         return apr_time_make(atoi(secs), atoi(usecs));
1534     }
1535     else {
1536         return apr_time_make(0, 0);
1537     }
1538 }
1539
1540 /**
1541  * I got tired of Typing. Meh.
1542  *
1543  * TODO: Convert it to static tables to make it cooler.
1544  */
1545
1546 #define mc_stat_cmp(name) \
1547     strncmp(STAT_ ## name, conn->buffer, STAT_ ## name ## _LEN) == 0
1548
1549 #define mc_stat_str(name) \
1550     stat_read_string(p, conn->buffer + name, \
1551                      conn->blen - name)
1552
1553 #define mc_stat_uint32(name) \
1554     stat_read_uint32(p, conn->buffer + name, \
1555                      conn->blen - name)
1556
1557 #define mc_stat_uint64(name) \
1558     stat_read_uint64(p, conn->buffer + name, \
1559                      conn->blen - name)
1560
1561 #define mc_stat_time(name) \
1562     stat_read_time(p, conn->buffer + name, \
1563                      conn->blen - name)
1564
1565 #define mc_stat_rtime(name) \
1566     stat_read_rtime(p, conn->buffer + name, \
1567                      conn->blen - name)
1568
1569
1570 #define mc_do_stat(name, type) \
1571     if (mc_stat_cmp(name)) { \
1572         stats-> name = mc_stat_ ## type ((STAT_ ## name ## _LEN)); \
1573     }
1574
1575 static void update_stats(apr_pool_t *p, apr_memcache_conn_t *conn,
1576                          apr_memcache_stats_t *stats)
1577 {
1578
1579     mc_do_stat(version, str)
1580     else mc_do_stat(pid, uint32)
1581     else mc_do_stat(uptime, uint32)
1582     else mc_do_stat(pointer_size, uint32)
1583     else mc_do_stat(time, time)
1584     else mc_do_stat(rusage_user, rtime)
1585     else mc_do_stat(rusage_system, rtime)
1586     else mc_do_stat(curr_items, uint32)
1587     else mc_do_stat(total_items, uint32)
1588     else mc_do_stat(bytes, uint64)
1589     else mc_do_stat(curr_connections, uint32)
1590     else mc_do_stat(total_connections, uint32)
1591     else mc_do_stat(connection_structures, uint32)
1592     else mc_do_stat(cmd_get, uint32)
1593     else mc_do_stat(cmd_set, uint32)
1594     else mc_do_stat(get_hits, uint32)
1595     else mc_do_stat(get_misses, uint32)
1596     else mc_do_stat(evictions, uint64)
1597     else mc_do_stat(bytes_read, uint64)
1598     else mc_do_stat(bytes_written, uint64)
1599     else mc_do_stat(limit_maxbytes, uint32)
1600     else mc_do_stat(threads, uint32)
1601 }
1602
1603 APU_DECLARE(apr_status_t)
1604 apr_memcache_stats(apr_memcache_server_t *ms,
1605                   apr_pool_t *p,
1606                   apr_memcache_stats_t **stats)
1607 {
1608     apr_memcache_stats_t *ret;
1609     apr_status_t rv;
1610     apr_memcache_conn_t *conn;
1611     apr_size_t written;
1612     struct iovec vec[2];
1613
1614     rv = ms_find_conn(ms, &conn);
1615
1616     if (rv != APR_SUCCESS) {
1617         return rv;
1618     }
1619
1620     /* version\r\n */
1621     vec[0].iov_base = MC_STATS;
1622     vec[0].iov_len  = MC_STATS_LEN;
1623
1624     vec[1].iov_base = MC_EOL;
1625     vec[1].iov_len  = MC_EOL_LEN;
1626
1627     rv = apr_socket_sendv(conn->sock, vec, 2, &written);
1628
1629     if (rv != APR_SUCCESS) {
1630         ms_bad_conn(ms, conn);
1631         return rv;
1632     }
1633
1634     ret = apr_pcalloc(p, sizeof(apr_memcache_stats_t));
1635
1636     do {
1637         rv = get_server_line(conn);
1638         if (rv != APR_SUCCESS) {
1639             ms_bad_conn(ms, conn);
1640             return rv;
1641         }
1642
1643         if (strncmp(MS_END, conn->buffer, MS_END_LEN) == 0) {
1644             rv = APR_SUCCESS;
1645             break;
1646         }
1647         else if (strncmp(MS_STAT, conn->buffer, MS_STAT_LEN) == 0) {
1648             update_stats(p, conn, ret);
1649             continue;
1650         }
1651         else {
1652             rv = APR_EGENERAL;
1653             break;
1654         }
1655
1656     } while(1);
1657
1658     ms_release_conn(ms, conn);
1659
1660     if (stats) {
1661         *stats = ret;
1662     }
1663
1664     return rv;
1665 }
1666