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