1 /* -*- Mode: C; tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*- */ 2 /* 3 * memcached - memory caching daemon 4 * 5 * http://www.danga.com/memcached/ 6 * 7 * Copyright 2003 Danga Interactive, Inc. All rights reserved. 8 * 9 * Use and distribution licensed under the BSD license. See 10 * the LICENSE file for full text. 11 * 12 * Authors: 13 * Anatoly Vorobey <[email protected]> 14 * Brad Fitzpatrick <[email protected]> 15 */ 16 #include "memcached.h" 17 #include <sys/stat.h> 18 #include <sys/socket.h> 19 #include <sys/un.h> 20 #include <signal.h> 21 #include <sys/resource.h> 22 #include <sys/uio.h> 23 #include <ctype.h> 24 #include <stdarg.h> 25 26 /* some POSIX systems need the following definition 27 * to get mlockall flags out of sys/mman.h. */ 28 #ifndef _P1003_1B_VISIBLE 29 #define _P1003_1B_VISIBLE 30 #endif 31 /* need this to get IOV_MAX on some platforms. */ 32 #ifndef __need_IOV_MAX 33 #define __need_IOV_MAX 34 #endif 35 #include <pwd.h> 36 #include <sys/mman.h> 37 #include <fcntl.h> 38 #include <netinet/tcp.h> 39 #include <arpa/inet.h> 40 #include <errno.h> 41 #include <stdlib.h> 42 #include <stdio.h> 43 #include <string.h> 44 #include <time.h> 45 #include <assert.h> 46 #include <limits.h> 47 #include <sysexits.h> 48 #include <stddef.h> 49 50 /* FreeBSD 4.x doesn't have IOV_MAX exposed. */ 51 #ifndef IOV_MAX 52 #if defined(__FreeBSD__) || defined(__APPLE__) 53 # define IOV_MAX 1024 54 #endif 55 #endif 56 57 /* 58 * forward declarations 59 */ 60 static void drive_machine(conn *c); 61 static int new_socket(struct addrinfo *ai); 62 static int try_read_command(conn *c); 63 64 enum try_read_result { 65 READ_DATA_RECEIVED, 66 READ_NO_DATA_RECEIVED, 67 READ_ERROR, /** an error occured (on the socket) (or client closed connection) */ 68 READ_MEMORY_ERROR /** failed to allocate more memory */ 69 }; 70 71 static enum try_read_result try_read_network(conn *c); 72 static enum try_read_result try_read_udp(conn *c); 73 74 static void conn_set_state(conn *c, enum conn_states state); 75 76 /* stats */ 77 static void stats_init(void); 78 static void server_stats(ADD_STAT add_stats, conn *c); 79 static void process_stat_settings(ADD_STAT add_stats, void *c); 80 81 82 /* defaults */ 83 static void settings_init(void); 84 85 /* event handling, network IO */ 86 static void event_handler(const int fd, const short which, void *arg); 87 static void conn_close(conn *c); 88 static void conn_init(void); 89 static bool update_event(conn *c, const int new_flags); 90 static void complete_nread(conn *c); 91 static void process_command(conn *c, char *command); 92 static void write_and_free(conn *c, char *buf, int bytes); 93 static int ensure_iov_space(conn *c); 94 static int add_iov(conn *c, const void *buf, int len); 95 static int add_msghdr(conn *c); 96 97 /* time handling */ 98 static void set_current_time(void); /* update the global variable holding 99 global 32-bit seconds-since-start time 100 (to avoid 64 bit time_t) */ 101 102 static void conn_free(conn *c); 103 104 /** exported globals **/ 105 struct stats stats; 106 struct settings settings; 107 time_t process_started; /* when the process was started */ 108 109 /** file scope variables **/ 110 static conn *listen_conn = NULL; 111 static struct event_base *main_base; 112 113 enum transmit_result { 114 TRANSMIT_COMPLETE, /** All done writing. */ 115 TRANSMIT_INCOMPLETE, /** More data remaining to write. */ 116 TRANSMIT_SOFT_ERROR, /** Can't write any more right now. */ 117 TRANSMIT_HARD_ERROR /** Can't write (c->state is set to conn_closing) */ 118 }; 119 120 static enum transmit_result transmit(conn *c); 121 122 #define REALTIME_MAXDELTA 60*60*24*30 123 124 /* 125 * given time value that's either unix time or delta from current unix time, return 126 * unix time. Use the fact that delta can't exceed one month (and real time value can't 127 * be that low). 128 */ 129 static rel_time_t realtime(const time_t exptime) { 130 /* no. of seconds in 30 days - largest possible delta exptime */ 131 132 if (exptime == 0) return 0; /* 0 means never expire */ 133 134 if (exptime > REALTIME_MAXDELTA) { 135 /* if item expiration is at/before the server started, give it an 136 expiration time of 1 second after the server started. 137 (because 0 means don't expire). without this, we'd 138 underflow and wrap around to some large value way in the 139 future, effectively making items expiring in the past 140 really expiring never */ 141 if (exptime <= process_started) 142 return (rel_time_t)1; 143 return (rel_time_t)(exptime - process_started); 144 } else { 145 return (rel_time_t)(exptime + current_time); 146 } 147 } 148 149 static void stats_init(void) { 150 stats.curr_items = stats.total_items = stats.curr_conns = stats.total_conns = stats.conn_structs = 0; 151 stats.get_cmds = stats.set_cmds = stats.get_hits = stats.get_misses = stats.evictions = 0; 152 stats.curr_bytes = stats.listen_disabled_num = 0; 153 stats.accepting_conns = true; /* assuming we start in this state. */ 154 155 /* make the time we started always be 2 seconds before we really 156 did, so time(0) - time.started is never zero. if so, things 157 like 'settings.oldest_live' which act as booleans as well as 158 values are now false in boolean context... */ 159 process_started = time(0) - 2; 160 stats_prefix_init(); 161 } 162 163 static void stats_reset(void) { 164 STATS_LOCK(); 165 stats.total_items = stats.total_conns = 0; 166 stats.evictions = 0; 167 stats.listen_disabled_num = 0; 168 stats_prefix_clear(); 169 STATS_UNLOCK(); 170 threadlocal_stats_reset(); 171 item_stats_reset(); 172 } 173 174 static void settings_init(void) { 175 settings.use_cas = true; 176 settings.access = 0700; 177 settings.port = 11211; 178 settings.udpport = 11211; 179 /* By default this string should be NULL for getaddrinfo() */ 180 settings.inter = NULL; 181 settings.maxbytes = 64 * 1024 * 1024; /* default is 64MB */ 182 settings.maxconns = 1024; /* to limit connections-related memory to about 5MB */ 183 settings.verbose = 0; 184 settings.oldest_live = 0; 185 settings.evict_to_free = 1; /* push old items out of cache when memory runs out */ 186 settings.socketpath = NULL; /* by default, not using a unix socket */ 187 settings.factor = 1.25; 188 settings.chunk_size = 48; /* space for a modest key and value */ 189 settings.num_threads = 4; /* N workers */ 190 settings.prefix_delimiter = ':'; 191 settings.detail_enabled = 0; 192 settings.reqs_per_event = 20; 193 settings.backlog = 1024; 194 settings.binding_protocol = negotiating_prot; 195 settings.item_size_max = 1024 * 1024; /* The famous 1MB upper limit. */ 196 } 197 198 /* 199 * Adds a message header to a connection. 200 * 201 * Returns 0 on success, -1 on out-of-memory. 202 */ 203 static int add_msghdr(conn *c) 204 { 205 struct msghdr *msg; 206 207 assert(c != NULL); 208 209 if (c->msgsize == c->msgused) { 210 msg = realloc(c->msglist, c->msgsize * 2 * sizeof(struct msghdr)); 211 if (! msg) 212 return -1; 213 c->msglist = msg; 214 c->msgsize *= 2; 215 } 216 217 msg = c->msglist + c->msgused; 218 219 /* this wipes msg_iovlen, msg_control, msg_controllen, and 220 msg_flags, the last 3 of which aren't defined on solaris: */ 221 memset(msg, 0, sizeof(struct msghdr)); 222 223 msg->msg_iov = &c->iov[c->iovused]; 224 225 if (c->request_addr_size > 0) { 226 msg->msg_name = &c->request_addr; 227 msg->msg_namelen = c->request_addr_size; 228 } 229 230 c->msgbytes = 0; 231 c->msgused++; 232 233 if (IS_UDP(c->transport)) { 234 /* Leave room for the UDP header, which we'll fill in later. */ 235 return add_iov(c, NULL, UDP_HEADER_SIZE); 236 } 237 238 return 0; 239 } 240 241 242 /* 243 * Free list management for connections. 244 */ 245 246 static conn **freeconns; 247 static int freetotal; 248 static int freecurr; 249 /* Lock for connection freelist */ 250 static pthread_mutex_t conn_lock = PTHREAD_MUTEX_INITIALIZER; 251 252 253 static void conn_init(void) { 254 freetotal = 200; 255 freecurr = 0; 256 if ((freeconns = calloc(freetotal, sizeof(conn *))) == NULL) { 257 fprintf(stderr, "Failed to allocate connection structures\n"); 258 } 259 return; 260 } 261 262 /* 263 * Returns a connection from the freelist, if any. 264 */ 265 conn *conn_from_freelist() { 266 conn *c; 267 268 pthread_mutex_lock(&conn_lock); 269 if (freecurr > 0) { 270 c = freeconns[--freecurr]; 271 } else { 272 c = NULL; 273 } 274 pthread_mutex_unlock(&conn_lock); 275 276 return c; 277 } 278 279 /* 280 * Adds a connection to the freelist. 0 = success. 281 */ 282 bool conn_add_to_freelist(conn *c) { 283 bool ret = true; 284 pthread_mutex_lock(&conn_lock); 285 if (freecurr < freetotal) { 286 freeconns[freecurr++] = c; 287 ret = false; 288 } else { 289 /* try to enlarge free connections array */ 290 size_t newsize = freetotal * 2; 291 conn **new_freeconns = realloc(freeconns, sizeof(conn *) * newsize); 292 if (new_freeconns) { 293 freetotal = newsize; 294 freeconns = new_freeconns; 295 freeconns[freecurr++] = c; 296 ret = false; 297 } 298 } 299 pthread_mutex_unlock(&conn_lock); 300 return ret; 301 } 302 303 static const char *prot_text(enum protocol prot) { 304 char *rv = "unknown"; 305 switch(prot) { 306 case ascii_prot: 307 rv = "ascii"; 308 break; 309 case binary_prot: 310 rv = "binary"; 311 break; 312 case negotiating_prot: 313 rv = "auto-negotiate"; 314 break; 315 } 316 return rv; 317 } 318 319 conn *conn_new(const int sfd, enum conn_states init_state, 320 const int event_flags, 321 const int read_buffer_size, enum network_transport transport, 322 struct event_base *base) { 323 conn *c = conn_from_freelist(); 324 325 if (NULL == c) { 326 if (!(c = (conn *)calloc(1, sizeof(conn)))) { 327 fprintf(stderr, "calloc()\n"); 328 return NULL; 329 } 330 MEMCACHED_CONN_CREATE(c); 331 332 c->rbuf = c->wbuf = 0; 333 c->ilist = 0; 334 c->suffixlist = 0; 335 c->iov = 0; 336 c->msglist = 0; 337 c->hdrbuf = 0; 338 339 c->rsize = read_buffer_size; 340 c->wsize = DATA_BUFFER_SIZE; 341 c->isize = ITEM_LIST_INITIAL; 342 c->suffixsize = SUFFIX_LIST_INITIAL; 343 c->iovsize = IOV_LIST_INITIAL; 344 c->msgsize = MSG_LIST_INITIAL; 345 c->hdrsize = 0; 346 347 c->rbuf = (char *)malloc((size_t)c->rsize); 348 c->wbuf = (char *)malloc((size_t)c->wsize); 349 c->ilist = (item **)malloc(sizeof(item *) * c->isize); 350 c->suffixlist = (char **)malloc(sizeof(char *) * c->suffixsize); 351 c->iov = (struct iovec *)malloc(sizeof(struct iovec) * c->iovsize); 352 c->msglist = (struct msghdr *)malloc(sizeof(struct msghdr) * c->msgsize); 353 354 if (c->rbuf == 0 || c->wbuf == 0 || c->ilist == 0 || c->iov == 0 || 355 c->msglist == 0 || c->suffixlist == 0) { 356 conn_free(c); 357 fprintf(stderr, "malloc()\n"); 358 return NULL; 359 } 360 361 STATS_LOCK(); 362 stats.conn_structs++; 363 STATS_UNLOCK(); 364 } 365 366 c->transport = transport; 367 c->protocol = settings.binding_protocol; 368 369 /* unix socket mode doesn't need this, so zeroed out. but why 370 * is this done for every command? presumably for UDP 371 * mode. */ 372 if (!settings.socketpath) { 373 c->request_addr_size = sizeof(c->request_addr); 374 } else { 375 c->request_addr_size = 0; 376 } 377 378 if (settings.verbose > 1) { 379 if (init_state == conn_listening) { 380 fprintf(stderr, "<%d server listening (%s)\n", sfd, 381 prot_text(c->protocol)); 382 } else if (IS_UDP(transport)) { 383 fprintf(stderr, "<%d server listening (udp)\n", sfd); 384 } else if (c->protocol == negotiating_prot) { 385 fprintf(stderr, "<%d new auto-negotiating client connection\n", 386 sfd); 387 } else if (c->protocol == ascii_prot) { 388 fprintf(stderr, "<%d new ascii client connection.\n", sfd); 389 } else if (c->protocol == binary_prot) { 390 fprintf(stderr, "<%d new binary client connection.\n", sfd); 391 } else { 392 fprintf(stderr, "<%d new unknown (%d) client connection\n", 393 sfd, c->protocol); 394 assert(false); 395 } 396 } 397 398 c->sfd = sfd; 399 c->state = init_state; 400 c->rlbytes = 0; 401 c->cmd = -1; 402 c->rbytes = c->wbytes = 0; 403 c->wcurr = c->wbuf; 404 c->rcurr = c->rbuf; 405 c->ritem = 0; 406 c->icurr = c->ilist; 407 c->suffixcurr = c->suffixlist; 408 c->ileft = 0; 409 c->suffixleft = 0; 410 c->iovused = 0; 411 c->msgcurr = 0; 412 c->msgused = 0; 413 414 c->write_and_go = init_state; 415 c->write_and_free = 0; 416 c->item = 0; 417 418 c->noreply = false; 419 420 event_set(&c->event, sfd, event_flags, event_handler, (void *)c); 421 event_base_set(base, &c->event); 422 c->ev_flags = event_flags; 423 424 if (event_add(&c->event, 0) == -1) { 425 if (conn_add_to_freelist(c)) { 426 conn_free(c); 427 } 428 perror("event_add"); 429 return NULL; 430 } 431 432 STATS_LOCK(); 433 stats.curr_conns++; 434 stats.total_conns++; 435 STATS_UNLOCK(); 436 437 MEMCACHED_CONN_ALLOCATE(c->sfd); 438 439 return c; 440 } 441 442 static void conn_cleanup(conn *c) { 443 assert(c != NULL); 444 445 if (c->item) { 446 item_remove(c->item); 447 c->item = 0; 448 } 449 450 if (c->ileft != 0) { 451 for (; c->ileft > 0; c->ileft--,c->icurr++) { 452 item_remove(*(c->icurr)); 453 } 454 } 455 456 if (c->suffixleft != 0) { 457 for (; c->suffixleft > 0; c->suffixleft--, c->suffixcurr++) { 458 cache_free(c->thread->suffix_cache, *(c->suffixcurr)); 459 } 460 } 461 462 if (c->write_and_free) { 463 free(c->write_and_free); 464 c->write_and_free = 0; 465 } 466 } 467 468 /* 469 * Frees a connection. 470 */ 471 void conn_free(conn *c) { 472 if (c) { 473 MEMCACHED_CONN_DESTROY(c); 474 if (c->hdrbuf) 475 free(c->hdrbuf); 476 if (c->msglist) 477 free(c->msglist); 478 if (c->rbuf) 479 free(c->rbuf); 480 if (c->wbuf) 481 free(c->wbuf); 482 if (c->ilist) 483 free(c->ilist); 484 if (c->suffixlist) 485 free(c->suffixlist); 486 if (c->iov) 487 free(c->iov); 488 free(c); 489 } 490 } 491 492 static void conn_close(conn *c) { 493 assert(c != NULL); 494 495 /* delete the event, the socket and the conn */ 496 event_del(&c->event); 497 498 if (settings.verbose > 1) 499 fprintf(stderr, "<%d connection closed.\n", c->sfd); 500 501 MEMCACHED_CONN_RELEASE(c->sfd); 502 close(c->sfd); 503 accept_new_conns(true); 504 conn_cleanup(c); 505 506 /* if the connection has big buffers, just free it */ 507 if (c->rsize > READ_BUFFER_HIGHWAT || conn_add_to_freelist(c)) { 508 conn_free(c); 509 } 510 511 STATS_LOCK(); 512 stats.curr_conns--; 513 STATS_UNLOCK(); 514 515 return; 516 } 517 518 /* 519 * Shrinks a connection's buffers if they're too big. This prevents 520 * periodic large "get" requests from permanently chewing lots of server 521 * memory. 522 * 523 * This should only be called in between requests since it can wipe output 524 * buffers! 525 */ 526 static void conn_shrink(conn *c) { 527 assert(c != NULL); 528 529 if (IS_UDP(c->transport)) 530 return; 531 532 if (c->rsize > READ_BUFFER_HIGHWAT && c->rbytes < DATA_BUFFER_SIZE) { 533 char *newbuf; 534 535 if (c->rcurr != c->rbuf) 536 memmove(c->rbuf, c->rcurr, (size_t)c->rbytes); 537 538 newbuf = (char *)realloc((void *)c->rbuf, DATA_BUFFER_SIZE); 539 540 if (newbuf) { 541 c->rbuf = newbuf; 542 c->rsize = DATA_BUFFER_SIZE; 543 } 544 /* TODO check other branch... */ 545 c->rcurr = c->rbuf; 546 } 547 548 if (c->isize > ITEM_LIST_HIGHWAT) { 549 item **newbuf = (item**) realloc((void *)c->ilist, ITEM_LIST_INITIAL * sizeof(c->ilist[0])); 550 if (newbuf) { 551 c->ilist = newbuf; 552 c->isize = ITEM_LIST_INITIAL; 553 } 554 /* TODO check error condition? */ 555 } 556 557 if (c->msgsize > MSG_LIST_HIGHWAT) { 558 struct msghdr *newbuf = (struct msghdr *) realloc((void *)c->msglist, MSG_LIST_INITIAL * sizeof(c->msglist[0])); 559 if (newbuf) { 560 c->msglist = newbuf; 561 c->msgsize = MSG_LIST_INITIAL; 562 } 563 /* TODO check error condition? */ 564 } 565 566 if (c->iovsize > IOV_LIST_HIGHWAT) { 567 struct iovec *newbuf = (struct iovec *) realloc((void *)c->iov, IOV_LIST_INITIAL * sizeof(c->iov[0])); 568 if (newbuf) { 569 c->iov = newbuf; 570 c->iovsize = IOV_LIST_INITIAL; 571 } 572 /* TODO check return value */ 573 } 574 } 575 576 /** 577 * Convert a state name to a human readable form. 578 */ 579 static const char *state_text(enum conn_states state) { 580 const char* const statenames[] = { "conn_listening", 581 "conn_new_cmd", 582 "conn_waiting", 583 "conn_read", 584 "conn_parse_cmd", 585 "conn_write", 586 "conn_nread", 587 "conn_swallow", 588 "conn_closing", 589 "conn_mwrite" }; 590 return statenames[state]; 591 } 592 593 /* 594 * Sets a connection's current state in the state machine. Any special 595 * processing that needs to happen on certain state transitions can 596 * happen here. 597 */ 598 static void conn_set_state(conn *c, enum conn_states state) { 599 assert(c != NULL); 600 assert(state >= conn_listening && state < conn_max_state); 601 602 if (state != c->state) { 603 if (settings.verbose > 2) { 604 fprintf(stderr, "%d: going from %s to %s\n", 605 c->sfd, state_text(c->state), 606 state_text(state)); 607 } 608 609 c->state = state; 610 611 if (state == conn_write || state == conn_mwrite) { 612 MEMCACHED_PROCESS_COMMAND_END(c->sfd, c->wbuf, c->wbytes); 613 } 614 } 615 } 616 617 /* 618 * Ensures that there is room for another struct iovec in a connection's 619 * iov list. 620 * 621 * Returns 0 on success, -1 on out-of-memory. 622 */ 623 static int ensure_iov_space(conn *c) { 624 assert(c != NULL); 625 626 if (c->iovused >= c->iovsize) { 627 int i, iovnum; 628 struct iovec *new_iov = (struct iovec *)realloc(c->iov, 629 (c->iovsize * 2) * sizeof(struct iovec)); 630 if (! new_iov) 631 return -1; 632 c->iov = new_iov; 633 c->iovsize *= 2; 634 635 /* Point all the msghdr structures at the new list. */ 636 for (i = 0, iovnum = 0; i < c->msgused; i++) { 637 c->msglist[i].msg_iov = &c->iov[iovnum]; 638 iovnum += c->msglist[i].msg_iovlen; 639 } 640 } 641 642 return 0; 643 } 644 645 646 /* 647 * Adds data to the list of pending data that will be written out to a 648 * connection. 649 * 650 * Returns 0 on success, -1 on out-of-memory. 651 */ 652 653 static int add_iov(conn *c, const void *buf, int len) { 654 struct msghdr *m; 655 int leftover; 656 bool limit_to_mtu; 657 658 assert(c != NULL); 659 660 do { 661 m = &c->msglist[c->msgused - 1]; 662 663 /* 664 * Limit UDP packets, and the first payloads of TCP replies, to 665 * UDP_MAX_PAYLOAD_SIZE bytes. 666 */ 667 limit_to_mtu = IS_UDP(c->transport) || (1 == c->msgused); 668 669 /* We may need to start a new msghdr if this one is full. */ 670 if (m->msg_iovlen == IOV_MAX || 671 (limit_to_mtu && c->msgbytes >= UDP_MAX_PAYLOAD_SIZE)) { 672 add_msghdr(c); 673 m = &c->msglist[c->msgused - 1]; 674 } 675 676 if (ensure_iov_space(c) != 0) 677 return -1; 678 679 /* If the fragment is too big to fit in the datagram, split it up */ 680 if (limit_to_mtu && len + c->msgbytes > UDP_MAX_PAYLOAD_SIZE) { 681 leftover = len + c->msgbytes - UDP_MAX_PAYLOAD_SIZE; 682 len -= leftover; 683 } else { 684 leftover = 0; 685 } 686 687 m = &c->msglist[c->msgused - 1]; 688 m->msg_iov[m->msg_iovlen].iov_base = (void *)buf; 689 m->msg_iov[m->msg_iovlen].iov_len = len; 690 691 c->msgbytes += len; 692 c->iovused++; 693 m->msg_iovlen++; 694 695 buf = ((char *)buf) + len; 696 len = leftover; 697 } while (leftover > 0); 698 699 return 0; 700 } 701 702 703 /* 704 * Constructs a set of UDP headers and attaches them to the outgoing messages. 705 */ 706 static int build_udp_headers(conn *c) { 707 int i; 708 unsigned char *hdr; 709 710 assert(c != NULL); 711 712 if (c->msgused > c->hdrsize) { 713 void *new_hdrbuf; 714 if (c->hdrbuf) 715 new_hdrbuf = realloc(c->hdrbuf, c->msgused * 2 * UDP_HEADER_SIZE); 716 else 717 new_hdrbuf = malloc(c->msgused * 2 * UDP_HEADER_SIZE); 718 if (! new_hdrbuf) 719 return -1; 720 c->hdrbuf = (unsigned char *)new_hdrbuf; 721 c->hdrsize = c->msgused * 2; 722 } 723 724 hdr = c->hdrbuf; 725 for (i = 0; i < c->msgused; i++) { 726 c->msglist[i].msg_iov[0].iov_base = (void*)hdr; 727 c->msglist[i].msg_iov[0].iov_len = UDP_HEADER_SIZE; 728 *hdr++ = c->request_id / 256; 729 *hdr++ = c->request_id % 256; 730 *hdr++ = i / 256; 731 *hdr++ = i % 256; 732 *hdr++ = c->msgused / 256; 733 *hdr++ = c->msgused % 256; 734 *hdr++ = 0; 735 *hdr++ = 0; 736 assert((void *) hdr == (caddr_t)c->msglist[i].msg_iov[0].iov_base + UDP_HEADER_SIZE); 737 } 738 739 return 0; 740 } 741 742 743 static void out_string(conn *c, const char *str) { 744 size_t len; 745 746 assert(c != NULL); 747 748 if (c->noreply) { 749 if (settings.verbose > 1) 750 fprintf(stderr, ">%d NOREPLY %s\n", c->sfd, str); 751 c->noreply = false; 752 conn_set_state(c, conn_new_cmd); 753 return; 754 } 755 756 if (settings.verbose > 1) 757 fprintf(stderr, ">%d %s\n", c->sfd, str); 758 759 len = strlen(str); 760 if ((len + 2) > c->wsize) { 761 /* ought to be always enough. just fail for simplicity */ 762 str = "SERVER_ERROR output line too long"; 763 len = strlen(str); 764 } 765 766 memcpy(c->wbuf, str, len); 767 memcpy(c->wbuf + len, "\r\n", 2); 768 c->wbytes = len + 2; 769 c->wcurr = c->wbuf; 770 771 conn_set_state(c, conn_write); 772 c->write_and_go = conn_new_cmd; 773 return; 774 } 775 776 /* 777 * we get here after reading the value in set/add/replace commands. The command 778 * has been stored in c->cmd, and the item is ready in c->item. 779 */ 780 static void complete_nread_ascii(conn *c) { 781 assert(c != NULL); 782 783 item *it = c->item; 784 int comm = c->cmd; 785 enum store_item_type ret; 786 787 pthread_mutex_lock(&c->thread->stats.mutex); 788 c->thread->stats.slab_stats[it->slabs_clsid].set_cmds++; 789 pthread_mutex_unlock(&c->thread->stats.mutex); 790 791 if (strncmp(ITEM_data(it) + it->nbytes - 2, "\r\n", 2) != 0) { 792 out_string(c, "CLIENT_ERROR bad data chunk"); 793 } else { 794 ret = store_item(it, comm, c); 795 796 #ifdef ENABLE_DTRACE 797 uint64_t cas = ITEM_get_cas(it); 798 switch (c->cmd) { 799 case NREAD_ADD: 800 MEMCACHED_COMMAND_ADD(c->sfd, ITEM_key(it), it->nkey, 801 (ret == 1) ? it->nbytes : -1, cas); 802 break; 803 case NREAD_REPLACE: 804 MEMCACHED_COMMAND_REPLACE(c->sfd, ITEM_key(it), it->nkey, 805 (ret == 1) ? it->nbytes : -1, cas); 806 break; 807 case NREAD_APPEND: 808 MEMCACHED_COMMAND_APPEND(c->sfd, ITEM_key(it), it->nkey, 809 (ret == 1) ? it->nbytes : -1, cas); 810 break; 811 case NREAD_PREPEND: 812 MEMCACHED_COMMAND_PREPEND(c->sfd, ITEM_key(it), it->nkey, 813 (ret == 1) ? it->nbytes : -1, cas); 814 break; 815 case NREAD_SET: 816 MEMCACHED_COMMAND_SET(c->sfd, ITEM_key(it), it->nkey, 817 (ret == 1) ? it->nbytes : -1, cas); 818 break; 819 case NREAD_CAS: 820 MEMCACHED_COMMAND_CAS(c->sfd, ITEM_key(it), it->nkey, it->nbytes, 821 cas); 822 break; 823 } 824 #endif 825 826 switch (ret) { 827 case STORED: 828 out_string(c, "STORED"); 829 break; 830 case EXISTS: 831 out_string(c, "EXISTS"); 832 break; 833 case NOT_FOUND: 834 out_string(c, "NOT_FOUND"); 835 break; 836 case NOT_STORED: 837 out_string(c, "NOT_STORED"); 838 break; 839 default: 840 out_string(c, "SERVER_ERROR Unhandled storage type."); 841 } 842 843 } 844 845 item_remove(c->item); /* release the c->item reference */ 846 c->item = 0; 847 } 848 849 /** 850 * get a pointer to the start of the request struct for the current command 851 */ 852 static void* binary_get_request(conn *c) { 853 char *ret = c->rcurr; 854 ret -= (sizeof(c->binary_header) + c->binary_header.request.keylen + 855 c->binary_header.request.extlen); 856 857 assert(ret >= c->rbuf); 858 return ret; 859 } 860 861 /** 862 * get a pointer to the key in this request 863 */ 864 static char* binary_get_key(conn *c) { 865 return c->rcurr - (c->binary_header.request.keylen); 866 } 867 868 static void add_bin_header(conn *c, uint16_t err, uint8_t hdr_len, uint16_t key_len, uint32_t body_len) { 869 protocol_binary_response_header* header; 870 871 assert(c); 872 873 c->msgcurr = 0; 874 c->msgused = 0; 875 c->iovused = 0; 876 if (add_msghdr(c) != 0) { 877 /* XXX: out_string is inappropriate here */ 878 out_string(c, "SERVER_ERROR out of memory"); 879 return; 880 } 881 882 header = (protocol_binary_response_header *)c->wbuf; 883 884 header->response.magic = (uint8_t)PROTOCOL_BINARY_RES; 885 header->response.opcode = c->binary_header.request.opcode; 886 header->response.keylen = (uint16_t)htons(key_len); 887 888 header->response.extlen = (uint8_t)hdr_len; 889 header->response.datatype = (uint8_t)PROTOCOL_BINARY_RAW_BYTES; 890 header->response.status = (uint16_t)htons(err); 891 892 header->response.bodylen = htonl(body_len); 893 header->response.opaque = c->opaque; 894 header->response.cas = htonll(c->cas); 895 896 if (settings.verbose > 1) { 897 int ii; 898 fprintf(stderr, ">%d Writing bin response:", c->sfd); 899 for (ii = 0; ii < sizeof(header->bytes); ++ii) { 900 if (ii % 4 == 0) { 901 fprintf(stderr, "\n>%d ", c->sfd); 902 } 903 fprintf(stderr, " 0x%02x", header->bytes[ii]); 904 } 905 fprintf(stderr, "\n"); 906 } 907 908 add_iov(c, c->wbuf, sizeof(header->response)); 909 } 910 911 static void write_bin_error(conn *c, protocol_binary_response_status err, int swallow) { 912 const char *errstr = "Unknown error"; 913 size_t len; 914 915 switch (err) { 916 case PROTOCOL_BINARY_RESPONSE_ENOMEM: 917 errstr = "Out of memory"; 918 break; 919 case PROTOCOL_BINARY_RESPONSE_UNKNOWN_COMMAND: 920 errstr = "Unknown command"; 921 break; 922 case PROTOCOL_BINARY_RESPONSE_KEY_ENOENT: 923 errstr = "Not found"; 924 break; 925 case PROTOCOL_BINARY_RESPONSE_EINVAL: 926 errstr = "Invalid arguments"; 927 break; 928 case PROTOCOL_BINARY_RESPONSE_KEY_EEXISTS: 929 errstr = "Data exists for key."; 930 break; 931 case PROTOCOL_BINARY_RESPONSE_E2BIG: 932 errstr = "Too large."; 933 break; 934 case PROTOCOL_BINARY_RESPONSE_DELTA_BADVAL: 935 errstr = "Non-numeric server-side value for incr or decr"; 936 break; 937 case PROTOCOL_BINARY_RESPONSE_NOT_STORED: 938 errstr = "Not stored."; 939 break; 940 default: 941 assert(false); 942 errstr = "UNHANDLED ERROR"; 943 fprintf(stderr, ">%d UNHANDLED ERROR: %d\n", c->sfd, err); 944 } 945 946 if (settings.verbose > 0) { 947 fprintf(stderr, ">%d Writing an error: %s\n", c->sfd, errstr); 948 } 949 950 len = strlen(errstr); 951 add_bin_header(c, err, 0, 0, len); 952 if (len > 0) { 953 add_iov(c, errstr, len); 954 } 955 conn_set_state(c, conn_mwrite); 956 if(swallow > 0) { 957 c->sbytes = swallow; 958 c->write_and_go = conn_swallow; 959 } else { 960 c->write_and_go = conn_new_cmd; 961 } 962 } 963 964 /* Form and send a response to a command over the binary protocol */ 965 static void write_bin_response(conn *c, void *d, int hlen, int keylen, int dlen) { 966 if (!c->noreply || c->cmd == PROTOCOL_BINARY_CMD_GET || 967 c->cmd == PROTOCOL_BINARY_CMD_GETK) { 968 add_bin_header(c, 0, hlen, keylen, dlen); 969 if(dlen > 0) { 970 add_iov(c, d, dlen); 971 } 972 conn_set_state(c, conn_mwrite); 973 c->write_and_go = conn_new_cmd; 974 } else { 975 conn_set_state(c, conn_new_cmd); 976 } 977 } 978 979 static void complete_incr_bin(conn *c) { 980 item *it; 981 char *key; 982 size_t nkey; 983 984 protocol_binary_response_incr* rsp = (protocol_binary_response_incr*)c->wbuf; 985 protocol_binary_request_incr* req = binary_get_request(c); 986 987 assert(c != NULL); 988 assert(c->wsize >= sizeof(*rsp)); 989 990 /* fix byteorder in the request */ 991 req->message.body.delta = ntohll(req->message.body.delta); 992 req->message.body.initial = ntohll(req->message.body.initial); 993 req->message.body.expiration = ntohl(req->message.body.expiration); 994 key = binary_get_key(c); 995 nkey = c->binary_header.request.keylen; 996 997 if (settings.verbose) { 998 int i; 999 fprintf(stderr, "incr "); 1000 1001 for (i = 0; i < nkey; i++) { 1002 fprintf(stderr, "%c", key[i]); 1003 } 1004 fprintf(stderr, " %lld, %llu, %d\n", 1005 (long long)req->message.body.delta, 1006 (long long)req->message.body.initial, 1007 req->message.body.expiration); 1008 } 1009 1010 it = item_get(key, nkey); 1011 if (it && (c->binary_header.request.cas == 0 || 1012 c->binary_header.request.cas == ITEM_get_cas(it))) { 1013 /* Weird magic in add_delta forces me to pad here */ 1014 char tmpbuf[INCR_MAX_STORAGE_LEN]; 1015 protocol_binary_response_status st = PROTOCOL_BINARY_RESPONSE_SUCCESS; 1016 1017 switch(add_delta(c, it, c->cmd == PROTOCOL_BINARY_CMD_INCREMENT, 1018 req->message.body.delta, tmpbuf)) { 1019 case OK: 1020 break; 1021 case NON_NUMERIC: 1022 st = PROTOCOL_BINARY_RESPONSE_DELTA_BADVAL; 1023 break; 1024 case EOM: 1025 st = PROTOCOL_BINARY_RESPONSE_ENOMEM; 1026 break; 1027 } 1028 1029 if (st != PROTOCOL_BINARY_RESPONSE_SUCCESS) { 1030 write_bin_error(c, st, 0); 1031 } else { 1032 rsp->message.body.value = htonll(strtoull(tmpbuf, NULL, 10)); 1033 c->cas = ITEM_get_cas(it); 1034 write_bin_response(c, &rsp->message.body, 0, 0, 1035 sizeof(rsp->message.body.value)); 1036 } 1037 1038 item_remove(it); /* release our reference */ 1039 } else if (!it && req->message.body.expiration != 0xffffffff) { 1040 /* Save some room for the response */ 1041 rsp->message.body.value = htonll(req->message.body.initial); 1042 it = item_alloc(key, nkey, 0, realtime(req->message.body.expiration), 1043 INCR_MAX_STORAGE_LEN); 1044 1045 if (it != NULL) { 1046 snprintf(ITEM_data(it), INCR_MAX_STORAGE_LEN, "%llu", 1047 (unsigned long long)req->message.body.initial); 1048 1049 if (store_item(it, NREAD_SET, c)) { 1050 c->cas = ITEM_get_cas(it); 1051 write_bin_response(c, &rsp->message.body, 0, 0, sizeof(rsp->message.body.value)); 1052 } else { 1053 write_bin_error(c, PROTOCOL_BINARY_RESPONSE_NOT_STORED, 0); 1054 } 1055 item_remove(it); /* release our reference */ 1056 } else { 1057 write_bin_error(c, PROTOCOL_BINARY_RESPONSE_ENOMEM, 0); 1058 } 1059 } else if (it) { 1060 /* incorrect CAS */ 1061 item_remove(it); /* release our reference */ 1062 write_bin_error(c, PROTOCOL_BINARY_RESPONSE_KEY_EEXISTS, 0); 1063 } else { 1064 1065 pthread_mutex_lock(&c->thread->stats.mutex); 1066 if (c->cmd == PROTOCOL_BINARY_CMD_INCREMENT) { 1067 c->thread->stats.incr_misses++; 1068 } else { 1069 c->thread->stats.decr_misses++; 1070 } 1071 pthread_mutex_unlock(&c->thread->stats.mutex); 1072 1073 write_bin_error(c, PROTOCOL_BINARY_RESPONSE_KEY_ENOENT, 0); 1074 } 1075 } 1076 1077 static void complete_update_bin(conn *c) { 1078 protocol_binary_response_status eno = PROTOCOL_BINARY_RESPONSE_EINVAL; 1079 enum store_item_type ret = NOT_STORED; 1080 assert(c != NULL); 1081 1082 item *it = c->item; 1083 1084 pthread_mutex_lock(&c->thread->stats.mutex); 1085 c->thread->stats.slab_stats[it->slabs_clsid].set_cmds++; 1086 pthread_mutex_unlock(&c->thread->stats.mutex); 1087 1088 /* We don't actually receive the trailing two characters in the bin 1089 * protocol, so we're going to just set them here */ 1090 *(ITEM_data(it) + it->nbytes - 2) = '\r'; 1091 *(ITEM_data(it) + it->nbytes - 1) = '\n'; 1092 1093 ret = store_item(it, c->cmd, c); 1094 1095 #ifdef ENABLE_DTRACE 1096 uint64_t cas = ITEM_get_cas(it); 1097 switch (c->cmd) { 1098 case NREAD_ADD: 1099 MEMCACHED_COMMAND_ADD(c->sfd, ITEM_key(it), it->nkey, 1100 (ret == STORED) ? it->nbytes : -1, cas); 1101 break; 1102 case NREAD_REPLACE: 1103 MEMCACHED_COMMAND_REPLACE(c->sfd, ITEM_key(it), it->nkey, 1104 (ret == STORED) ? it->nbytes : -1, cas); 1105 break; 1106 case NREAD_APPEND: 1107 MEMCACHED_COMMAND_APPEND(c->sfd, ITEM_key(it), it->nkey, 1108 (ret == STORED) ? it->nbytes : -1, cas); 1109 break; 1110 case NREAD_PREPEND: 1111 MEMCACHED_COMMAND_PREPEND(c->sfd, ITEM_key(it), it->nkey, 1112 (ret == STORED) ? it->nbytes : -1, cas); 1113 break; 1114 case NREAD_SET: 1115 MEMCACHED_COMMAND_SET(c->sfd, ITEM_key(it), it->nkey, 1116 (ret == STORED) ? it->nbytes : -1, cas); 1117 break; 1118 } 1119 #endif 1120 1121 switch (ret) { 1122 case STORED: 1123 /* Stored */ 1124 write_bin_response(c, NULL, 0, 0, 0); 1125 break; 1126 case EXISTS: 1127 write_bin_error(c, PROTOCOL_BINARY_RESPONSE_KEY_EEXISTS, 0); 1128 break; 1129 case NOT_FOUND: 1130 write_bin_error(c, PROTOCOL_BINARY_RESPONSE_KEY_ENOENT, 0); 1131 break; 1132 case NOT_STORED: 1133 if (c->cmd == NREAD_ADD) { 1134 eno = PROTOCOL_BINARY_RESPONSE_KEY_EEXISTS; 1135 } else if(c->cmd == NREAD_REPLACE) { 1136 eno = PROTOCOL_BINARY_RESPONSE_KEY_ENOENT; 1137 } else { 1138 eno = PROTOCOL_BINARY_RESPONSE_NOT_STORED; 1139 } 1140 write_bin_error(c, eno, 0); 1141 } 1142 1143 item_remove(c->item); /* release the c->item reference */ 1144 c->item = 0; 1145 } 1146 1147 static void process_bin_get(conn *c) { 1148 item *it; 1149 1150 protocol_binary_response_get* rsp = (protocol_binary_response_get*)c->wbuf; 1151 char* key = binary_get_key(c); 1152 size_t nkey = c->binary_header.request.keylen; 1153 1154 if (settings.verbose) { 1155 int ii; 1156 fprintf(stderr, "<%d GET ", c->sfd); 1157 for (ii = 0; ii < nkey; ++ii) { 1158 fprintf(stderr, "%c", key[ii]); 1159 } 1160 fprintf(stderr, "\n"); 1161 } 1162 1163 it = item_get(key, nkey); 1164 if (it) { 1165 /* the length has two unnecessary bytes ("\r\n") */ 1166 uint16_t keylen = 0; 1167 uint32_t bodylen = sizeof(rsp->message.body) + (it->nbytes - 2); 1168 1169 pthread_mutex_lock(&c->thread->stats.mutex); 1170 c->thread->stats.get_cmds++; 1171 c->thread->stats.slab_stats[it->slabs_clsid].get_hits++; 1172 pthread_mutex_unlock(&c->thread->stats.mutex); 1173 1174 MEMCACHED_COMMAND_GET(c->sfd, ITEM_key(it), it->nkey, 1175 it->nbytes, ITEM_get_cas(it)); 1176 1177 if (c->cmd == PROTOCOL_BINARY_CMD_GETK) { 1178 bodylen += nkey; 1179 keylen = nkey; 1180 } 1181 add_bin_header(c, 0, sizeof(rsp->message.body), keylen, bodylen); 1182 rsp->message.header.response.cas = htonll(ITEM_get_cas(it)); 1183 1184 // add the flags 1185 rsp->message.body.flags = htonl(strtoul(ITEM_suffix(it), NULL, 10)); 1186 add_iov(c, &rsp->message.body, sizeof(rsp->message.body)); 1187 1188 if (c->cmd == PROTOCOL_BINARY_CMD_GETK) { 1189 add_iov(c, ITEM_key(it), nkey); 1190 } 1191 1192 /* Add the data minus the CRLF */ 1193 add_iov(c, ITEM_data(it), it->nbytes - 2); 1194 conn_set_state(c, conn_mwrite); 1195 /* Remember this command so we can garbage collect it later */ 1196 c->item = it; 1197 } else { 1198 pthread_mutex_lock(&c->thread->stats.mutex); 1199 c->thread->stats.get_cmds++; 1200 c->thread->stats.get_misses++; 1201 pthread_mutex_unlock(&c->thread->stats.mutex); 1202 1203 MEMCACHED_COMMAND_GET(c->sfd, key, nkey, -1, 0); 1204 1205 if (c->noreply) { 1206 conn_set_state(c, conn_new_cmd); 1207 } else { 1208 if (c->cmd == PROTOCOL_BINARY_CMD_GETK) { 1209 char *ofs = c->wbuf + sizeof(protocol_binary_response_header); 1210 add_bin_header(c, PROTOCOL_BINARY_RESPONSE_KEY_ENOENT, 1211 0, nkey, nkey); 1212 memcpy(ofs, key, nkey); 1213 add_iov(c, ofs, nkey); 1214 conn_set_state(c, conn_mwrite); 1215 } else { 1216 write_bin_error(c, PROTOCOL_BINARY_RESPONSE_KEY_ENOENT, 0); 1217 } 1218 } 1219 } 1220 1221 if (settings.detail_enabled) { 1222 stats_prefix_record_get(key, nkey, NULL != it); 1223 } 1224 } 1225 1226 static void append_bin_stats(const char *key, const uint16_t klen, 1227 const char *val, const uint32_t vlen, 1228 conn *c) { 1229 char *buf = c->stats.buffer + c->stats.offset; 1230 uint32_t bodylen = klen + vlen; 1231 protocol_binary_response_header header = { 1232 .response.magic = (uint8_t)PROTOCOL_BINARY_RES, 1233 .response.opcode = PROTOCOL_BINARY_CMD_STAT, 1234 .response.keylen = (uint16_t)htons(klen), 1235 .response.datatype = (uint8_t)PROTOCOL_BINARY_RAW_BYTES, 1236 .response.bodylen = htonl(bodylen), 1237 .response.opaque = c->opaque 1238 }; 1239 1240 memcpy(buf, header.bytes, sizeof(header.response)); 1241 buf += sizeof(header.response); 1242 1243 if (klen > 0) { 1244 memcpy(buf, key, klen); 1245 buf += klen; 1246 1247 if (vlen > 0) { 1248 memcpy(buf, val, vlen); 1249 } 1250 } 1251 1252 c->stats.offset += sizeof(header.response) + bodylen; 1253 } 1254 1255 static void append_ascii_stats(const char *key, const uint16_t klen, 1256 const char *val, const uint32_t vlen, 1257 conn *c) { 1258 char *pos = c->stats.buffer + c->stats.offset; 1259 uint32_t nbytes = 0; 1260 int remaining = c->stats.size - c->stats.offset; 1261 int room = remaining - 1; 1262 1263 if (klen == 0 && vlen == 0) { 1264 nbytes = snprintf(pos, room, "END\r\n"); 1265 } else if (vlen == 0) { 1266 nbytes = snprintf(pos, room, "STAT %s\r\n", key); 1267 } else { 1268 nbytes = snprintf(pos, room, "STAT %s %s\r\n", key, val); 1269 } 1270 1271 c->stats.offset += nbytes; 1272 } 1273 1274 static bool grow_stats_buf(conn *c, size_t needed) { 1275 size_t nsize = c->stats.size; 1276 size_t available = nsize - c->stats.offset; 1277 bool rv = true; 1278 1279 /* Special case: No buffer -- need to allocate fresh */ 1280 if (c->stats.buffer == NULL) { 1281 nsize = 1024; 1282 available = c->stats.size = c->stats.offset = 0; 1283 } 1284 1285 while (needed > available) { 1286 assert(nsize > 0); 1287 nsize = nsize << 1; 1288 available = nsize - c->stats.offset; 1289 } 1290 1291 if (nsize != c->stats.size) { 1292 char *ptr = realloc(c->stats.buffer, nsize); 1293 if (ptr) { 1294 c->stats.buffer = ptr; 1295 c->stats.size = nsize; 1296 } else { 1297 rv = false; 1298 } 1299 } 1300 1301 return rv; 1302 } 1303 1304 static void append_stats(const char *key, const uint16_t klen, 1305 const char *val, const uint32_t vlen, 1306 const void *cookie) 1307 { 1308 /* value without a key is invalid */ 1309 if (klen == 0 && vlen > 0) { 1310 return ; 1311 } 1312 1313 conn *c = (conn*)cookie; 1314 1315 if (c->protocol == binary_prot) { 1316 size_t needed = vlen + klen + sizeof(protocol_binary_response_header); 1317 if (!grow_stats_buf(c, needed)) { 1318 return ; 1319 } 1320 append_bin_stats(key, klen, val, vlen, c); 1321 } else { 1322 size_t needed = vlen + klen + 10; // 10 == "STAT = \r\n" 1323 if (!grow_stats_buf(c, needed)) { 1324 return ; 1325 } 1326 append_ascii_stats(key, klen, val, vlen, c); 1327 } 1328 1329 assert(c->stats.offset <= c->stats.size); 1330 } 1331 1332 static void process_bin_stat(conn *c) { 1333 char *subcommand = binary_get_key(c); 1334 size_t nkey = c->binary_header.request.keylen; 1335 1336 if (settings.verbose) { 1337 int ii; 1338 fprintf(stderr, "<%d STATS ", c->sfd); 1339 for (ii = 0; ii < nkey; ++ii) { 1340 fprintf(stderr, "%c", subcommand[ii]); 1341 } 1342 fprintf(stderr, "\n"); 1343 } 1344 1345 if (nkey == 0) { 1346 /* request all statistics */ 1347 server_stats(&append_stats, c); 1348 (void)get_stats(NULL, 0, &append_stats, c); 1349 } else if (strncmp(subcommand, "reset", 5) == 0) { 1350 stats_reset(); 1351 } else if (strncmp(subcommand, "settings", 8) == 0) { 1352 process_stat_settings(&append_stats, c); 1353 } else if (strncmp(subcommand, "detail", 6) == 0) { 1354 char *subcmd_pos = subcommand + 6; 1355 if (strncmp(subcmd_pos, " dump", 5) == 0) { 1356 int len; 1357 char *dump_buf = stats_prefix_dump(&len); 1358 if (dump_buf == NULL || len <= 0) { 1359 write_bin_error(c, PROTOCOL_BINARY_RESPONSE_ENOMEM, 0); 1360 return ; 1361 } else { 1362 append_stats("detailed", strlen("detailed"), dump_buf, len, c); 1363 free(dump_buf); 1364 } 1365 } else if (strncmp(subcmd_pos, " on", 3) == 0) { 1366 settings.detail_enabled = 1; 1367 } else if (strncmp(subcmd_pos, " off", 4) == 0) { 1368 settings.detail_enabled = 0; 1369 } else { 1370 write_bin_error(c, PROTOCOL_BINARY_RESPONSE_KEY_ENOENT, 0); 1371 return; 1372 } 1373 } else { 1374 if (get_stats(subcommand, nkey, &append_stats, c)) { 1375 if (c->stats.buffer == NULL) { 1376 write_bin_error(c, PROTOCOL_BINARY_RESPONSE_ENOMEM, 0); 1377 } else { 1378 write_and_free(c, c->stats.buffer, c->stats.offset); 1379 c->stats.buffer = NULL; 1380 } 1381 } else { 1382 write_bin_error(c, PROTOCOL_BINARY_RESPONSE_KEY_ENOENT, 0); 1383 } 1384 1385 return; 1386 } 1387 1388 /* Append termination package and start the transfer */ 1389 append_stats(NULL, 0, NULL, 0, c); 1390 if (c->stats.buffer == NULL) { 1391 write_bin_error(c, PROTOCOL_BINARY_RESPONSE_ENOMEM, 0); 1392 } else { 1393 write_and_free(c, c->stats.buffer, c->stats.offset); 1394 c->stats.buffer = NULL; 1395 } 1396 } 1397 1398 static void bin_read_key(conn *c, enum bin_substates next_substate, int extra) { 1399 assert(c); 1400 c->substate = next_substate; 1401 c->rlbytes = c->keylen + extra; 1402 1403 /* Ok... do we have room for the extras and the key in the input buffer? */ 1404 ptrdiff_t offset = c->rcurr + sizeof(protocol_binary_request_header) - c->rbuf; 1405 if (c->rlbytes > c->rsize - offset) { 1406 size_t nsize = c->rsize; 1407 size_t size = c->rlbytes + sizeof(protocol_binary_request_header); 1408 1409 while (size > nsize) { 1410 nsize *= 2; 1411 } 1412 1413 if (nsize != c->rsize) { 1414 if (settings.verbose) { 1415 fprintf(stderr, "%d: Need to grow buffer from %lu to %lu\n", 1416 c->sfd, (unsigned long)c->rsize, (unsigned long)nsize); 1417 } 1418 char *newm = realloc(c->rbuf, nsize); 1419 if (newm == NULL) { 1420 if (settings.verbose) { 1421 fprintf(stderr, "%d: Failed to grow buffer.. closing connection\n", 1422 c->sfd); 1423 } 1424 conn_set_state(c, conn_closing); 1425 return; 1426 } 1427 1428 c->rbuf= newm; 1429 /* rcurr should point to the same offset in the packet */ 1430 c->rcurr = c->rbuf + offset - sizeof(protocol_binary_request_header); 1431 c->rsize = nsize; 1432 } 1433 if (c->rbuf != c->rcurr) { 1434 memmove(c->rbuf, c->rcurr, c->rbytes); 1435 c->rcurr = c->rbuf; 1436 if (settings.verbose) { 1437 fprintf(stderr, "%d: Repack input buffer\n", c->sfd); 1438 } 1439 } 1440 } 1441 1442 /* preserve the header in the buffer.. */ 1443 c->ritem = c->rcurr + sizeof(protocol_binary_request_header); 1444 conn_set_state(c, conn_nread); 1445 } 1446 1447 static void dispatch_bin_command(conn *c) { 1448 int protocol_error = 0; 1449 1450 int extlen = c->binary_header.request.extlen; 1451 int keylen = c->binary_header.request.keylen; 1452 uint32_t bodylen = c->binary_header.request.bodylen; 1453 1454 MEMCACHED_PROCESS_COMMAND_START(c->sfd, c->rcurr, c->rbytes); 1455 c->noreply = true; 1456 switch (c->cmd) { 1457 case PROTOCOL_BINARY_CMD_SETQ: 1458 c->cmd = PROTOCOL_BINARY_CMD_SET; 1459 break; 1460 case PROTOCOL_BINARY_CMD_ADDQ: 1461 c->cmd = PROTOCOL_BINARY_CMD_ADD; 1462 break; 1463 case PROTOCOL_BINARY_CMD_REPLACEQ: 1464 c->cmd = PROTOCOL_BINARY_CMD_REPLACE; 1465 break; 1466 case PROTOCOL_BINARY_CMD_DELETEQ: 1467 c->cmd = PROTOCOL_BINARY_CMD_DELETE; 1468 break; 1469 case PROTOCOL_BINARY_CMD_INCREMENTQ: 1470 c->cmd = PROTOCOL_BINARY_CMD_INCREMENT; 1471 break; 1472 case PROTOCOL_BINARY_CMD_DECREMENTQ: 1473 c->cmd = PROTOCOL_BINARY_CMD_DECREMENT; 1474 break; 1475 case PROTOCOL_BINARY_CMD_QUITQ: 1476 c->cmd = PROTOCOL_BINARY_CMD_QUIT; 1477 break; 1478 case PROTOCOL_BINARY_CMD_FLUSHQ: 1479 c->cmd = PROTOCOL_BINARY_CMD_FLUSH; 1480 break; 1481 case PROTOCOL_BINARY_CMD_APPENDQ: 1482 c->cmd = PROTOCOL_BINARY_CMD_APPEND; 1483 break; 1484 case PROTOCOL_BINARY_CMD_PREPENDQ: 1485 c->cmd = PROTOCOL_BINARY_CMD_PREPEND; 1486 break; 1487 case PROTOCOL_BINARY_CMD_GETQ: 1488 c->cmd = PROTOCOL_BINARY_CMD_GET; 1489 break; 1490 case PROTOCOL_BINARY_CMD_GETKQ: 1491 c->cmd = PROTOCOL_BINARY_CMD_GETK; 1492 break; 1493 default: 1494 c->noreply = false; 1495 } 1496 1497 switch (c->cmd) { 1498 case PROTOCOL_BINARY_CMD_VERSION: 1499 if (extlen == 0 && keylen == 0 && bodylen == 0) { 1500 write_bin_response(c, VERSION, 0, 0, strlen(VERSION)); 1501 } else { 1502 protocol_error = 1; 1503 } 1504 break; 1505 case PROTOCOL_BINARY_CMD_FLUSH: 1506 if (keylen == 0 && bodylen == extlen && (extlen == 0 || extlen == 4)) { 1507 bin_read_key(c, bin_read_flush_exptime, extlen); 1508 } else { 1509 protocol_error = 1; 1510 } 1511 break; 1512 case PROTOCOL_BINARY_CMD_NOOP: 1513 if (extlen == 0 && keylen == 0 && bodylen == 0) { 1514 write_bin_response(c, NULL, 0, 0, 0); 1515 } else { 1516 protocol_error = 1; 1517 } 1518 break; 1519 case PROTOCOL_BINARY_CMD_SET: /* FALLTHROUGH */ 1520 case PROTOCOL_BINARY_CMD_ADD: /* FALLTHROUGH */ 1521 case PROTOCOL_BINARY_CMD_REPLACE: 1522 if (extlen == 8 && keylen != 0 && bodylen >= (keylen + 8)) { 1523 bin_read_key(c, bin_reading_set_header, 8); 1524 } else { 1525 protocol_error = 1; 1526 } 1527 break; 1528 case PROTOCOL_BINARY_CMD_GETQ: /* FALLTHROUGH */ 1529 case PROTOCOL_BINARY_CMD_GET: /* FALLTHROUGH */ 1530 case PROTOCOL_BINARY_CMD_GETKQ: /* FALLTHROUGH */ 1531 case PROTOCOL_BINARY_CMD_GETK: 1532 if (extlen == 0 && bodylen == keylen && keylen > 0) { 1533 bin_read_key(c, bin_reading_get_key, 0); 1534 } else { 1535 protocol_error = 1; 1536 } 1537 break; 1538 case PROTOCOL_BINARY_CMD_DELETE: 1539 if (keylen > 0 && extlen == 0 && bodylen == keylen) { 1540 bin_read_key(c, bin_reading_del_header, extlen); 1541 } else { 1542 protocol_error = 1; 1543 } 1544 break; 1545 case PROTOCOL_BINARY_CMD_INCREMENT: 1546 case PROTOCOL_BINARY_CMD_DECREMENT: 1547 if (keylen > 0 && extlen == 20 && bodylen == (keylen + extlen)) { 1548 bin_read_key(c, bin_reading_incr_header, 20); 1549 } else { 1550 protocol_error = 1; 1551 } 1552 break; 1553 case PROTOCOL_BINARY_CMD_APPEND: 1554 case PROTOCOL_BINARY_CMD_PREPEND: 1555 if (keylen > 0 && extlen == 0) { 1556 bin_read_key(c, bin_reading_set_header, 0); 1557 } else { 1558 protocol_error = 1; 1559 } 1560 break; 1561 case PROTOCOL_BINARY_CMD_STAT: 1562 if (extlen == 0) { 1563 bin_read_key(c, bin_reading_stat, 0); 1564 } else { 1565 protocol_error = 1; 1566 } 1567 break; 1568 case PROTOCOL_BINARY_CMD_QUIT: 1569 if (keylen == 0 && extlen == 0 && bodylen == 0) { 1570 write_bin_response(c, NULL, 0, 0, 0); 1571 c->write_and_go = conn_closing; 1572 if (c->noreply) { 1573 conn_set_state(c, conn_closing); 1574 } 1575 } else { 1576 protocol_error = 1; 1577 } 1578 break; 1579 default: 1580 write_bin_error(c, PROTOCOL_BINARY_RESPONSE_UNKNOWN_COMMAND, bodylen); 1581 } 1582 1583 if (protocol_error) { 1584 /* Just write an error message and disconnect the client */ 1585 write_bin_error(c, PROTOCOL_BINARY_RESPONSE_EINVAL, 0); 1586 if (settings.verbose) { 1587 fprintf(stderr, "Protocol error (opcode %02x), close connection %d\n", 1588 c->binary_header.request.opcode, c->sfd); 1589 } 1590 c->write_and_go = conn_closing; 1591 } 1592 } 1593 1594 static void process_bin_update(conn *c) { 1595 char *key; 1596 int nkey; 1597 int vlen; 1598 item *it; 1599 protocol_binary_request_set* req = binary_get_request(c); 1600 1601 assert(c != NULL); 1602 1603 key = binary_get_key(c); 1604 nkey = c->binary_header.request.keylen; 1605 1606 /* fix byteorder in the request */ 1607 req->message.body.flags = ntohl(req->message.body.flags); 1608 req->message.body.expiration = ntohl(req->message.body.expiration); 1609 1610 vlen = c->binary_header.request.bodylen - (nkey + c->binary_header.request.extlen); 1611 1612 if (settings.verbose) { 1613 int ii; 1614 if (c->cmd == PROTOCOL_BINARY_CMD_ADD) { 1615 fprintf(stderr, "<%d ADD ", c->sfd); 1616 } else if (c->cmd == PROTOCOL_BINARY_CMD_SET) { 1617 fprintf(stderr, "<%d SET ", c->sfd); 1618 } else { 1619 fprintf(stderr, "<%d REPLACE ", c->sfd); 1620 } 1621 for (ii = 0; ii < nkey; ++ii) { 1622 fprintf(stderr, "%c", key[ii]); 1623 } 1624 1625 if (settings.verbose > 1) { 1626 fprintf(stderr, " Value len is %d", vlen); 1627 } 1628 fprintf(stderr, "\n"); 1629 } 1630 1631 if (settings.detail_enabled) { 1632 stats_prefix_record_set(key, nkey); 1633 } 1634 1635 it = item_alloc(key, nkey, req->message.body.flags, 1636 realtime(req->message.body.expiration), vlen+2); 1637 1638 if (it == 0) { 1639 if (! item_size_ok(nkey, req->message.body.flags, vlen + 2)) { 1640 write_bin_error(c, PROTOCOL_BINARY_RESPONSE_E2BIG, vlen); 1641 } else { 1642 write_bin_error(c, PROTOCOL_BINARY_RESPONSE_ENOMEM, vlen); 1643 } 1644 1645 /* Avoid stale data persisting in cache because we failed alloc. 1646 * Unacceptable for SET. Anywhere else too? */ 1647 if (c->cmd == PROTOCOL_BINARY_CMD_SET) { 1648 it = item_get(key, nkey); 1649 if (it) { 1650 item_unlink(it); 1651 item_remove(it); 1652 } 1653 } 1654 1655 /* swallow the data line */ 1656 c->write_and_go = conn_swallow; 1657 return; 1658 } 1659 1660 ITEM_set_cas(it, c->binary_header.request.cas); 1661 1662 switch (c->cmd) { 1663 case PROTOCOL_BINARY_CMD_ADD: 1664 c->cmd = NREAD_ADD; 1665 break; 1666 case PROTOCOL_BINARY_CMD_SET: 1667 c->cmd = NREAD_SET; 1668 break; 1669 case PROTOCOL_BINARY_CMD_REPLACE: 1670 c->cmd = NREAD_REPLACE; 1671 break; 1672 default: 1673 assert(0); 1674 } 1675 1676 if (ITEM_get_cas(it) != 0) { 1677 c->cmd = NREAD_CAS; 1678 } 1679 1680 c->item = it; 1681 c->ritem = ITEM_data(it); 1682 c->rlbytes = vlen; 1683 conn_set_state(c, conn_nread); 1684 c->substate = bin_read_set_value; 1685 } 1686 1687 static void process_bin_append_prepend(conn *c) { 1688 char *key; 1689 int nkey; 1690 int vlen; 1691 item *it; 1692 1693 assert(c != NULL); 1694 1695 key = binary_get_key(c); 1696 nkey = c->binary_header.request.keylen; 1697 vlen = c->binary_header.request.bodylen - nkey; 1698 1699 if (settings.verbose > 1) { 1700 fprintf(stderr, "Value len is %d\n", vlen); 1701 } 1702 1703 if (settings.detail_enabled) { 1704 stats_prefix_record_set(key, nkey); 1705 } 1706 1707 it = item_alloc(key, nkey, 0, 0, vlen+2); 1708 1709 if (it == 0) { 1710 if (! item_size_ok(nkey, 0, vlen + 2)) { 1711 write_bin_error(c, PROTOCOL_BINARY_RESPONSE_E2BIG, vlen); 1712 } else { 1713 write_bin_error(c, PROTOCOL_BINARY_RESPONSE_ENOMEM, vlen); 1714 } 1715 /* swallow the data line */ 1716 c->write_and_go = conn_swallow; 1717 return; 1718 } 1719 1720 ITEM_set_cas(it, c->binary_header.request.cas); 1721 1722 switch (c->cmd) { 1723 case PROTOCOL_BINARY_CMD_APPEND: 1724 c->cmd = NREAD_APPEND; 1725 break; 1726 case PROTOCOL_BINARY_CMD_PREPEND: 1727 c->cmd = NREAD_PREPEND; 1728 break; 1729 default: 1730 assert(0); 1731 } 1732 1733 c->item = it; 1734 c->ritem = ITEM_data(it); 1735 c->rlbytes = vlen; 1736 conn_set_state(c, conn_nread); 1737 c->substate = bin_read_set_value; 1738 } 1739 1740 static void process_bin_flush(conn *c) { 1741 time_t exptime = 0; 1742 protocol_binary_request_flush* req = binary_get_request(c); 1743 1744 if (c->binary_header.request.extlen == sizeof(req->message.body)) { 1745 exptime = ntohl(req->message.body.expiration); 1746 } 1747 1748 set_current_time(); 1749 1750 if (exptime > 0) { 1751 settings.oldest_live = realtime(exptime) - 1; 1752 } else { 1753 settings.oldest_live = current_time - 1; 1754 } 1755 item_flush_expired(); 1756 1757 pthread_mutex_lock(&c->thread->stats.mutex); 1758 c->thread->stats.flush_cmds++; 1759 pthread_mutex_unlock(&c->thread->stats.mutex); 1760 1761 write_bin_response(c, NULL, 0, 0, 0); 1762 } 1763 1764 static void process_bin_delete(conn *c) { 1765 item *it; 1766 1767 protocol_binary_request_delete* req = binary_get_request(c); 1768 1769 char* key = binary_get_key(c); 1770 size_t nkey = c->binary_header.request.keylen; 1771 1772 assert(c != NULL); 1773 1774 if (settings.verbose) { 1775 fprintf(stderr, "Deleting %s\n", key); 1776 } 1777 1778 if (settings.detail_enabled) { 1779 stats_prefix_record_delete(key, nkey); 1780 } 1781 1782 it = item_get(key, nkey); 1783 if (it) { 1784 uint64_t cas = ntohll(req->message.header.request.cas); 1785 if (cas == 0 || cas == ITEM_get_cas(it)) { 1786 MEMCACHED_COMMAND_DELETE(c->sfd, ITEM_key(it), it->nkey); 1787 item_unlink(it); 1788 write_bin_response(c, NULL, 0, 0, 0); 1789 } else { 1790 write_bin_error(c, PROTOCOL_BINARY_RESPONSE_KEY_EEXISTS, 0); 1791 } 1792 item_remove(it); /* release our reference */ 1793 } else { 1794 write_bin_error(c, PROTOCOL_BINARY_RESPONSE_KEY_ENOENT, 0); 1795 } 1796 } 1797 1798 static void complete_nread_binary(conn *c) { 1799 assert(c != NULL); 1800 assert(c->cmd >= 0); 1801 1802 switch(c->substate) { 1803 case bin_reading_set_header: 1804 if (c->cmd == PROTOCOL_BINARY_CMD_APPEND || 1805 c->cmd == PROTOCOL_BINARY_CMD_PREPEND) { 1806 process_bin_append_prepend(c); 1807 } else { 1808 process_bin_update(c); 1809 } 1810 break; 1811 case bin_read_set_value: 1812 complete_update_bin(c); 1813 break; 1814 case bin_reading_get_key: 1815 process_bin_get(c); 1816 break; 1817 case bin_reading_stat: 1818 process_bin_stat(c); 1819 break; 1820 case bin_reading_del_header: 1821 process_bin_delete(c); 1822 break; 1823 case bin_reading_incr_header: 1824 complete_incr_bin(c); 1825 break; 1826 case bin_read_flush_exptime: 1827 process_bin_flush(c); 1828 break; 1829 default: 1830 fprintf(stderr, "Not handling substate %d\n", c->substate); 1831 assert(0); 1832 } 1833 } 1834 1835 static void reset_cmd_handler(conn *c) { 1836 c->cmd = -1; 1837 c->substate = bin_no_state; 1838 if(c->item != NULL) { 1839 item_remove(c->item); 1840 c->item = NULL; 1841 } 1842 conn_shrink(c); 1843 if (c->rbytes > 0) { 1844 conn_set_state(c, conn_parse_cmd); 1845 } else { 1846 conn_set_state(c, conn_waiting); 1847 } 1848 } 1849 1850 static void complete_nread(conn *c) { 1851 assert(c != NULL); 1852 assert(c->protocol == ascii_prot 1853 || c->protocol == binary_prot); 1854 1855 if (c->protocol == ascii_prot) { 1856 complete_nread_ascii(c); 1857 } else if (c->protocol == binary_prot) { 1858 complete_nread_binary(c); 1859 } 1860 } 1861 1862 /* 1863 * Stores an item in the cache according to the semantics of one of the set 1864 * commands. In threaded mode, this is protected by the cache lock. 1865 * 1866 * Returns the state of storage. 1867 */ 1868 enum store_item_type do_store_item(item *it, int comm, conn *c) { 1869 char *key = ITEM_key(it); 1870 item *old_it = do_item_get(key, it->nkey); 1871 enum store_item_type stored = NOT_STORED; 1872 1873 item *new_it = NULL; 1874 int flags; 1875 1876 if (old_it != NULL && comm == NREAD_ADD) { 1877 /* add only adds a nonexistent item, but promote to head of LRU */ 1878 do_item_update(old_it); 1879 } else if (!old_it && (comm == NREAD_REPLACE 1880 || comm == NREAD_APPEND || comm == NREAD_PREPEND)) 1881 { 1882 /* replace only replaces an existing value; don't store */ 1883 } else if (comm == NREAD_CAS) { 1884 /* validate cas operation */ 1885 if(old_it == NULL) { 1886 // LRU expired 1887 stored = NOT_FOUND; 1888 pthread_mutex_lock(&c->thread->stats.mutex); 1889 c->thread->stats.cas_misses++; 1890 pthread_mutex_unlock(&c->thread->stats.mutex); 1891 } 1892 else if (ITEM_get_cas(it) == ITEM_get_cas(old_it)) { 1893 // cas validates 1894 // it and old_it may belong to different classes. 1895 // I'm updating the stats for the one that's getting pushed out 1896 pthread_mutex_lock(&c->thread->stats.mutex); 1897 c->thread->stats.slab_stats[old_it->slabs_clsid].cas_hits++; 1898 pthread_mutex_unlock(&c->thread->stats.mutex); 1899 1900 item_replace(old_it, it); 1901 stored = STORED; 1902 } else { 1903 pthread_mutex_lock(&c->thread->stats.mutex); 1904 c->thread->stats.slab_stats[old_it->slabs_clsid].cas_badval++; 1905 pthread_mutex_unlock(&c->thread->stats.mutex); 1906 1907 if(settings.verbose > 1) { 1908 fprintf(stderr, "CAS: failure: expected %llu, got %llu\n", 1909 (unsigned long long)ITEM_get_cas(old_it), 1910 (unsigned long long)ITEM_get_cas(it)); 1911 } 1912 stored = EXISTS; 1913 } 1914 } else { 1915 /* 1916 * Append - combine new and old record into single one. Here it's 1917 * atomic and thread-safe. 1918 */ 1919 if (comm == NREAD_APPEND || comm == NREAD_PREPEND) { 1920 /* 1921 * Validate CAS 1922 */ 1923 if (ITEM_get_cas(it) != 0) { 1924 // CAS much be equal 1925 if (ITEM_get_cas(it) != ITEM_get_cas(old_it)) { 1926 stored = EXISTS; 1927 } 1928 } 1929 1930 if (stored == NOT_STORED) { 1931 /* we have it and old_it here - alloc memory to hold both */ 1932 /* flags was already lost - so recover them from ITEM_suffix(it) */ 1933 1934 flags = (int) strtol(ITEM_suffix(old_it), (char **) NULL, 10); 1935 1936 new_it = do_item_alloc(key, it->nkey, flags, old_it->exptime, it->nbytes + old_it->nbytes - 2 /* CRLF */); 1937 1938 if (new_it == NULL) { 1939 /* SERVER_ERROR out of memory */ 1940 if (old_it != NULL) 1941 do_item_remove(old_it); 1942 1943 return NOT_STORED; 1944 } 1945 1946 /* copy data from it and old_it to new_it */ 1947 1948 if (comm == NREAD_APPEND) { 1949 memcpy(ITEM_data(new_it), ITEM_data(old_it), old_it->nbytes); 1950 memcpy(ITEM_data(new_it) + old_it->nbytes - 2 /* CRLF */, ITEM_data(it), it->nbytes); 1951 } else { 1952 /* NREAD_PREPEND */ 1953 memcpy(ITEM_data(new_it), ITEM_data(it), it->nbytes); 1954 memcpy(ITEM_data(new_it) + it->nbytes - 2 /* CRLF */, ITEM_data(old_it), old_it->nbytes); 1955 } 1956 1957 it = new_it; 1958 } 1959 } 1960 1961 if (stored == NOT_STORED) { 1962 if (old_it != NULL) 1963 item_replace(old_it, it); 1964 else 1965 do_item_link(it); 1966 1967 c->cas = ITEM_get_cas(it); 1968 1969 stored = STORED; 1970 } 1971 } 1972 1973 if (old_it != NULL) 1974 do_item_remove(old_it); /* release our reference */ 1975 if (new_it != NULL) 1976 do_item_remove(new_it); 1977 1978 if (stored == STORED) { 1979 c->cas = ITEM_get_cas(it); 1980 } 1981 1982 return stored; 1983 } 1984 1985 typedef struct token_s { 1986 char *value; 1987 size_t length; 1988 } token_t; 1989 1990 #define COMMAND_TOKEN 0 1991 #define SUBCOMMAND_TOKEN 1 1992 #define KEY_TOKEN 1 1993 1994 #define MAX_TOKENS 8 1995 1996 /* 1997 * Tokenize the command string by replacing whitespace with '\0' and update 1998 * the token array tokens with pointer to start of each token and length. 1999 * Returns total number of tokens. The last valid token is the terminal 2000 * token (value points to the first unprocessed character of the string and 2001 * length zero). 2002 * 2003 * Usage example: 2004 * 2005 * while(tokenize_command(command, ncommand, tokens, max_tokens) > 0) { 2006 * for(int ix = 0; tokens[ix].length != 0; ix++) { 2007 * ... 2008 * } 2009 * ncommand = tokens[ix].value - command; 2010 * command = tokens[ix].value; 2011 * } 2012 */ 2013 static size_t tokenize_command(char *command, token_t *tokens, const size_t max_tokens) { 2014 char *s, *e; 2015 size_t ntokens = 0; 2016 2017 assert(command != NULL && tokens != NULL && max_tokens > 1); 2018 2019 for (s = e = command; ntokens < max_tokens - 1; ++e) { 2020 if (*e == ' ') { 2021 if (s != e) { 2022 tokens[ntokens].value = s; 2023 tokens[ntokens].length = e - s; 2024 ntokens++; 2025 *e = '\0'; 2026 } 2027 s = e + 1; 2028 } 2029 else if (*e == '\0') { 2030 if (s != e) { 2031 tokens[ntokens].value = s; 2032 tokens[ntokens].length = e - s; 2033 ntokens++; 2034 } 2035 2036 break; /* string end */ 2037 } 2038 } 2039 2040 /* 2041 * If we scanned the whole string, the terminal value pointer is null, 2042 * otherwise it is the first unprocessed character. 2043 */ 2044 tokens[ntokens].value = *e == '\0' ? NULL : e; 2045 tokens[ntokens].length = 0; 2046 ntokens++; 2047 2048 return ntokens; 2049 } 2050 2051 /* set up a connection to write a buffer then free it, used for stats */ 2052 static void write_and_free(conn *c, char *buf, int bytes) { 2053 if (buf) { 2054 c->write_and_free = buf; 2055 c->wcurr = buf; 2056 c->wbytes = bytes; 2057 conn_set_state(c, conn_write); 2058 c->write_and_go = conn_new_cmd; 2059 } else { 2060 out_string(c, "SERVER_ERROR out of memory writing stats"); 2061 } 2062 } 2063 2064 static inline void set_noreply_maybe(conn *c, token_t *tokens, size_t ntokens) 2065 { 2066 int noreply_index = ntokens - 2; 2067 2068 /* 2069 NOTE: this function is not the first place where we are going to 2070 send the reply. We could send it instead from process_command() 2071 if the request line has wrong number of tokens. However parsing 2072 malformed line for "noreply" option is not reliable anyway, so 2073 it can't be helped. 2074 */ 2075 if (tokens[noreply_index].value 2076 && strcmp(tokens[noreply_index].value, "noreply") == 0) { 2077 c->noreply = true; 2078 } 2079 } 2080 2081 void append_stat(const char *name, ADD_STAT add_stats, conn *c, 2082 const char *fmt, ...) { 2083 char val_str[STAT_VAL_LEN]; 2084 int vlen; 2085 va_list ap; 2086 2087 assert(name); 2088 assert(add_stats); 2089 assert(c); 2090 assert(fmt); 2091 2092 va_start(ap, fmt); 2093 vlen = vsnprintf(val_str, sizeof(val_str) - 1, fmt, ap); 2094 va_end(ap); 2095 2096 add_stats(name, strlen(name), val_str, vlen, c); 2097 } 2098 2099 inline static void process_stats_detail(conn *c, const char *command) { 2100 assert(c != NULL); 2101 2102 if (strcmp(command, "on") == 0) { 2103 settings.detail_enabled = 1; 2104 out_string(c, "OK"); 2105 } 2106 else if (strcmp(command, "off") == 0) { 2107 settings.detail_enabled = 0; 2108 out_string(c, "OK"); 2109 } 2110 else if (strcmp(command, "dump") == 0) { 2111 int len; 2112 char *stats = stats_prefix_dump(&len); 2113 write_and_free(c, stats, len); 2114 } 2115 else { 2116 out_string(c, "CLIENT_ERROR usage: stats detail on|off|dump"); 2117 } 2118 } 2119 2120 /* return server specific stats only */ 2121 static void server_stats(ADD_STAT add_stats, conn *c) { 2122 pid_t pid = getpid(); 2123 rel_time_t now = current_time; 2124 2125 struct thread_stats thread_stats; 2126 threadlocal_stats_aggregate(&thread_stats); 2127 struct slab_stats slab_stats; 2128 slab_stats_aggregate(&thread_stats, &slab_stats); 2129 2130 #ifndef WIN32 2131 struct rusage usage; 2132 getrusage(RUSAGE_SELF, &usage); 2133 #endif /* !WIN32 */ 2134 2135 STATS_LOCK(); 2136 2137 APPEND_STAT("pid", "%lu", (long)pid); 2138 APPEND_STAT("uptime", "%u", now); 2139 APPEND_STAT("time", "%ld", now + (long)process_started); 2140 APPEND_STAT("version", "%s", VERSION); 2141 APPEND_STAT("pointer_size", "%d", (int)(8 * sizeof(void *))); 2142 2143 #ifndef WIN32 2144 append_stat("rusage_user", add_stats, c, "%ld.%06ld", 2145 (long)usage.ru_utime.tv_sec, 2146 (long)usage.ru_utime.tv_usec); 2147 append_stat("rusage_system", add_stats, c, "%ld.%06ld", 2148 (long)usage.ru_stime.tv_sec, 2149 (long)usage.ru_stime.tv_usec); 2150 #endif /* !WIN32 */ 2151 2152 APPEND_STAT("curr_connections", "%u", stats.curr_conns - 1); 2153 APPEND_STAT("total_connections", "%u", stats.total_conns); 2154 APPEND_STAT("connection_structures", "%u", stats.conn_structs); 2155 APPEND_STAT("cmd_get", "%llu", (unsigned long long)thread_stats.get_cmds); 2156 APPEND_STAT("cmd_set", "%llu", (unsigned long long)slab_stats.set_cmds); 2157 APPEND_STAT("cmd_flush", "%llu", (unsigned long long)thread_stats.flush_cmds); 2158 APPEND_STAT("get_hits", "%llu", (unsigned long long)slab_stats.get_hits); 2159 APPEND_STAT("get_misses", "%llu", (unsigned long long)thread_stats.get_misses); 2160 APPEND_STAT("delete_misses", "%llu", (unsigned long long)thread_stats.delete_misses); 2161 APPEND_STAT("delete_hits", "%llu", (unsigned long long)slab_stats.delete_hits); 2162 APPEND_STAT("incr_misses", "%llu", (unsigned long long)thread_stats.incr_misses); 2163 APPEND_STAT("incr_hits", "%llu", (unsigned long long)slab_stats.incr_hits); 2164 APPEND_STAT("decr_misses", "%llu", (unsigned long long)thread_stats.decr_misses); 2165 APPEND_STAT("decr_hits", "%llu", (unsigned long long)slab_stats.decr_hits); 2166 APPEND_STAT("cas_misses", "%llu", (unsigned long long)thread_stats.cas_misses); 2167 APPEND_STAT("cas_hits", "%llu", (unsigned long long)slab_stats.cas_hits); 2168 APPEND_STAT("cas_badval", "%llu", (unsigned long long)slab_stats.cas_badval); 2169 APPEND_STAT("bytes_read", "%llu", (unsigned long long)thread_stats.bytes_read); 2170 APPEND_STAT("bytes_written", "%llu", (unsigned long long)thread_stats.bytes_written); 2171 APPEND_STAT("limit_maxbytes", "%llu", (unsigned long long)settings.maxbytes); 2172 APPEND_STAT("accepting_conns", "%u", stats.accepting_conns); 2173 APPEND_STAT("listen_disabled_num", "%llu", (unsigned long long)stats.listen_disabled_num); 2174 APPEND_STAT("threads", "%d", settings.num_threads); 2175 APPEND_STAT("conn_yields", "%llu", (unsigned long long)thread_stats.conn_yields); 2176 STATS_UNLOCK(); 2177 } 2178 2179 static void process_stat_settings(ADD_STAT add_stats, void *c) { 2180 assert(add_stats); 2181 APPEND_STAT("maxbytes", "%u", (unsigned int)settings.maxbytes); 2182 APPEND_STAT("maxconns", "%d", settings.maxconns); 2183 APPEND_STAT("tcpport", "%d", settings.port); 2184 APPEND_STAT("udpport", "%d", settings.udpport); 2185 APPEND_STAT("inter", "%s", settings.inter ? settings.inter : "NULL"); 2186 APPEND_STAT("verbosity", "%d", settings.verbose); 2187 APPEND_STAT("oldest", "%lu", (unsigned long)settings.oldest_live); 2188 APPEND_STAT("evictions", "%s", settings.evict_to_free ? "on" : "off"); 2189 APPEND_STAT("domain_socket", "%s", 2190 settings.socketpath ? settings.socketpath : "NULL"); 2191 APPEND_STAT("umask", "%o", settings.access); 2192 APPEND_STAT("growth_factor", "%.2f", settings.factor); 2193 APPEND_STAT("chunk_size", "%d", settings.chunk_size); 2194 APPEND_STAT("num_threads", "%d", settings.num_threads); 2195 APPEND_STAT("stat_key_prefix", "%c", settings.prefix_delimiter); 2196 APPEND_STAT("detail_enabled", "%s", 2197 settings.detail_enabled ? "yes" : "no"); 2198 APPEND_STAT("reqs_per_event", "%d", settings.reqs_per_event); 2199 APPEND_STAT("cas_enabled", "%s", settings.use_cas ? "yes" : "no"); 2200 APPEND_STAT("tcp_backlog", "%d", settings.backlog); 2201 APPEND_STAT("binding_protocol", "%s", 2202 prot_text(settings.binding_protocol)); 2203 APPEND_STAT("item_size_max", "%d", settings.item_size_max); 2204 } 2205 2206 static void process_stat(conn *c, token_t *tokens, const size_t ntokens) { 2207 const char *subcommand = tokens[SUBCOMMAND_TOKEN].value; 2208 assert(c != NULL); 2209 2210 if (ntokens < 2) { 2211 out_string(c, "CLIENT_ERROR bad command line"); 2212 return; 2213 } 2214 2215 if (ntokens == 2) { 2216 server_stats(&append_stats, c); 2217 (void)get_stats(NULL, 0, &append_stats, c); 2218 } else if (strcmp(subcommand, "reset") == 0) { 2219 stats_reset(); 2220 out_string(c, "RESET"); 2221 return ; 2222 } else if (strcmp(subcommand, "detail") == 0) { 2223 /* NOTE: how to tackle detail with binary? */ 2224 if (ntokens < 4) 2225 process_stats_detail(c, ""); /* outputs the error message */ 2226 else 2227 process_stats_detail(c, tokens[2].value); 2228 /* Output already generated */ 2229 return ; 2230 } else if (strcmp(subcommand, "settings") == 0) { 2231 process_stat_settings(&append_stats, c); 2232 } else if (strcmp(subcommand, "cachedump") == 0) { 2233 char *buf; 2234 unsigned int bytes, id, limit = 0; 2235 2236 if (ntokens < 5) { 2237 out_string(c, "CLIENT_ERROR bad command line"); 2238 return; 2239 } 2240 2241 if (!safe_strtoul(tokens[2].value, &id) || 2242 !safe_strtoul(tokens[3].value, &limit)) { 2243 out_string(c, "CLIENT_ERROR bad command line format"); 2244 return; 2245 } 2246 2247 if (id >= POWER_LARGEST) { 2248 out_string(c, "CLIENT_ERROR Illegal slab id"); 2249 return; 2250 } 2251 2252 buf = item_cachedump(id, limit, &bytes); 2253 write_and_free(c, buf, bytes); 2254 return ; 2255 } else { 2256 /* getting here means that the subcommand is either engine specific or 2257 is invalid. query the engine and see. */ 2258 if (get_stats(subcommand, strlen(subcommand), &append_stats, c)) { 2259 if (c->stats.buffer == NULL) { 2260 out_string(c, "SERVER_ERROR out of memory writing stats"); 2261 } else { 2262 write_and_free(c, c->stats.buffer, c->stats.offset); 2263 c->stats.buffer = NULL; 2264 } 2265 } else { 2266 out_string(c, "ERROR"); 2267 } 2268 return ; 2269 } 2270 2271 /* append terminator and start the transfer */ 2272 append_stats(NULL, 0, NULL, 0, c); 2273 2274 if (c->stats.buffer == NULL) { 2275 out_string(c, "SERVER_ERROR out of memory writing stats"); 2276 } else { 2277 write_and_free(c, c->stats.buffer, c->stats.offset); 2278 c->stats.buffer = NULL; 2279 } 2280 } 2281 2282 /* ntokens is overwritten here... shrug.. */ 2283 static inline void process_get_command(conn *c, token_t *tokens, size_t ntokens, bool return_cas) { 2284 char *key; 2285 size_t nkey; 2286 int i = 0; 2287 item *it; 2288 token_t *key_token = &tokens[KEY_TOKEN]; 2289 char *suffix; 2290 assert(c != NULL); 2291 2292 do { 2293 while(key_token->length != 0) { 2294 2295 key = key_token->value; 2296 nkey = key_token->length; 2297 2298 if(nkey > KEY_MAX_LENGTH) { 2299 out_string(c, "CLIENT_ERROR bad command line format"); 2300 return; 2301 } 2302 2303 it = item_get(key, nkey); 2304 if (settings.detail_enabled) { 2305 stats_prefix_record_get(key, nkey, NULL != it); 2306 } 2307 if (it) { 2308 if (i >= c->isize) { 2309 item **new_list = realloc(c->ilist, sizeof(item *) * c->isize * 2); 2310 if (new_list) { 2311 c->isize *= 2; 2312 c->ilist = new_list; 2313 } else { 2314 item_remove(it); 2315 break; 2316 } 2317 } 2318 2319 /* 2320 * Construct the response. Each hit adds three elements to the 2321 * outgoing data list: 2322 * "VALUE " 2323 * key 2324 * " " + flags + " " + data length + "\r\n" + data (with \r\n) 2325 */ 2326 2327 if (return_cas) 2328 { 2329 MEMCACHED_COMMAND_GET(c->sfd, ITEM_key(it), it->nkey, 2330 it->nbytes, ITEM_get_cas(it)); 2331 /* Goofy mid-flight realloc. */ 2332 if (i >= c->suffixsize) { 2333 char **new_suffix_list = realloc(c->suffixlist, 2334 sizeof(char *) * c->suffixsize * 2); 2335 if (new_suffix_list) { 2336 c->suffixsize *= 2; 2337 c->suffixlist = new_suffix_list; 2338 } else { 2339 item_remove(it); 2340 break; 2341 } 2342 } 2343 2344 suffix = cache_alloc(c->thread->suffix_cache); 2345 if (suffix == NULL) { 2346 out_string(c, "SERVER_ERROR out of memory making CAS suffix"); 2347 item_remove(it); 2348 return; 2349 } 2350 *(c->suffixlist + i) = suffix; 2351 int suffix_len = snprintf(suffix, SUFFIX_SIZE, 2352 " %llu\r\n", 2353 (unsigned long long)ITEM_get_cas(it)); 2354 if (add_iov(c, "VALUE ", 6) != 0 || 2355 add_iov(c, ITEM_key(it), it->nkey) != 0 || 2356 add_iov(c, ITEM_suffix(it), it->nsuffix - 2) != 0 || 2357 add_iov(c, suffix, suffix_len) != 0 || 2358 add_iov(c, ITEM_data(it), it->nbytes) != 0) 2359 { 2360 item_remove(it); 2361 break; 2362 } 2363 } 2364 else 2365 { 2366 MEMCACHED_COMMAND_GET(c->sfd, ITEM_key(it), it->nkey, 2367 it->nbytes, ITEM_get_cas(it)); 2368 if (add_iov(c, "VALUE ", 6) != 0 || 2369 add_iov(c, ITEM_key(it), it->nkey) != 0 || 2370 add_iov(c, ITEM_suffix(it), it->nsuffix + it->nbytes) != 0) 2371 { 2372 item_remove(it); 2373 break; 2374 } 2375 } 2376 2377 2378 if (settings.verbose > 1) 2379 fprintf(stderr, ">%d sending key %s\n", c->sfd, ITEM_key(it)); 2380 2381 /* item_get() has incremented it->refcount for us */ 2382 pthread_mutex_lock(&c->thread->stats.mutex); 2383 c->thread->stats.slab_stats[it->slabs_clsid].get_hits++; 2384 c->thread->stats.get_cmds++; 2385 pthread_mutex_unlock(&c->thread->stats.mutex); 2386 item_update(it); 2387 *(c->ilist + i) = it; 2388 i++; 2389 2390 } else { 2391 pthread_mutex_lock(&c->thread->stats.mutex); 2392 c->thread->stats.get_misses++; 2393 pthread_mutex_unlock(&c->thread->stats.mutex); 2394 MEMCACHED_COMMAND_GET(c->sfd, key, nkey, -1, 0); 2395 } 2396 2397 key_token++; 2398 } 2399 2400 /* 2401 * If the command string hasn't been fully processed, get the next set 2402 * of tokens. 2403 */ 2404 if(key_token->value != NULL) { 2405 ntokens = tokenize_command(key_token->value, tokens, MAX_TOKENS); 2406 key_token = tokens; 2407 } 2408 2409 } while(key_token->value != NULL); 2410 2411 c->icurr = c->ilist; 2412 c->ileft = i; 2413 if (return_cas) { 2414 c->suffixcurr = c->suffixlist; 2415 c->suffixleft = i; 2416 } 2417 2418 if (settings.verbose > 1) 2419 fprintf(stderr, ">%d END\n", c->sfd); 2420 2421 /* 2422 If the loop was terminated because of out-of-memory, it is not 2423 reliable to add END\r\n to the buffer, because it might not end 2424 in \r\n. So we send SERVER_ERROR instead. 2425 */ 2426 if (key_token->value != NULL || add_iov(c, "END\r\n", 5) != 0 2427 || (IS_UDP(c->transport) && build_udp_headers(c) != 0)) { 2428 out_string(c, "SERVER_ERROR out of memory writing get response"); 2429 } 2430 else { 2431 conn_set_state(c, conn_mwrite); 2432 c->msgcurr = 0; 2433 } 2434 2435 return; 2436 } 2437 2438 static void process_update_command(conn *c, token_t *tokens, const size_t ntokens, int comm, bool handle_cas) { 2439 char *key; 2440 size_t nkey; 2441 unsigned int flags; 2442 int32_t exptime_int = 0; 2443 time_t exptime; 2444 int vlen; 2445 uint64_t req_cas_id=0; 2446 item *it; 2447 2448 assert(c != NULL); 2449 2450 set_noreply_maybe(c, tokens, ntokens); 2451 2452 if (tokens[KEY_TOKEN].length > KEY_MAX_LENGTH) { 2453 out_string(c, "CLIENT_ERROR bad command line format"); 2454 return; 2455 } 2456 2457 key = tokens[KEY_TOKEN].value; 2458 nkey = tokens[KEY_TOKEN].length; 2459 2460 if (! (safe_strtoul(tokens[2].value, (uint32_t *)&flags) 2461 && safe_strtol(tokens[3].value, &exptime_int) 2462 && safe_strtol(tokens[4].value, (int32_t *)&vlen))) { 2463 out_string(c, "CLIENT_ERROR bad command line format"); 2464 return; 2465 } 2466 2467 /* Ubuntu 8.04 breaks when I pass exptime to safe_strtol */ 2468 exptime = exptime_int; 2469 2470 // does cas value exist? 2471 if (handle_cas) { 2472 if (!safe_strtoull(tokens[5].value, &req_cas_id)) { 2473 out_string(c, "CLIENT_ERROR bad command line format"); 2474 return; 2475 } 2476 } 2477 2478 vlen += 2; 2479 if (vlen < 0 || vlen - 2 < 0) { 2480 out_string(c, "CLIENT_ERROR bad command line format"); 2481 return; 2482 } 2483 2484 if (settings.detail_enabled) { 2485 stats_prefix_record_set(key, nkey); 2486 } 2487 2488 it = item_alloc(key, nkey, flags, realtime(exptime), vlen); 2489 2490 if (it == 0) { 2491 if (! item_size_ok(nkey, flags, vlen)) 2492 out_string(c, "SERVER_ERROR object too large for cache"); 2493 else 2494 out_string(c, "SERVER_ERROR out of memory storing object"); 2495 /* swallow the data line */ 2496 c->write_and_go = conn_swallow; 2497 c->sbytes = vlen; 2498 2499 /* Avoid stale data persisting in cache because we failed alloc. 2500 * Unacceptable for SET. Anywhere else too? */ 2501 if (comm == NREAD_SET) { 2502 it = item_get(key, nkey); 2503 if (it) { 2504 item_unlink(it); 2505 item_remove(it); 2506 } 2507 } 2508 2509 return; 2510 } 2511 ITEM_set_cas(it, req_cas_id); 2512 2513 c->item = it; 2514 c->ritem = ITEM_data(it); 2515 c->rlbytes = it->nbytes; 2516 c->cmd = comm; 2517 conn_set_state(c, conn_nread); 2518 } 2519 2520 static void process_arithmetic_command(conn *c, token_t *tokens, const size_t ntokens, const bool incr) { 2521 char temp[INCR_MAX_STORAGE_LEN]; 2522 item *it; 2523 uint64_t delta; 2524 char *key; 2525 size_t nkey; 2526 2527 assert(c != NULL); 2528 2529 set_noreply_maybe(c, tokens, ntokens); 2530 2531 if (tokens[KEY_TOKEN].length > KEY_MAX_LENGTH) { 2532 out_string(c, "CLIENT_ERROR bad command line format"); 2533 return; 2534 } 2535 2536 key = tokens[KEY_TOKEN].value; 2537 nkey = tokens[KEY_TOKEN].length; 2538 2539 if (!safe_strtoull(tokens[2].value, &delta)) { 2540 out_string(c, "CLIENT_ERROR invalid numeric delta argument"); 2541 return; 2542 } 2543 2544 it = item_get(key, nkey); 2545 if (!it) { 2546 pthread_mutex_lock(&c->thread->stats.mutex); 2547 if (incr) { 2548 c->thread->stats.incr_misses++; 2549 } else { 2550 c->thread->stats.decr_misses++; 2551 } 2552 pthread_mutex_unlock(&c->thread->stats.mutex); 2553 2554 out_string(c, "NOT_FOUND"); 2555 return; 2556 } 2557 2558 switch(add_delta(c, it, incr, delta, temp)) { 2559 case OK: 2560 out_string(c, temp); 2561 break; 2562 case NON_NUMERIC: 2563 out_string(c, "CLIENT_ERROR cannot increment or decrement non-numeric value"); 2564 break; 2565 case EOM: 2566 out_string(c, "SERVER_ERROR out of memory"); 2567 break; 2568 } 2569 item_remove(it); /* release our reference */ 2570 } 2571 2572 /* 2573 * adds a delta value to a numeric item. 2574 * 2575 * c connection requesting the operation 2576 * it item to adjust 2577 * incr true to increment value, false to decrement 2578 * delta amount to adjust value by 2579 * buf buffer for response string 2580 * 2581 * returns a response string to send back to the client. 2582 */ 2583 enum delta_result_type do_add_delta(conn *c, item *it, const bool incr, 2584 const int64_t delta, char *buf) { 2585 char *ptr; 2586 uint64_t value; 2587 int res; 2588 2589 ptr = ITEM_data(it); 2590 2591 if (!safe_strtoull(ptr, &value)) { 2592 return NON_NUMERIC; 2593 } 2594 2595 if (incr) { 2596 value += delta; 2597 MEMCACHED_COMMAND_INCR(c->sfd, ITEM_key(it), it->nkey, value); 2598 } else { 2599 if(delta > value) { 2600 value = 0; 2601 } else { 2602 value -= delta; 2603 } 2604 MEMCACHED_COMMAND_DECR(c->sfd, ITEM_key(it), it->nkey, value); 2605 } 2606 2607 pthread_mutex_lock(&c->thread->stats.mutex); 2608 if (incr) { 2609 c->thread->stats.slab_stats[it->slabs_clsid].incr_hits++; 2610 } else { 2611 c->thread->stats.slab_stats[it->slabs_clsid].decr_hits++; 2612 } 2613 pthread_mutex_unlock(&c->thread->stats.mutex); 2614 2615 snprintf(buf, INCR_MAX_STORAGE_LEN, "%llu", (unsigned long long)value); 2616 res = strlen(buf); 2617 if (res + 2 > it->nbytes) { /* need to realloc */ 2618 item *new_it; 2619 new_it = do_item_alloc(ITEM_key(it), it->nkey, atoi(ITEM_suffix(it) + 1), it->exptime, res + 2 ); 2620 if (new_it == 0) { 2621 return EOM; 2622 } 2623 memcpy(ITEM_data(new_it), buf, res); 2624 memcpy(ITEM_data(new_it) + res, "\r\n", 2); 2625 item_replace(it, new_it); 2626 do_item_remove(new_it); /* release our reference */ 2627 } else { /* replace in-place */ 2628 /* When changing the value without replacing the item, we 2629 need to update the CAS on the existing item. */ 2630 ITEM_set_cas(it, (settings.use_cas) ? get_cas_id() : 0); 2631 2632 memcpy(ITEM_data(it), buf, res); 2633 memset(ITEM_data(it) + res, ' ', it->nbytes - res - 2); 2634 } 2635 2636 return OK; 2637 } 2638 2639 static void process_delete_command(conn *c, token_t *tokens, const size_t ntokens) { 2640 char *key; 2641 size_t nkey; 2642 item *it; 2643 2644 assert(c != NULL); 2645 2646 set_noreply_maybe(c, tokens, ntokens); 2647 2648 key = tokens[KEY_TOKEN].value; 2649 nkey = tokens[KEY_TOKEN].length; 2650 2651 if(nkey > KEY_MAX_LENGTH) { 2652 out_string(c, "CLIENT_ERROR bad command line format"); 2653 return; 2654 } 2655 2656 if (settings.detail_enabled) { 2657 stats_prefix_record_delete(key, nkey); 2658 } 2659 2660 it = item_get(key, nkey); 2661 if (it) { 2662 MEMCACHED_COMMAND_DELETE(c->sfd, ITEM_key(it), it->nkey); 2663 2664 pthread_mutex_lock(&c->thread->stats.mutex); 2665 c->thread->stats.slab_stats[it->slabs_clsid].delete_hits++; 2666 pthread_mutex_unlock(&c->thread->stats.mutex); 2667 2668 item_unlink(it); 2669 item_remove(it); /* release our reference */ 2670 out_string(c, "DELETED"); 2671 } else { 2672 pthread_mutex_lock(&c->thread->stats.mutex); 2673 c->thread->stats.delete_misses++; 2674 pthread_mutex_unlock(&c->thread->stats.mutex); 2675 2676 out_string(c, "NOT_FOUND"); 2677 } 2678 } 2679 2680 static void process_verbosity_command(conn *c, token_t *tokens, const size_t ntokens) { 2681 unsigned int level; 2682 2683 assert(c != NULL); 2684 2685 set_noreply_maybe(c, tokens, ntokens); 2686 2687 level = strtoul(tokens[1].value, NULL, 10); 2688 settings.verbose = level > MAX_VERBOSITY_LEVEL ? MAX_VERBOSITY_LEVEL : level; 2689 out_string(c, "OK"); 2690 return; 2691 } 2692 2693 static void process_command(conn *c, char *command) { 2694 2695 token_t tokens[MAX_TOKENS]; 2696 size_t ntokens; 2697 int comm; 2698 2699 assert(c != NULL); 2700 2701 MEMCACHED_PROCESS_COMMAND_START(c->sfd, c->rcurr, c->rbytes); 2702 2703 if (settings.verbose > 1) 2704 fprintf(stderr, "<%d %s\n", c->sfd, command); 2705 2706 /* 2707 * for commands set/add/replace, we build an item and read the data 2708 * directly into it, then continue in nread_complete(). 2709 */ 2710 2711 c->msgcurr = 0; 2712 c->msgused = 0; 2713 c->iovused = 0; 2714 if (add_msghdr(c) != 0) { 2715 out_string(c, "SERVER_ERROR out of memory preparing response"); 2716 return; 2717 } 2718 2719 ntokens = tokenize_command(command, tokens, MAX_TOKENS); 2720 if (ntokens >= 3 && 2721 ((strcmp(tokens[COMMAND_TOKEN].value, "get") == 0) || 2722 (strcmp(tokens[COMMAND_TOKEN].value, "bget") == 0))) { 2723 2724 process_get_command(c, tokens, ntokens, false); 2725 2726 } else if ((ntokens == 6 || ntokens == 7) && 2727 ((strcmp(tokens[COMMAND_TOKEN].value, "add") == 0 && (comm = NREAD_ADD)) || 2728 (strcmp(tokens[COMMAND_TOKEN].value, "set") == 0 && (comm = NREAD_SET)) || 2729 (strcmp(tokens[COMMAND_TOKEN].value, "replace") == 0 && (comm = NREAD_REPLACE)) || 2730 (strcmp(tokens[COMMAND_TOKEN].value, "prepend") == 0 && (comm = NREAD_PREPEND)) || 2731 (strcmp(tokens[COMMAND_TOKEN].value, "append") == 0 && (comm = NREAD_APPEND)) )) { 2732 2733 process_update_command(c, tokens, ntokens, comm, false); 2734 2735 } else if ((ntokens == 7 || ntokens == 8) && (strcmp(tokens[COMMAND_TOKEN].value, "cas") == 0 && (comm = NREAD_CAS))) { 2736 2737 process_update_command(c, tokens, ntokens, comm, true); 2738 2739 } else if ((ntokens == 4 || ntokens == 5) && (strcmp(tokens[COMMAND_TOKEN].value, "incr") == 0)) { 2740 2741 process_arithmetic_command(c, tokens, ntokens, 1); 2742 2743 } else if (ntokens >= 3 && (strcmp(tokens[COMMAND_TOKEN].value, "gets") == 0)) { 2744 2745 process_get_command(c, tokens, ntokens, true); 2746 2747 } else if ((ntokens == 4 || ntokens == 5) && (strcmp(tokens[COMMAND_TOKEN].value, "decr") == 0)) { 2748 2749 process_arithmetic_command(c, tokens, ntokens, 0); 2750 2751 } else if (ntokens >= 3 && ntokens <= 4 && (strcmp(tokens[COMMAND_TOKEN].value, "delete") == 0)) { 2752 2753 process_delete_command(c, tokens, ntokens); 2754 2755 } else if (ntokens >= 2 && (strcmp(tokens[COMMAND_TOKEN].value, "stats") == 0)) { 2756 2757 process_stat(c, tokens, ntokens); 2758 2759 } else if (ntokens >= 2 && ntokens <= 4 && (strcmp(tokens[COMMAND_TOKEN].value, "flush_all") == 0)) { 2760 time_t exptime = 0; 2761 set_current_time(); 2762 2763 set_noreply_maybe(c, tokens, ntokens); 2764 2765 pthread_mutex_lock(&c->thread->stats.mutex); 2766 c->thread->stats.flush_cmds++; 2767 pthread_mutex_unlock(&c->thread->stats.mutex); 2768 2769 if(ntokens == (c->noreply ? 3 : 2)) { 2770 settings.oldest_live = current_time - 1; 2771 item_flush_expired(); 2772 out_string(c, "OK"); 2773 return; 2774 } 2775 2776 exptime = strtol(tokens[1].value, NULL, 10); 2777 if(errno == ERANGE) { 2778 out_string(c, "CLIENT_ERROR bad command line format"); 2779 return; 2780 } 2781 2782 /* 2783 If exptime is zero realtime() would return zero too, and 2784 realtime(exptime) - 1 would overflow to the max unsigned 2785 value. So we process exptime == 0 the same way we do when 2786 no delay is given at all. 2787 */ 2788 if (exptime > 0) 2789 settings.oldest_live = realtime(exptime) - 1; 2790 else /* exptime == 0 */ 2791 settings.oldest_live = current_time - 1; 2792 item_flush_expired(); 2793 out_string(c, "OK"); 2794 return; 2795 2796 } else if (ntokens == 2 && (strcmp(tokens[COMMAND_TOKEN].value, "version") == 0)) { 2797 2798 out_string(c, "VERSION " VERSION); 2799 2800 } else if (ntokens == 2 && (strcmp(tokens[COMMAND_TOKEN].value, "quit") == 0)) { 2801 2802 conn_set_state(c, conn_closing); 2803 2804 } else if (ntokens == 5 && (strcmp(tokens[COMMAND_TOKEN].value, "slabs") == 0 && 2805 strcmp(tokens[COMMAND_TOKEN + 1].value, "reassign") == 0)) { 2806 #ifdef ALLOW_SLABS_REASSIGN 2807 2808 int src, dst, rv; 2809 2810 src = strtol(tokens[2].value, NULL, 10); 2811 dst = strtol(tokens[3].value, NULL, 10); 2812 2813 if(errno == ERANGE) { 2814 out_string(c, "CLIENT_ERROR bad command line format"); 2815 return; 2816 } 2817 2818 rv = slabs_reassign(src, dst); 2819 if (rv == 1) { 2820 out_string(c, "DONE"); 2821 return; 2822 } 2823 if (rv == 0) { 2824 out_string(c, "CANT"); 2825 return; 2826 } 2827 if (rv == -1) { 2828 out_string(c, "BUSY"); 2829 return; 2830 } 2831 #else 2832 out_string(c, "CLIENT_ERROR Slab reassignment not supported"); 2833 #endif 2834 } else if ((ntokens == 3 || ntokens == 4) && (strcmp(tokens[COMMAND_TOKEN].value, "verbosity") == 0)) { 2835 process_verbosity_command(c, tokens, ntokens); 2836 } else { 2837 out_string(c, "ERROR"); 2838 } 2839 return; 2840 } 2841 2842 /* 2843 * if we have a complete line in the buffer, process it. 2844 */ 2845 static int try_read_command(conn *c) { 2846 assert(c != NULL); 2847 assert(c->rcurr <= (c->rbuf + c->rsize)); 2848 assert(c->rbytes > 0); 2849 2850 if (c->protocol == negotiating_prot || c->transport == udp_transport) { 2851 if ((unsigned char)c->rbuf[0] == (unsigned char)PROTOCOL_BINARY_REQ) { 2852 c->protocol = binary_prot; 2853 } else { 2854 c->protocol = ascii_prot; 2855 } 2856 2857 if (settings.verbose) { 2858 fprintf(stderr, "%d: Client using the %s protocol\n", c->sfd, 2859 prot_text(c->protocol)); 2860 } 2861 } 2862 2863 if (c->protocol == binary_prot) { 2864 /* Do we have the complete packet header? */ 2865 if (c->rbytes < sizeof(c->binary_header)) { 2866 /* need more data! */ 2867 return 0; 2868 } else { 2869 #ifdef NEED_ALIGN 2870 if (((long)(c->rcurr)) % 8 != 0) { 2871 /* must realign input buffer */ 2872 memmove(c->rbuf, c->rcurr, c->rbytes); 2873 c->rcurr = c->rbuf; 2874 if (settings.verbose) { 2875 fprintf(stderr, "%d: Realign input buffer\n", c->sfd); 2876 } 2877 } 2878 #endif 2879 protocol_binary_request_header* req; 2880 req = (protocol_binary_request_header*)c->rcurr; 2881 2882 if (settings.verbose > 1) { 2883 /* Dump the packet before we convert it to host order */ 2884 int ii; 2885 fprintf(stderr, "<%d Read binary protocol data:", c->sfd); 2886 for (ii = 0; ii < sizeof(req->bytes); ++ii) { 2887 if (ii % 4 == 0) { 2888 fprintf(stderr, "\n<%d ", c->sfd); 2889 } 2890 fprintf(stderr, " 0x%02x", req->bytes[ii]); 2891 } 2892 fprintf(stderr, "\n"); 2893 } 2894 2895 c->binary_header = *req; 2896 c->binary_header.request.keylen = ntohs(req->request.keylen); 2897 c->binary_header.request.bodylen = ntohl(req->request.bodylen); 2898 c->binary_header.request.cas = ntohll(req->request.cas); 2899 2900 if (c->binary_header.request.magic != PROTOCOL_BINARY_REQ) { 2901 if (settings.verbose) { 2902 fprintf(stderr, "Invalid magic: %x\n", 2903 c->binary_header.request.magic); 2904 } 2905 conn_set_state(c, conn_closing); 2906 return -1; 2907 } 2908 2909 c->msgcurr = 0; 2910 c->msgused = 0; 2911 c->iovused = 0; 2912 if (add_msghdr(c) != 0) { 2913 out_string(c, "SERVER_ERROR out of memory"); 2914 return 0; 2915 } 2916 2917 c->cmd = c->binary_header.request.opcode; 2918 c->keylen = c->binary_header.request.keylen; 2919 c->opaque = c->binary_header.request.opaque; 2920 /* clear the returned cas value */ 2921 c->cas = 0; 2922 2923 dispatch_bin_command(c); 2924 2925 c->rbytes -= sizeof(c->binary_header); 2926 c->rcurr += sizeof(c->binary_header); 2927 } 2928 } else { 2929 char *el, *cont; 2930 2931 if (c->rbytes == 0) 2932 return 0; 2933 el = memchr(c->rcurr, '\n', c->rbytes); 2934 if (!el) 2935 return 0; 2936 cont = el + 1; 2937 if ((el - c->rcurr) > 1 && *(el - 1) == '\r') { 2938 el--; 2939 } 2940 *el = '\0'; 2941 2942 assert(cont <= (c->rcurr + c->rbytes)); 2943 2944 process_command(c, c->rcurr); 2945 2946 c->rbytes -= (cont - c->rcurr); 2947 c->rcurr = cont; 2948 2949 assert(c->rcurr <= (c->rbuf + c->rsize)); 2950 } 2951 2952 return 1; 2953 } 2954 2955 /* 2956 * read a UDP request. 2957 */ 2958 static enum try_read_result try_read_udp(conn *c) { 2959 int res; 2960 2961 assert(c != NULL); 2962 2963 c->request_addr_size = sizeof(c->request_addr); 2964 res = recvfrom(c->sfd, c->rbuf, c->rsize, 2965 0, &c->request_addr, &c->request_addr_size); 2966 if (res > 8) { 2967 unsigned char *buf = (unsigned char *)c->rbuf; 2968 pthread_mutex_lock(&c->thread->stats.mutex); 2969 c->thread->stats.bytes_read += res; 2970 pthread_mutex_unlock(&c->thread->stats.mutex); 2971 2972 /* Beginning of UDP packet is the request ID; save it. */ 2973 c->request_id = buf[0] * 256 + buf[1]; 2974 2975 /* If this is a multi-packet request, drop it. */ 2976 if (buf[4] != 0 || buf[5] != 1) { 2977 out_string(c, "SERVER_ERROR multi-packet request not supported"); 2978 return READ_NO_DATA_RECEIVED; 2979 } 2980 2981 /* Don't care about any of the rest of the header. */ 2982 res -= 8; 2983 memmove(c->rbuf, c->rbuf + 8, res); 2984 2985 c->rbytes += res; 2986 c->rcurr = c->rbuf; 2987 return READ_DATA_RECEIVED; 2988 } 2989 return READ_NO_DATA_RECEIVED; 2990 } 2991 2992 /* 2993 * read from network as much as we can, handle buffer overflow and connection 2994 * close. 2995 * before reading, move the remaining incomplete fragment of a command 2996 * (if any) to the beginning of the buffer. 2997 * @return enum try_read_result 2998 */ 2999 static enum try_read_result try_read_network(conn *c) { 3000 enum try_read_result gotdata = READ_NO_DATA_RECEIVED; 3001 int res; 3002 3003 assert(c != NULL); 3004 3005 if (c->rcurr != c->rbuf) { 3006 if (c->rbytes != 0) /* otherwise there's nothing to copy */ 3007 memmove(c->rbuf, c->rcurr, c->rbytes); 3008 c->rcurr = c->rbuf; 3009 } 3010 3011 while (1) { 3012 if (c->rbytes >= c->rsize) { 3013 char *new_rbuf = realloc(c->rbuf, c->rsize * 2); 3014 if (!new_rbuf) { 3015 if (settings.verbose > 0) 3016 fprintf(stderr, "Couldn't realloc input buffer\n"); 3017 c->rbytes = 0; /* ignore what we read */ 3018 out_string(c, "SERVER_ERROR out of memory reading request"); 3019 c->write_and_go = conn_closing; 3020 return READ_MEMORY_ERROR; 3021 } 3022 c->rcurr = c->rbuf = new_rbuf; 3023 c->rsize *= 2; 3024 } 3025 3026 int avail = c->rsize - c->rbytes; 3027 res = read(c->sfd, c->rbuf + c->rbytes, avail); 3028 if (res > 0) { 3029 pthread_mutex_lock(&c->thread->stats.mutex); 3030 c->thread->stats.bytes_read += res; 3031 pthread_mutex_unlock(&c->thread->stats.mutex); 3032 gotdata = READ_DATA_RECEIVED; 3033 c->rbytes += res; 3034 if (res == avail) { 3035 continue; 3036 } else { 3037 break; 3038 } 3039 } 3040 if (res == 0) { 3041 return READ_ERROR; 3042 } 3043 if (res == -1) { 3044 if (errno == EAGAIN || errno == EWOULDBLOCK) { 3045 break; 3046 } 3047 return READ_ERROR; 3048 } 3049 } 3050 return gotdata; 3051 } 3052 3053 static bool update_event(conn *c, const int new_flags) { 3054 assert(c != NULL); 3055 3056 struct event_base *base = c->event.ev_base; 3057 if (c->ev_flags == new_flags) 3058 return true; 3059 if (event_del(&c->event) == -1) return false; 3060 event_set(&c->event, c->sfd, new_flags, event_handler, (void *)c); 3061 event_base_set(base, &c->event); 3062 c->ev_flags = new_flags; 3063 if (event_add(&c->event, 0) == -1) return false; 3064 return true; 3065 } 3066 3067 /* 3068 * Sets whether we are listening for new connections or not. 3069 */ 3070 void do_accept_new_conns(const bool do_accept) { 3071 conn *next; 3072 3073 for (next = listen_conn; next; next = next->next) { 3074 if (do_accept) { 3075 update_event(next, EV_READ | EV_PERSIST); 3076 if (listen(next->sfd, settings.backlog) != 0) { 3077 perror("listen"); 3078 } 3079 } 3080 else { 3081 update_event(next, 0); 3082 if (listen(next->sfd, 0) != 0) { 3083 perror("listen"); 3084 } 3085 } 3086 } 3087 3088 if (do_accept) { 3089 STATS_LOCK(); 3090 stats.accepting_conns = true; 3091 STATS_UNLOCK(); 3092 } else { 3093 STATS_LOCK(); 3094 stats.accepting_conns = false; 3095 stats.listen_disabled_num++; 3096 STATS_UNLOCK(); 3097 } 3098 } 3099 3100 /* 3101 * Transmit the next chunk of data from our list of msgbuf structures. 3102 * 3103 * Returns: 3104 * TRANSMIT_COMPLETE All done writing. 3105 * TRANSMIT_INCOMPLETE More data remaining to write. 3106 * TRANSMIT_SOFT_ERROR Can't write any more right now. 3107 * TRANSMIT_HARD_ERROR Can't write (c->state is set to conn_closing) 3108 */ 3109 static enum transmit_result transmit(conn *c) { 3110 assert(c != NULL); 3111 3112 if (c->msgcurr < c->msgused && 3113 c->msglist[c->msgcurr].msg_iovlen == 0) { 3114 /* Finished writing the current msg; advance to the next. */ 3115 c->msgcurr++; 3116 } 3117 if (c->msgcurr < c->msgused) { 3118 ssize_t res; 3119 struct msghdr *m = &c->msglist[c->msgcurr]; 3120 3121 res = sendmsg(c->sfd, m, 0); 3122 if (res > 0) { 3123 pthread_mutex_lock(&c->thread->stats.mutex); 3124 c->thread->stats.bytes_written += res; 3125 pthread_mutex_unlock(&c->thread->stats.mutex); 3126 3127 /* We've written some of the data. Remove the completed 3128 iovec entries from the list of pending writes. */ 3129 while (m->msg_iovlen > 0 && res >= m->msg_iov->iov_len) { 3130 res -= m->msg_iov->iov_len; 3131 m->msg_iovlen--; 3132 m->msg_iov++; 3133 } 3134 3135 /* Might have written just part of the last iovec entry; 3136 adjust it so the next write will do the rest. */ 3137 if (res > 0) { 3138 m->msg_iov->iov_base = (caddr_t)m->msg_iov->iov_base + res; 3139 m->msg_iov->iov_len -= res; 3140 } 3141 return TRANSMIT_INCOMPLETE; 3142 } 3143 if (res == -1 && (errno == EAGAIN || errno == EWOULDBLOCK)) { 3144 if (!update_event(c, EV_WRITE | EV_PERSIST)) { 3145 if (settings.verbose > 0) 3146 fprintf(stderr, "Couldn't update event\n"); 3147 conn_set_state(c, conn_closing); 3148 return TRANSMIT_HARD_ERROR; 3149 } 3150 return TRANSMIT_SOFT_ERROR; 3151 } 3152 /* if res == 0 or res == -1 and error is not EAGAIN or EWOULDBLOCK, 3153 we have a real error, on which we close the connection */ 3154 if (settings.verbose > 0) 3155 perror("Failed to write, and not due to blocking"); 3156 3157 if (IS_UDP(c->transport)) 3158 conn_set_state(c, conn_read); 3159 else 3160 conn_set_state(c, conn_closing); 3161 return TRANSMIT_HARD_ERROR; 3162 } else { 3163 return TRANSMIT_COMPLETE; 3164 } 3165 } 3166 3167 static void drive_machine(conn *c) { 3168 bool stop = false; 3169 int sfd, flags = 1; 3170 socklen_t addrlen; 3171 struct sockaddr_storage addr; 3172 int nreqs = settings.reqs_per_event; 3173 int res; 3174 3175 assert(c != NULL); 3176 3177 while (!stop) { 3178 3179 switch(c->state) { 3180 case conn_listening: 3181 addrlen = sizeof(addr); 3182 if ((sfd = accept(c->sfd, (struct sockaddr *)&addr, &addrlen)) == -1) { 3183 if (errno == EAGAIN || errno == EWOULDBLOCK) { 3184 /* these are transient, so don't log anything */ 3185 stop = true; 3186 } else if (errno == EMFILE) { 3187 if (settings.verbose > 0) 3188 fprintf(stderr, "Too many open connections\n"); 3189 accept_new_conns(false); 3190 stop = true; 3191 } else { 3192 perror("accept()"); 3193 stop = true; 3194 } 3195 break; 3196 } 3197 if ((flags = fcntl(sfd, F_GETFL, 0)) < 0 || 3198 fcntl(sfd, F_SETFL, flags | O_NONBLOCK) < 0) { 3199 perror("setting O_NONBLOCK"); 3200 close(sfd); 3201 break; 3202 } 3203 3204 dispatch_conn_new(sfd, conn_new_cmd, EV_READ | EV_PERSIST, 3205 DATA_BUFFER_SIZE, tcp_transport); 3206 stop = true; 3207 break; 3208 3209 case conn_waiting: 3210 if (!update_event(c, EV_READ | EV_PERSIST)) { 3211 if (settings.verbose > 0) 3212 fprintf(stderr, "Couldn't update event\n"); 3213 conn_set_state(c, conn_closing); 3214 break; 3215 } 3216 3217 conn_set_state(c, conn_read); 3218 stop = true; 3219 break; 3220 3221 case conn_read: 3222 res = IS_UDP(c->transport) ? try_read_udp(c) : try_read_network(c); 3223 3224 switch (res) { 3225 case READ_NO_DATA_RECEIVED: 3226 conn_set_state(c, conn_waiting); 3227 break; 3228 case READ_DATA_RECEIVED: 3229 conn_set_state(c, conn_parse_cmd); 3230 break; 3231 case READ_ERROR: 3232 conn_set_state(c, conn_closing); 3233 break; 3234 case READ_MEMORY_ERROR: /* Failed to allocate more memory */ 3235 /* State already set by try_read_network */ 3236 break; 3237 } 3238 break; 3239 3240 case conn_parse_cmd : 3241 if (try_read_command(c) == 0) { 3242 /* wee need more data! */ 3243 conn_set_state(c, conn_waiting); 3244 } 3245 3246 break; 3247 3248 case conn_new_cmd: 3249 /* Only process nreqs at a time to avoid starving other 3250 connections */ 3251 3252 --nreqs; 3253 if (nreqs >= 0) { 3254 reset_cmd_handler(c); 3255 } else { 3256 pthread_mutex_lock(&c->thread->stats.mutex); 3257 c->thread->stats.conn_yields++; 3258 pthread_mutex_unlock(&c->thread->stats.mutex); 3259 if (c->rbytes > 0) { 3260 /* We have already read in data into the input buffer, 3261 so libevent will most likely not signal read events 3262 on the socket (unless more data is available. As a 3263 hack we should just put in a request to write data, 3264 because that should be possible ;-) 3265 */ 3266 if (!update_event(c, EV_WRITE | EV_PERSIST)) { 3267 if (settings.verbose > 0) 3268 fprintf(stderr, "Couldn't update event\n"); 3269 conn_set_state(c, conn_closing); 3270 } 3271 } 3272 stop = true; 3273 } 3274 break; 3275 3276 case conn_nread: 3277 if (c->rlbytes == 0) { 3278 complete_nread(c); 3279 break; 3280 } 3281 /* first check if we have leftovers in the conn_read buffer */ 3282 if (c->rbytes > 0) { 3283 int tocopy = c->rbytes > c->rlbytes ? c->rlbytes : c->rbytes; 3284 if (c->ritem != c->rcurr) { 3285 memmove(c->ritem, c->rcurr, tocopy); 3286 } 3287 c->ritem += tocopy; 3288 c->rlbytes -= tocopy; 3289 c->rcurr += tocopy; 3290 c->rbytes -= tocopy; 3291 if (c->rlbytes == 0) { 3292 break; 3293 } 3294 } 3295 3296 /* now try reading from the socket */ 3297 res = read(c->sfd, c->ritem, c->rlbytes); 3298 if (res > 0) { 3299 pthread_mutex_lock(&c->thread->stats.mutex); 3300 c->thread->stats.bytes_read += res; 3301 pthread_mutex_unlock(&c->thread->stats.mutex); 3302 if (c->rcurr == c->ritem) { 3303 c->rcurr += res; 3304 } 3305 c->ritem += res; 3306 c->rlbytes -= res; 3307 break; 3308 } 3309 if (res == 0) { /* end of stream */ 3310 conn_set_state(c, conn_closing); 3311 break; 3312 } 3313 if (res == -1 && (errno == EAGAIN || errno == EWOULDBLOCK)) { 3314 if (!update_event(c, EV_READ | EV_PERSIST)) { 3315 if (settings.verbose > 0) 3316 fprintf(stderr, "Couldn't update event\n"); 3317 conn_set_state(c, conn_closing); 3318 break; 3319 } 3320 stop = true; 3321 break; 3322 } 3323 /* otherwise we have a real error, on which we close the connection */ 3324 if (settings.verbose > 0) { 3325 fprintf(stderr, "Failed to read, and not due to blocking:\n" 3326 "errno: %d %s \n" 3327 "rcurr=%lx ritem=%lx rbuf=%lx rlbytes=%d rsize=%d\n", 3328 errno, strerror(errno), 3329 (long)c->rcurr, (long)c->ritem, (long)c->rbuf, 3330 (int)c->rlbytes, (int)c->rsize); 3331 } 3332 conn_set_state(c, conn_closing); 3333 break; 3334 3335 case conn_swallow: 3336 /* we are reading sbytes and throwing them away */ 3337 if (c->sbytes == 0) { 3338 conn_set_state(c, conn_new_cmd); 3339 break; 3340 } 3341 3342 /* first check if we have leftovers in the conn_read buffer */ 3343 if (c->rbytes > 0) { 3344 int tocopy = c->rbytes > c->sbytes ? c->sbytes : c->rbytes; 3345 c->sbytes -= tocopy; 3346 c->rcurr += tocopy; 3347 c->rbytes -= tocopy; 3348 break; 3349 } 3350 3351 /* now try reading from the socket */ 3352 res = read(c->sfd, c->rbuf, c->rsize > c->sbytes ? c->sbytes : c->rsize); 3353 if (res > 0) { 3354 pthread_mutex_lock(&c->thread->stats.mutex); 3355 c->thread->stats.bytes_read += res; 3356 pthread_mutex_unlock(&c->thread->stats.mutex); 3357 c->sbytes -= res; 3358 break; 3359 } 3360 if (res == 0) { /* end of stream */ 3361 conn_set_state(c, conn_closing); 3362 break; 3363 } 3364 if (res == -1 && (errno == EAGAIN || errno == EWOULDBLOCK)) { 3365 if (!update_event(c, EV_READ | EV_PERSIST)) { 3366 if (settings.verbose > 0) 3367 fprintf(stderr, "Couldn't update event\n"); 3368 conn_set_state(c, conn_closing); 3369 break; 3370 } 3371 stop = true; 3372 break; 3373 } 3374 /* otherwise we have a real error, on which we close the connection */ 3375 if (settings.verbose > 0) 3376 fprintf(stderr, "Failed to read, and not due to blocking\n"); 3377 conn_set_state(c, conn_closing); 3378 break; 3379 3380 case conn_write: 3381 /* 3382 * We want to write out a simple response. If we haven't already, 3383 * assemble it into a msgbuf list (this will be a single-entry 3384 * list for TCP or a two-entry list for UDP). 3385 */ 3386 if (c->iovused == 0 || (IS_UDP(c->transport) && c->iovused == 1)) { 3387 if (add_iov(c, c->wcurr, c->wbytes) != 0) { 3388 if (settings.verbose > 0) 3389 fprintf(stderr, "Couldn't build response\n"); 3390 conn_set_state(c, conn_closing); 3391 break; 3392 } 3393 } 3394 3395 /* fall through... */ 3396 3397 case conn_mwrite: 3398 if (IS_UDP(c->transport) && c->msgcurr == 0 && build_udp_headers(c) != 0) { 3399 if (settings.verbose > 0) 3400 fprintf(stderr, "Failed to build UDP headers\n"); 3401 conn_set_state(c, conn_closing); 3402 break; 3403 } 3404 switch (transmit(c)) { 3405 case TRANSMIT_COMPLETE: 3406 if (c->state == conn_mwrite) { 3407 while (c->ileft > 0) { 3408 item *it = *(c->icurr); 3409 assert((it->it_flags & ITEM_SLABBED) == 0); 3410 item_remove(it); 3411 c->icurr++; 3412 c->ileft--; 3413 } 3414 while (c->suffixleft > 0) { 3415 char *suffix = *(c->suffixcurr); 3416 cache_free(c->thread->suffix_cache, suffix); 3417 c->suffixcurr++; 3418 c->suffixleft--; 3419 } 3420 /* XXX: I don't know why this wasn't the general case */ 3421 if(c->protocol == binary_prot) { 3422 conn_set_state(c, c->write_and_go); 3423 } else { 3424 conn_set_state(c, conn_new_cmd); 3425 } 3426 } else if (c->state == conn_write) { 3427 if (c->write_and_free) { 3428 free(c->write_and_free); 3429 c->write_and_free = 0; 3430 } 3431 conn_set_state(c, c->write_and_go); 3432 } else { 3433 if (settings.verbose > 0) 3434 fprintf(stderr, "Unexpected state %d\n", c->state); 3435 conn_set_state(c, conn_closing); 3436 } 3437 break; 3438 3439 case TRANSMIT_INCOMPLETE: 3440 case TRANSMIT_HARD_ERROR: 3441 break; /* Continue in state machine. */ 3442 3443 case TRANSMIT_SOFT_ERROR: 3444 stop = true; 3445 break; 3446 } 3447 break; 3448 3449 case conn_closing: 3450 if (IS_UDP(c->transport)) 3451 conn_cleanup(c); 3452 else 3453 conn_close(c); 3454 stop = true; 3455 break; 3456 3457 case conn_max_state: 3458 assert(false); 3459 break; 3460 } 3461 } 3462 3463 return; 3464 } 3465 3466 void event_handler(const int fd, const short which, void *arg) { 3467 conn *c; 3468 3469 c = (conn *)arg; 3470 assert(c != NULL); 3471 3472 c->which = which; 3473 3474 /* sanity */ 3475 if (fd != c->sfd) { 3476 if (settings.verbose > 0) 3477 fprintf(stderr, "Catastrophic: event fd doesn't match conn fd!\n"); 3478 conn_close(c); 3479 return; 3480 } 3481 3482 drive_machine(c); 3483 3484 /* wait for next event */ 3485 return; 3486 } 3487 3488 static int new_socket(struct addrinfo *ai) { 3489 int sfd; 3490 int flags; 3491 3492 if ((sfd = socket(ai->ai_family, ai->ai_socktype, ai->ai_protocol)) == -1) { 3493 return -1; 3494 } 3495 3496 if ((flags = fcntl(sfd, F_GETFL, 0)) < 0 || 3497 fcntl(sfd, F_SETFL, flags | O_NONBLOCK) < 0) { 3498 perror("setting O_NONBLOCK"); 3499 close(sfd); 3500 return -1; 3501 } 3502 return sfd; 3503 } 3504 3505 3506 /* 3507 * Sets a socket's send buffer size to the maximum allowed by the system. 3508 */ 3509 static void maximize_sndbuf(const int sfd) { 3510 socklen_t intsize = sizeof(int); 3511 int last_good = 0; 3512 int min, max, avg; 3513 int old_size; 3514 3515 /* Start with the default size. */ 3516 if (getsockopt(sfd, SOL_SOCKET, SO_SNDBUF, &old_size, &intsize) != 0) { 3517 if (settings.verbose > 0) 3518 perror("getsockopt(SO_SNDBUF)"); 3519 return; 3520 } 3521 3522 /* Binary-search for the real maximum. */ 3523 min = old_size; 3524 max = MAX_SENDBUF_SIZE; 3525 3526 while (min <= max) { 3527 avg = ((unsigned int)(min + max)) / 2; 3528 if (setsockopt(sfd, SOL_SOCKET, SO_SNDBUF, (void *)&avg, intsize) == 0) { 3529 last_good = avg; 3530 min = avg + 1; 3531 } else { 3532 max = avg - 1; 3533 } 3534 } 3535 3536 if (settings.verbose > 1) 3537 fprintf(stderr, "<%d send buffer was %d, now %d\n", sfd, old_size, last_good); 3538 } 3539 3540 /** 3541 * Create a socket and bind it to a specific port number 3542 * @param port the port number to bind to 3543 * @param transport the transport protocol (TCP / UDP) 3544 * @param portnumber_file A filepointer to write the port numbers to 3545 * when they are successfully added to the list of ports we 3546 * listen on. 3547 */ 3548 static int server_socket(int port, enum network_transport transport, 3549 FILE *portnumber_file) { 3550 int sfd; 3551 struct linger ling = {0, 0}; 3552 struct addrinfo *ai; 3553 struct addrinfo *next; 3554 struct addrinfo hints = { .ai_flags = AI_PASSIVE, 3555 .ai_family = AF_UNSPEC }; 3556 char port_buf[NI_MAXSERV]; 3557 int error; 3558 int success = 0; 3559 int flags =1; 3560 3561 hints.ai_socktype = IS_UDP(transport) ? SOCK_DGRAM : SOCK_STREAM; 3562 3563 if (port == -1) { 3564 port = 0; 3565 } 3566 snprintf(port_buf, sizeof(port_buf), "%d", port); 3567 error= getaddrinfo(settings.inter, port_buf, &hints, &ai); 3568 if (error != 0) { 3569 if (error != EAI_SYSTEM) 3570 fprintf(stderr, "getaddrinfo(): %s\n", gai_strerror(error)); 3571 else 3572 perror("getaddrinfo()"); 3573 return 1; 3574 } 3575 3576 for (next= ai; next; next= next->ai_next) { 3577 conn *listen_conn_add; 3578 if ((sfd = new_socket(next)) == -1) { 3579 /* getaddrinfo can return "junk" addresses, 3580 * we make sure at least one works before erroring. 3581 */ 3582 continue; 3583 } 3584 3585 #ifdef IPV6_V6ONLY 3586 if (next->ai_family == AF_INET6) { 3587 error = setsockopt(sfd, IPPROTO_IPV6, IPV6_V6ONLY, (char *) &flags, sizeof(flags)); 3588 if (error != 0) { 3589 perror("setsockopt"); 3590 close(sfd); 3591 continue; 3592 } 3593 } 3594 #endif 3595 3596 setsockopt(sfd, SOL_SOCKET, SO_REUSEADDR, (void *)&flags, sizeof(flags)); 3597 if (IS_UDP(transport)) { 3598 maximize_sndbuf(sfd); 3599 } else { 3600 error = setsockopt(sfd, SOL_SOCKET, SO_KEEPALIVE, (void *)&flags, sizeof(flags)); 3601 if (error != 0) 3602 perror("setsockopt"); 3603 3604 error = setsockopt(sfd, SOL_SOCKET, SO_LINGER, (void *)&ling, sizeof(ling)); 3605 if (error != 0) 3606 perror("setsockopt"); 3607 3608 error = setsockopt(sfd, IPPROTO_TCP, TCP_NODELAY, (void *)&flags, sizeof(flags)); 3609 if (error != 0) 3610 perror("setsockopt"); 3611 } 3612 3613 if (bind(sfd, next->ai_addr, next->ai_addrlen) == -1) { 3614 if (errno != EADDRINUSE) { 3615 perror("bind()"); 3616 close(sfd); 3617 freeaddrinfo(ai); 3618 return 1; 3619 } 3620 close(sfd); 3621 continue; 3622 } else { 3623 success++; 3624 if (!IS_UDP(transport) && listen(sfd, settings.backlog) == -1) { 3625 perror("listen()"); 3626 close(sfd); 3627 freeaddrinfo(ai); 3628 return 1; 3629 } 3630 if (portnumber_file != NULL && 3631 (next->ai_addr->sa_family == AF_INET || 3632 next->ai_addr->sa_family == AF_INET6)) { 3633 union { 3634 struct sockaddr_in in; 3635 struct sockaddr_in6 in6; 3636 } my_sockaddr; 3637 socklen_t len = sizeof(my_sockaddr); 3638 if (getsockname(sfd, (struct sockaddr*)&my_sockaddr, &len)==0) { 3639 if (next->ai_addr->sa_family == AF_INET) { 3640 fprintf(portnumber_file, "%s INET: %u\n", 3641 IS_UDP(transport) ? "UDP" : "TCP", 3642 ntohs(my_sockaddr.in.sin_port)); 3643 } else { 3644 fprintf(portnumber_file, "%s INET6: %u\n", 3645 IS_UDP(transport) ? "UDP" : "TCP", 3646 ntohs(my_sockaddr.in6.sin6_port)); 3647 } 3648 } 3649 } 3650 } 3651 3652 if (IS_UDP(transport)) { 3653 int c; 3654 3655 for (c = 0; c < settings.num_threads; c++) { 3656 /* this is guaranteed to hit all threads because we round-robin */ 3657 dispatch_conn_new(sfd, conn_read, EV_READ | EV_PERSIST, 3658 UDP_READ_BUFFER_SIZE, transport); 3659 } 3660 } else { 3661 if (!(listen_conn_add = conn_new(sfd, conn_listening, 3662 EV_READ | EV_PERSIST, 1, 3663 transport, main_base))) { 3664 fprintf(stderr, "failed to create listening connection\n"); 3665 exit(EXIT_FAILURE); 3666 } 3667 listen_conn_add->next = listen_conn; 3668 listen_conn = listen_conn_add; 3669 } 3670 } 3671 3672 freeaddrinfo(ai); 3673 3674 /* Return zero iff we detected no errors in starting up connections */ 3675 return success == 0; 3676 } 3677 3678 static int new_socket_unix(void) { 3679 int sfd; 3680 int flags; 3681 3682 if ((sfd = socket(AF_UNIX, SOCK_STREAM, 0)) == -1) { 3683 perror("socket()"); 3684 return -1; 3685 } 3686 3687 if ((flags = fcntl(sfd, F_GETFL, 0)) < 0 || 3688 fcntl(sfd, F_SETFL, flags | O_NONBLOCK) < 0) { 3689 perror("setting O_NONBLOCK"); 3690 close(sfd); 3691 return -1; 3692 } 3693 return sfd; 3694 } 3695 3696 static int server_socket_unix(const char *path, int access_mask) { 3697 int sfd; 3698 struct linger ling = {0, 0}; 3699 struct sockaddr_un addr; 3700 struct stat tstat; 3701 int flags =1; 3702 int old_umask; 3703 3704 if (!path) { 3705 return 1; 3706 } 3707 3708 if ((sfd = new_socket_unix()) == -1) { 3709 return 1; 3710 } 3711 3712 /* 3713 * Clean up a previous socket file if we left it around 3714 */ 3715 if (lstat(path, &tstat) == 0) { 3716 if (S_ISSOCK(tstat.st_mode)) 3717 unlink(path); 3718 } 3719 3720 setsockopt(sfd, SOL_SOCKET, SO_REUSEADDR, (void *)&flags, sizeof(flags)); 3721 setsockopt(sfd, SOL_SOCKET, SO_KEEPALIVE, (void *)&flags, sizeof(flags)); 3722 setsockopt(sfd, SOL_SOCKET, SO_LINGER, (void *)&ling, sizeof(ling)); 3723 3724 /* 3725 * the memset call clears nonstandard fields in some impementations 3726 * that otherwise mess things up. 3727 */ 3728 memset(&addr, 0, sizeof(addr)); 3729 3730 addr.sun_family = AF_UNIX; 3731 strncpy(addr.sun_path, path, sizeof(addr.sun_path) - 1); 3732 assert(strcmp(addr.sun_path, path) == 0); 3733 old_umask = umask( ~(access_mask&0777)); 3734 if (bind(sfd, (struct sockaddr *)&addr, sizeof(addr)) == -1) { 3735 perror("bind()"); 3736 close(sfd); 3737 umask(old_umask); 3738 return 1; 3739 } 3740 umask(old_umask); 3741 if (listen(sfd, settings.backlog) == -1) { 3742 perror("listen()"); 3743 close(sfd); 3744 return 1; 3745 } 3746 if (!(listen_conn = conn_new(sfd, conn_listening, 3747 EV_READ | EV_PERSIST, 1, 3748 local_transport, main_base))) { 3749 fprintf(stderr, "failed to create listening connection\n"); 3750 exit(EXIT_FAILURE); 3751 } 3752 3753 return 0; 3754 } 3755 3756 /* 3757 * We keep the current time of day in a global variable that's updated by a 3758 * timer event. This saves us a bunch of time() system calls (we really only 3759 * need to get the time once a second, whereas there can be tens of thousands 3760 * of requests a second) and allows us to use server-start-relative timestamps 3761 * rather than absolute UNIX timestamps, a space savings on systems where 3762 * sizeof(time_t) > sizeof(unsigned int). 3763 */ 3764 volatile rel_time_t current_time; 3765 static struct event clockevent; 3766 3767 /* time-sensitive callers can call it by hand with this, outside the normal ever-1-second timer */ 3768 static void set_current_time(void) { 3769 struct timeval timer; 3770 3771 gettimeofday(&timer, NULL); 3772 current_time = (rel_time_t) (timer.tv_sec - process_started); 3773 } 3774 3775 static void clock_handler(const int fd, const short which, void *arg) { 3776 struct timeval t = {.tv_sec = 1, .tv_usec = 0}; 3777 static bool initialized = false; 3778 3779 if (initialized) { 3780 /* only delete the event if it's actually there. */ 3781 evtimer_del(&clockevent); 3782 } else { 3783 initialized = true; 3784 } 3785 3786 evtimer_set(&clockevent, clock_handler, 0); 3787 event_base_set(main_base, &clockevent); 3788 evtimer_add(&clockevent, &t); 3789 3790 set_current_time(); 3791 } 3792 3793 static void usage(void) { 3794 printf(PACKAGE " " VERSION "\n"); 3795 printf("-p <num> TCP port number to listen on (default: 11211)\n" 3796 "-U <num> UDP port number to listen on (default: 11211, 0 is off)\n" 3797 "-s <file> UNIX socket path to listen on (disables network support)\n" 3798 "-a <mask> access mask for UNIX socket, in octal (default: 0700)\n" 3799 "-l <ip_addr> interface to listen on (default: INADDR_ANY, all addresses)\n" 3800 "-d run as a daemon\n" 3801 "-r maximize core file limit\n" 3802 "-u <username> assume identity of <username> (only when run as root)\n" 3803 "-m <num> max memory to use for items in megabytes (default: 64 MB)\n" 3804 "-M return error on memory exhausted (rather than removing items)\n" 3805 "-c <num> max simultaneous connections (default: 1024)\n" 3806 "-k lock down all paged memory. Note that there is a\n" 3807 " limit on how much memory you may lock. Trying to\n" 3808 " allocate more than that would fail, so be sure you\n" 3809 " set the limit correctly for the user you started\n" 3810 " the daemon with (not for -u <username> user;\n" 3811 " under sh this is done with 'ulimit -S -l NUM_KB').\n" 3812 "-v verbose (print errors/warnings while in event loop)\n" 3813 "-vv very verbose (also print client commands/reponses)\n" 3814 "-vvv extremely verbose (also print internal state transitions)\n" 3815 "-h print this help and exit\n" 3816 "-i print memcached and libevent license\n" 3817 "-P <file> save PID in <file>, only used with -d option\n" 3818 "-f <factor> chunk size growth factor (default: 1.25)\n" 3819 "-n <bytes> minimum space allocated for key+value+flags (default: 48)\n"); 3820 printf("-L Try to use large memory pages (if available). Increasing\n" 3821 " the memory page size could reduce the number of TLB misses\n" 3822 " and improve the performance. In order to get large pages\n" 3823 " from the OS, memcached will allocate the total item-cache\n" 3824 " in one large chunk.\n"); 3825 printf("-D <char> Use <char> as the delimiter between key prefixes and IDs.\n" 3826 " This is used for per-prefix stats reporting. The default is\n" 3827 " \":\" (colon). If this option is specified, stats collection\n" 3828 " is turned on automatically; if not, then it may be turned on\n" 3829 " by sending the \"stats detail on\" command to the server.\n"); 3830 printf("-t <num> number of threads to use (default: 4)\n"); 3831 printf("-R Maximum number of requests per event, limits the number of\n" 3832 " requests process for a given connection to prevent \n" 3833 " starvation (default: 20)\n"); 3834 printf("-C Disable use of CAS\n"); 3835 printf("-b Set the backlog queue limit (default: 1024)\n"); 3836 printf("-B Binding protocol - one of ascii, binary, or auto (default)\n"); 3837 printf("-I Override the size of each slab page. Adjusts max item size\n"); 3838 return; 3839 } 3840 3841 static void usage_license(void) { 3842 printf(PACKAGE " " VERSION "\n\n"); 3843 printf( 3844 "Copyright (c) 2003, Danga Interactive, Inc. <http://www.danga.com/>\n" 3845 "All rights reserved.\n" 3846 "\n" 3847 "Redistribution and use in source and binary forms, with or without\n" 3848 "modification, are permitted provided that the following conditions are\n" 3849 "met:\n" 3850 "\n" 3851 " * Redistributions of source code must retain the above copyright\n" 3852 "notice, this list of conditions and the following disclaimer.\n" 3853 "\n" 3854 " * Redistributions in binary form must reproduce the above\n" 3855 "copyright notice, this list of conditions and the following disclaimer\n" 3856 "in the documentation and/or other materials provided with the\n" 3857 "distribution.\n" 3858 "\n" 3859 " * Neither the name of the Danga Interactive nor the names of its\n" 3860 "contributors may be used to endorse or promote products derived from\n" 3861 "this software without specific prior written permission.\n" 3862 "\n" 3863 "THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS\n" 3864 "\"AS IS\" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT\n" 3865 "LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR\n" 3866 "A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT\n" 3867 "OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,\n" 3868 "SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT\n" 3869 "LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,\n" 3870 "DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY\n" 3871 "THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT\n" 3872 "(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE\n" 3873 "OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.\n" 3874 "\n" 3875 "\n" 3876 "This product includes software developed by Niels Provos.\n" 3877 "\n" 3878 "[ libevent ]\n" 3879 "\n" 3880 "Copyright 2000-2003 Niels Provos <[email protected]>\n" 3881 "All rights reserved.\n" 3882 "\n" 3883 "Redistribution and use in source and binary forms, with or without\n" 3884 "modification, are permitted provided that the following conditions\n" 3885 "are met:\n" 3886 "1. Redistributions of source code must retain the above copyright\n" 3887 " notice, this list of conditions and the following disclaimer.\n" 3888 "2. Redistributions in binary form must reproduce the above copyright\n" 3889 " notice, this list of conditions and the following disclaimer in the\n" 3890 " documentation and/or other materials provided with the distribution.\n" 3891 "3. All advertising materials mentioning features or use of this software\n" 3892 " must display the following acknowledgement:\n" 3893 " This product includes software developed by Niels Provos.\n" 3894 "4. The name of the author may not be used to endorse or promote products\n" 3895 " derived from this software without specific prior written permission.\n" 3896 "\n" 3897 "THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR\n" 3898 "IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES\n" 3899 "OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.\n" 3900 "IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT,\n" 3901 "INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT\n" 3902 "NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,\n" 3903 "DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY\n" 3904 "THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT\n" 3905 "(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF\n" 3906 "THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.\n" 3907 ); 3908 3909 return; 3910 } 3911 3912 static void save_pid(const pid_t pid, const char *pid_file) { 3913 FILE *fp; 3914 if (pid_file == NULL) 3915 return; 3916 3917 if ((fp = fopen(pid_file, "w")) == NULL) { 3918 fprintf(stderr, "Could not open the pid file %s for writing\n", pid_file); 3919 return; 3920 } 3921 3922 fprintf(fp,"%ld\n", (long)pid); 3923 if (fclose(fp) == -1) { 3924 fprintf(stderr, "Could not close the pid file %s.\n", pid_file); 3925 return; 3926 } 3927 } 3928 3929 static void remove_pidfile(const char *pid_file) { 3930 if (pid_file == NULL) 3931 return; 3932 3933 if (unlink(pid_file) != 0) { 3934 fprintf(stderr, "Could not remove the pid file %s.\n", pid_file); 3935 } 3936 3937 } 3938 3939 static void sig_handler(const int sig) { 3940 printf("SIGINT handled.\n"); 3941 exit(EXIT_SUCCESS); 3942 } 3943 3944 #ifndef HAVE_SIGIGNORE 3945 static int sigignore(int sig) { 3946 struct sigaction sa = { .sa_handler = SIG_IGN, .sa_flags = 0 }; 3947 3948 if (sigemptyset(&sa.sa_mask) == -1 || sigaction(sig, &sa, 0) == -1) { 3949 return -1; 3950 } 3951 return 0; 3952 } 3953 #endif 3954 3955 3956 /* 3957 * On systems that supports multiple page sizes we may reduce the 3958 * number of TLB-misses by using the biggest available page size 3959 */ 3960 static int enable_large_pages(void) { 3961 #if defined(HAVE_GETPAGESIZES) && defined(HAVE_MEMCNTL) 3962 int ret = -1; 3963 size_t sizes[32]; 3964 int avail = getpagesizes(sizes, 32); 3965 if (avail != -1) { 3966 size_t max = sizes[0]; 3967 struct memcntl_mha arg = {0}; 3968 int ii; 3969 3970 for (ii = 1; ii < avail; ++ii) { 3971 if (max < sizes[ii]) { 3972 max = sizes[ii]; 3973 } 3974 } 3975 3976 arg.mha_flags = 0; 3977 arg.mha_pagesize = max; 3978 arg.mha_cmd = MHA_MAPSIZE_BSSBRK; 3979 3980 if (memcntl(0, 0, MC_HAT_ADVISE, (caddr_t)&arg, 0, 0) == -1) { 3981 fprintf(stderr, "Failed to set large pages: %s\n", 3982 strerror(errno)); 3983 fprintf(stderr, "Will use default page size\n"); 3984 } else { 3985 ret = 0; 3986 } 3987 } else { 3988 fprintf(stderr, "Failed to get supported pagesizes: %s\n", 3989 strerror(errno)); 3990 fprintf(stderr, "Will use default page size\n"); 3991 } 3992 3993 return ret; 3994 #else 3995 return 0; 3996 #endif 3997 } 3998 3999 int main (int argc, char **argv) { 4000 int c; 4001 bool lock_memory = false; 4002 bool do_daemonize = false; 4003 bool preallocate = false; 4004 int maxcore = 0; 4005 char *username = NULL; 4006 char *pid_file = NULL; 4007 struct passwd *pw; 4008 struct rlimit rlim; 4009 char unit = '\0'; 4010 int size_max = 0; 4011 /* listening sockets */ 4012 static int *l_socket = NULL; 4013 4014 /* udp socket */ 4015 static int *u_socket = NULL; 4016 4017 /* handle SIGINT */ 4018 signal(SIGINT, sig_handler); 4019 4020 /* init settings */ 4021 settings_init(); 4022 4023 /* set stderr non-buffering (for running under, say, daemontools) */ 4024 setbuf(stderr, NULL); 4025 4026 /* process arguments */ 4027 while (-1 != (c = getopt(argc, argv, 4028 "a:" /* access mask for unix socket */ 4029 "p:" /* TCP port number to listen on */ 4030 "s:" /* unix socket path to listen on */ 4031 "U:" /* UDP port number to listen on */ 4032 "m:" /* max memory to use for items in megabytes */ 4033 "M" /* return error on memory exhausted */ 4034 "c:" /* max simultaneous connections */ 4035 "k" /* lock down all paged memory */ 4036 "hi" /* help, licence info */ 4037 "r" /* maximize core file limit */ 4038 "v" /* verbose */ 4039 "d" /* daemon mode */ 4040 "l:" /* interface to listen on */ 4041 "u:" /* user identity to run as */ 4042 "P:" /* save PID in file */ 4043 "f:" /* factor? */ 4044 "n:" /* minimum space allocated for key+value+flags */ 4045 "t:" /* threads */ 4046 "D:" /* prefix delimiter? */ 4047 "L" /* Large memory pages */ 4048 "R:" /* max requests per event */ 4049 "C" /* Disable use of CAS */ 4050 "b:" /* backlog queue limit */ 4051 "B:" /* Binding protocol */ 4052 "I:" /* Max item size */ 4053 ))) { 4054 switch (c) { 4055 case 'a': 4056 /* access for unix domain socket, as octal mask (like chmod)*/ 4057 settings.access= strtol(optarg,NULL,8); 4058 break; 4059 4060 case 'U': 4061 settings.udpport = atoi(optarg); 4062 break; 4063 case 'p': 4064 settings.port = atoi(optarg); 4065 break; 4066 case 's': 4067 settings.socketpath = optarg; 4068 break; 4069 case 'm': 4070 settings.maxbytes = ((size_t)atoi(optarg)) * 1024 * 1024; 4071 break; 4072 case 'M': 4073 settings.evict_to_free = 0; 4074 break; 4075 case 'c': 4076 settings.maxconns = atoi(optarg); 4077 break; 4078 case 'h': 4079 usage(); 4080 exit(EXIT_SUCCESS); 4081 case 'i': 4082 usage_license(); 4083 exit(EXIT_SUCCESS); 4084 case 'k': 4085 lock_memory = true; 4086 break; 4087 case 'v': 4088 settings.verbose++; 4089 break; 4090 case 'l': 4091 settings.inter= strdup(optarg); 4092 break; 4093 case 'd': 4094 do_daemonize = true; 4095 break; 4096 case 'r': 4097 maxcore = 1; 4098 break; 4099 case 'R': 4100 settings.reqs_per_event = atoi(optarg); 4101 if (settings.reqs_per_event == 0) { 4102 fprintf(stderr, "Number of requests per event must be greater than 0\n"); 4103 return 1; 4104 } 4105 break; 4106 case 'u': 4107 username = optarg; 4108 break; 4109 case 'P': 4110 pid_file = optarg; 4111 break; 4112 case 'f': 4113 settings.factor = atof(optarg); 4114 if (settings.factor <= 1.0) { 4115 fprintf(stderr, "Factor must be greater than 1\n"); 4116 return 1; 4117 } 4118 break; 4119 case 'n': 4120 settings.chunk_size = atoi(optarg); 4121 if (settings.chunk_size == 0) { 4122 fprintf(stderr, "Chunk size must be greater than 0\n"); 4123 return 1; 4124 } 4125 break; 4126 case 't': 4127 settings.num_threads = atoi(optarg); 4128 if (settings.num_threads == 0) { 4129 fprintf(stderr, "Number of threads must be greater than 0\n"); 4130 return 1; 4131 } 4132 break; 4133 case 'D': 4134 if (! optarg || ! optarg[0]) { 4135 fprintf(stderr, "No delimiter specified\n"); 4136 return 1; 4137 } 4138 settings.prefix_delimiter = optarg[0]; 4139 settings.detail_enabled = 1; 4140 break; 4141 case 'L' : 4142 if (enable_large_pages() == 0) { 4143 preallocate = true; 4144 } 4145 break; 4146 case 'C' : 4147 settings.use_cas = false; 4148 break; 4149 case 'b' : 4150 settings.backlog = atoi(optarg); 4151 break; 4152 case 'B': 4153 if (strcmp(optarg, "auto") == 0) { 4154 settings.binding_protocol = negotiating_prot; 4155 } else if (strcmp(optarg, "binary") == 0) { 4156 settings.binding_protocol = binary_prot; 4157 } else if (strcmp(optarg, "ascii") == 0) { 4158 settings.binding_protocol = ascii_prot; 4159 } else { 4160 fprintf(stderr, "Invalid value for binding protocol: %s\n" 4161 " -- should be one of auto, binary, or ascii\n", optarg); 4162 exit(EX_USAGE); 4163 } 4164 break; 4165 case 'I': 4166 unit = optarg[strlen(optarg)-1]; 4167 if (unit == 'k' || unit == 'm' || 4168 unit == 'K' || unit == 'M') { 4169 optarg[strlen(optarg)-1] = '\0'; 4170 size_max = atoi(optarg); 4171 if (unit == 'k' || unit == 'K') 4172 size_max *= 1024; 4173 if (unit == 'm' || unit == 'M') 4174 size_max *= 1024 * 1024; 4175 settings.item_size_max = size_max; 4176 } else { 4177 settings.item_size_max = atoi(optarg); 4178 } 4179 if (settings.item_size_max < 1024) { 4180 fprintf(stderr, "Item max size cannot be less than 1024 bytes.\n"); 4181 return 1; 4182 } 4183 if (settings.item_size_max > 1024 * 1024 * 128) { 4184 fprintf(stderr, "Cannot set item size limit higher than 128 mb.\n"); 4185 return 1; 4186 } 4187 if (settings.item_size_max > 1024 * 1024) { 4188 fprintf(stderr, "WARNING: Setting item max size above 1MB is not" 4189 " recommended!\n" 4190 " Raising this limit increases the minimum memory requirements\n" 4191 " and will decrease your memory efficiency.\n" 4192 ); 4193 } 4194 break; 4195 default: 4196 fprintf(stderr, "Illegal argument \"%c\"\n", c); 4197 return 1; 4198 } 4199 } 4200 4201 if (maxcore != 0) { 4202 struct rlimit rlim_new; 4203 /* 4204 * First try raising to infinity; if that fails, try bringing 4205 * the soft limit to the hard. 4206 */ 4207 if (getrlimit(RLIMIT_CORE, &rlim) == 0) { 4208 rlim_new.rlim_cur = rlim_new.rlim_max = RLIM_INFINITY; 4209 if (setrlimit(RLIMIT_CORE, &rlim_new)!= 0) { 4210 /* failed. try raising just to the old max */ 4211 rlim_new.rlim_cur = rlim_new.rlim_max = rlim.rlim_max; 4212 (void)setrlimit(RLIMIT_CORE, &rlim_new); 4213 } 4214 } 4215 /* 4216 * getrlimit again to see what we ended up with. Only fail if 4217 * the soft limit ends up 0, because then no core files will be 4218 * created at all. 4219 */ 4220 4221 if ((getrlimit(RLIMIT_CORE, &rlim) != 0) || rlim.rlim_cur == 0) { 4222 fprintf(stderr, "failed to ensure corefile creation\n"); 4223 exit(EX_OSERR); 4224 } 4225 } 4226 4227 /* 4228 * If needed, increase rlimits to allow as many connections 4229 * as needed. 4230 */ 4231 4232 if (getrlimit(RLIMIT_NOFILE, &rlim) != 0) { 4233 fprintf(stderr, "failed to getrlimit number of files\n"); 4234 exit(EX_OSERR); 4235 } else { 4236 int maxfiles = settings.maxconns; 4237 if (rlim.rlim_cur < maxfiles) 4238 rlim.rlim_cur = maxfiles; 4239 if (rlim.rlim_max < rlim.rlim_cur) 4240 rlim.rlim_max = rlim.rlim_cur; 4241 if (setrlimit(RLIMIT_NOFILE, &rlim) != 0) { 4242 fprintf(stderr, "failed to set rlimit for open files. Try running as root or requesting smaller maxconns value.\n"); 4243 exit(EX_OSERR); 4244 } 4245 } 4246 4247 /* lose root privileges if we have them */ 4248 if (getuid() == 0 || geteuid() == 0) { 4249 if (username == 0 || *username == '\0') { 4250 fprintf(stderr, "can't run as root without the -u switch\n"); 4251 exit(EX_USAGE); 4252 } 4253 if ((pw = getpwnam(username)) == 0) { 4254 fprintf(stderr, "can't find the user %s to switch to\n", username); 4255 exit(EX_NOUSER); 4256 } 4257 if (setgid(pw->pw_gid) < 0 || setuid(pw->pw_uid) < 0) { 4258 fprintf(stderr, "failed to assume identity of user %s\n", username); 4259 exit(EX_OSERR); 4260 } 4261 } 4262 4263 /* daemonize if requested */ 4264 /* if we want to ensure our ability to dump core, don't chdir to / */ 4265 if (do_daemonize) { 4266 if (sigignore(SIGHUP) == -1) { 4267 perror("Failed to ignore SIGHUP"); 4268 } 4269 if (daemonize(maxcore, settings.verbose) == -1) { 4270 fprintf(stderr, "failed to daemon() in order to daemonize\n"); 4271 exit(EXIT_FAILURE); 4272 } 4273 } 4274 4275 /* lock paged memory if needed */ 4276 if (lock_memory) { 4277 #ifdef HAVE_MLOCKALL 4278 int res = mlockall(MCL_CURRENT | MCL_FUTURE); 4279 if (res != 0) { 4280 fprintf(stderr, "warning: -k invalid, mlockall() failed: %s\n", 4281 strerror(errno)); 4282 } 4283 #else 4284 fprintf(stderr, "warning: -k invalid, mlockall() not supported on this platform. proceeding without.\n"); 4285 #endif 4286 } 4287 4288 /* initialize main thread libevent instance */ 4289 main_base = event_init(); 4290 4291 /* initialize other stuff */ 4292 stats_init(); 4293 assoc_init(); 4294 conn_init(); 4295 slabs_init(settings.maxbytes, settings.factor, preallocate); 4296 4297 /* 4298 * ignore SIGPIPE signals; we can use errno == EPIPE if we 4299 * need that information 4300 */ 4301 if (sigignore(SIGPIPE) == -1) { 4302 perror("failed to ignore SIGPIPE; sigaction"); 4303 exit(EX_OSERR); 4304 } 4305 /* start up worker threads if MT mode */ 4306 thread_init(settings.num_threads, main_base); 4307 /* save the PID in if we're a daemon, do this after thread_init due to 4308 a file descriptor handling bug somewhere in libevent */ 4309 4310 if (start_assoc_maintenance_thread() == -1) { 4311 exit(EXIT_FAILURE); 4312 } 4313 4314 if (do_daemonize) 4315 save_pid(getpid(), pid_file); 4316 /* initialise clock event */ 4317 clock_handler(0, 0, 0); 4318 4319 /* create unix mode sockets after dropping privileges */ 4320 if (settings.socketpath != NULL) { 4321 errno = 0; 4322 if (server_socket_unix(settings.socketpath,settings.access)) { 4323 vperror("failed to listen on UNIX socket: %s", settings.socketpath); 4324 exit(EX_OSERR); 4325 } 4326 } 4327 4328 /* create the listening socket, bind it, and init */ 4329 if (settings.socketpath == NULL) { 4330 int udp_port; 4331 4332 const char *portnumber_filename = getenv("MEMCACHED_PORT_FILENAME"); 4333 char temp_portnumber_filename[PATH_MAX]; 4334 FILE *portnumber_file = NULL; 4335 4336 if (portnumber_filename != NULL) { 4337 snprintf(temp_portnumber_filename, 4338 sizeof(temp_portnumber_filename), 4339 "%s.lck", portnumber_filename); 4340 4341 portnumber_file = fopen(temp_portnumber_filename, "a"); 4342 if (portnumber_file == NULL) { 4343 fprintf(stderr, "Failed to open \"%s\": %s\n", 4344 temp_portnumber_filename, strerror(errno)); 4345 } 4346 } 4347 4348 errno = 0; 4349 if (settings.port && server_socket(settings.port, tcp_transport, 4350 portnumber_file)) { 4351 vperror("failed to listen on TCP port %d", settings.port); 4352 exit(EX_OSERR); 4353 } 4354 4355 /* 4356 * initialization order: first create the listening sockets 4357 * (may need root on low ports), then drop root if needed, 4358 * then daemonise if needed, then init libevent (in some cases 4359 * descriptors created by libevent wouldn't survive forking). 4360 */ 4361 udp_port = settings.udpport ? settings.udpport : settings.port; 4362 4363 /* create the UDP listening socket and bind it */ 4364 errno = 0; 4365 if (settings.udpport && server_socket(settings.udpport, udp_transport, 4366 portnumber_file)) { 4367 vperror("failed to listen on UDP port %d", settings.udpport); 4368 exit(EX_OSERR); 4369 } 4370 4371 if (portnumber_file) { 4372 fclose(portnumber_file); 4373 rename(temp_portnumber_filename, portnumber_filename); 4374 } 4375 } 4376 4377 /* Drop privileges no longer needed */ 4378 drop_privileges(); 4379 4380 /* enter the event loop */ 4381 event_base_loop(main_base, 0); 4382 4383 stop_assoc_maintenance_thread(); 4384 4385 /* remove the PID file if we're a daemon */ 4386 if (do_daemonize) 4387 remove_pidfile(pid_file); 4388 /* Clean up strdup() call for bind() address */ 4389 if (settings.inter) 4390 free(settings.inter); 4391 if (l_socket) 4392 free(l_socket); 4393 if (u_socket) 4394 free(u_socket); 4395 4396 return EXIT_SUCCESS; 4397 } 4398