1 /* -*- Mode: C; tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*- */ 2 /* 3 * memcached - memory caching daemon 4 * 5 * http://www.memcached.org/ 6 * 7 * Copyright 2003 Danga Interactive, Inc. All rights reserved. 8 * 9 * Use and distribution licensed under the BSD license. See 10 * the LICENSE file for full text. 11 * 12 * Authors: 13 * Anatoly Vorobey <[email protected]> 14 * Brad Fitzpatrick <[email protected]> 15 */ 16 #include "memcached.h" 17 #include <sys/stat.h> 18 #include <sys/socket.h> 19 #include <sys/un.h> 20 #include <signal.h> 21 #include <sys/param.h> 22 #include <sys/resource.h> 23 #include <sys/uio.h> 24 #include <ctype.h> 25 #include <stdarg.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 #include <sysexits.h> 49 #include <stddef.h> 50 51 /* FreeBSD 4.x doesn't have IOV_MAX exposed. */ 52 #ifndef IOV_MAX 53 #if defined(__FreeBSD__) || defined(__APPLE__) 54 # define IOV_MAX 1024 55 #endif 56 #endif 57 58 /* 59 * forward declarations 60 */ 61 static void drive_machine(conn *c); 62 static int new_socket(struct addrinfo *ai); 63 static int try_read_command(conn *c); 64 65 enum try_read_result { 66 READ_DATA_RECEIVED, 67 READ_NO_DATA_RECEIVED, 68 READ_ERROR, /** an error occured (on the socket) (or client closed connection) */ 69 READ_MEMORY_ERROR /** failed to allocate more memory */ 70 }; 71 72 static enum try_read_result try_read_network(conn *c); 73 static enum try_read_result try_read_udp(conn *c); 74 75 static void conn_set_state(conn *c, enum conn_states state); 76 77 /* stats */ 78 static void stats_init(void); 79 static void server_stats(ADD_STAT add_stats, conn *c); 80 static void process_stat_settings(ADD_STAT add_stats, void *c); 81 static void conn_to_str(const conn *c, char *buf); 82 83 84 /* defaults */ 85 static void settings_init(void); 86 87 /* event handling, network IO */ 88 static void event_handler(const int fd, const short which, void *arg); 89 static void conn_close(conn *c); 90 static void conn_init(void); 91 static bool update_event(conn *c, const int new_flags); 92 static void complete_nread(conn *c); 93 static void process_command(conn *c, char *command); 94 static void write_and_free(conn *c, char *buf, int bytes); 95 static int ensure_iov_space(conn *c); 96 static int add_iov(conn *c, const void *buf, int len); 97 static int add_msghdr(conn *c); 98 static void write_bin_error(conn *c, protocol_binary_response_status err, 99 const char *errstr, int swallow); 100 101 static void conn_free(conn *c); 102 103 /** exported globals **/ 104 struct stats stats; 105 struct settings settings; 106 time_t process_started; /* when the process was started */ 107 conn **conns; 108 109 struct slab_rebalance slab_rebal; 110 volatile int slab_rebalance_signal; 111 112 /** file scope variables **/ 113 static conn *listen_conn = NULL; 114 static int max_fds; 115 static struct event_base *main_base; 116 117 enum transmit_result { 118 TRANSMIT_COMPLETE, /** All done writing. */ 119 TRANSMIT_INCOMPLETE, /** More data remaining to write. */ 120 TRANSMIT_SOFT_ERROR, /** Can't write any more right now. */ 121 TRANSMIT_HARD_ERROR /** Can't write (c->state is set to conn_closing) */ 122 }; 123 124 static enum transmit_result transmit(conn *c); 125 126 /* This reduces the latency without adding lots of extra wiring to be able to 127 * notify the listener thread of when to listen again. 128 * Also, the clock timer could be broken out into its own thread and we 129 * can block the listener via a condition. 130 */ 131 static volatile bool allow_new_conns = true; 132 static struct event maxconnsevent; 133 static void maxconns_handler(const int fd, const short which, void *arg) { 134 struct timeval t = {.tv_sec = 0, .tv_usec = 10000}; 135 136 if (fd == -42 || allow_new_conns == false) { 137 /* reschedule in 10ms if we need to keep polling */ 138 evtimer_set(&maxconnsevent, maxconns_handler, 0); 139 event_base_set(main_base, &maxconnsevent); 140 evtimer_add(&maxconnsevent, &t); 141 } else { 142 evtimer_del(&maxconnsevent); 143 accept_new_conns(true); 144 } 145 } 146 147 #define REALTIME_MAXDELTA 60*60*24*30 148 149 /* 150 * given time value that's either unix time or delta from current unix time, return 151 * unix time. Use the fact that delta can't exceed one month (and real time value can't 152 * be that low). 153 */ 154 static rel_time_t realtime(const time_t exptime) { 155 /* no. of seconds in 30 days - largest possible delta exptime */ 156 157 if (exptime == 0) return 0; /* 0 means never expire */ 158 159 if (exptime > REALTIME_MAXDELTA) { 160 /* if item expiration is at/before the server started, give it an 161 expiration time of 1 second after the server started. 162 (because 0 means don't expire). without this, we'd 163 underflow and wrap around to some large value way in the 164 future, effectively making items expiring in the past 165 really expiring never */ 166 if (exptime <= process_started) 167 return (rel_time_t)1; 168 return (rel_time_t)(exptime - process_started); 169 } else { 170 return (rel_time_t)(exptime + current_time); 171 } 172 } 173 174 static void stats_init(void) { 175 stats.curr_items = stats.total_items = stats.curr_conns = stats.total_conns = stats.conn_structs = 0; 176 stats.get_cmds = stats.set_cmds = stats.get_hits = stats.get_misses = stats.evictions = stats.reclaimed = 0; 177 stats.touch_cmds = stats.touch_misses = stats.touch_hits = stats.rejected_conns = 0; 178 stats.malloc_fails = 0; 179 stats.curr_bytes = stats.listen_disabled_num = 0; 180 stats.hash_power_level = stats.hash_bytes = stats.hash_is_expanding = 0; 181 stats.expired_unfetched = stats.evicted_unfetched = 0; 182 stats.slabs_moved = 0; 183 stats.accepting_conns = true; /* assuming we start in this state. */ 184 stats.slab_reassign_running = false; 185 stats.lru_crawler_running = false; 186 187 /* make the time we started always be 2 seconds before we really 188 did, so time(0) - time.started is never zero. if so, things 189 like 'settings.oldest_live' which act as booleans as well as 190 values are now false in boolean context... */ 191 process_started = time(0) - ITEM_UPDATE_INTERVAL - 2; 192 stats_prefix_init(); 193 } 194 195 static void stats_reset(void) { 196 STATS_LOCK(); 197 stats.total_items = stats.total_conns = 0; 198 stats.rejected_conns = 0; 199 stats.malloc_fails = 0; 200 stats.evictions = 0; 201 stats.reclaimed = 0; 202 stats.listen_disabled_num = 0; 203 stats_prefix_clear(); 204 STATS_UNLOCK(); 205 threadlocal_stats_reset(); 206 item_stats_reset(); 207 } 208 209 static void settings_init(void) { 210 settings.use_cas = true; 211 settings.access = 0700; 212 settings.port = 11211; 213 settings.udpport = 11211; 214 /* By default this string should be NULL for getaddrinfo() */ 215 settings.inter = NULL; 216 settings.maxbytes = 64 * 1024 * 1024; /* default is 64MB */ 217 settings.maxconns = 1024; /* to limit connections-related memory to about 5MB */ 218 settings.verbose = 0; 219 settings.oldest_live = 0; 220 settings.evict_to_free = 1; /* push old items out of cache when memory runs out */ 221 settings.socketpath = NULL; /* by default, not using a unix socket */ 222 settings.factor = 1.25; 223 settings.chunk_size = 48; /* space for a modest key and value */ 224 settings.num_threads = 4; /* N workers */ 225 settings.num_threads_per_udp = 0; 226 settings.prefix_delimiter = ':'; 227 settings.detail_enabled = 0; 228 settings.reqs_per_event = 20; 229 settings.backlog = 1024; 230 settings.binding_protocol = negotiating_prot; 231 settings.item_size_max = 1024 * 1024; /* The famous 1MB upper limit. */ 232 settings.maxconns_fast = false; 233 settings.lru_crawler = false; 234 settings.lru_crawler_sleep = 100; 235 settings.lru_crawler_tocrawl = 0; 236 settings.hashpower_init = 0; 237 settings.slab_reassign = false; 238 settings.slab_automove = 0; 239 settings.shutdown_command = false; 240 settings.tail_repair_time = TAIL_REPAIR_TIME_DEFAULT; 241 settings.flush_enabled = true; 242 } 243 244 /* 245 * Adds a message header to a connection. 246 * 247 * Returns 0 on success, -1 on out-of-memory. 248 */ 249 static int add_msghdr(conn *c) 250 { 251 struct msghdr *msg; 252 253 assert(c != NULL); 254 255 if (c->msgsize == c->msgused) { 256 msg = realloc(c->msglist, c->msgsize * 2 * sizeof(struct msghdr)); 257 if (! msg) { 258 STATS_LOCK(); 259 stats.malloc_fails++; 260 STATS_UNLOCK(); 261 return -1; 262 } 263 c->msglist = msg; 264 c->msgsize *= 2; 265 } 266 267 msg = c->msglist + c->msgused; 268 269 /* this wipes msg_iovlen, msg_control, msg_controllen, and 270 msg_flags, the last 3 of which aren't defined on solaris: */ 271 memset(msg, 0, sizeof(struct msghdr)); 272 273 msg->msg_iov = &c->iov[c->iovused]; 274 275 if (IS_UDP(c->transport) && c->request_addr_size > 0) { 276 msg->msg_name = &c->request_addr; 277 msg->msg_namelen = c->request_addr_size; 278 } 279 280 c->msgbytes = 0; 281 c->msgused++; 282 283 if (IS_UDP(c->transport)) { 284 /* Leave room for the UDP header, which we'll fill in later. */ 285 return add_iov(c, NULL, UDP_HEADER_SIZE); 286 } 287 288 return 0; 289 } 290 291 extern pthread_mutex_t conn_lock; 292 293 /* 294 * Initializes the connections array. We don't actually allocate connection 295 * structures until they're needed, so as to avoid wasting memory when the 296 * maximum connection count is much higher than the actual number of 297 * connections. 298 * 299 * This does end up wasting a few pointers' worth of memory for FDs that are 300 * used for things other than connections, but that's worth it in exchange for 301 * being able to directly index the conns array by FD. 302 */ 303 static void conn_init(void) { 304 /* We're unlikely to see an FD much higher than maxconns. */ 305 int next_fd = dup(1); 306 int headroom = 10; /* account for extra unexpected open FDs */ 307 struct rlimit rl; 308 309 max_fds = settings.maxconns + headroom + next_fd; 310 311 /* But if possible, get the actual highest FD we can possibly ever see. */ 312 if (getrlimit(RLIMIT_NOFILE, &rl) == 0) { 313 max_fds = rl.rlim_max; 314 } else { 315 fprintf(stderr, "Failed to query maximum file descriptor; " 316 "falling back to maxconns\n"); 317 } 318 319 close(next_fd); 320 321 if ((conns = calloc(max_fds, sizeof(conn *))) == NULL) { 322 fprintf(stderr, "Failed to allocate connection structures\n"); 323 /* This is unrecoverable so bail out early. */ 324 exit(1); 325 } 326 } 327 328 static const char *prot_text(enum protocol prot) { 329 char *rv = "unknown"; 330 switch(prot) { 331 case ascii_prot: 332 rv = "ascii"; 333 break; 334 case binary_prot: 335 rv = "binary"; 336 break; 337 case negotiating_prot: 338 rv = "auto-negotiate"; 339 break; 340 } 341 return rv; 342 } 343 344 conn *conn_new(const int sfd, enum conn_states init_state, 345 const int event_flags, 346 const int read_buffer_size, enum network_transport transport, 347 struct event_base *base) { 348 conn *c; 349 350 assert(sfd >= 0 && sfd < max_fds); 351 c = conns[sfd]; 352 353 if (NULL == c) { 354 if (!(c = (conn *)calloc(1, sizeof(conn)))) { 355 STATS_LOCK(); 356 stats.malloc_fails++; 357 STATS_UNLOCK(); 358 fprintf(stderr, "Failed to allocate connection object\n"); 359 return NULL; 360 } 361 MEMCACHED_CONN_CREATE(c); 362 363 c->rbuf = c->wbuf = 0; 364 c->ilist = 0; 365 c->suffixlist = 0; 366 c->iov = 0; 367 c->msglist = 0; 368 c->hdrbuf = 0; 369 370 c->rsize = read_buffer_size; 371 c->wsize = DATA_BUFFER_SIZE; 372 c->isize = ITEM_LIST_INITIAL; 373 c->suffixsize = SUFFIX_LIST_INITIAL; 374 c->iovsize = IOV_LIST_INITIAL; 375 c->msgsize = MSG_LIST_INITIAL; 376 c->hdrsize = 0; 377 378 c->rbuf = (char *)malloc((size_t)c->rsize); 379 c->wbuf = (char *)malloc((size_t)c->wsize); 380 c->ilist = (item **)malloc(sizeof(item *) * c->isize); 381 c->suffixlist = (char **)malloc(sizeof(char *) * c->suffixsize); 382 c->iov = (struct iovec *)malloc(sizeof(struct iovec) * c->iovsize); 383 c->msglist = (struct msghdr *)malloc(sizeof(struct msghdr) * c->msgsize); 384 385 if (c->rbuf == 0 || c->wbuf == 0 || c->ilist == 0 || c->iov == 0 || 386 c->msglist == 0 || c->suffixlist == 0) { 387 conn_free(c); 388 STATS_LOCK(); 389 stats.malloc_fails++; 390 STATS_UNLOCK(); 391 fprintf(stderr, "Failed to allocate buffers for connection\n"); 392 return NULL; 393 } 394 395 STATS_LOCK(); 396 stats.conn_structs++; 397 STATS_UNLOCK(); 398 399 c->sfd = sfd; 400 conns[sfd] = c; 401 } 402 403 c->transport = transport; 404 c->protocol = settings.binding_protocol; 405 406 /* unix socket mode doesn't need this, so zeroed out. but why 407 * is this done for every command? presumably for UDP 408 * mode. */ 409 if (!settings.socketpath) { 410 c->request_addr_size = sizeof(c->request_addr); 411 } else { 412 c->request_addr_size = 0; 413 } 414 415 if (transport == tcp_transport && init_state == conn_new_cmd) { 416 if (getpeername(sfd, (struct sockaddr *) &c->request_addr, 417 &c->request_addr_size)) { 418 perror("getpeername"); 419 memset(&c->request_addr, 0, sizeof(c->request_addr)); 420 } 421 } 422 423 if (settings.verbose > 1) { 424 if (init_state == conn_listening) { 425 fprintf(stderr, "<%d server listening (%s)\n", sfd, 426 prot_text(c->protocol)); 427 } else if (IS_UDP(transport)) { 428 fprintf(stderr, "<%d server listening (udp)\n", sfd); 429 } else if (c->protocol == negotiating_prot) { 430 fprintf(stderr, "<%d new auto-negotiating client connection\n", 431 sfd); 432 } else if (c->protocol == ascii_prot) { 433 fprintf(stderr, "<%d new ascii client connection.\n", sfd); 434 } else if (c->protocol == binary_prot) { 435 fprintf(stderr, "<%d new binary client connection.\n", sfd); 436 } else { 437 fprintf(stderr, "<%d new unknown (%d) client connection\n", 438 sfd, c->protocol); 439 assert(false); 440 } 441 } 442 443 c->state = init_state; 444 c->rlbytes = 0; 445 c->cmd = -1; 446 c->rbytes = c->wbytes = 0; 447 c->wcurr = c->wbuf; 448 c->rcurr = c->rbuf; 449 c->ritem = 0; 450 c->icurr = c->ilist; 451 c->suffixcurr = c->suffixlist; 452 c->ileft = 0; 453 c->suffixleft = 0; 454 c->iovused = 0; 455 c->msgcurr = 0; 456 c->msgused = 0; 457 c->authenticated = false; 458 459 c->write_and_go = init_state; 460 c->write_and_free = 0; 461 c->item = 0; 462 463 c->noreply = false; 464 465 event_set(&c->event, sfd, event_flags, event_handler, (void *)c); 466 event_base_set(base, &c->event); 467 c->ev_flags = event_flags; 468 469 if (event_add(&c->event, 0) == -1) { 470 perror("event_add"); 471 return NULL; 472 } 473 474 STATS_LOCK(); 475 stats.curr_conns++; 476 stats.total_conns++; 477 STATS_UNLOCK(); 478 479 MEMCACHED_CONN_ALLOCATE(c->sfd); 480 481 return c; 482 } 483 484 static void conn_release_items(conn *c) { 485 assert(c != NULL); 486 487 if (c->item) { 488 item_remove(c->item); 489 c->item = 0; 490 } 491 492 while (c->ileft > 0) { 493 item *it = *(c->icurr); 494 assert((it->it_flags & ITEM_SLABBED) == 0); 495 item_remove(it); 496 c->icurr++; 497 c->ileft--; 498 } 499 500 if (c->suffixleft != 0) { 501 for (; c->suffixleft > 0; c->suffixleft--, c->suffixcurr++) { 502 cache_free(c->thread->suffix_cache, *(c->suffixcurr)); 503 } 504 } 505 506 c->icurr = c->ilist; 507 c->suffixcurr = c->suffixlist; 508 } 509 510 static void conn_cleanup(conn *c) { 511 assert(c != NULL); 512 513 conn_release_items(c); 514 515 if (c->write_and_free) { 516 free(c->write_and_free); 517 c->write_and_free = 0; 518 } 519 520 if (c->sasl_conn) { 521 assert(settings.sasl); 522 sasl_dispose(&c->sasl_conn); 523 c->sasl_conn = NULL; 524 } 525 526 if (IS_UDP(c->transport)) { 527 conn_set_state(c, conn_read); 528 } 529 } 530 531 /* 532 * Frees a connection. 533 */ 534 void conn_free(conn *c) { 535 if (c) { 536 assert(c != NULL); 537 assert(c->sfd >= 0 && c->sfd < max_fds); 538 539 MEMCACHED_CONN_DESTROY(c); 540 conns[c->sfd] = NULL; 541 if (c->hdrbuf) 542 free(c->hdrbuf); 543 if (c->msglist) 544 free(c->msglist); 545 if (c->rbuf) 546 free(c->rbuf); 547 if (c->wbuf) 548 free(c->wbuf); 549 if (c->ilist) 550 free(c->ilist); 551 if (c->suffixlist) 552 free(c->suffixlist); 553 if (c->iov) 554 free(c->iov); 555 free(c); 556 } 557 } 558 559 static void conn_close(conn *c) { 560 assert(c != NULL); 561 562 /* delete the event, the socket and the conn */ 563 event_del(&c->event); 564 565 if (settings.verbose > 1) 566 fprintf(stderr, "<%d connection closed.\n", c->sfd); 567 568 conn_cleanup(c); 569 570 MEMCACHED_CONN_RELEASE(c->sfd); 571 conn_set_state(c, conn_closed); 572 close(c->sfd); 573 574 pthread_mutex_lock(&conn_lock); 575 allow_new_conns = true; 576 pthread_mutex_unlock(&conn_lock); 577 578 STATS_LOCK(); 579 stats.curr_conns--; 580 STATS_UNLOCK(); 581 582 return; 583 } 584 585 /* 586 * Shrinks a connection's buffers if they're too big. This prevents 587 * periodic large "get" requests from permanently chewing lots of server 588 * memory. 589 * 590 * This should only be called in between requests since it can wipe output 591 * buffers! 592 */ 593 static void conn_shrink(conn *c) { 594 assert(c != NULL); 595 596 if (IS_UDP(c->transport)) 597 return; 598 599 if (c->rsize > READ_BUFFER_HIGHWAT && c->rbytes < DATA_BUFFER_SIZE) { 600 char *newbuf; 601 602 if (c->rcurr != c->rbuf) 603 memmove(c->rbuf, c->rcurr, (size_t)c->rbytes); 604 605 newbuf = (char *)realloc((void *)c->rbuf, DATA_BUFFER_SIZE); 606 607 if (newbuf) { 608 c->rbuf = newbuf; 609 c->rsize = DATA_BUFFER_SIZE; 610 } 611 /* TODO check other branch... */ 612 c->rcurr = c->rbuf; 613 } 614 615 if (c->isize > ITEM_LIST_HIGHWAT) { 616 item **newbuf = (item**) realloc((void *)c->ilist, ITEM_LIST_INITIAL * sizeof(c->ilist[0])); 617 if (newbuf) { 618 c->ilist = newbuf; 619 c->isize = ITEM_LIST_INITIAL; 620 } 621 /* TODO check error condition? */ 622 } 623 624 if (c->msgsize > MSG_LIST_HIGHWAT) { 625 struct msghdr *newbuf = (struct msghdr *) realloc((void *)c->msglist, MSG_LIST_INITIAL * sizeof(c->msglist[0])); 626 if (newbuf) { 627 c->msglist = newbuf; 628 c->msgsize = MSG_LIST_INITIAL; 629 } 630 /* TODO check error condition? */ 631 } 632 633 if (c->iovsize > IOV_LIST_HIGHWAT) { 634 struct iovec *newbuf = (struct iovec *) realloc((void *)c->iov, IOV_LIST_INITIAL * sizeof(c->iov[0])); 635 if (newbuf) { 636 c->iov = newbuf; 637 c->iovsize = IOV_LIST_INITIAL; 638 } 639 /* TODO check return value */ 640 } 641 } 642 643 /** 644 * Convert a state name to a human readable form. 645 */ 646 static const char *state_text(enum conn_states state) { 647 const char* const statenames[] = { "conn_listening", 648 "conn_new_cmd", 649 "conn_waiting", 650 "conn_read", 651 "conn_parse_cmd", 652 "conn_write", 653 "conn_nread", 654 "conn_swallow", 655 "conn_closing", 656 "conn_mwrite", 657 "conn_closed" }; 658 return statenames[state]; 659 } 660 661 /* 662 * Sets a connection's current state in the state machine. Any special 663 * processing that needs to happen on certain state transitions can 664 * happen here. 665 */ 666 static void conn_set_state(conn *c, enum conn_states state) { 667 assert(c != NULL); 668 assert(state >= conn_listening && state < conn_max_state); 669 670 if (state != c->state) { 671 if (settings.verbose > 2) { 672 fprintf(stderr, "%d: going from %s to %s\n", 673 c->sfd, state_text(c->state), 674 state_text(state)); 675 } 676 677 if (state == conn_write || state == conn_mwrite) { 678 MEMCACHED_PROCESS_COMMAND_END(c->sfd, c->wbuf, c->wbytes); 679 } 680 c->state = state; 681 } 682 } 683 684 /* 685 * Ensures that there is room for another struct iovec in a connection's 686 * iov list. 687 * 688 * Returns 0 on success, -1 on out-of-memory. 689 */ 690 static int ensure_iov_space(conn *c) { 691 assert(c != NULL); 692 693 if (c->iovused >= c->iovsize) { 694 int i, iovnum; 695 struct iovec *new_iov = (struct iovec *)realloc(c->iov, 696 (c->iovsize * 2) * sizeof(struct iovec)); 697 if (! new_iov) { 698 STATS_LOCK(); 699 stats.malloc_fails++; 700 STATS_UNLOCK(); 701 return -1; 702 } 703 c->iov = new_iov; 704 c->iovsize *= 2; 705 706 /* Point all the msghdr structures at the new list. */ 707 for (i = 0, iovnum = 0; i < c->msgused; i++) { 708 c->msglist[i].msg_iov = &c->iov[iovnum]; 709 iovnum += c->msglist[i].msg_iovlen; 710 } 711 } 712 713 return 0; 714 } 715 716 717 /* 718 * Adds data to the list of pending data that will be written out to a 719 * connection. 720 * 721 * Returns 0 on success, -1 on out-of-memory. 722 */ 723 724 static int add_iov(conn *c, const void *buf, int len) { 725 struct msghdr *m; 726 int leftover; 727 bool limit_to_mtu; 728 729 assert(c != NULL); 730 731 do { 732 m = &c->msglist[c->msgused - 1]; 733 734 /* 735 * Limit UDP packets, and the first payloads of TCP replies, to 736 * UDP_MAX_PAYLOAD_SIZE bytes. 737 */ 738 limit_to_mtu = IS_UDP(c->transport) || (1 == c->msgused); 739 740 /* We may need to start a new msghdr if this one is full. */ 741 if (m->msg_iovlen == IOV_MAX || 742 (limit_to_mtu && c->msgbytes >= UDP_MAX_PAYLOAD_SIZE)) { 743 add_msghdr(c); 744 m = &c->msglist[c->msgused - 1]; 745 } 746 747 if (ensure_iov_space(c) != 0) 748 return -1; 749 750 /* If the fragment is too big to fit in the datagram, split it up */ 751 if (limit_to_mtu && len + c->msgbytes > UDP_MAX_PAYLOAD_SIZE) { 752 leftover = len + c->msgbytes - UDP_MAX_PAYLOAD_SIZE; 753 len -= leftover; 754 } else { 755 leftover = 0; 756 } 757 758 m = &c->msglist[c->msgused - 1]; 759 m->msg_iov[m->msg_iovlen].iov_base = (void *)buf; 760 m->msg_iov[m->msg_iovlen].iov_len = len; 761 762 c->msgbytes += len; 763 c->iovused++; 764 m->msg_iovlen++; 765 766 buf = ((char *)buf) + len; 767 len = leftover; 768 } while (leftover > 0); 769 770 return 0; 771 } 772 773 774 /* 775 * Constructs a set of UDP headers and attaches them to the outgoing messages. 776 */ 777 static int build_udp_headers(conn *c) { 778 int i; 779 unsigned char *hdr; 780 781 assert(c != NULL); 782 783 if (c->msgused > c->hdrsize) { 784 void *new_hdrbuf; 785 if (c->hdrbuf) { 786 new_hdrbuf = realloc(c->hdrbuf, c->msgused * 2 * UDP_HEADER_SIZE); 787 } else { 788 new_hdrbuf = malloc(c->msgused * 2 * UDP_HEADER_SIZE); 789 } 790 791 if (! new_hdrbuf) { 792 STATS_LOCK(); 793 stats.malloc_fails++; 794 STATS_UNLOCK(); 795 return -1; 796 } 797 c->hdrbuf = (unsigned char *)new_hdrbuf; 798 c->hdrsize = c->msgused * 2; 799 } 800 801 hdr = c->hdrbuf; 802 for (i = 0; i < c->msgused; i++) { 803 c->msglist[i].msg_iov[0].iov_base = (void*)hdr; 804 c->msglist[i].msg_iov[0].iov_len = UDP_HEADER_SIZE; 805 *hdr++ = c->request_id / 256; 806 *hdr++ = c->request_id % 256; 807 *hdr++ = i / 256; 808 *hdr++ = i % 256; 809 *hdr++ = c->msgused / 256; 810 *hdr++ = c->msgused % 256; 811 *hdr++ = 0; 812 *hdr++ = 0; 813 assert((void *) hdr == (caddr_t)c->msglist[i].msg_iov[0].iov_base + UDP_HEADER_SIZE); 814 } 815 816 return 0; 817 } 818 819 820 static void out_string(conn *c, const char *str) { 821 size_t len; 822 823 assert(c != NULL); 824 825 if (c->noreply) { 826 if (settings.verbose > 1) 827 fprintf(stderr, ">%d NOREPLY %s\n", c->sfd, str); 828 c->noreply = false; 829 conn_set_state(c, conn_new_cmd); 830 return; 831 } 832 833 if (settings.verbose > 1) 834 fprintf(stderr, ">%d %s\n", c->sfd, str); 835 836 /* Nuke a partial output... */ 837 c->msgcurr = 0; 838 c->msgused = 0; 839 c->iovused = 0; 840 add_msghdr(c); 841 842 len = strlen(str); 843 if ((len + 2) > c->wsize) { 844 /* ought to be always enough. just fail for simplicity */ 845 str = "SERVER_ERROR output line too long"; 846 len = strlen(str); 847 } 848 849 memcpy(c->wbuf, str, len); 850 memcpy(c->wbuf + len, "\r\n", 2); 851 c->wbytes = len + 2; 852 c->wcurr = c->wbuf; 853 854 conn_set_state(c, conn_write); 855 c->write_and_go = conn_new_cmd; 856 return; 857 } 858 859 /* 860 * Outputs a protocol-specific "out of memory" error. For ASCII clients, 861 * this is equivalent to out_string(). 862 */ 863 static void out_of_memory(conn *c, char *ascii_error) { 864 const static char error_prefix[] = "SERVER_ERROR "; 865 const static int error_prefix_len = sizeof(error_prefix) - 1; 866 867 if (c->protocol == binary_prot) { 868 /* Strip off the generic error prefix; it's irrelevant in binary */ 869 if (!strncmp(ascii_error, error_prefix, error_prefix_len)) { 870 ascii_error += error_prefix_len; 871 } 872 write_bin_error(c, PROTOCOL_BINARY_RESPONSE_ENOMEM, ascii_error, 0); 873 } else { 874 out_string(c, ascii_error); 875 } 876 } 877 878 /* 879 * we get here after reading the value in set/add/replace commands. The command 880 * has been stored in c->cmd, and the item is ready in c->item. 881 */ 882 static void complete_nread_ascii(conn *c) { 883 assert(c != NULL); 884 885 item *it = c->item; 886 int comm = c->cmd; 887 enum store_item_type ret; 888 889 pthread_mutex_lock(&c->thread->stats.mutex); 890 c->thread->stats.slab_stats[it->slabs_clsid].set_cmds++; 891 pthread_mutex_unlock(&c->thread->stats.mutex); 892 893 if (strncmp(ITEM_data(it) + it->nbytes - 2, "\r\n", 2) != 0) { 894 out_string(c, "CLIENT_ERROR bad data chunk"); 895 } else { 896 ret = store_item(it, comm, c); 897 898 #ifdef ENABLE_DTRACE 899 uint64_t cas = ITEM_get_cas(it); 900 switch (c->cmd) { 901 case NREAD_ADD: 902 MEMCACHED_COMMAND_ADD(c->sfd, ITEM_key(it), it->nkey, 903 (ret == 1) ? it->nbytes : -1, cas); 904 break; 905 case NREAD_REPLACE: 906 MEMCACHED_COMMAND_REPLACE(c->sfd, ITEM_key(it), it->nkey, 907 (ret == 1) ? it->nbytes : -1, cas); 908 break; 909 case NREAD_APPEND: 910 MEMCACHED_COMMAND_APPEND(c->sfd, ITEM_key(it), it->nkey, 911 (ret == 1) ? it->nbytes : -1, cas); 912 break; 913 case NREAD_PREPEND: 914 MEMCACHED_COMMAND_PREPEND(c->sfd, ITEM_key(it), it->nkey, 915 (ret == 1) ? it->nbytes : -1, cas); 916 break; 917 case NREAD_SET: 918 MEMCACHED_COMMAND_SET(c->sfd, ITEM_key(it), it->nkey, 919 (ret == 1) ? it->nbytes : -1, cas); 920 break; 921 case NREAD_CAS: 922 MEMCACHED_COMMAND_CAS(c->sfd, ITEM_key(it), it->nkey, it->nbytes, 923 cas); 924 break; 925 } 926 #endif 927 928 switch (ret) { 929 case STORED: 930 out_string(c, "STORED"); 931 break; 932 case EXISTS: 933 out_string(c, "EXISTS"); 934 break; 935 case NOT_FOUND: 936 out_string(c, "NOT_FOUND"); 937 break; 938 case NOT_STORED: 939 out_string(c, "NOT_STORED"); 940 break; 941 default: 942 out_string(c, "SERVER_ERROR Unhandled storage type."); 943 } 944 945 } 946 947 item_remove(c->item); /* release the c->item reference */ 948 c->item = 0; 949 } 950 951 /** 952 * get a pointer to the start of the request struct for the current command 953 */ 954 static void* binary_get_request(conn *c) { 955 char *ret = c->rcurr; 956 ret -= (sizeof(c->binary_header) + c->binary_header.request.keylen + 957 c->binary_header.request.extlen); 958 959 assert(ret >= c->rbuf); 960 return ret; 961 } 962 963 /** 964 * get a pointer to the key in this request 965 */ 966 static char* binary_get_key(conn *c) { 967 return c->rcurr - (c->binary_header.request.keylen); 968 } 969 970 static void add_bin_header(conn *c, uint16_t err, uint8_t hdr_len, uint16_t key_len, uint32_t body_len) { 971 protocol_binary_response_header* header; 972 973 assert(c); 974 975 c->msgcurr = 0; 976 c->msgused = 0; 977 c->iovused = 0; 978 if (add_msghdr(c) != 0) { 979 /* This should never run out of memory because iov and msg lists 980 * have minimum sizes big enough to hold an error response. 981 */ 982 out_of_memory(c, "SERVER_ERROR out of memory adding binary header"); 983 return; 984 } 985 986 header = (protocol_binary_response_header *)c->wbuf; 987 988 header->response.magic = (uint8_t)PROTOCOL_BINARY_RES; 989 header->response.opcode = c->binary_header.request.opcode; 990 header->response.keylen = (uint16_t)htons(key_len); 991 992 header->response.extlen = (uint8_t)hdr_len; 993 header->response.datatype = (uint8_t)PROTOCOL_BINARY_RAW_BYTES; 994 header->response.status = (uint16_t)htons(err); 995 996 header->response.bodylen = htonl(body_len); 997 header->response.opaque = c->opaque; 998 header->response.cas = htonll(c->cas); 999 1000 if (settings.verbose > 1) { 1001 int ii; 1002 fprintf(stderr, ">%d Writing bin response:", c->sfd); 1003 for (ii = 0; ii < sizeof(header->bytes); ++ii) { 1004 if (ii % 4 == 0) { 1005 fprintf(stderr, "\n>%d ", c->sfd); 1006 } 1007 fprintf(stderr, " 0x%02x", header->bytes[ii]); 1008 } 1009 fprintf(stderr, "\n"); 1010 } 1011 1012 add_iov(c, c->wbuf, sizeof(header->response)); 1013 } 1014 1015 /** 1016 * Writes a binary error response. If errstr is supplied, it is used as the 1017 * error text; otherwise a generic description of the error status code is 1018 * included. 1019 */ 1020 static void write_bin_error(conn *c, protocol_binary_response_status err, 1021 const char *errstr, int swallow) { 1022 size_t len; 1023 1024 if (!errstr) { 1025 switch (err) { 1026 case PROTOCOL_BINARY_RESPONSE_ENOMEM: 1027 errstr = "Out of memory"; 1028 break; 1029 case PROTOCOL_BINARY_RESPONSE_UNKNOWN_COMMAND: 1030 errstr = "Unknown command"; 1031 break; 1032 case PROTOCOL_BINARY_RESPONSE_KEY_ENOENT: 1033 errstr = "Not found"; 1034 break; 1035 case PROTOCOL_BINARY_RESPONSE_EINVAL: 1036 errstr = "Invalid arguments"; 1037 break; 1038 case PROTOCOL_BINARY_RESPONSE_KEY_EEXISTS: 1039 errstr = "Data exists for key."; 1040 break; 1041 case PROTOCOL_BINARY_RESPONSE_E2BIG: 1042 errstr = "Too large."; 1043 break; 1044 case PROTOCOL_BINARY_RESPONSE_DELTA_BADVAL: 1045 errstr = "Non-numeric server-side value for incr or decr"; 1046 break; 1047 case PROTOCOL_BINARY_RESPONSE_NOT_STORED: 1048 errstr = "Not stored."; 1049 break; 1050 case PROTOCOL_BINARY_RESPONSE_AUTH_ERROR: 1051 errstr = "Auth failure."; 1052 break; 1053 default: 1054 assert(false); 1055 errstr = "UNHANDLED ERROR"; 1056 fprintf(stderr, ">%d UNHANDLED ERROR: %d\n", c->sfd, err); 1057 } 1058 } 1059 1060 if (settings.verbose > 1) { 1061 fprintf(stderr, ">%d Writing an error: %s\n", c->sfd, errstr); 1062 } 1063 1064 len = strlen(errstr); 1065 add_bin_header(c, err, 0, 0, len); 1066 if (len > 0) { 1067 add_iov(c, errstr, len); 1068 } 1069 conn_set_state(c, conn_mwrite); 1070 if(swallow > 0) { 1071 c->sbytes = swallow; 1072 c->write_and_go = conn_swallow; 1073 } else { 1074 c->write_and_go = conn_new_cmd; 1075 } 1076 } 1077 1078 /* Form and send a response to a command over the binary protocol */ 1079 static void write_bin_response(conn *c, void *d, int hlen, int keylen, int dlen) { 1080 if (!c->noreply || c->cmd == PROTOCOL_BINARY_CMD_GET || 1081 c->cmd == PROTOCOL_BINARY_CMD_GETK) { 1082 add_bin_header(c, 0, hlen, keylen, dlen); 1083 if(dlen > 0) { 1084 add_iov(c, d, dlen); 1085 } 1086 conn_set_state(c, conn_mwrite); 1087 c->write_and_go = conn_new_cmd; 1088 } else { 1089 conn_set_state(c, conn_new_cmd); 1090 } 1091 } 1092 1093 static void complete_incr_bin(conn *c) { 1094 item *it; 1095 char *key; 1096 size_t nkey; 1097 /* Weird magic in add_delta forces me to pad here */ 1098 char tmpbuf[INCR_MAX_STORAGE_LEN]; 1099 uint64_t cas = 0; 1100 1101 protocol_binary_response_incr* rsp = (protocol_binary_response_incr*)c->wbuf; 1102 protocol_binary_request_incr* req = binary_get_request(c); 1103 1104 assert(c != NULL); 1105 assert(c->wsize >= sizeof(*rsp)); 1106 1107 /* fix byteorder in the request */ 1108 req->message.body.delta = ntohll(req->message.body.delta); 1109 req->message.body.initial = ntohll(req->message.body.initial); 1110 req->message.body.expiration = ntohl(req->message.body.expiration); 1111 key = binary_get_key(c); 1112 nkey = c->binary_header.request.keylen; 1113 1114 if (settings.verbose > 1) { 1115 int i; 1116 fprintf(stderr, "incr "); 1117 1118 for (i = 0; i < nkey; i++) { 1119 fprintf(stderr, "%c", key[i]); 1120 } 1121 fprintf(stderr, " %lld, %llu, %d\n", 1122 (long long)req->message.body.delta, 1123 (long long)req->message.body.initial, 1124 req->message.body.expiration); 1125 } 1126 1127 if (c->binary_header.request.cas != 0) { 1128 cas = c->binary_header.request.cas; 1129 } 1130 switch(add_delta(c, key, nkey, c->cmd == PROTOCOL_BINARY_CMD_INCREMENT, 1131 req->message.body.delta, tmpbuf, 1132 &cas)) { 1133 case OK: 1134 rsp->message.body.value = htonll(strtoull(tmpbuf, NULL, 10)); 1135 if (cas) { 1136 c->cas = cas; 1137 } 1138 write_bin_response(c, &rsp->message.body, 0, 0, 1139 sizeof(rsp->message.body.value)); 1140 break; 1141 case NON_NUMERIC: 1142 write_bin_error(c, PROTOCOL_BINARY_RESPONSE_DELTA_BADVAL, NULL, 0); 1143 break; 1144 case EOM: 1145 out_of_memory(c, "SERVER_ERROR Out of memory incrementing value"); 1146 break; 1147 case DELTA_ITEM_NOT_FOUND: 1148 if (req->message.body.expiration != 0xffffffff) { 1149 /* Save some room for the response */ 1150 rsp->message.body.value = htonll(req->message.body.initial); 1151 1152 snprintf(tmpbuf, INCR_MAX_STORAGE_LEN, "%llu", 1153 (unsigned long long)req->message.body.initial); 1154 int res = strlen(tmpbuf); 1155 it = item_alloc(key, nkey, 0, realtime(req->message.body.expiration), 1156 res + 2); 1157 1158 if (it != NULL) { 1159 memcpy(ITEM_data(it), tmpbuf, res); 1160 memcpy(ITEM_data(it) + res, "\r\n", 2); 1161 1162 if (store_item(it, NREAD_ADD, c)) { 1163 c->cas = ITEM_get_cas(it); 1164 write_bin_response(c, &rsp->message.body, 0, 0, sizeof(rsp->message.body.value)); 1165 } else { 1166 write_bin_error(c, PROTOCOL_BINARY_RESPONSE_NOT_STORED, 1167 NULL, 0); 1168 } 1169 item_remove(it); /* release our reference */ 1170 } else { 1171 out_of_memory(c, 1172 "SERVER_ERROR Out of memory allocating new item"); 1173 } 1174 } else { 1175 pthread_mutex_lock(&c->thread->stats.mutex); 1176 if (c->cmd == PROTOCOL_BINARY_CMD_INCREMENT) { 1177 c->thread->stats.incr_misses++; 1178 } else { 1179 c->thread->stats.decr_misses++; 1180 } 1181 pthread_mutex_unlock(&c->thread->stats.mutex); 1182 1183 write_bin_error(c, PROTOCOL_BINARY_RESPONSE_KEY_ENOENT, NULL, 0); 1184 } 1185 break; 1186 case DELTA_ITEM_CAS_MISMATCH: 1187 write_bin_error(c, PROTOCOL_BINARY_RESPONSE_KEY_EEXISTS, NULL, 0); 1188 break; 1189 } 1190 } 1191 1192 static void complete_update_bin(conn *c) { 1193 protocol_binary_response_status eno = PROTOCOL_BINARY_RESPONSE_EINVAL; 1194 enum store_item_type ret = NOT_STORED; 1195 assert(c != NULL); 1196 1197 item *it = c->item; 1198 1199 pthread_mutex_lock(&c->thread->stats.mutex); 1200 c->thread->stats.slab_stats[it->slabs_clsid].set_cmds++; 1201 pthread_mutex_unlock(&c->thread->stats.mutex); 1202 1203 /* We don't actually receive the trailing two characters in the bin 1204 * protocol, so we're going to just set them here */ 1205 *(ITEM_data(it) + it->nbytes - 2) = '\r'; 1206 *(ITEM_data(it) + it->nbytes - 1) = '\n'; 1207 1208 ret = store_item(it, c->cmd, c); 1209 1210 #ifdef ENABLE_DTRACE 1211 uint64_t cas = ITEM_get_cas(it); 1212 switch (c->cmd) { 1213 case NREAD_ADD: 1214 MEMCACHED_COMMAND_ADD(c->sfd, ITEM_key(it), it->nkey, 1215 (ret == STORED) ? it->nbytes : -1, cas); 1216 break; 1217 case NREAD_REPLACE: 1218 MEMCACHED_COMMAND_REPLACE(c->sfd, ITEM_key(it), it->nkey, 1219 (ret == STORED) ? it->nbytes : -1, cas); 1220 break; 1221 case NREAD_APPEND: 1222 MEMCACHED_COMMAND_APPEND(c->sfd, ITEM_key(it), it->nkey, 1223 (ret == STORED) ? it->nbytes : -1, cas); 1224 break; 1225 case NREAD_PREPEND: 1226 MEMCACHED_COMMAND_PREPEND(c->sfd, ITEM_key(it), it->nkey, 1227 (ret == STORED) ? it->nbytes : -1, cas); 1228 break; 1229 case NREAD_SET: 1230 MEMCACHED_COMMAND_SET(c->sfd, ITEM_key(it), it->nkey, 1231 (ret == STORED) ? it->nbytes : -1, cas); 1232 break; 1233 } 1234 #endif 1235 1236 switch (ret) { 1237 case STORED: 1238 /* Stored */ 1239 write_bin_response(c, NULL, 0, 0, 0); 1240 break; 1241 case EXISTS: 1242 write_bin_error(c, PROTOCOL_BINARY_RESPONSE_KEY_EEXISTS, NULL, 0); 1243 break; 1244 case NOT_FOUND: 1245 write_bin_error(c, PROTOCOL_BINARY_RESPONSE_KEY_ENOENT, NULL, 0); 1246 break; 1247 case NOT_STORED: 1248 if (c->cmd == NREAD_ADD) { 1249 eno = PROTOCOL_BINARY_RESPONSE_KEY_EEXISTS; 1250 } else if(c->cmd == NREAD_REPLACE) { 1251 eno = PROTOCOL_BINARY_RESPONSE_KEY_ENOENT; 1252 } else { 1253 eno = PROTOCOL_BINARY_RESPONSE_NOT_STORED; 1254 } 1255 write_bin_error(c, eno, NULL, 0); 1256 } 1257 1258 item_remove(c->item); /* release the c->item reference */ 1259 c->item = 0; 1260 } 1261 1262 static void process_bin_get_or_touch(conn *c) { 1263 item *it; 1264 1265 protocol_binary_response_get* rsp = (protocol_binary_response_get*)c->wbuf; 1266 char* key = binary_get_key(c); 1267 size_t nkey = c->binary_header.request.keylen; 1268 int should_touch = (c->cmd == PROTOCOL_BINARY_CMD_TOUCH || 1269 c->cmd == PROTOCOL_BINARY_CMD_GAT || 1270 c->cmd == PROTOCOL_BINARY_CMD_GATK); 1271 int should_return_key = (c->cmd == PROTOCOL_BINARY_CMD_GETK || 1272 c->cmd == PROTOCOL_BINARY_CMD_GATK); 1273 int should_return_value = (c->cmd != PROTOCOL_BINARY_CMD_TOUCH); 1274 1275 if (settings.verbose > 1) { 1276 fprintf(stderr, "<%d %s ", c->sfd, should_touch ? "TOUCH" : "GET"); 1277 fwrite(key, 1, nkey, stderr); 1278 fputc('\n', stderr); 1279 } 1280 1281 if (should_touch) { 1282 protocol_binary_request_touch *t = binary_get_request(c); 1283 time_t exptime = ntohl(t->message.body.expiration); 1284 1285 it = item_touch(key, nkey, realtime(exptime)); 1286 } else { 1287 it = item_get(key, nkey); 1288 } 1289 1290 if (it) { 1291 /* the length has two unnecessary bytes ("\r\n") */ 1292 uint16_t keylen = 0; 1293 uint32_t bodylen = sizeof(rsp->message.body) + (it->nbytes - 2); 1294 1295 item_update(it); 1296 pthread_mutex_lock(&c->thread->stats.mutex); 1297 if (should_touch) { 1298 c->thread->stats.touch_cmds++; 1299 c->thread->stats.slab_stats[it->slabs_clsid].touch_hits++; 1300 } else { 1301 c->thread->stats.get_cmds++; 1302 c->thread->stats.slab_stats[it->slabs_clsid].get_hits++; 1303 } 1304 pthread_mutex_unlock(&c->thread->stats.mutex); 1305 1306 if (should_touch) { 1307 MEMCACHED_COMMAND_TOUCH(c->sfd, ITEM_key(it), it->nkey, 1308 it->nbytes, ITEM_get_cas(it)); 1309 } else { 1310 MEMCACHED_COMMAND_GET(c->sfd, ITEM_key(it), it->nkey, 1311 it->nbytes, ITEM_get_cas(it)); 1312 } 1313 1314 if (c->cmd == PROTOCOL_BINARY_CMD_TOUCH) { 1315 bodylen -= it->nbytes - 2; 1316 } else if (should_return_key) { 1317 bodylen += nkey; 1318 keylen = nkey; 1319 } 1320 1321 add_bin_header(c, 0, sizeof(rsp->message.body), keylen, bodylen); 1322 rsp->message.header.response.cas = htonll(ITEM_get_cas(it)); 1323 1324 // add the flags 1325 rsp->message.body.flags = htonl(strtoul(ITEM_suffix(it), NULL, 10)); 1326 add_iov(c, &rsp->message.body, sizeof(rsp->message.body)); 1327 1328 if (should_return_key) { 1329 add_iov(c, ITEM_key(it), nkey); 1330 } 1331 1332 if (should_return_value) { 1333 /* Add the data minus the CRLF */ 1334 add_iov(c, ITEM_data(it), it->nbytes - 2); 1335 } 1336 1337 conn_set_state(c, conn_mwrite); 1338 c->write_and_go = conn_new_cmd; 1339 /* Remember this command so we can garbage collect it later */ 1340 c->item = it; 1341 } else { 1342 pthread_mutex_lock(&c->thread->stats.mutex); 1343 if (should_touch) { 1344 c->thread->stats.touch_cmds++; 1345 c->thread->stats.touch_misses++; 1346 } else { 1347 c->thread->stats.get_cmds++; 1348 c->thread->stats.get_misses++; 1349 } 1350 pthread_mutex_unlock(&c->thread->stats.mutex); 1351 1352 if (should_touch) { 1353 MEMCACHED_COMMAND_TOUCH(c->sfd, key, nkey, -1, 0); 1354 } else { 1355 MEMCACHED_COMMAND_GET(c->sfd, key, nkey, -1, 0); 1356 } 1357 1358 if (c->noreply) { 1359 conn_set_state(c, conn_new_cmd); 1360 } else { 1361 if (should_return_key) { 1362 char *ofs = c->wbuf + sizeof(protocol_binary_response_header); 1363 add_bin_header(c, PROTOCOL_BINARY_RESPONSE_KEY_ENOENT, 1364 0, nkey, nkey); 1365 memcpy(ofs, key, nkey); 1366 add_iov(c, ofs, nkey); 1367 conn_set_state(c, conn_mwrite); 1368 c->write_and_go = conn_new_cmd; 1369 } else { 1370 write_bin_error(c, PROTOCOL_BINARY_RESPONSE_KEY_ENOENT, 1371 NULL, 0); 1372 } 1373 } 1374 } 1375 1376 if (settings.detail_enabled) { 1377 stats_prefix_record_get(key, nkey, NULL != it); 1378 } 1379 } 1380 1381 static void append_bin_stats(const char *key, const uint16_t klen, 1382 const char *val, const uint32_t vlen, 1383 conn *c) { 1384 char *buf = c->stats.buffer + c->stats.offset; 1385 uint32_t bodylen = klen + vlen; 1386 protocol_binary_response_header header = { 1387 .response.magic = (uint8_t)PROTOCOL_BINARY_RES, 1388 .response.opcode = PROTOCOL_BINARY_CMD_STAT, 1389 .response.keylen = (uint16_t)htons(klen), 1390 .response.datatype = (uint8_t)PROTOCOL_BINARY_RAW_BYTES, 1391 .response.bodylen = htonl(bodylen), 1392 .response.opaque = c->opaque 1393 }; 1394 1395 memcpy(buf, header.bytes, sizeof(header.response)); 1396 buf += sizeof(header.response); 1397 1398 if (klen > 0) { 1399 memcpy(buf, key, klen); 1400 buf += klen; 1401 1402 if (vlen > 0) { 1403 memcpy(buf, val, vlen); 1404 } 1405 } 1406 1407 c->stats.offset += sizeof(header.response) + bodylen; 1408 } 1409 1410 static void append_ascii_stats(const char *key, const uint16_t klen, 1411 const char *val, const uint32_t vlen, 1412 conn *c) { 1413 char *pos = c->stats.buffer + c->stats.offset; 1414 uint32_t nbytes = 0; 1415 int remaining = c->stats.size - c->stats.offset; 1416 int room = remaining - 1; 1417 1418 if (klen == 0 && vlen == 0) { 1419 nbytes = snprintf(pos, room, "END\r\n"); 1420 } else if (vlen == 0) { 1421 nbytes = snprintf(pos, room, "STAT %s\r\n", key); 1422 } else { 1423 nbytes = snprintf(pos, room, "STAT %s %s\r\n", key, val); 1424 } 1425 1426 c->stats.offset += nbytes; 1427 } 1428 1429 static bool grow_stats_buf(conn *c, size_t needed) { 1430 size_t nsize = c->stats.size; 1431 size_t available = nsize - c->stats.offset; 1432 bool rv = true; 1433 1434 /* Special case: No buffer -- need to allocate fresh */ 1435 if (c->stats.buffer == NULL) { 1436 nsize = 1024; 1437 available = c->stats.size = c->stats.offset = 0; 1438 } 1439 1440 while (needed > available) { 1441 assert(nsize > 0); 1442 nsize = nsize << 1; 1443 available = nsize - c->stats.offset; 1444 } 1445 1446 if (nsize != c->stats.size) { 1447 char *ptr = realloc(c->stats.buffer, nsize); 1448 if (ptr) { 1449 c->stats.buffer = ptr; 1450 c->stats.size = nsize; 1451 } else { 1452 STATS_LOCK(); 1453 stats.malloc_fails++; 1454 STATS_UNLOCK(); 1455 rv = false; 1456 } 1457 } 1458 1459 return rv; 1460 } 1461 1462 static void append_stats(const char *key, const uint16_t klen, 1463 const char *val, const uint32_t vlen, 1464 const void *cookie) 1465 { 1466 /* value without a key is invalid */ 1467 if (klen == 0 && vlen > 0) { 1468 return ; 1469 } 1470 1471 conn *c = (conn*)cookie; 1472 1473 if (c->protocol == binary_prot) { 1474 size_t needed = vlen + klen + sizeof(protocol_binary_response_header); 1475 if (!grow_stats_buf(c, needed)) { 1476 return ; 1477 } 1478 append_bin_stats(key, klen, val, vlen, c); 1479 } else { 1480 size_t needed = vlen + klen + 10; // 10 == "STAT = \r\n" 1481 if (!grow_stats_buf(c, needed)) { 1482 return ; 1483 } 1484 append_ascii_stats(key, klen, val, vlen, c); 1485 } 1486 1487 assert(c->stats.offset <= c->stats.size); 1488 } 1489 1490 static void process_bin_stat(conn *c) { 1491 char *subcommand = binary_get_key(c); 1492 size_t nkey = c->binary_header.request.keylen; 1493 1494 if (settings.verbose > 1) { 1495 int ii; 1496 fprintf(stderr, "<%d STATS ", c->sfd); 1497 for (ii = 0; ii < nkey; ++ii) { 1498 fprintf(stderr, "%c", subcommand[ii]); 1499 } 1500 fprintf(stderr, "\n"); 1501 } 1502 1503 if (nkey == 0) { 1504 /* request all statistics */ 1505 server_stats(&append_stats, c); 1506 (void)get_stats(NULL, 0, &append_stats, c); 1507 } else if (strncmp(subcommand, "reset", 5) == 0) { 1508 stats_reset(); 1509 } else if (strncmp(subcommand, "settings", 8) == 0) { 1510 process_stat_settings(&append_stats, c); 1511 } else if (strncmp(subcommand, "detail", 6) == 0) { 1512 char *subcmd_pos = subcommand + 6; 1513 if (strncmp(subcmd_pos, " dump", 5) == 0) { 1514 int len; 1515 char *dump_buf = stats_prefix_dump(&len); 1516 if (dump_buf == NULL || len <= 0) { 1517 out_of_memory(c, "SERVER_ERROR Out of memory generating stats"); 1518 return ; 1519 } else { 1520 append_stats("detailed", strlen("detailed"), dump_buf, len, c); 1521 free(dump_buf); 1522 } 1523 } else if (strncmp(subcmd_pos, " on", 3) == 0) { 1524 settings.detail_enabled = 1; 1525 } else if (strncmp(subcmd_pos, " off", 4) == 0) { 1526 settings.detail_enabled = 0; 1527 } else { 1528 write_bin_error(c, PROTOCOL_BINARY_RESPONSE_KEY_ENOENT, NULL, 0); 1529 return; 1530 } 1531 } else { 1532 if (get_stats(subcommand, nkey, &append_stats, c)) { 1533 if (c->stats.buffer == NULL) { 1534 out_of_memory(c, "SERVER_ERROR Out of memory generating stats"); 1535 } else { 1536 write_and_free(c, c->stats.buffer, c->stats.offset); 1537 c->stats.buffer = NULL; 1538 } 1539 } else { 1540 write_bin_error(c, PROTOCOL_BINARY_RESPONSE_KEY_ENOENT, NULL, 0); 1541 } 1542 1543 return; 1544 } 1545 1546 /* Append termination package and start the transfer */ 1547 append_stats(NULL, 0, NULL, 0, c); 1548 if (c->stats.buffer == NULL) { 1549 out_of_memory(c, "SERVER_ERROR Out of memory preparing to send stats"); 1550 } else { 1551 write_and_free(c, c->stats.buffer, c->stats.offset); 1552 c->stats.buffer = NULL; 1553 } 1554 } 1555 1556 static void bin_read_key(conn *c, enum bin_substates next_substate, int extra) { 1557 assert(c); 1558 c->substate = next_substate; 1559 c->rlbytes = c->keylen + extra; 1560 1561 /* Ok... do we have room for the extras and the key in the input buffer? */ 1562 ptrdiff_t offset = c->rcurr + sizeof(protocol_binary_request_header) - c->rbuf; 1563 if (c->rlbytes > c->rsize - offset) { 1564 size_t nsize = c->rsize; 1565 size_t size = c->rlbytes + sizeof(protocol_binary_request_header); 1566 1567 while (size > nsize) { 1568 nsize *= 2; 1569 } 1570 1571 if (nsize != c->rsize) { 1572 if (settings.verbose > 1) { 1573 fprintf(stderr, "%d: Need to grow buffer from %lu to %lu\n", 1574 c->sfd, (unsigned long)c->rsize, (unsigned long)nsize); 1575 } 1576 char *newm = realloc(c->rbuf, nsize); 1577 if (newm == NULL) { 1578 STATS_LOCK(); 1579 stats.malloc_fails++; 1580 STATS_UNLOCK(); 1581 if (settings.verbose) { 1582 fprintf(stderr, "%d: Failed to grow buffer.. closing connection\n", 1583 c->sfd); 1584 } 1585 conn_set_state(c, conn_closing); 1586 return; 1587 } 1588 1589 c->rbuf= newm; 1590 /* rcurr should point to the same offset in the packet */ 1591 c->rcurr = c->rbuf + offset - sizeof(protocol_binary_request_header); 1592 c->rsize = nsize; 1593 } 1594 if (c->rbuf != c->rcurr) { 1595 memmove(c->rbuf, c->rcurr, c->rbytes); 1596 c->rcurr = c->rbuf; 1597 if (settings.verbose > 1) { 1598 fprintf(stderr, "%d: Repack input buffer\n", c->sfd); 1599 } 1600 } 1601 } 1602 1603 /* preserve the header in the buffer.. */ 1604 c->ritem = c->rcurr + sizeof(protocol_binary_request_header); 1605 conn_set_state(c, conn_nread); 1606 } 1607 1608 /* Just write an error message and disconnect the client */ 1609 static void handle_binary_protocol_error(conn *c) { 1610 write_bin_error(c, PROTOCOL_BINARY_RESPONSE_EINVAL, NULL, 0); 1611 if (settings.verbose) { 1612 fprintf(stderr, "Protocol error (opcode %02x), close connection %d\n", 1613 c->binary_header.request.opcode, c->sfd); 1614 } 1615 c->write_and_go = conn_closing; 1616 } 1617 1618 static void init_sasl_conn(conn *c) { 1619 assert(c); 1620 /* should something else be returned? */ 1621 if (!settings.sasl) 1622 return; 1623 1624 c->authenticated = false; 1625 1626 if (!c->sasl_conn) { 1627 int result=sasl_server_new("memcached", 1628 NULL, 1629 my_sasl_hostname[0] ? my_sasl_hostname : NULL, 1630 NULL, NULL, 1631 NULL, 0, &c->sasl_conn); 1632 if (result != SASL_OK) { 1633 if (settings.verbose) { 1634 fprintf(stderr, "Failed to initialize SASL conn.\n"); 1635 } 1636 c->sasl_conn = NULL; 1637 } 1638 } 1639 } 1640 1641 static void bin_list_sasl_mechs(conn *c) { 1642 // Guard against a disabled SASL. 1643 if (!settings.sasl) { 1644 write_bin_error(c, PROTOCOL_BINARY_RESPONSE_UNKNOWN_COMMAND, NULL, 1645 c->binary_header.request.bodylen 1646 - c->binary_header.request.keylen); 1647 return; 1648 } 1649 1650 init_sasl_conn(c); 1651 const char *result_string = NULL; 1652 unsigned int string_length = 0; 1653 int result=sasl_listmech(c->sasl_conn, NULL, 1654 "", /* What to prepend the string with */ 1655 " ", /* What to separate mechanisms with */ 1656 "", /* What to append to the string */ 1657 &result_string, &string_length, 1658 NULL); 1659 if (result != SASL_OK) { 1660 /* Perhaps there's a better error for this... */ 1661 if (settings.verbose) { 1662 fprintf(stderr, "Failed to list SASL mechanisms.\n"); 1663 } 1664 write_bin_error(c, PROTOCOL_BINARY_RESPONSE_AUTH_ERROR, NULL, 0); 1665 return; 1666 } 1667 write_bin_response(c, (char*)result_string, 0, 0, string_length); 1668 } 1669 1670 static void process_bin_sasl_auth(conn *c) { 1671 // Guard for handling disabled SASL on the server. 1672 if (!settings.sasl) { 1673 write_bin_error(c, PROTOCOL_BINARY_RESPONSE_UNKNOWN_COMMAND, NULL, 1674 c->binary_header.request.bodylen 1675 - c->binary_header.request.keylen); 1676 return; 1677 } 1678 1679 assert(c->binary_header.request.extlen == 0); 1680 1681 int nkey = c->binary_header.request.keylen; 1682 int vlen = c->binary_header.request.bodylen - nkey; 1683 1684 if (nkey > MAX_SASL_MECH_LEN) { 1685 write_bin_error(c, PROTOCOL_BINARY_RESPONSE_EINVAL, NULL, vlen); 1686 c->write_and_go = conn_swallow; 1687 return; 1688 } 1689 1690 char *key = binary_get_key(c); 1691 assert(key); 1692 1693 item *it = item_alloc(key, nkey, 0, 0, vlen); 1694 1695 if (it == 0) { 1696 write_bin_error(c, PROTOCOL_BINARY_RESPONSE_ENOMEM, NULL, vlen); 1697 c->write_and_go = conn_swallow; 1698 return; 1699 } 1700 1701 c->item = it; 1702 c->ritem = ITEM_data(it); 1703 c->rlbytes = vlen; 1704 conn_set_state(c, conn_nread); 1705 c->substate = bin_reading_sasl_auth_data; 1706 } 1707 1708 static void process_bin_complete_sasl_auth(conn *c) { 1709 assert(settings.sasl); 1710 const char *out = NULL; 1711 unsigned int outlen = 0; 1712 1713 assert(c->item); 1714 init_sasl_conn(c); 1715 1716 int nkey = c->binary_header.request.keylen; 1717 int vlen = c->binary_header.request.bodylen - nkey; 1718 1719 char mech[nkey+1]; 1720 memcpy(mech, ITEM_key((item*)c->item), nkey); 1721 mech[nkey] = 0x00; 1722 1723 if (settings.verbose) 1724 fprintf(stderr, "mech: ``%s'' with %d bytes of data\n", mech, vlen); 1725 1726 const char *challenge = vlen == 0 ? NULL : ITEM_data((item*) c->item); 1727 1728 int result=-1; 1729 1730 switch (c->cmd) { 1731 case PROTOCOL_BINARY_CMD_SASL_AUTH: 1732 result = sasl_server_start(c->sasl_conn, mech, 1733 challenge, vlen, 1734 &out, &outlen); 1735 break; 1736 case PROTOCOL_BINARY_CMD_SASL_STEP: 1737 result = sasl_server_step(c->sasl_conn, 1738 challenge, vlen, 1739 &out, &outlen); 1740 break; 1741 default: 1742 assert(false); /* CMD should be one of the above */ 1743 /* This code is pretty much impossible, but makes the compiler 1744 happier */ 1745 if (settings.verbose) { 1746 fprintf(stderr, "Unhandled command %d with challenge %s\n", 1747 c->cmd, challenge); 1748 } 1749 break; 1750 } 1751 1752 item_unlink(c->item); 1753 1754 if (settings.verbose) { 1755 fprintf(stderr, "sasl result code: %d\n", result); 1756 } 1757 1758 switch(result) { 1759 case SASL_OK: 1760 c->authenticated = true; 1761 write_bin_response(c, "Authenticated", 0, 0, strlen("Authenticated")); 1762 pthread_mutex_lock(&c->thread->stats.mutex); 1763 c->thread->stats.auth_cmds++; 1764 pthread_mutex_unlock(&c->thread->stats.mutex); 1765 break; 1766 case SASL_CONTINUE: 1767 add_bin_header(c, PROTOCOL_BINARY_RESPONSE_AUTH_CONTINUE, 0, 0, outlen); 1768 if(outlen > 0) { 1769 add_iov(c, out, outlen); 1770 } 1771 conn_set_state(c, conn_mwrite); 1772 c->write_and_go = conn_new_cmd; 1773 break; 1774 default: 1775 if (settings.verbose) 1776 fprintf(stderr, "Unknown sasl response: %d\n", result); 1777 write_bin_error(c, PROTOCOL_BINARY_RESPONSE_AUTH_ERROR, NULL, 0); 1778 pthread_mutex_lock(&c->thread->stats.mutex); 1779 c->thread->stats.auth_cmds++; 1780 c->thread->stats.auth_errors++; 1781 pthread_mutex_unlock(&c->thread->stats.mutex); 1782 } 1783 } 1784 1785 static bool authenticated(conn *c) { 1786 assert(settings.sasl); 1787 bool rv = false; 1788 1789 switch (c->cmd) { 1790 case PROTOCOL_BINARY_CMD_SASL_LIST_MECHS: /* FALLTHROUGH */ 1791 case PROTOCOL_BINARY_CMD_SASL_AUTH: /* FALLTHROUGH */ 1792 case PROTOCOL_BINARY_CMD_SASL_STEP: /* FALLTHROUGH */ 1793 case PROTOCOL_BINARY_CMD_VERSION: /* FALLTHROUGH */ 1794 rv = true; 1795 break; 1796 default: 1797 rv = c->authenticated; 1798 } 1799 1800 if (settings.verbose > 1) { 1801 fprintf(stderr, "authenticated() in cmd 0x%02x is %s\n", 1802 c->cmd, rv ? "true" : "false"); 1803 } 1804 1805 return rv; 1806 } 1807 1808 static void dispatch_bin_command(conn *c) { 1809 int protocol_error = 0; 1810 1811 int extlen = c->binary_header.request.extlen; 1812 int keylen = c->binary_header.request.keylen; 1813 uint32_t bodylen = c->binary_header.request.bodylen; 1814 1815 if (settings.sasl && !authenticated(c)) { 1816 write_bin_error(c, PROTOCOL_BINARY_RESPONSE_AUTH_ERROR, NULL, 0); 1817 c->write_and_go = conn_closing; 1818 return; 1819 } 1820 1821 MEMCACHED_PROCESS_COMMAND_START(c->sfd, c->rcurr, c->rbytes); 1822 c->noreply = true; 1823 1824 /* binprot supports 16bit keys, but internals are still 8bit */ 1825 if (keylen > KEY_MAX_LENGTH) { 1826 handle_binary_protocol_error(c); 1827 return; 1828 } 1829 1830 switch (c->cmd) { 1831 case PROTOCOL_BINARY_CMD_SETQ: 1832 c->cmd = PROTOCOL_BINARY_CMD_SET; 1833 break; 1834 case PROTOCOL_BINARY_CMD_ADDQ: 1835 c->cmd = PROTOCOL_BINARY_CMD_ADD; 1836 break; 1837 case PROTOCOL_BINARY_CMD_REPLACEQ: 1838 c->cmd = PROTOCOL_BINARY_CMD_REPLACE; 1839 break; 1840 case PROTOCOL_BINARY_CMD_DELETEQ: 1841 c->cmd = PROTOCOL_BINARY_CMD_DELETE; 1842 break; 1843 case PROTOCOL_BINARY_CMD_INCREMENTQ: 1844 c->cmd = PROTOCOL_BINARY_CMD_INCREMENT; 1845 break; 1846 case PROTOCOL_BINARY_CMD_DECREMENTQ: 1847 c->cmd = PROTOCOL_BINARY_CMD_DECREMENT; 1848 break; 1849 case PROTOCOL_BINARY_CMD_QUITQ: 1850 c->cmd = PROTOCOL_BINARY_CMD_QUIT; 1851 break; 1852 case PROTOCOL_BINARY_CMD_FLUSHQ: 1853 c->cmd = PROTOCOL_BINARY_CMD_FLUSH; 1854 break; 1855 case PROTOCOL_BINARY_CMD_APPENDQ: 1856 c->cmd = PROTOCOL_BINARY_CMD_APPEND; 1857 break; 1858 case PROTOCOL_BINARY_CMD_PREPENDQ: 1859 c->cmd = PROTOCOL_BINARY_CMD_PREPEND; 1860 break; 1861 case PROTOCOL_BINARY_CMD_GETQ: 1862 c->cmd = PROTOCOL_BINARY_CMD_GET; 1863 break; 1864 case PROTOCOL_BINARY_CMD_GETKQ: 1865 c->cmd = PROTOCOL_BINARY_CMD_GETK; 1866 break; 1867 case PROTOCOL_BINARY_CMD_GATQ: 1868 c->cmd = PROTOCOL_BINARY_CMD_GAT; 1869 break; 1870 case PROTOCOL_BINARY_CMD_GATKQ: 1871 c->cmd = PROTOCOL_BINARY_CMD_GAT; 1872 break; 1873 default: 1874 c->noreply = false; 1875 } 1876 1877 switch (c->cmd) { 1878 case PROTOCOL_BINARY_CMD_VERSION: 1879 if (extlen == 0 && keylen == 0 && bodylen == 0) { 1880 write_bin_response(c, VERSION, 0, 0, strlen(VERSION)); 1881 } else { 1882 protocol_error = 1; 1883 } 1884 break; 1885 case PROTOCOL_BINARY_CMD_FLUSH: 1886 if (keylen == 0 && bodylen == extlen && (extlen == 0 || extlen == 4)) { 1887 bin_read_key(c, bin_read_flush_exptime, extlen); 1888 } else { 1889 protocol_error = 1; 1890 } 1891 break; 1892 case PROTOCOL_BINARY_CMD_NOOP: 1893 if (extlen == 0 && keylen == 0 && bodylen == 0) { 1894 write_bin_response(c, NULL, 0, 0, 0); 1895 } else { 1896 protocol_error = 1; 1897 } 1898 break; 1899 case PROTOCOL_BINARY_CMD_SET: /* FALLTHROUGH */ 1900 case PROTOCOL_BINARY_CMD_ADD: /* FALLTHROUGH */ 1901 case PROTOCOL_BINARY_CMD_REPLACE: 1902 if (extlen == 8 && keylen != 0 && bodylen >= (keylen + 8)) { 1903 bin_read_key(c, bin_reading_set_header, 8); 1904 } else { 1905 protocol_error = 1; 1906 } 1907 break; 1908 case PROTOCOL_BINARY_CMD_GETQ: /* FALLTHROUGH */ 1909 case PROTOCOL_BINARY_CMD_GET: /* FALLTHROUGH */ 1910 case PROTOCOL_BINARY_CMD_GETKQ: /* FALLTHROUGH */ 1911 case PROTOCOL_BINARY_CMD_GETK: 1912 if (extlen == 0 && bodylen == keylen && keylen > 0) { 1913 bin_read_key(c, bin_reading_get_key, 0); 1914 } else { 1915 protocol_error = 1; 1916 } 1917 break; 1918 case PROTOCOL_BINARY_CMD_DELETE: 1919 if (keylen > 0 && extlen == 0 && bodylen == keylen) { 1920 bin_read_key(c, bin_reading_del_header, extlen); 1921 } else { 1922 protocol_error = 1; 1923 } 1924 break; 1925 case PROTOCOL_BINARY_CMD_INCREMENT: 1926 case PROTOCOL_BINARY_CMD_DECREMENT: 1927 if (keylen > 0 && extlen == 20 && bodylen == (keylen + extlen)) { 1928 bin_read_key(c, bin_reading_incr_header, 20); 1929 } else { 1930 protocol_error = 1; 1931 } 1932 break; 1933 case PROTOCOL_BINARY_CMD_APPEND: 1934 case PROTOCOL_BINARY_CMD_PREPEND: 1935 if (keylen > 0 && extlen == 0) { 1936 bin_read_key(c, bin_reading_set_header, 0); 1937 } else { 1938 protocol_error = 1; 1939 } 1940 break; 1941 case PROTOCOL_BINARY_CMD_STAT: 1942 if (extlen == 0) { 1943 bin_read_key(c, bin_reading_stat, 0); 1944 } else { 1945 protocol_error = 1; 1946 } 1947 break; 1948 case PROTOCOL_BINARY_CMD_QUIT: 1949 if (keylen == 0 && extlen == 0 && bodylen == 0) { 1950 write_bin_response(c, NULL, 0, 0, 0); 1951 c->write_and_go = conn_closing; 1952 if (c->noreply) { 1953 conn_set_state(c, conn_closing); 1954 } 1955 } else { 1956 protocol_error = 1; 1957 } 1958 break; 1959 case PROTOCOL_BINARY_CMD_SASL_LIST_MECHS: 1960 if (extlen == 0 && keylen == 0 && bodylen == 0) { 1961 bin_list_sasl_mechs(c); 1962 } else { 1963 protocol_error = 1; 1964 } 1965 break; 1966 case PROTOCOL_BINARY_CMD_SASL_AUTH: 1967 case PROTOCOL_BINARY_CMD_SASL_STEP: 1968 if (extlen == 0 && keylen != 0) { 1969 bin_read_key(c, bin_reading_sasl_auth, 0); 1970 } else { 1971 protocol_error = 1; 1972 } 1973 break; 1974 case PROTOCOL_BINARY_CMD_TOUCH: 1975 case PROTOCOL_BINARY_CMD_GAT: 1976 case PROTOCOL_BINARY_CMD_GATQ: 1977 case PROTOCOL_BINARY_CMD_GATK: 1978 case PROTOCOL_BINARY_CMD_GATKQ: 1979 if (extlen == 4 && keylen != 0) { 1980 bin_read_key(c, bin_reading_touch_key, 4); 1981 } else { 1982 protocol_error = 1; 1983 } 1984 break; 1985 default: 1986 write_bin_error(c, PROTOCOL_BINARY_RESPONSE_UNKNOWN_COMMAND, NULL, 1987 bodylen); 1988 } 1989 1990 if (protocol_error) 1991 handle_binary_protocol_error(c); 1992 } 1993 1994 static void process_bin_update(conn *c) { 1995 char *key; 1996 int nkey; 1997 int vlen; 1998 item *it; 1999 protocol_binary_request_set* req = binary_get_request(c); 2000 2001 assert(c != NULL); 2002 2003 key = binary_get_key(c); 2004 nkey = c->binary_header.request.keylen; 2005 2006 /* fix byteorder in the request */ 2007 req->message.body.flags = ntohl(req->message.body.flags); 2008 req->message.body.expiration = ntohl(req->message.body.expiration); 2009 2010 vlen = c->binary_header.request.bodylen - (nkey + c->binary_header.request.extlen); 2011 2012 if (settings.verbose > 1) { 2013 int ii; 2014 if (c->cmd == PROTOCOL_BINARY_CMD_ADD) { 2015 fprintf(stderr, "<%d ADD ", c->sfd); 2016 } else if (c->cmd == PROTOCOL_BINARY_CMD_SET) { 2017 fprintf(stderr, "<%d SET ", c->sfd); 2018 } else { 2019 fprintf(stderr, "<%d REPLACE ", c->sfd); 2020 } 2021 for (ii = 0; ii < nkey; ++ii) { 2022 fprintf(stderr, "%c", key[ii]); 2023 } 2024 2025 fprintf(stderr, " Value len is %d", vlen); 2026 fprintf(stderr, "\n"); 2027 } 2028 2029 if (settings.detail_enabled) { 2030 stats_prefix_record_set(key, nkey); 2031 } 2032 2033 it = item_alloc(key, nkey, req->message.body.flags, 2034 realtime(req->message.body.expiration), vlen+2); 2035 2036 if (it == 0) { 2037 if (! item_size_ok(nkey, req->message.body.flags, vlen + 2)) { 2038 write_bin_error(c, PROTOCOL_BINARY_RESPONSE_E2BIG, NULL, vlen); 2039 } else { 2040 out_of_memory(c, "SERVER_ERROR Out of memory allocating item"); 2041 } 2042 2043 /* Avoid stale data persisting in cache because we failed alloc. 2044 * Unacceptable for SET. Anywhere else too? */ 2045 if (c->cmd == PROTOCOL_BINARY_CMD_SET) { 2046 it = item_get(key, nkey); 2047 if (it) { 2048 item_unlink(it); 2049 item_remove(it); 2050 } 2051 } 2052 2053 /* swallow the data line */ 2054 c->write_and_go = conn_swallow; 2055 return; 2056 } 2057 2058 ITEM_set_cas(it, c->binary_header.request.cas); 2059 2060 switch (c->cmd) { 2061 case PROTOCOL_BINARY_CMD_ADD: 2062 c->cmd = NREAD_ADD; 2063 break; 2064 case PROTOCOL_BINARY_CMD_SET: 2065 c->cmd = NREAD_SET; 2066 break; 2067 case PROTOCOL_BINARY_CMD_REPLACE: 2068 c->cmd = NREAD_REPLACE; 2069 break; 2070 default: 2071 assert(0); 2072 } 2073 2074 if (ITEM_get_cas(it) != 0) { 2075 c->cmd = NREAD_CAS; 2076 } 2077 2078 c->item = it; 2079 c->ritem = ITEM_data(it); 2080 c->rlbytes = vlen; 2081 conn_set_state(c, conn_nread); 2082 c->substate = bin_read_set_value; 2083 } 2084 2085 static void process_bin_append_prepend(conn *c) { 2086 char *key; 2087 int nkey; 2088 int vlen; 2089 item *it; 2090 2091 assert(c != NULL); 2092 2093 key = binary_get_key(c); 2094 nkey = c->binary_header.request.keylen; 2095 vlen = c->binary_header.request.bodylen - nkey; 2096 2097 if (settings.verbose > 1) { 2098 fprintf(stderr, "Value len is %d\n", vlen); 2099 } 2100 2101 if (settings.detail_enabled) { 2102 stats_prefix_record_set(key, nkey); 2103 } 2104 2105 it = item_alloc(key, nkey, 0, 0, vlen+2); 2106 2107 if (it == 0) { 2108 if (! item_size_ok(nkey, 0, vlen + 2)) { 2109 write_bin_error(c, PROTOCOL_BINARY_RESPONSE_E2BIG, NULL, vlen); 2110 } else { 2111 out_of_memory(c, "SERVER_ERROR Out of memory allocating item"); 2112 } 2113 /* swallow the data line */ 2114 c->write_and_go = conn_swallow; 2115 return; 2116 } 2117 2118 ITEM_set_cas(it, c->binary_header.request.cas); 2119 2120 switch (c->cmd) { 2121 case PROTOCOL_BINARY_CMD_APPEND: 2122 c->cmd = NREAD_APPEND; 2123 break; 2124 case PROTOCOL_BINARY_CMD_PREPEND: 2125 c->cmd = NREAD_PREPEND; 2126 break; 2127 default: 2128 assert(0); 2129 } 2130 2131 c->item = it; 2132 c->ritem = ITEM_data(it); 2133 c->rlbytes = vlen; 2134 conn_set_state(c, conn_nread); 2135 c->substate = bin_read_set_value; 2136 } 2137 2138 static void process_bin_flush(conn *c) { 2139 time_t exptime = 0; 2140 protocol_binary_request_flush* req = binary_get_request(c); 2141 2142 if (!settings.flush_enabled) { 2143 // flush_all is not allowed but we log it on stats 2144 write_bin_error(c, PROTOCOL_BINARY_RESPONSE_AUTH_ERROR, NULL, 0); 2145 return; 2146 } 2147 2148 if (c->binary_header.request.extlen == sizeof(req->message.body)) { 2149 exptime = ntohl(req->message.body.expiration); 2150 } 2151 2152 if (exptime > 0) { 2153 settings.oldest_live = realtime(exptime) - 1; 2154 } else { 2155 settings.oldest_live = current_time - 1; 2156 } 2157 item_flush_expired(); 2158 2159 pthread_mutex_lock(&c->thread->stats.mutex); 2160 c->thread->stats.flush_cmds++; 2161 pthread_mutex_unlock(&c->thread->stats.mutex); 2162 2163 write_bin_response(c, NULL, 0, 0, 0); 2164 } 2165 2166 static void process_bin_delete(conn *c) { 2167 item *it; 2168 2169 protocol_binary_request_delete* req = binary_get_request(c); 2170 2171 char* key = binary_get_key(c); 2172 size_t nkey = c->binary_header.request.keylen; 2173 2174 assert(c != NULL); 2175 2176 if (settings.verbose > 1) { 2177 int ii; 2178 fprintf(stderr, "Deleting "); 2179 for (ii = 0; ii < nkey; ++ii) { 2180 fprintf(stderr, "%c", key[ii]); 2181 } 2182 fprintf(stderr, "\n"); 2183 } 2184 2185 if (settings.detail_enabled) { 2186 stats_prefix_record_delete(key, nkey); 2187 } 2188 2189 it = item_get(key, nkey); 2190 if (it) { 2191 uint64_t cas = ntohll(req->message.header.request.cas); 2192 if (cas == 0 || cas == ITEM_get_cas(it)) { 2193 MEMCACHED_COMMAND_DELETE(c->sfd, ITEM_key(it), it->nkey); 2194 pthread_mutex_lock(&c->thread->stats.mutex); 2195 c->thread->stats.slab_stats[it->slabs_clsid].delete_hits++; 2196 pthread_mutex_unlock(&c->thread->stats.mutex); 2197 item_unlink(it); 2198 write_bin_response(c, NULL, 0, 0, 0); 2199 } else { 2200 write_bin_error(c, PROTOCOL_BINARY_RESPONSE_KEY_EEXISTS, NULL, 0); 2201 } 2202 item_remove(it); /* release our reference */ 2203 } else { 2204 write_bin_error(c, PROTOCOL_BINARY_RESPONSE_KEY_ENOENT, NULL, 0); 2205 pthread_mutex_lock(&c->thread->stats.mutex); 2206 c->thread->stats.delete_misses++; 2207 pthread_mutex_unlock(&c->thread->stats.mutex); 2208 } 2209 } 2210 2211 static void complete_nread_binary(conn *c) { 2212 assert(c != NULL); 2213 assert(c->cmd >= 0); 2214 2215 switch(c->substate) { 2216 case bin_reading_set_header: 2217 if (c->cmd == PROTOCOL_BINARY_CMD_APPEND || 2218 c->cmd == PROTOCOL_BINARY_CMD_PREPEND) { 2219 process_bin_append_prepend(c); 2220 } else { 2221 process_bin_update(c); 2222 } 2223 break; 2224 case bin_read_set_value: 2225 complete_update_bin(c); 2226 break; 2227 case bin_reading_get_key: 2228 case bin_reading_touch_key: 2229 process_bin_get_or_touch(c); 2230 break; 2231 case bin_reading_stat: 2232 process_bin_stat(c); 2233 break; 2234 case bin_reading_del_header: 2235 process_bin_delete(c); 2236 break; 2237 case bin_reading_incr_header: 2238 complete_incr_bin(c); 2239 break; 2240 case bin_read_flush_exptime: 2241 process_bin_flush(c); 2242 break; 2243 case bin_reading_sasl_auth: 2244 process_bin_sasl_auth(c); 2245 break; 2246 case bin_reading_sasl_auth_data: 2247 process_bin_complete_sasl_auth(c); 2248 break; 2249 default: 2250 fprintf(stderr, "Not handling substate %d\n", c->substate); 2251 assert(0); 2252 } 2253 } 2254 2255 static void reset_cmd_handler(conn *c) { 2256 c->cmd = -1; 2257 c->substate = bin_no_state; 2258 if(c->item != NULL) { 2259 item_remove(c->item); 2260 c->item = NULL; 2261 } 2262 conn_shrink(c); 2263 if (c->rbytes > 0) { 2264 conn_set_state(c, conn_parse_cmd); 2265 } else { 2266 conn_set_state(c, conn_waiting); 2267 } 2268 } 2269 2270 static void complete_nread(conn *c) { 2271 assert(c != NULL); 2272 assert(c->protocol == ascii_prot 2273 || c->protocol == binary_prot); 2274 2275 if (c->protocol == ascii_prot) { 2276 complete_nread_ascii(c); 2277 } else if (c->protocol == binary_prot) { 2278 complete_nread_binary(c); 2279 } 2280 } 2281 2282 /* 2283 * Stores an item in the cache according to the semantics of one of the set 2284 * commands. In threaded mode, this is protected by the cache lock. 2285 * 2286 * Returns the state of storage. 2287 */ 2288 enum store_item_type do_store_item(item *it, int comm, conn *c, const uint32_t hv) { 2289 char *key = ITEM_key(it); 2290 item *old_it = do_item_get(key, it->nkey, hv); 2291 enum store_item_type stored = NOT_STORED; 2292 2293 item *new_it = NULL; 2294 int flags; 2295 2296 if (old_it != NULL && comm == NREAD_ADD) { 2297 /* add only adds a nonexistent item, but promote to head of LRU */ 2298 do_item_update(old_it); 2299 } else if (!old_it && (comm == NREAD_REPLACE 2300 || comm == NREAD_APPEND || comm == NREAD_PREPEND)) 2301 { 2302 /* replace only replaces an existing value; don't store */ 2303 } else if (comm == NREAD_CAS) { 2304 /* validate cas operation */ 2305 if(old_it == NULL) { 2306 // LRU expired 2307 stored = NOT_FOUND; 2308 pthread_mutex_lock(&c->thread->stats.mutex); 2309 c->thread->stats.cas_misses++; 2310 pthread_mutex_unlock(&c->thread->stats.mutex); 2311 } 2312 else if (ITEM_get_cas(it) == ITEM_get_cas(old_it)) { 2313 // cas validates 2314 // it and old_it may belong to different classes. 2315 // I'm updating the stats for the one that's getting pushed out 2316 pthread_mutex_lock(&c->thread->stats.mutex); 2317 c->thread->stats.slab_stats[old_it->slabs_clsid].cas_hits++; 2318 pthread_mutex_unlock(&c->thread->stats.mutex); 2319 2320 item_replace(old_it, it, hv); 2321 stored = STORED; 2322 } else { 2323 pthread_mutex_lock(&c->thread->stats.mutex); 2324 c->thread->stats.slab_stats[old_it->slabs_clsid].cas_badval++; 2325 pthread_mutex_unlock(&c->thread->stats.mutex); 2326 2327 if(settings.verbose > 1) { 2328 fprintf(stderr, "CAS: failure: expected %llu, got %llu\n", 2329 (unsigned long long)ITEM_get_cas(old_it), 2330 (unsigned long long)ITEM_get_cas(it)); 2331 } 2332 stored = EXISTS; 2333 } 2334 } else { 2335 /* 2336 * Append - combine new and old record into single one. Here it's 2337 * atomic and thread-safe. 2338 */ 2339 if (comm == NREAD_APPEND || comm == NREAD_PREPEND) { 2340 /* 2341 * Validate CAS 2342 */ 2343 if (ITEM_get_cas(it) != 0) { 2344 // CAS much be equal 2345 if (ITEM_get_cas(it) != ITEM_get_cas(old_it)) { 2346 stored = EXISTS; 2347 } 2348 } 2349 2350 if (stored == NOT_STORED) { 2351 /* we have it and old_it here - alloc memory to hold both */ 2352 /* flags was already lost - so recover them from ITEM_suffix(it) */ 2353 2354 flags = (int) strtol(ITEM_suffix(old_it), (char **) NULL, 10); 2355 2356 new_it = do_item_alloc(key, it->nkey, flags, old_it->exptime, it->nbytes + old_it->nbytes - 2 /* CRLF */, hv); 2357 2358 if (new_it == NULL) { 2359 /* SERVER_ERROR out of memory */ 2360 if (old_it != NULL) 2361 do_item_remove(old_it); 2362 2363 return NOT_STORED; 2364 } 2365 2366 /* copy data from it and old_it to new_it */ 2367 2368 if (comm == NREAD_APPEND) { 2369 memcpy(ITEM_data(new_it), ITEM_data(old_it), old_it->nbytes); 2370 memcpy(ITEM_data(new_it) + old_it->nbytes - 2 /* CRLF */, ITEM_data(it), it->nbytes); 2371 } else { 2372 /* NREAD_PREPEND */ 2373 memcpy(ITEM_data(new_it), ITEM_data(it), it->nbytes); 2374 memcpy(ITEM_data(new_it) + it->nbytes - 2 /* CRLF */, ITEM_data(old_it), old_it->nbytes); 2375 } 2376 2377 it = new_it; 2378 } 2379 } 2380 2381 if (stored == NOT_STORED) { 2382 if (old_it != NULL) 2383 item_replace(old_it, it, hv); 2384 else 2385 do_item_link(it, hv); 2386 2387 c->cas = ITEM_get_cas(it); 2388 2389 stored = STORED; 2390 } 2391 } 2392 2393 if (old_it != NULL) 2394 do_item_remove(old_it); /* release our reference */ 2395 if (new_it != NULL) 2396 do_item_remove(new_it); 2397 2398 if (stored == STORED) { 2399 c->cas = ITEM_get_cas(it); 2400 } 2401 2402 return stored; 2403 } 2404 2405 typedef struct token_s { 2406 char *value; 2407 size_t length; 2408 } token_t; 2409 2410 #define COMMAND_TOKEN 0 2411 #define SUBCOMMAND_TOKEN 1 2412 #define KEY_TOKEN 1 2413 2414 #define MAX_TOKENS 8 2415 2416 /* 2417 * Tokenize the command string by replacing whitespace with '\0' and update 2418 * the token array tokens with pointer to start of each token and length. 2419 * Returns total number of tokens. The last valid token is the terminal 2420 * token (value points to the first unprocessed character of the string and 2421 * length zero). 2422 * 2423 * Usage example: 2424 * 2425 * while(tokenize_command(command, ncommand, tokens, max_tokens) > 0) { 2426 * for(int ix = 0; tokens[ix].length != 0; ix++) { 2427 * ... 2428 * } 2429 * ncommand = tokens[ix].value - command; 2430 * command = tokens[ix].value; 2431 * } 2432 */ 2433 static size_t tokenize_command(char *command, token_t *tokens, const size_t max_tokens) { 2434 char *s, *e; 2435 size_t ntokens = 0; 2436 size_t len = strlen(command); 2437 unsigned int i = 0; 2438 2439 assert(command != NULL && tokens != NULL && max_tokens > 1); 2440 2441 s = e = command; 2442 for (i = 0; i < len; i++) { 2443 if (*e == ' ') { 2444 if (s != e) { 2445 tokens[ntokens].value = s; 2446 tokens[ntokens].length = e - s; 2447 ntokens++; 2448 *e = '\0'; 2449 if (ntokens == max_tokens - 1) { 2450 e++; 2451 s = e; /* so we don't add an extra token */ 2452 break; 2453 } 2454 } 2455 s = e + 1; 2456 } 2457 e++; 2458 } 2459 2460 if (s != e) { 2461 tokens[ntokens].value = s; 2462 tokens[ntokens].length = e - s; 2463 ntokens++; 2464 } 2465 2466 /* 2467 * If we scanned the whole string, the terminal value pointer is null, 2468 * otherwise it is the first unprocessed character. 2469 */ 2470 tokens[ntokens].value = *e == '\0' ? NULL : e; 2471 tokens[ntokens].length = 0; 2472 ntokens++; 2473 2474 return ntokens; 2475 } 2476 2477 /* set up a connection to write a buffer then free it, used for stats */ 2478 static void write_and_free(conn *c, char *buf, int bytes) { 2479 if (buf) { 2480 c->write_and_free = buf; 2481 c->wcurr = buf; 2482 c->wbytes = bytes; 2483 conn_set_state(c, conn_write); 2484 c->write_and_go = conn_new_cmd; 2485 } else { 2486 out_of_memory(c, "SERVER_ERROR out of memory writing stats"); 2487 } 2488 } 2489 2490 static inline bool set_noreply_maybe(conn *c, token_t *tokens, size_t ntokens) 2491 { 2492 int noreply_index = ntokens - 2; 2493 2494 /* 2495 NOTE: this function is not the first place where we are going to 2496 send the reply. We could send it instead from process_command() 2497 if the request line has wrong number of tokens. However parsing 2498 malformed line for "noreply" option is not reliable anyway, so 2499 it can't be helped. 2500 */ 2501 if (tokens[noreply_index].value 2502 && strcmp(tokens[noreply_index].value, "noreply") == 0) { 2503 c->noreply = true; 2504 } 2505 return c->noreply; 2506 } 2507 2508 void append_stat(const char *name, ADD_STAT add_stats, conn *c, 2509 const char *fmt, ...) { 2510 char val_str[STAT_VAL_LEN]; 2511 int vlen; 2512 va_list ap; 2513 2514 assert(name); 2515 assert(add_stats); 2516 assert(c); 2517 assert(fmt); 2518 2519 va_start(ap, fmt); 2520 vlen = vsnprintf(val_str, sizeof(val_str) - 1, fmt, ap); 2521 va_end(ap); 2522 2523 add_stats(name, strlen(name), val_str, vlen, c); 2524 } 2525 2526 inline static void process_stats_detail(conn *c, const char *command) { 2527 assert(c != NULL); 2528 2529 if (strcmp(command, "on") == 0) { 2530 settings.detail_enabled = 1; 2531 out_string(c, "OK"); 2532 } 2533 else if (strcmp(command, "off") == 0) { 2534 settings.detail_enabled = 0; 2535 out_string(c, "OK"); 2536 } 2537 else if (strcmp(command, "dump") == 0) { 2538 int len; 2539 char *stats = stats_prefix_dump(&len); 2540 write_and_free(c, stats, len); 2541 } 2542 else { 2543 out_string(c, "CLIENT_ERROR usage: stats detail on|off|dump"); 2544 } 2545 } 2546 2547 /* return server specific stats only */ 2548 static void server_stats(ADD_STAT add_stats, conn *c) { 2549 pid_t pid = getpid(); 2550 rel_time_t now = current_time; 2551 2552 struct thread_stats thread_stats; 2553 threadlocal_stats_aggregate(&thread_stats); 2554 struct slab_stats slab_stats; 2555 slab_stats_aggregate(&thread_stats, &slab_stats); 2556 2557 #ifndef WIN32 2558 struct rusage usage; 2559 getrusage(RUSAGE_SELF, &usage); 2560 #endif /* !WIN32 */ 2561 2562 STATS_LOCK(); 2563 2564 APPEND_STAT("pid", "%lu", (long)pid); 2565 APPEND_STAT("uptime", "%u", now - ITEM_UPDATE_INTERVAL); 2566 APPEND_STAT("time", "%ld", now + (long)process_started); 2567 APPEND_STAT("version", "%s", VERSION); 2568 APPEND_STAT("libevent", "%s", event_get_version()); 2569 APPEND_STAT("pointer_size", "%d", (int)(8 * sizeof(void *))); 2570 2571 #ifndef WIN32 2572 append_stat("rusage_user", add_stats, c, "%ld.%06ld", 2573 (long)usage.ru_utime.tv_sec, 2574 (long)usage.ru_utime.tv_usec); 2575 append_stat("rusage_system", add_stats, c, "%ld.%06ld", 2576 (long)usage.ru_stime.tv_sec, 2577 (long)usage.ru_stime.tv_usec); 2578 #endif /* !WIN32 */ 2579 2580 APPEND_STAT("curr_connections", "%u", stats.curr_conns - 1); 2581 APPEND_STAT("total_connections", "%u", stats.total_conns); 2582 if (settings.maxconns_fast) { 2583 APPEND_STAT("rejected_connections", "%llu", (unsigned long long)stats.rejected_conns); 2584 } 2585 APPEND_STAT("connection_structures", "%u", stats.conn_structs); 2586 APPEND_STAT("reserved_fds", "%u", stats.reserved_fds); 2587 APPEND_STAT("cmd_get", "%llu", (unsigned long long)thread_stats.get_cmds); 2588 APPEND_STAT("cmd_set", "%llu", (unsigned long long)slab_stats.set_cmds); 2589 APPEND_STAT("cmd_flush", "%llu", (unsigned long long)thread_stats.flush_cmds); 2590 APPEND_STAT("cmd_touch", "%llu", (unsigned long long)thread_stats.touch_cmds); 2591 APPEND_STAT("get_hits", "%llu", (unsigned long long)slab_stats.get_hits); 2592 APPEND_STAT("get_misses", "%llu", (unsigned long long)thread_stats.get_misses); 2593 APPEND_STAT("delete_misses", "%llu", (unsigned long long)thread_stats.delete_misses); 2594 APPEND_STAT("delete_hits", "%llu", (unsigned long long)slab_stats.delete_hits); 2595 APPEND_STAT("incr_misses", "%llu", (unsigned long long)thread_stats.incr_misses); 2596 APPEND_STAT("incr_hits", "%llu", (unsigned long long)slab_stats.incr_hits); 2597 APPEND_STAT("decr_misses", "%llu", (unsigned long long)thread_stats.decr_misses); 2598 APPEND_STAT("decr_hits", "%llu", (unsigned long long)slab_stats.decr_hits); 2599 APPEND_STAT("cas_misses", "%llu", (unsigned long long)thread_stats.cas_misses); 2600 APPEND_STAT("cas_hits", "%llu", (unsigned long long)slab_stats.cas_hits); 2601 APPEND_STAT("cas_badval", "%llu", (unsigned long long)slab_stats.cas_badval); 2602 APPEND_STAT("touch_hits", "%llu", (unsigned long long)slab_stats.touch_hits); 2603 APPEND_STAT("touch_misses", "%llu", (unsigned long long)thread_stats.touch_misses); 2604 APPEND_STAT("auth_cmds", "%llu", (unsigned long long)thread_stats.auth_cmds); 2605 APPEND_STAT("auth_errors", "%llu", (unsigned long long)thread_stats.auth_errors); 2606 APPEND_STAT("bytes_read", "%llu", (unsigned long long)thread_stats.bytes_read); 2607 APPEND_STAT("bytes_written", "%llu", (unsigned long long)thread_stats.bytes_written); 2608 APPEND_STAT("limit_maxbytes", "%llu", (unsigned long long)settings.maxbytes); 2609 APPEND_STAT("accepting_conns", "%u", stats.accepting_conns); 2610 APPEND_STAT("listen_disabled_num", "%llu", (unsigned long long)stats.listen_disabled_num); 2611 APPEND_STAT("threads", "%d", settings.num_threads); 2612 APPEND_STAT("conn_yields", "%llu", (unsigned long long)thread_stats.conn_yields); 2613 APPEND_STAT("hash_power_level", "%u", stats.hash_power_level); 2614 APPEND_STAT("hash_bytes", "%llu", (unsigned long long)stats.hash_bytes); 2615 APPEND_STAT("hash_is_expanding", "%u", stats.hash_is_expanding); 2616 if (settings.slab_reassign) { 2617 APPEND_STAT("slab_reassign_running", "%u", stats.slab_reassign_running); 2618 APPEND_STAT("slabs_moved", "%llu", stats.slabs_moved); 2619 } 2620 if (settings.lru_crawler) { 2621 APPEND_STAT("lru_crawler_running", "%u", stats.lru_crawler_running); 2622 } 2623 APPEND_STAT("malloc_fails", "%llu", 2624 (unsigned long long)stats.malloc_fails); 2625 STATS_UNLOCK(); 2626 } 2627 2628 static void process_stat_settings(ADD_STAT add_stats, void *c) { 2629 assert(add_stats); 2630 APPEND_STAT("maxbytes", "%llu", (unsigned long long)settings.maxbytes); 2631 APPEND_STAT("maxconns", "%d", settings.maxconns); 2632 APPEND_STAT("tcpport", "%d", settings.port); 2633 APPEND_STAT("udpport", "%d", settings.udpport); 2634 APPEND_STAT("inter", "%s", settings.inter ? settings.inter : "NULL"); 2635 APPEND_STAT("verbosity", "%d", settings.verbose); 2636 APPEND_STAT("oldest", "%lu", (unsigned long)settings.oldest_live); 2637 APPEND_STAT("evictions", "%s", settings.evict_to_free ? "on" : "off"); 2638 APPEND_STAT("domain_socket", "%s", 2639 settings.socketpath ? settings.socketpath : "NULL"); 2640 APPEND_STAT("umask", "%o", settings.access); 2641 APPEND_STAT("growth_factor", "%.2f", settings.factor); 2642 APPEND_STAT("chunk_size", "%d", settings.chunk_size); 2643 APPEND_STAT("num_threads", "%d", settings.num_threads); 2644 APPEND_STAT("num_threads_per_udp", "%d", settings.num_threads_per_udp); 2645 APPEND_STAT("stat_key_prefix", "%c", settings.prefix_delimiter); 2646 APPEND_STAT("detail_enabled", "%s", 2647 settings.detail_enabled ? "yes" : "no"); 2648 APPEND_STAT("reqs_per_event", "%d", settings.reqs_per_event); 2649 APPEND_STAT("cas_enabled", "%s", settings.use_cas ? "yes" : "no"); 2650 APPEND_STAT("tcp_backlog", "%d", settings.backlog); 2651 APPEND_STAT("binding_protocol", "%s", 2652 prot_text(settings.binding_protocol)); 2653 APPEND_STAT("auth_enabled_sasl", "%s", settings.sasl ? "yes" : "no"); 2654 APPEND_STAT("item_size_max", "%d", settings.item_size_max); 2655 APPEND_STAT("maxconns_fast", "%s", settings.maxconns_fast ? "yes" : "no"); 2656 APPEND_STAT("hashpower_init", "%d", settings.hashpower_init); 2657 APPEND_STAT("slab_reassign", "%s", settings.slab_reassign ? "yes" : "no"); 2658 APPEND_STAT("slab_automove", "%d", settings.slab_automove); 2659 APPEND_STAT("lru_crawler", "%s", settings.lru_crawler ? "yes" : "no"); 2660 APPEND_STAT("lru_crawler_sleep", "%d", settings.lru_crawler_sleep); 2661 APPEND_STAT("lru_crawler_tocrawl", "%lu", (unsigned long)settings.lru_crawler_tocrawl); 2662 APPEND_STAT("tail_repair_time", "%d", settings.tail_repair_time); 2663 APPEND_STAT("flush_enabled", "%s", settings.flush_enabled ? "yes" : "no"); 2664 APPEND_STAT("hash_algorithm", "%s", settings.hash_algorithm); 2665 } 2666 2667 static void conn_to_str(const conn *c, char *buf) { 2668 char addr_text[MAXPATHLEN]; 2669 2670 if (!c) { 2671 strcpy(buf, "<null>"); 2672 } else if (c->state == conn_closed) { 2673 strcpy(buf, "<closed>"); 2674 } else { 2675 const char *protoname = "?"; 2676 struct sockaddr_in6 local_addr; 2677 struct sockaddr *addr = (void *)&c->request_addr; 2678 int af; 2679 unsigned short port = 0; 2680 2681 /* For listen ports and idle UDP ports, show listen address */ 2682 if (c->state == conn_listening || 2683 (IS_UDP(c->transport) && 2684 c->state == conn_read)) { 2685 socklen_t local_addr_len = sizeof(local_addr); 2686 2687 if (getsockname(c->sfd, 2688 (struct sockaddr *)&local_addr, 2689 &local_addr_len) == 0) { 2690 addr = (struct sockaddr *)&local_addr; 2691 } 2692 } 2693 2694 af = addr->sa_family; 2695 addr_text[0] = '\0'; 2696 2697 switch (af) { 2698 case AF_INET: 2699 (void) inet_ntop(af, 2700 &((struct sockaddr_in *)addr)->sin_addr, 2701 addr_text, 2702 sizeof(addr_text) - 1); 2703 port = ntohs(((struct sockaddr_in *)addr)->sin_port); 2704 protoname = IS_UDP(c->transport) ? "udp" : "tcp"; 2705 break; 2706 2707 case AF_INET6: 2708 addr_text[0] = '['; 2709 addr_text[1] = '\0'; 2710 if (inet_ntop(af, 2711 &((struct sockaddr_in6 *)addr)->sin6_addr, 2712 addr_text + 1, 2713 sizeof(addr_text) - 2)) { 2714 strcat(addr_text, "]"); 2715 } 2716 port = ntohs(((struct sockaddr_in6 *)addr)->sin6_port); 2717 protoname = IS_UDP(c->transport) ? "udp6" : "tcp6"; 2718 break; 2719 2720 case AF_UNIX: 2721 strncpy(addr_text, 2722 ((struct sockaddr_un *)addr)->sun_path, 2723 sizeof(addr_text) - 1); 2724 addr_text[sizeof(addr_text)-1] = '\0'; 2725 protoname = "unix"; 2726 break; 2727 } 2728 2729 if (strlen(addr_text) < 2) { 2730 /* Most likely this is a connected UNIX-domain client which 2731 * has no peer socket address, but there's no portable way 2732 * to tell for sure. 2733 */ 2734 sprintf(addr_text, "<AF %d>", af); 2735 } 2736 2737 if (port) { 2738 sprintf(buf, "%s:%s:%u", protoname, addr_text, port); 2739 } else { 2740 sprintf(buf, "%s:%s", protoname, addr_text); 2741 } 2742 } 2743 } 2744 2745 static void process_stats_conns(ADD_STAT add_stats, void *c) { 2746 int i; 2747 char key_str[STAT_KEY_LEN]; 2748 char val_str[STAT_VAL_LEN]; 2749 char conn_name[MAXPATHLEN + sizeof("unix:")]; 2750 int klen = 0, vlen = 0; 2751 2752 assert(add_stats); 2753 2754 for (i = 0; i < max_fds; i++) { 2755 if (conns[i]) { 2756 /* This is safe to do unlocked because conns are never freed; the 2757 * worst that'll happen will be a minor inconsistency in the 2758 * output -- not worth the complexity of the locking that'd be 2759 * required to prevent it. 2760 */ 2761 if (conns[i]->state != conn_closed) { 2762 conn_to_str(conns[i], conn_name); 2763 2764 APPEND_NUM_STAT(i, "addr", "%s", conn_name); 2765 APPEND_NUM_STAT(i, "state", "%s", 2766 state_text(conns[i]->state)); 2767 APPEND_NUM_STAT(i, "secs_since_last_cmd", "%d", 2768 current_time - conns[i]->last_cmd_time); 2769 } 2770 } 2771 } 2772 } 2773 2774 static void process_stat(conn *c, token_t *tokens, const size_t ntokens) { 2775 const char *subcommand = tokens[SUBCOMMAND_TOKEN].value; 2776 assert(c != NULL); 2777 2778 if (ntokens < 2) { 2779 out_string(c, "CLIENT_ERROR bad command line"); 2780 return; 2781 } 2782 2783 if (ntokens == 2) { 2784 server_stats(&append_stats, c); 2785 (void)get_stats(NULL, 0, &append_stats, c); 2786 } else if (strcmp(subcommand, "reset") == 0) { 2787 stats_reset(); 2788 out_string(c, "RESET"); 2789 return ; 2790 } else if (strcmp(subcommand, "detail") == 0) { 2791 /* NOTE: how to tackle detail with binary? */ 2792 if (ntokens < 4) 2793 process_stats_detail(c, ""); /* outputs the error message */ 2794 else 2795 process_stats_detail(c, tokens[2].value); 2796 /* Output already generated */ 2797 return ; 2798 } else if (strcmp(subcommand, "settings") == 0) { 2799 process_stat_settings(&append_stats, c); 2800 } else if (strcmp(subcommand, "cachedump") == 0) { 2801 char *buf; 2802 unsigned int bytes, id, limit = 0; 2803 2804 if (ntokens < 5) { 2805 out_string(c, "CLIENT_ERROR bad command line"); 2806 return; 2807 } 2808 2809 if (!safe_strtoul(tokens[2].value, &id) || 2810 !safe_strtoul(tokens[3].value, &limit)) { 2811 out_string(c, "CLIENT_ERROR bad command line format"); 2812 return; 2813 } 2814 2815 if (id >= POWER_LARGEST) { 2816 out_string(c, "CLIENT_ERROR Illegal slab id"); 2817 return; 2818 } 2819 2820 buf = item_cachedump(id, limit, &bytes); 2821 write_and_free(c, buf, bytes); 2822 return ; 2823 } else if (strcmp(subcommand, "conns") == 0) { 2824 process_stats_conns(&append_stats, c); 2825 } else { 2826 /* getting here means that the subcommand is either engine specific or 2827 is invalid. query the engine and see. */ 2828 if (get_stats(subcommand, strlen(subcommand), &append_stats, c)) { 2829 if (c->stats.buffer == NULL) { 2830 out_of_memory(c, "SERVER_ERROR out of memory writing stats"); 2831 } else { 2832 write_and_free(c, c->stats.buffer, c->stats.offset); 2833 c->stats.buffer = NULL; 2834 } 2835 } else { 2836 out_string(c, "ERROR"); 2837 } 2838 return ; 2839 } 2840 2841 /* append terminator and start the transfer */ 2842 append_stats(NULL, 0, NULL, 0, c); 2843 2844 if (c->stats.buffer == NULL) { 2845 out_of_memory(c, "SERVER_ERROR out of memory writing stats"); 2846 } else { 2847 write_and_free(c, c->stats.buffer, c->stats.offset); 2848 c->stats.buffer = NULL; 2849 } 2850 } 2851 2852 /* ntokens is overwritten here... shrug.. */ 2853 static inline void process_get_command(conn *c, token_t *tokens, size_t ntokens, bool return_cas) { 2854 char *key; 2855 size_t nkey; 2856 int i = 0; 2857 item *it; 2858 token_t *key_token = &tokens[KEY_TOKEN]; 2859 char *suffix; 2860 assert(c != NULL); 2861 2862 do { 2863 while(key_token->length != 0) { 2864 2865 key = key_token->value; 2866 nkey = key_token->length; 2867 2868 if(nkey > KEY_MAX_LENGTH) { 2869 out_string(c, "CLIENT_ERROR bad command line format"); 2870 while (i-- > 0) { 2871 item_remove(*(c->ilist + i)); 2872 } 2873 return; 2874 } 2875 2876 it = item_get(key, nkey); 2877 if (settings.detail_enabled) { 2878 stats_prefix_record_get(key, nkey, NULL != it); 2879 } 2880 if (it) { 2881 if (i >= c->isize) { 2882 item **new_list = realloc(c->ilist, sizeof(item *) * c->isize * 2); 2883 if (new_list) { 2884 c->isize *= 2; 2885 c->ilist = new_list; 2886 } else { 2887 STATS_LOCK(); 2888 stats.malloc_fails++; 2889 STATS_UNLOCK(); 2890 item_remove(it); 2891 break; 2892 } 2893 } 2894 2895 /* 2896 * Construct the response. Each hit adds three elements to the 2897 * outgoing data list: 2898 * "VALUE " 2899 * key 2900 * " " + flags + " " + data length + "\r\n" + data (with \r\n) 2901 */ 2902 2903 if (return_cas) 2904 { 2905 MEMCACHED_COMMAND_GET(c->sfd, ITEM_key(it), it->nkey, 2906 it->nbytes, ITEM_get_cas(it)); 2907 /* Goofy mid-flight realloc. */ 2908 if (i >= c->suffixsize) { 2909 char **new_suffix_list = realloc(c->suffixlist, 2910 sizeof(char *) * c->suffixsize * 2); 2911 if (new_suffix_list) { 2912 c->suffixsize *= 2; 2913 c->suffixlist = new_suffix_list; 2914 } else { 2915 STATS_LOCK(); 2916 stats.malloc_fails++; 2917 STATS_UNLOCK(); 2918 item_remove(it); 2919 break; 2920 } 2921 } 2922 2923 suffix = cache_alloc(c->thread->suffix_cache); 2924 if (suffix == NULL) { 2925 STATS_LOCK(); 2926 stats.malloc_fails++; 2927 STATS_UNLOCK(); 2928 out_of_memory(c, "SERVER_ERROR out of memory making CAS suffix"); 2929 item_remove(it); 2930 while (i-- > 0) { 2931 item_remove(*(c->ilist + i)); 2932 } 2933 return; 2934 } 2935 *(c->suffixlist + i) = suffix; 2936 int suffix_len = snprintf(suffix, SUFFIX_SIZE, 2937 " %llu\r\n", 2938 (unsigned long long)ITEM_get_cas(it)); 2939 if (add_iov(c, "VALUE ", 6) != 0 || 2940 add_iov(c, ITEM_key(it), it->nkey) != 0 || 2941 add_iov(c, ITEM_suffix(it), it->nsuffix - 2) != 0 || 2942 add_iov(c, suffix, suffix_len) != 0 || 2943 add_iov(c, ITEM_data(it), it->nbytes) != 0) 2944 { 2945 item_remove(it); 2946 break; 2947 } 2948 } 2949 else 2950 { 2951 MEMCACHED_COMMAND_GET(c->sfd, ITEM_key(it), it->nkey, 2952 it->nbytes, ITEM_get_cas(it)); 2953 if (add_iov(c, "VALUE ", 6) != 0 || 2954 add_iov(c, ITEM_key(it), it->nkey) != 0 || 2955 add_iov(c, ITEM_suffix(it), it->nsuffix + it->nbytes) != 0) 2956 { 2957 item_remove(it); 2958 break; 2959 } 2960 } 2961 2962 2963 if (settings.verbose > 1) { 2964 int ii; 2965 fprintf(stderr, ">%d sending key ", c->sfd); 2966 for (ii = 0; ii < it->nkey; ++ii) { 2967 fprintf(stderr, "%c", key[ii]); 2968 } 2969 fprintf(stderr, "\n"); 2970 } 2971 2972 /* item_get() has incremented it->refcount for us */ 2973 pthread_mutex_lock(&c->thread->stats.mutex); 2974 c->thread->stats.slab_stats[it->slabs_clsid].get_hits++; 2975 c->thread->stats.get_cmds++; 2976 pthread_mutex_unlock(&c->thread->stats.mutex); 2977 item_update(it); 2978 *(c->ilist + i) = it; 2979 i++; 2980 2981 } else { 2982 pthread_mutex_lock(&c->thread->stats.mutex); 2983 c->thread->stats.get_misses++; 2984 c->thread->stats.get_cmds++; 2985 pthread_mutex_unlock(&c->thread->stats.mutex); 2986 MEMCACHED_COMMAND_GET(c->sfd, key, nkey, -1, 0); 2987 } 2988 2989 key_token++; 2990 } 2991 2992 /* 2993 * If the command string hasn't been fully processed, get the next set 2994 * of tokens. 2995 */ 2996 if(key_token->value != NULL) { 2997 ntokens = tokenize_command(key_token->value, tokens, MAX_TOKENS); 2998 key_token = tokens; 2999 } 3000 3001 } while(key_token->value != NULL); 3002 3003 c->icurr = c->ilist; 3004 c->ileft = i; 3005 if (return_cas) { 3006 c->suffixcurr = c->suffixlist; 3007 c->suffixleft = i; 3008 } 3009 3010 if (settings.verbose > 1) 3011 fprintf(stderr, ">%d END\n", c->sfd); 3012 3013 /* 3014 If the loop was terminated because of out-of-memory, it is not 3015 reliable to add END\r\n to the buffer, because it might not end 3016 in \r\n. So we send SERVER_ERROR instead. 3017 */ 3018 if (key_token->value != NULL || add_iov(c, "END\r\n", 5) != 0 3019 || (IS_UDP(c->transport) && build_udp_headers(c) != 0)) { 3020 out_of_memory(c, "SERVER_ERROR out of memory writing get response"); 3021 } 3022 else { 3023 conn_set_state(c, conn_mwrite); 3024 c->msgcurr = 0; 3025 } 3026 } 3027 3028 static void process_update_command(conn *c, token_t *tokens, const size_t ntokens, int comm, bool handle_cas) { 3029 char *key; 3030 size_t nkey; 3031 unsigned int flags; 3032 int32_t exptime_int = 0; 3033 time_t exptime; 3034 int vlen; 3035 uint64_t req_cas_id=0; 3036 item *it; 3037 3038 assert(c != NULL); 3039 3040 set_noreply_maybe(c, tokens, ntokens); 3041 3042 if (tokens[KEY_TOKEN].length > KEY_MAX_LENGTH) { 3043 out_string(c, "CLIENT_ERROR bad command line format"); 3044 return; 3045 } 3046 3047 key = tokens[KEY_TOKEN].value; 3048 nkey = tokens[KEY_TOKEN].length; 3049 3050 if (! (safe_strtoul(tokens[2].value, (uint32_t *)&flags) 3051 && safe_strtol(tokens[3].value, &exptime_int) 3052 && safe_strtol(tokens[4].value, (int32_t *)&vlen))) { 3053 out_string(c, "CLIENT_ERROR bad command line format"); 3054 return; 3055 } 3056 3057 /* Ubuntu 8.04 breaks when I pass exptime to safe_strtol */ 3058 exptime = exptime_int; 3059 3060 /* Negative exptimes can underflow and end up immortal. realtime() will 3061 immediately expire values that are greater than REALTIME_MAXDELTA, but less 3062 than process_started, so lets aim for that. */ 3063 if (exptime < 0) 3064 exptime = REALTIME_MAXDELTA + 1; 3065 3066 // does cas value exist? 3067 if (handle_cas) { 3068 if (!safe_strtoull(tokens[5].value, &req_cas_id)) { 3069 out_string(c, "CLIENT_ERROR bad command line format"); 3070 return; 3071 } 3072 } 3073 3074 vlen += 2; 3075 if (vlen < 0 || vlen - 2 < 0) { 3076 out_string(c, "CLIENT_ERROR bad command line format"); 3077 return; 3078 } 3079 3080 if (settings.detail_enabled) { 3081 stats_prefix_record_set(key, nkey); 3082 } 3083 3084 it = item_alloc(key, nkey, flags, realtime(exptime), vlen); 3085 3086 if (it == 0) { 3087 if (! item_size_ok(nkey, flags, vlen)) 3088 out_string(c, "SERVER_ERROR object too large for cache"); 3089 else 3090 out_of_memory(c, "SERVER_ERROR out of memory storing object"); 3091 /* swallow the data line */ 3092 c->write_and_go = conn_swallow; 3093 c->sbytes = vlen; 3094 3095 /* Avoid stale data persisting in cache because we failed alloc. 3096 * Unacceptable for SET. Anywhere else too? */ 3097 if (comm == NREAD_SET) { 3098 it = item_get(key, nkey); 3099 if (it) { 3100 item_unlink(it); 3101 item_remove(it); 3102 } 3103 } 3104 3105 return; 3106 } 3107 ITEM_set_cas(it, req_cas_id); 3108 3109 c->item = it; 3110 c->ritem = ITEM_data(it); 3111 c->rlbytes = it->nbytes; 3112 c->cmd = comm; 3113 conn_set_state(c, conn_nread); 3114 } 3115 3116 static void process_touch_command(conn *c, token_t *tokens, const size_t ntokens) { 3117 char *key; 3118 size_t nkey; 3119 int32_t exptime_int = 0; 3120 item *it; 3121 3122 assert(c != NULL); 3123 3124 set_noreply_maybe(c, tokens, ntokens); 3125 3126 if (tokens[KEY_TOKEN].length > KEY_MAX_LENGTH) { 3127 out_string(c, "CLIENT_ERROR bad command line format"); 3128 return; 3129 } 3130 3131 key = tokens[KEY_TOKEN].value; 3132 nkey = tokens[KEY_TOKEN].length; 3133 3134 if (!safe_strtol(tokens[2].value, &exptime_int)) { 3135 out_string(c, "CLIENT_ERROR invalid exptime argument"); 3136 return; 3137 } 3138 3139 it = item_touch(key, nkey, realtime(exptime_int)); 3140 if (it) { 3141 item_update(it); 3142 pthread_mutex_lock(&c->thread->stats.mutex); 3143 c->thread->stats.touch_cmds++; 3144 c->thread->stats.slab_stats[it->slabs_clsid].touch_hits++; 3145 pthread_mutex_unlock(&c->thread->stats.mutex); 3146 3147 out_string(c, "TOUCHED"); 3148 item_remove(it); 3149 } else { 3150 pthread_mutex_lock(&c->thread->stats.mutex); 3151 c->thread->stats.touch_cmds++; 3152 c->thread->stats.touch_misses++; 3153 pthread_mutex_unlock(&c->thread->stats.mutex); 3154 3155 out_string(c, "NOT_FOUND"); 3156 } 3157 } 3158 3159 static void process_arithmetic_command(conn *c, token_t *tokens, const size_t ntokens, const bool incr) { 3160 char temp[INCR_MAX_STORAGE_LEN]; 3161 uint64_t delta; 3162 char *key; 3163 size_t nkey; 3164 3165 assert(c != NULL); 3166 3167 set_noreply_maybe(c, tokens, ntokens); 3168 3169 if (tokens[KEY_TOKEN].length > KEY_MAX_LENGTH) { 3170 out_string(c, "CLIENT_ERROR bad command line format"); 3171 return; 3172 } 3173 3174 key = tokens[KEY_TOKEN].value; 3175 nkey = tokens[KEY_TOKEN].length; 3176 3177 if (!safe_strtoull(tokens[2].value, &delta)) { 3178 out_string(c, "CLIENT_ERROR invalid numeric delta argument"); 3179 return; 3180 } 3181 3182 switch(add_delta(c, key, nkey, incr, delta, temp, NULL)) { 3183 case OK: 3184 out_string(c, temp); 3185 break; 3186 case NON_NUMERIC: 3187 out_string(c, "CLIENT_ERROR cannot increment or decrement non-numeric value"); 3188 break; 3189 case EOM: 3190 out_of_memory(c, "SERVER_ERROR out of memory"); 3191 break; 3192 case DELTA_ITEM_NOT_FOUND: 3193 pthread_mutex_lock(&c->thread->stats.mutex); 3194 if (incr) { 3195 c->thread->stats.incr_misses++; 3196 } else { 3197 c->thread->stats.decr_misses++; 3198 } 3199 pthread_mutex_unlock(&c->thread->stats.mutex); 3200 3201 out_string(c, "NOT_FOUND"); 3202 break; 3203 case DELTA_ITEM_CAS_MISMATCH: 3204 break; /* Should never get here */ 3205 } 3206 } 3207 3208 /* 3209 * adds a delta value to a numeric item. 3210 * 3211 * c connection requesting the operation 3212 * it item to adjust 3213 * incr true to increment value, false to decrement 3214 * delta amount to adjust value by 3215 * buf buffer for response string 3216 * 3217 * returns a response string to send back to the client. 3218 */ 3219 enum delta_result_type do_add_delta(conn *c, const char *key, const size_t nkey, 3220 const bool incr, const int64_t delta, 3221 char *buf, uint64_t *cas, 3222 const uint32_t hv) { 3223 char *ptr; 3224 uint64_t value; 3225 int res; 3226 item *it; 3227 3228 it = do_item_get(key, nkey, hv); 3229 if (!it) { 3230 return DELTA_ITEM_NOT_FOUND; 3231 } 3232 3233 if (cas != NULL && *cas != 0 && ITEM_get_cas(it) != *cas) { 3234 do_item_remove(it); 3235 return DELTA_ITEM_CAS_MISMATCH; 3236 } 3237 3238 ptr = ITEM_data(it); 3239 3240 if (!safe_strtoull(ptr, &value)) { 3241 do_item_remove(it); 3242 return NON_NUMERIC; 3243 } 3244 3245 if (incr) { 3246 value += delta; 3247 MEMCACHED_COMMAND_INCR(c->sfd, ITEM_key(it), it->nkey, value); 3248 } else { 3249 if(delta > value) { 3250 value = 0; 3251 } else { 3252 value -= delta; 3253 } 3254 MEMCACHED_COMMAND_DECR(c->sfd, ITEM_key(it), it->nkey, value); 3255 } 3256 3257 pthread_mutex_lock(&c->thread->stats.mutex); 3258 if (incr) { 3259 c->thread->stats.slab_stats[it->slabs_clsid].incr_hits++; 3260 } else { 3261 c->thread->stats.slab_stats[it->slabs_clsid].decr_hits++; 3262 } 3263 pthread_mutex_unlock(&c->thread->stats.mutex); 3264 3265 snprintf(buf, INCR_MAX_STORAGE_LEN, "%llu", (unsigned long long)value); 3266 res = strlen(buf); 3267 /* refcount == 2 means we are the only ones holding the item, and it is 3268 * linked. We hold the item's lock in this function, so refcount cannot 3269 * increase. */ 3270 if (res + 2 <= it->nbytes && it->refcount == 2) { /* replace in-place */ 3271 /* When changing the value without replacing the item, we 3272 need to update the CAS on the existing item. */ 3273 mutex_lock(&cache_lock); /* FIXME */ 3274 ITEM_set_cas(it, (settings.use_cas) ? get_cas_id() : 0); 3275 mutex_unlock(&cache_lock); 3276 3277 memcpy(ITEM_data(it), buf, res); 3278 memset(ITEM_data(it) + res, ' ', it->nbytes - res - 2); 3279 do_item_update(it); 3280 } else if (it->refcount > 1) { 3281 item *new_it; 3282 new_it = do_item_alloc(ITEM_key(it), it->nkey, atoi(ITEM_suffix(it) + 1), it->exptime, res + 2, hv); 3283 if (new_it == 0) { 3284 do_item_remove(it); 3285 return EOM; 3286 } 3287 memcpy(ITEM_data(new_it), buf, res); 3288 memcpy(ITEM_data(new_it) + res, "\r\n", 2); 3289 item_replace(it, new_it, hv); 3290 // Overwrite the older item's CAS with our new CAS since we're 3291 // returning the CAS of the old item below. 3292 ITEM_set_cas(it, (settings.use_cas) ? ITEM_get_cas(new_it) : 0); 3293 do_item_remove(new_it); /* release our reference */ 3294 } else { 3295 /* Should never get here. This means we somehow fetched an unlinked 3296 * item. TODO: Add a counter? */ 3297 if (settings.verbose) { 3298 fprintf(stderr, "Tried to do incr/decr on invalid item\n"); 3299 } 3300 if (it->refcount == 1) 3301 do_item_remove(it); 3302 return DELTA_ITEM_NOT_FOUND; 3303 } 3304 3305 if (cas) { 3306 *cas = ITEM_get_cas(it); /* swap the incoming CAS value */ 3307 } 3308 do_item_remove(it); /* release our reference */ 3309 return OK; 3310 } 3311 3312 static void process_delete_command(conn *c, token_t *tokens, const size_t ntokens) { 3313 char *key; 3314 size_t nkey; 3315 item *it; 3316 3317 assert(c != NULL); 3318 3319 if (ntokens > 3) { 3320 bool hold_is_zero = strcmp(tokens[KEY_TOKEN+1].value, "0") == 0; 3321 bool sets_noreply = set_noreply_maybe(c, tokens, ntokens); 3322 bool valid = (ntokens == 4 && (hold_is_zero || sets_noreply)) 3323 || (ntokens == 5 && hold_is_zero && sets_noreply); 3324 if (!valid) { 3325 out_string(c, "CLIENT_ERROR bad command line format. " 3326 "Usage: delete <key> [noreply]"); 3327 return; 3328 } 3329 } 3330 3331 3332 key = tokens[KEY_TOKEN].value; 3333 nkey = tokens[KEY_TOKEN].length; 3334 3335 if(nkey > KEY_MAX_LENGTH) { 3336 out_string(c, "CLIENT_ERROR bad command line format"); 3337 return; 3338 } 3339 3340 if (settings.detail_enabled) { 3341 stats_prefix_record_delete(key, nkey); 3342 } 3343 3344 it = item_get(key, nkey); 3345 if (it) { 3346 MEMCACHED_COMMAND_DELETE(c->sfd, ITEM_key(it), it->nkey); 3347 3348 pthread_mutex_lock(&c->thread->stats.mutex); 3349 c->thread->stats.slab_stats[it->slabs_clsid].delete_hits++; 3350 pthread_mutex_unlock(&c->thread->stats.mutex); 3351 3352 item_unlink(it); 3353 item_remove(it); /* release our reference */ 3354 out_string(c, "DELETED"); 3355 } else { 3356 pthread_mutex_lock(&c->thread->stats.mutex); 3357 c->thread->stats.delete_misses++; 3358 pthread_mutex_unlock(&c->thread->stats.mutex); 3359 3360 out_string(c, "NOT_FOUND"); 3361 } 3362 } 3363 3364 static void process_verbosity_command(conn *c, token_t *tokens, const size_t ntokens) { 3365 unsigned int level; 3366 3367 assert(c != NULL); 3368 3369 set_noreply_maybe(c, tokens, ntokens); 3370 3371 level = strtoul(tokens[1].value, NULL, 10); 3372 settings.verbose = level > MAX_VERBOSITY_LEVEL ? MAX_VERBOSITY_LEVEL : level; 3373 out_string(c, "OK"); 3374 return; 3375 } 3376 3377 static void process_slabs_automove_command(conn *c, token_t *tokens, const size_t ntokens) { 3378 unsigned int level; 3379 3380 assert(c != NULL); 3381 3382 set_noreply_maybe(c, tokens, ntokens); 3383 3384 level = strtoul(tokens[2].value, NULL, 10); 3385 if (level == 0) { 3386 settings.slab_automove = 0; 3387 } else if (level == 1 || level == 2) { 3388 settings.slab_automove = level; 3389 } else { 3390 out_string(c, "ERROR"); 3391 return; 3392 } 3393 out_string(c, "OK"); 3394 return; 3395 } 3396 3397 static void process_command(conn *c, char *command) { 3398 3399 token_t tokens[MAX_TOKENS]; 3400 size_t ntokens; 3401 int comm; 3402 3403 assert(c != NULL); 3404 3405 MEMCACHED_PROCESS_COMMAND_START(c->sfd, c->rcurr, c->rbytes); 3406 3407 if (settings.verbose > 1) 3408 fprintf(stderr, "<%d %s\n", c->sfd, command); 3409 3410 /* 3411 * for commands set/add/replace, we build an item and read the data 3412 * directly into it, then continue in nread_complete(). 3413 */ 3414 3415 c->msgcurr = 0; 3416 c->msgused = 0; 3417 c->iovused = 0; 3418 if (add_msghdr(c) != 0) { 3419 out_of_memory(c, "SERVER_ERROR out of memory preparing response"); 3420 return; 3421 } 3422 3423 ntokens = tokenize_command(command, tokens, MAX_TOKENS); 3424 if (ntokens >= 3 && 3425 ((strcmp(tokens[COMMAND_TOKEN].value, "get") == 0) || 3426 (strcmp(tokens[COMMAND_TOKEN].value, "bget") == 0))) { 3427 3428 process_get_command(c, tokens, ntokens, false); 3429 3430 } else if ((ntokens == 6 || ntokens == 7) && 3431 ((strcmp(tokens[COMMAND_TOKEN].value, "add") == 0 && (comm = NREAD_ADD)) || 3432 (strcmp(tokens[COMMAND_TOKEN].value, "set") == 0 && (comm = NREAD_SET)) || 3433 (strcmp(tokens[COMMAND_TOKEN].value, "replace") == 0 && (comm = NREAD_REPLACE)) || 3434 (strcmp(tokens[COMMAND_TOKEN].value, "prepend") == 0 && (comm = NREAD_PREPEND)) || 3435 (strcmp(tokens[COMMAND_TOKEN].value, "append") == 0 && (comm = NREAD_APPEND)) )) { 3436 3437 process_update_command(c, tokens, ntokens, comm, false); 3438 3439 } else if ((ntokens == 7 || ntokens == 8) && (strcmp(tokens[COMMAND_TOKEN].value, "cas") == 0 && (comm = NREAD_CAS))) { 3440 3441 process_update_command(c, tokens, ntokens, comm, true); 3442 3443 } else if ((ntokens == 4 || ntokens == 5) && (strcmp(tokens[COMMAND_TOKEN].value, "incr") == 0)) { 3444 3445 process_arithmetic_command(c, tokens, ntokens, 1); 3446 3447 } else if (ntokens >= 3 && (strcmp(tokens[COMMAND_TOKEN].value, "gets") == 0)) { 3448 3449 process_get_command(c, tokens, ntokens, true); 3450 3451 } else if ((ntokens == 4 || ntokens == 5) && (strcmp(tokens[COMMAND_TOKEN].value, "decr") == 0)) { 3452 3453 process_arithmetic_command(c, tokens, ntokens, 0); 3454 3455 } else if (ntokens >= 3 && ntokens <= 5 && (strcmp(tokens[COMMAND_TOKEN].value, "delete") == 0)) { 3456 3457 process_delete_command(c, tokens, ntokens); 3458 3459 } else if ((ntokens == 4 || ntokens == 5) && (strcmp(tokens[COMMAND_TOKEN].value, "touch") == 0)) { 3460 3461 process_touch_command(c, tokens, ntokens); 3462 3463 } else if (ntokens >= 2 && (strcmp(tokens[COMMAND_TOKEN].value, "stats") == 0)) { 3464 3465 process_stat(c, tokens, ntokens); 3466 3467 } else if (ntokens >= 2 && ntokens <= 4 && (strcmp(tokens[COMMAND_TOKEN].value, "flush_all") == 0)) { 3468 time_t exptime = 0; 3469 3470 set_noreply_maybe(c, tokens, ntokens); 3471 3472 pthread_mutex_lock(&c->thread->stats.mutex); 3473 c->thread->stats.flush_cmds++; 3474 pthread_mutex_unlock(&c->thread->stats.mutex); 3475 3476 if (!settings.flush_enabled) { 3477 // flush_all is not allowed but we log it on stats 3478 out_string(c, "CLIENT_ERROR flush_all not allowed"); 3479 return; 3480 } 3481 3482 if(ntokens == (c->noreply ? 3 : 2)) { 3483 settings.oldest_live = current_time - 1; 3484 item_flush_expired(); 3485 out_string(c, "OK"); 3486 return; 3487 } 3488 3489 exptime = strtol(tokens[1].value, NULL, 10); 3490 if(errno == ERANGE) { 3491 out_string(c, "CLIENT_ERROR bad command line format"); 3492 return; 3493 } 3494 3495 /* 3496 If exptime is zero realtime() would return zero too, and 3497 realtime(exptime) - 1 would overflow to the max unsigned 3498 value. So we process exptime == 0 the same way we do when 3499 no delay is given at all. 3500 */ 3501 if (exptime > 0) 3502 settings.oldest_live = realtime(exptime) - 1; 3503 else /* exptime == 0 */ 3504 settings.oldest_live = current_time - 1; 3505 item_flush_expired(); 3506 out_string(c, "OK"); 3507 return; 3508 3509 } else if (ntokens == 2 && (strcmp(tokens[COMMAND_TOKEN].value, "version") == 0)) { 3510 3511 out_string(c, "VERSION " VERSION); 3512 3513 } else if (ntokens == 2 && (strcmp(tokens[COMMAND_TOKEN].value, "quit") == 0)) { 3514 3515 conn_set_state(c, conn_closing); 3516 3517 } else if (ntokens == 2 && (strcmp(tokens[COMMAND_TOKEN].value, "shutdown") == 0)) { 3518 3519 if (settings.shutdown_command) { 3520 conn_set_state(c, conn_closing); 3521 raise(SIGINT); 3522 } else { 3523 out_string(c, "ERROR: shutdown not enabled"); 3524 } 3525 3526 } else if (ntokens > 1 && strcmp(tokens[COMMAND_TOKEN].value, "slabs") == 0) { 3527 if (ntokens == 5 && strcmp(tokens[COMMAND_TOKEN + 1].value, "reassign") == 0) { 3528 int src, dst, rv; 3529 3530 if (settings.slab_reassign == false) { 3531 out_string(c, "CLIENT_ERROR slab reassignment disabled"); 3532 return; 3533 } 3534 3535 src = strtol(tokens[2].value, NULL, 10); 3536 dst = strtol(tokens[3].value, NULL, 10); 3537 3538 if (errno == ERANGE) { 3539 out_string(c, "CLIENT_ERROR bad command line format"); 3540 return; 3541 } 3542 3543 rv = slabs_reassign(src, dst); 3544 switch (rv) { 3545 case REASSIGN_OK: 3546 out_string(c, "OK"); 3547 break; 3548 case REASSIGN_RUNNING: 3549 out_string(c, "BUSY currently processing reassign request"); 3550 break; 3551 case REASSIGN_BADCLASS: 3552 out_string(c, "BADCLASS invalid src or dst class id"); 3553 break; 3554 case REASSIGN_NOSPARE: 3555 out_string(c, "NOSPARE source class has no spare pages"); 3556 break; 3557 case REASSIGN_SRC_DST_SAME: 3558 out_string(c, "SAME src and dst class are identical"); 3559 break; 3560 } 3561 return; 3562 } else if (ntokens == 4 && 3563 (strcmp(tokens[COMMAND_TOKEN + 1].value, "automove") == 0)) { 3564 process_slabs_automove_command(c, tokens, ntokens); 3565 } else { 3566 out_string(c, "ERROR"); 3567 } 3568 } else if (ntokens > 1 && strcmp(tokens[COMMAND_TOKEN].value, "lru_crawler") == 0) { 3569 if (ntokens == 4 && strcmp(tokens[COMMAND_TOKEN + 1].value, "crawl") == 0) { 3570 int rv; 3571 if (settings.lru_crawler == false) { 3572 out_string(c, "CLIENT_ERROR lru crawler disabled"); 3573 return; 3574 } 3575 3576 rv = lru_crawler_crawl(tokens[2].value); 3577 switch(rv) { 3578 case CRAWLER_OK: 3579 out_string(c, "OK"); 3580 break; 3581 case CRAWLER_RUNNING: 3582 out_string(c, "BUSY currently processing crawler request"); 3583 break; 3584 case CRAWLER_BADCLASS: 3585 out_string(c, "BADCLASS invalid class id"); 3586 break; 3587 } 3588 return; 3589 } else if (ntokens == 4 && strcmp(tokens[COMMAND_TOKEN + 1].value, "tocrawl") == 0) { 3590 uint32_t tocrawl; 3591 if (!safe_strtoul(tokens[2].value, &tocrawl)) { 3592 out_string(c, "CLIENT_ERROR bad command line format"); 3593 return; 3594 } 3595 settings.lru_crawler_tocrawl = tocrawl; 3596 out_string(c, "OK"); 3597 return; 3598 } else if (ntokens == 4 && strcmp(tokens[COMMAND_TOKEN + 1].value, "sleep") == 0) { 3599 uint32_t tosleep; 3600 if (!safe_strtoul(tokens[2].value, &tosleep)) { 3601 out_string(c, "CLIENT_ERROR bad command line format"); 3602 return; 3603 } 3604 if (tosleep > 1000000) { 3605 out_string(c, "CLIENT_ERROR sleep must be one second or less"); 3606 return; 3607 } 3608 settings.lru_crawler_sleep = tosleep; 3609 out_string(c, "OK"); 3610 return; 3611 } else if (ntokens == 3) { 3612 if ((strcmp(tokens[COMMAND_TOKEN + 1].value, "enable") == 0)) { 3613 if (start_item_crawler_thread() == 0) { 3614 out_string(c, "OK"); 3615 } else { 3616 out_string(c, "ERROR failed to start lru crawler thread"); 3617 } 3618 } else if ((strcmp(tokens[COMMAND_TOKEN + 1].value, "disable") == 0)) { 3619 if (stop_item_crawler_thread() == 0) { 3620 out_string(c, "OK"); 3621 } else { 3622 out_string(c, "ERROR failed to stop lru crawler thread"); 3623 } 3624 } else { 3625 out_string(c, "ERROR"); 3626 } 3627 return; 3628 } else { 3629 out_string(c, "ERROR"); 3630 } 3631 } else if ((ntokens == 3 || ntokens == 4) && (strcmp(tokens[COMMAND_TOKEN].value, "verbosity") == 0)) { 3632 process_verbosity_command(c, tokens, ntokens); 3633 } else { 3634 out_string(c, "ERROR"); 3635 } 3636 return; 3637 } 3638 3639 /* 3640 * if we have a complete line in the buffer, process it. 3641 */ 3642 static int try_read_command(conn *c) { 3643 assert(c != NULL); 3644 assert(c->rcurr <= (c->rbuf + c->rsize)); 3645 assert(c->rbytes > 0); 3646 3647 if (c->protocol == negotiating_prot || c->transport == udp_transport) { 3648 if ((unsigned char)c->rbuf[0] == (unsigned char)PROTOCOL_BINARY_REQ) { 3649 c->protocol = binary_prot; 3650 } else { 3651 c->protocol = ascii_prot; 3652 } 3653 3654 if (settings.verbose > 1) { 3655 fprintf(stderr, "%d: Client using the %s protocol\n", c->sfd, 3656 prot_text(c->protocol)); 3657 } 3658 } 3659 3660 if (c->protocol == binary_prot) { 3661 /* Do we have the complete packet header? */ 3662 if (c->rbytes < sizeof(c->binary_header)) { 3663 /* need more data! */ 3664 return 0; 3665 } else { 3666 #ifdef NEED_ALIGN 3667 if (((long)(c->rcurr)) % 8 != 0) { 3668 /* must realign input buffer */ 3669 memmove(c->rbuf, c->rcurr, c->rbytes); 3670 c->rcurr = c->rbuf; 3671 if (settings.verbose > 1) { 3672 fprintf(stderr, "%d: Realign input buffer\n", c->sfd); 3673 } 3674 } 3675 #endif 3676 protocol_binary_request_header* req; 3677 req = (protocol_binary_request_header*)c->rcurr; 3678 3679 if (settings.verbose > 1) { 3680 /* Dump the packet before we convert it to host order */ 3681 int ii; 3682 fprintf(stderr, "<%d Read binary protocol data:", c->sfd); 3683 for (ii = 0; ii < sizeof(req->bytes); ++ii) { 3684 if (ii % 4 == 0) { 3685 fprintf(stderr, "\n<%d ", c->sfd); 3686 } 3687 fprintf(stderr, " 0x%02x", req->bytes[ii]); 3688 } 3689 fprintf(stderr, "\n"); 3690 } 3691 3692 c->binary_header = *req; 3693 c->binary_header.request.keylen = ntohs(req->request.keylen); 3694 c->binary_header.request.bodylen = ntohl(req->request.bodylen); 3695 c->binary_header.request.cas = ntohll(req->request.cas); 3696 3697 if (c->binary_header.request.magic != PROTOCOL_BINARY_REQ) { 3698 if (settings.verbose) { 3699 fprintf(stderr, "Invalid magic: %x\n", 3700 c->binary_header.request.magic); 3701 } 3702 conn_set_state(c, conn_closing); 3703 return -1; 3704 } 3705 3706 c->msgcurr = 0; 3707 c->msgused = 0; 3708 c->iovused = 0; 3709 if (add_msghdr(c) != 0) { 3710 out_of_memory(c, 3711 "SERVER_ERROR Out of memory allocating headers"); 3712 return 0; 3713 } 3714 3715 c->cmd = c->binary_header.request.opcode; 3716 c->keylen = c->binary_header.request.keylen; 3717 c->opaque = c->binary_header.request.opaque; 3718 /* clear the returned cas value */ 3719 c->cas = 0; 3720 3721 dispatch_bin_command(c); 3722 3723 c->rbytes -= sizeof(c->binary_header); 3724 c->rcurr += sizeof(c->binary_header); 3725 } 3726 } else { 3727 char *el, *cont; 3728 3729 if (c->rbytes == 0) 3730 return 0; 3731 3732 el = memchr(c->rcurr, '\n', c->rbytes); 3733 if (!el) { 3734 if (c->rbytes > 1024) { 3735 /* 3736 * We didn't have a '\n' in the first k. This _has_ to be a 3737 * large multiget, if not we should just nuke the connection. 3738 */ 3739 char *ptr = c->rcurr; 3740 while (*ptr == ' ') { /* ignore leading whitespaces */ 3741 ++ptr; 3742 } 3743 3744 if (ptr - c->rcurr > 100 || 3745 (strncmp(ptr, "get ", 4) && strncmp(ptr, "gets ", 5))) { 3746 3747 conn_set_state(c, conn_closing); 3748 return 1; 3749 } 3750 } 3751 3752 return 0; 3753 } 3754 cont = el + 1; 3755 if ((el - c->rcurr) > 1 && *(el - 1) == '\r') { 3756 el--; 3757 } 3758 *el = '\0'; 3759 3760 assert(cont <= (c->rcurr + c->rbytes)); 3761 3762 c->last_cmd_time = current_time; 3763 process_command(c, c->rcurr); 3764 3765 c->rbytes -= (cont - c->rcurr); 3766 c->rcurr = cont; 3767 3768 assert(c->rcurr <= (c->rbuf + c->rsize)); 3769 } 3770 3771 return 1; 3772 } 3773 3774 /* 3775 * read a UDP request. 3776 */ 3777 static enum try_read_result try_read_udp(conn *c) { 3778 int res; 3779 3780 assert(c != NULL); 3781 3782 c->request_addr_size = sizeof(c->request_addr); 3783 res = recvfrom(c->sfd, c->rbuf, c->rsize, 3784 0, (struct sockaddr *)&c->request_addr, 3785 &c->request_addr_size); 3786 if (res > 8) { 3787 unsigned char *buf = (unsigned char *)c->rbuf; 3788 pthread_mutex_lock(&c->thread->stats.mutex); 3789 c->thread->stats.bytes_read += res; 3790 pthread_mutex_unlock(&c->thread->stats.mutex); 3791 3792 /* Beginning of UDP packet is the request ID; save it. */ 3793 c->request_id = buf[0] * 256 + buf[1]; 3794 3795 /* If this is a multi-packet request, drop it. */ 3796 if (buf[4] != 0 || buf[5] != 1) { 3797 out_string(c, "SERVER_ERROR multi-packet request not supported"); 3798 return READ_NO_DATA_RECEIVED; 3799 } 3800 3801 /* Don't care about any of the rest of the header. */ 3802 res -= 8; 3803 memmove(c->rbuf, c->rbuf + 8, res); 3804 3805 c->rbytes = res; 3806 c->rcurr = c->rbuf; 3807 return READ_DATA_RECEIVED; 3808 } 3809 return READ_NO_DATA_RECEIVED; 3810 } 3811 3812 /* 3813 * read from network as much as we can, handle buffer overflow and connection 3814 * close. 3815 * before reading, move the remaining incomplete fragment of a command 3816 * (if any) to the beginning of the buffer. 3817 * 3818 * To protect us from someone flooding a connection with bogus data causing 3819 * the connection to eat up all available memory, break out and start looking 3820 * at the data I've got after a number of reallocs... 3821 * 3822 * @return enum try_read_result 3823 */ 3824 static enum try_read_result try_read_network(conn *c) { 3825 enum try_read_result gotdata = READ_NO_DATA_RECEIVED; 3826 int res; 3827 int num_allocs = 0; 3828 assert(c != NULL); 3829 3830 if (c->rcurr != c->rbuf) { 3831 if (c->rbytes != 0) /* otherwise there's nothing to copy */ 3832 memmove(c->rbuf, c->rcurr, c->rbytes); 3833 c->rcurr = c->rbuf; 3834 } 3835 3836 while (1) { 3837 if (c->rbytes >= c->rsize) { 3838 if (num_allocs == 4) { 3839 return gotdata; 3840 } 3841 ++num_allocs; 3842 char *new_rbuf = realloc(c->rbuf, c->rsize * 2); 3843 if (!new_rbuf) { 3844 STATS_LOCK(); 3845 stats.malloc_fails++; 3846 STATS_UNLOCK(); 3847 if (settings.verbose > 0) { 3848 fprintf(stderr, "Couldn't realloc input buffer\n"); 3849 } 3850 c->rbytes = 0; /* ignore what we read */ 3851 out_of_memory(c, "SERVER_ERROR out of memory reading request"); 3852 c->write_and_go = conn_closing; 3853 return READ_MEMORY_ERROR; 3854 } 3855 c->rcurr = c->rbuf = new_rbuf; 3856 c->rsize *= 2; 3857 } 3858 3859 int avail = c->rsize - c->rbytes; 3860 res = read(c->sfd, c->rbuf + c->rbytes, avail); 3861 if (res > 0) { 3862 pthread_mutex_lock(&c->thread->stats.mutex); 3863 c->thread->stats.bytes_read += res; 3864 pthread_mutex_unlock(&c->thread->stats.mutex); 3865 gotdata = READ_DATA_RECEIVED; 3866 c->rbytes += res; 3867 if (res == avail) { 3868 continue; 3869 } else { 3870 break; 3871 } 3872 } 3873 if (res == 0) { 3874 return READ_ERROR; 3875 } 3876 if (res == -1) { 3877 if (errno == EAGAIN || errno == EWOULDBLOCK) { 3878 break; 3879 } 3880 return READ_ERROR; 3881 } 3882 } 3883 return gotdata; 3884 } 3885 3886 static bool update_event(conn *c, const int new_flags) { 3887 assert(c != NULL); 3888 3889 struct event_base *base = c->event.ev_base; 3890 if (c->ev_flags == new_flags) 3891 return true; 3892 if (event_del(&c->event) == -1) return false; 3893 event_set(&c->event, c->sfd, new_flags, event_handler, (void *)c); 3894 event_base_set(base, &c->event); 3895 c->ev_flags = new_flags; 3896 if (event_add(&c->event, 0) == -1) return false; 3897 return true; 3898 } 3899 3900 /* 3901 * Sets whether we are listening for new connections or not. 3902 */ 3903 void do_accept_new_conns(const bool do_accept) { 3904 conn *next; 3905 3906 for (next = listen_conn; next; next = next->next) { 3907 if (do_accept) { 3908 update_event(next, EV_READ | EV_PERSIST); 3909 if (listen(next->sfd, settings.backlog) != 0) { 3910 perror("listen"); 3911 } 3912 } 3913 else { 3914 update_event(next, 0); 3915 if (listen(next->sfd, 0) != 0) { 3916 perror("listen"); 3917 } 3918 } 3919 } 3920 3921 if (do_accept) { 3922 STATS_LOCK(); 3923 stats.accepting_conns = true; 3924 STATS_UNLOCK(); 3925 } else { 3926 STATS_LOCK(); 3927 stats.accepting_conns = false; 3928 stats.listen_disabled_num++; 3929 STATS_UNLOCK(); 3930 allow_new_conns = false; 3931 maxconns_handler(-42, 0, 0); 3932 } 3933 } 3934 3935 /* 3936 * Transmit the next chunk of data from our list of msgbuf structures. 3937 * 3938 * Returns: 3939 * TRANSMIT_COMPLETE All done writing. 3940 * TRANSMIT_INCOMPLETE More data remaining to write. 3941 * TRANSMIT_SOFT_ERROR Can't write any more right now. 3942 * TRANSMIT_HARD_ERROR Can't write (c->state is set to conn_closing) 3943 */ 3944 static enum transmit_result transmit(conn *c) { 3945 assert(c != NULL); 3946 3947 if (c->msgcurr < c->msgused && 3948 c->msglist[c->msgcurr].msg_iovlen == 0) { 3949 /* Finished writing the current msg; advance to the next. */ 3950 c->msgcurr++; 3951 } 3952 if (c->msgcurr < c->msgused) { 3953 ssize_t res; 3954 struct msghdr *m = &c->msglist[c->msgcurr]; 3955 3956 res = sendmsg(c->sfd, m, 0); 3957 if (res > 0) { 3958 pthread_mutex_lock(&c->thread->stats.mutex); 3959 c->thread->stats.bytes_written += res; 3960 pthread_mutex_unlock(&c->thread->stats.mutex); 3961 3962 /* We've written some of the data. Remove the completed 3963 iovec entries from the list of pending writes. */ 3964 while (m->msg_iovlen > 0 && res >= m->msg_iov->iov_len) { 3965 res -= m->msg_iov->iov_len; 3966 m->msg_iovlen--; 3967 m->msg_iov++; 3968 } 3969 3970 /* Might have written just part of the last iovec entry; 3971 adjust it so the next write will do the rest. */ 3972 if (res > 0) { 3973 m->msg_iov->iov_base = (caddr_t)m->msg_iov->iov_base + res; 3974 m->msg_iov->iov_len -= res; 3975 } 3976 return TRANSMIT_INCOMPLETE; 3977 } 3978 if (res == -1 && (errno == EAGAIN || errno == EWOULDBLOCK)) { 3979 if (!update_event(c, EV_WRITE | EV_PERSIST)) { 3980 if (settings.verbose > 0) 3981 fprintf(stderr, "Couldn't update event\n"); 3982 conn_set_state(c, conn_closing); 3983 return TRANSMIT_HARD_ERROR; 3984 } 3985 return TRANSMIT_SOFT_ERROR; 3986 } 3987 /* if res == 0 or res == -1 and error is not EAGAIN or EWOULDBLOCK, 3988 we have a real error, on which we close the connection */ 3989 if (settings.verbose > 0) 3990 perror("Failed to write, and not due to blocking"); 3991 3992 if (IS_UDP(c->transport)) 3993 conn_set_state(c, conn_read); 3994 else 3995 conn_set_state(c, conn_closing); 3996 return TRANSMIT_HARD_ERROR; 3997 } else { 3998 return TRANSMIT_COMPLETE; 3999 } 4000 } 4001 4002 static void drive_machine(conn *c) { 4003 bool stop = false; 4004 int sfd; 4005 socklen_t addrlen; 4006 struct sockaddr_storage addr; 4007 int nreqs = settings.reqs_per_event; 4008 int res; 4009 const char *str; 4010 #ifdef HAVE_ACCEPT4 4011 static int use_accept4 = 1; 4012 #else 4013 static int use_accept4 = 0; 4014 #endif 4015 4016 assert(c != NULL); 4017 4018 while (!stop) { 4019 4020 switch(c->state) { 4021 case conn_listening: 4022 addrlen = sizeof(addr); 4023 #ifdef HAVE_ACCEPT4 4024 if (use_accept4) { 4025 sfd = accept4(c->sfd, (struct sockaddr *)&addr, &addrlen, SOCK_NONBLOCK); 4026 } else { 4027 sfd = accept(c->sfd, (struct sockaddr *)&addr, &addrlen); 4028 } 4029 #else 4030 sfd = accept(c->sfd, (struct sockaddr *)&addr, &addrlen); 4031 #endif 4032 if (sfd == -1) { 4033 if (use_accept4 && errno == ENOSYS) { 4034 use_accept4 = 0; 4035 continue; 4036 } 4037 perror(use_accept4 ? "accept4()" : "accept()"); 4038 if (errno == EAGAIN || errno == EWOULDBLOCK) { 4039 /* these are transient, so don't log anything */ 4040 stop = true; 4041 } else if (errno == EMFILE) { 4042 if (settings.verbose > 0) 4043 fprintf(stderr, "Too many open connections\n"); 4044 accept_new_conns(false); 4045 stop = true; 4046 } else { 4047 perror("accept()"); 4048 stop = true; 4049 } 4050 break; 4051 } 4052 if (!use_accept4) { 4053 if (fcntl(sfd, F_SETFL, fcntl(sfd, F_GETFL) | O_NONBLOCK) < 0) { 4054 perror("setting O_NONBLOCK"); 4055 close(sfd); 4056 break; 4057 } 4058 } 4059 4060 if (settings.maxconns_fast && 4061 stats.curr_conns + stats.reserved_fds >= settings.maxconns - 1) { 4062 str = "ERROR Too many open connections\r\n"; 4063 res = write(sfd, str, strlen(str)); 4064 close(sfd); 4065 STATS_LOCK(); 4066 stats.rejected_conns++; 4067 STATS_UNLOCK(); 4068 } else { 4069 dispatch_conn_new(sfd, conn_new_cmd, EV_READ | EV_PERSIST, 4070 DATA_BUFFER_SIZE, tcp_transport); 4071 } 4072 4073 stop = true; 4074 break; 4075 4076 case conn_waiting: 4077 if (!update_event(c, EV_READ | EV_PERSIST)) { 4078 if (settings.verbose > 0) 4079 fprintf(stderr, "Couldn't update event\n"); 4080 conn_set_state(c, conn_closing); 4081 break; 4082 } 4083 4084 conn_set_state(c, conn_read); 4085 stop = true; 4086 break; 4087 4088 case conn_read: 4089 res = IS_UDP(c->transport) ? try_read_udp(c) : try_read_network(c); 4090 4091 switch (res) { 4092 case READ_NO_DATA_RECEIVED: 4093 conn_set_state(c, conn_waiting); 4094 break; 4095 case READ_DATA_RECEIVED: 4096 conn_set_state(c, conn_parse_cmd); 4097 break; 4098 case READ_ERROR: 4099 conn_set_state(c, conn_closing); 4100 break; 4101 case READ_MEMORY_ERROR: /* Failed to allocate more memory */ 4102 /* State already set by try_read_network */ 4103 break; 4104 } 4105 break; 4106 4107 case conn_parse_cmd : 4108 if (try_read_command(c) == 0) { 4109 /* wee need more data! */ 4110 conn_set_state(c, conn_waiting); 4111 } 4112 4113 break; 4114 4115 case conn_new_cmd: 4116 /* Only process nreqs at a time to avoid starving other 4117 connections */ 4118 4119 --nreqs; 4120 if (nreqs >= 0) { 4121 reset_cmd_handler(c); 4122 } else { 4123 pthread_mutex_lock(&c->thread->stats.mutex); 4124 c->thread->stats.conn_yields++; 4125 pthread_mutex_unlock(&c->thread->stats.mutex); 4126 if (c->rbytes > 0) { 4127 /* We have already read in data into the input buffer, 4128 so libevent will most likely not signal read events 4129 on the socket (unless more data is available. As a 4130 hack we should just put in a request to write data, 4131 because that should be possible ;-) 4132 */ 4133 if (!update_event(c, EV_WRITE | EV_PERSIST)) { 4134 if (settings.verbose > 0) 4135 fprintf(stderr, "Couldn't update event\n"); 4136 conn_set_state(c, conn_closing); 4137 break; 4138 } 4139 } 4140 stop = true; 4141 } 4142 break; 4143 4144 case conn_nread: 4145 if (c->rlbytes == 0) { 4146 complete_nread(c); 4147 break; 4148 } 4149 4150 /* Check if rbytes < 0, to prevent crash */ 4151 if (c->rlbytes < 0) { 4152 if (settings.verbose) { 4153 fprintf(stderr, "Invalid rlbytes to read: len %d\n", c->rlbytes); 4154 } 4155 conn_set_state(c, conn_closing); 4156 break; 4157 } 4158 4159 /* first check if we have leftovers in the conn_read buffer */ 4160 if (c->rbytes > 0) { 4161 int tocopy = c->rbytes > c->rlbytes ? c->rlbytes : c->rbytes; 4162 if (c->ritem != c->rcurr) { 4163 memmove(c->ritem, c->rcurr, tocopy); 4164 } 4165 c->ritem += tocopy; 4166 c->rlbytes -= tocopy; 4167 c->rcurr += tocopy; 4168 c->rbytes -= tocopy; 4169 if (c->rlbytes == 0) { 4170 break; 4171 } 4172 } 4173 4174 /* now try reading from the socket */ 4175 res = read(c->sfd, c->ritem, c->rlbytes); 4176 if (res > 0) { 4177 pthread_mutex_lock(&c->thread->stats.mutex); 4178 c->thread->stats.bytes_read += res; 4179 pthread_mutex_unlock(&c->thread->stats.mutex); 4180 if (c->rcurr == c->ritem) { 4181 c->rcurr += res; 4182 } 4183 c->ritem += res; 4184 c->rlbytes -= res; 4185 break; 4186 } 4187 if (res == 0) { /* end of stream */ 4188 conn_set_state(c, conn_closing); 4189 break; 4190 } 4191 if (res == -1 && (errno == EAGAIN || errno == EWOULDBLOCK)) { 4192 if (!update_event(c, EV_READ | EV_PERSIST)) { 4193 if (settings.verbose > 0) 4194 fprintf(stderr, "Couldn't update event\n"); 4195 conn_set_state(c, conn_closing); 4196 break; 4197 } 4198 stop = true; 4199 break; 4200 } 4201 /* otherwise we have a real error, on which we close the connection */ 4202 if (settings.verbose > 0) { 4203 fprintf(stderr, "Failed to read, and not due to blocking:\n" 4204 "errno: %d %s \n" 4205 "rcurr=%lx ritem=%lx rbuf=%lx rlbytes=%d rsize=%d\n", 4206 errno, strerror(errno), 4207 (long)c->rcurr, (long)c->ritem, (long)c->rbuf, 4208 (int)c->rlbytes, (int)c->rsize); 4209 } 4210 conn_set_state(c, conn_closing); 4211 break; 4212 4213 case conn_swallow: 4214 /* we are reading sbytes and throwing them away */ 4215 if (c->sbytes == 0) { 4216 conn_set_state(c, conn_new_cmd); 4217 break; 4218 } 4219 4220 /* first check if we have leftovers in the conn_read buffer */ 4221 if (c->rbytes > 0) { 4222 int tocopy = c->rbytes > c->sbytes ? c->sbytes : c->rbytes; 4223 c->sbytes -= tocopy; 4224 c->rcurr += tocopy; 4225 c->rbytes -= tocopy; 4226 break; 4227 } 4228 4229 /* now try reading from the socket */ 4230 res = read(c->sfd, c->rbuf, c->rsize > c->sbytes ? c->sbytes : c->rsize); 4231 if (res > 0) { 4232 pthread_mutex_lock(&c->thread->stats.mutex); 4233 c->thread->stats.bytes_read += res; 4234 pthread_mutex_unlock(&c->thread->stats.mutex); 4235 c->sbytes -= res; 4236 break; 4237 } 4238 if (res == 0) { /* end of stream */ 4239 conn_set_state(c, conn_closing); 4240 break; 4241 } 4242 if (res == -1 && (errno == EAGAIN || errno == EWOULDBLOCK)) { 4243 if (!update_event(c, EV_READ | EV_PERSIST)) { 4244 if (settings.verbose > 0) 4245 fprintf(stderr, "Couldn't update event\n"); 4246 conn_set_state(c, conn_closing); 4247 break; 4248 } 4249 stop = true; 4250 break; 4251 } 4252 /* otherwise we have a real error, on which we close the connection */ 4253 if (settings.verbose > 0) 4254 fprintf(stderr, "Failed to read, and not due to blocking\n"); 4255 conn_set_state(c, conn_closing); 4256 break; 4257 4258 case conn_write: 4259 /* 4260 * We want to write out a simple response. If we haven't already, 4261 * assemble it into a msgbuf list (this will be a single-entry 4262 * list for TCP or a two-entry list for UDP). 4263 */ 4264 if (c->iovused == 0 || (IS_UDP(c->transport) && c->iovused == 1)) { 4265 if (add_iov(c, c->wcurr, c->wbytes) != 0) { 4266 if (settings.verbose > 0) 4267 fprintf(stderr, "Couldn't build response\n"); 4268 conn_set_state(c, conn_closing); 4269 break; 4270 } 4271 } 4272 4273 /* fall through... */ 4274 4275 case conn_mwrite: 4276 if (IS_UDP(c->transport) && c->msgcurr == 0 && build_udp_headers(c) != 0) { 4277 if (settings.verbose > 0) 4278 fprintf(stderr, "Failed to build UDP headers\n"); 4279 conn_set_state(c, conn_closing); 4280 break; 4281 } 4282 switch (transmit(c)) { 4283 case TRANSMIT_COMPLETE: 4284 if (c->state == conn_mwrite) { 4285 conn_release_items(c); 4286 /* XXX: I don't know why this wasn't the general case */ 4287 if(c->protocol == binary_prot) { 4288 conn_set_state(c, c->write_and_go); 4289 } else { 4290 conn_set_state(c, conn_new_cmd); 4291 } 4292 } else if (c->state == conn_write) { 4293 if (c->write_and_free) { 4294 free(c->write_and_free); 4295 c->write_and_free = 0; 4296 } 4297 conn_set_state(c, c->write_and_go); 4298 } else { 4299 if (settings.verbose > 0) 4300 fprintf(stderr, "Unexpected state %d\n", c->state); 4301 conn_set_state(c, conn_closing); 4302 } 4303 break; 4304 4305 case TRANSMIT_INCOMPLETE: 4306 case TRANSMIT_HARD_ERROR: 4307 break; /* Continue in state machine. */ 4308 4309 case TRANSMIT_SOFT_ERROR: 4310 stop = true; 4311 break; 4312 } 4313 break; 4314 4315 case conn_closing: 4316 if (IS_UDP(c->transport)) 4317 conn_cleanup(c); 4318 else 4319 conn_close(c); 4320 stop = true; 4321 break; 4322 4323 case conn_closed: 4324 /* This only happens if dormando is an idiot. */ 4325 abort(); 4326 break; 4327 4328 case conn_max_state: 4329 assert(false); 4330 break; 4331 } 4332 } 4333 4334 return; 4335 } 4336 4337 void event_handler(const int fd, const short which, void *arg) { 4338 conn *c; 4339 4340 c = (conn *)arg; 4341 assert(c != NULL); 4342 4343 c->which = which; 4344 4345 /* sanity */ 4346 if (fd != c->sfd) { 4347 if (settings.verbose > 0) 4348 fprintf(stderr, "Catastrophic: event fd doesn't match conn fd!\n"); 4349 conn_close(c); 4350 return; 4351 } 4352 4353 drive_machine(c); 4354 4355 /* wait for next event */ 4356 return; 4357 } 4358 4359 static int new_socket(struct addrinfo *ai) { 4360 int sfd; 4361 int flags; 4362 4363 if ((sfd = socket(ai->ai_family, ai->ai_socktype, ai->ai_protocol)) == -1) { 4364 return -1; 4365 } 4366 4367 if ((flags = fcntl(sfd, F_GETFL, 0)) < 0 || 4368 fcntl(sfd, F_SETFL, flags | O_NONBLOCK) < 0) { 4369 perror("setting O_NONBLOCK"); 4370 close(sfd); 4371 return -1; 4372 } 4373 return sfd; 4374 } 4375 4376 4377 /* 4378 * Sets a socket's send buffer size to the maximum allowed by the system. 4379 */ 4380 static void maximize_sndbuf(const int sfd) { 4381 socklen_t intsize = sizeof(int); 4382 int last_good = 0; 4383 int min, max, avg; 4384 int old_size; 4385 4386 /* Start with the default size. */ 4387 if (getsockopt(sfd, SOL_SOCKET, SO_SNDBUF, &old_size, &intsize) != 0) { 4388 if (settings.verbose > 0) 4389 perror("getsockopt(SO_SNDBUF)"); 4390 return; 4391 } 4392 4393 /* Binary-search for the real maximum. */ 4394 min = old_size; 4395 max = MAX_SENDBUF_SIZE; 4396 4397 while (min <= max) { 4398 avg = ((unsigned int)(min + max)) / 2; 4399 if (setsockopt(sfd, SOL_SOCKET, SO_SNDBUF, (void *)&avg, intsize) == 0) { 4400 last_good = avg; 4401 min = avg + 1; 4402 } else { 4403 max = avg - 1; 4404 } 4405 } 4406 4407 if (settings.verbose > 1) 4408 fprintf(stderr, "<%d send buffer was %d, now %d\n", sfd, old_size, last_good); 4409 } 4410 4411 /** 4412 * Create a socket and bind it to a specific port number 4413 * @param interface the interface to bind to 4414 * @param port the port number to bind to 4415 * @param transport the transport protocol (TCP / UDP) 4416 * @param portnumber_file A filepointer to write the port numbers to 4417 * when they are successfully added to the list of ports we 4418 * listen on. 4419 */ 4420 static int server_socket(const char *interface, 4421 int port, 4422 enum network_transport transport, 4423 FILE *portnumber_file) { 4424 int sfd; 4425 struct linger ling = {0, 0}; 4426 struct addrinfo *ai; 4427 struct addrinfo *next; 4428 struct addrinfo hints = { .ai_flags = AI_PASSIVE, 4429 .ai_family = AF_UNSPEC }; 4430 char port_buf[NI_MAXSERV]; 4431 int error; 4432 int success = 0; 4433 int flags =1; 4434 4435 hints.ai_socktype = IS_UDP(transport) ? SOCK_DGRAM : SOCK_STREAM; 4436 4437 if (port == -1) { 4438 port = 0; 4439 } 4440 snprintf(port_buf, sizeof(port_buf), "%d", port); 4441 error= getaddrinfo(interface, port_buf, &hints, &ai); 4442 if (error != 0) { 4443 if (error != EAI_SYSTEM) 4444 fprintf(stderr, "getaddrinfo(): %s\n", gai_strerror(error)); 4445 else 4446 perror("getaddrinfo()"); 4447 return 1; 4448 } 4449 4450 for (next= ai; next; next= next->ai_next) { 4451 conn *listen_conn_add; 4452 if ((sfd = new_socket(next)) == -1) { 4453 /* getaddrinfo can return "junk" addresses, 4454 * we make sure at least one works before erroring. 4455 */ 4456 if (errno == EMFILE) { 4457 /* ...unless we're out of fds */ 4458 perror("server_socket"); 4459 exit(EX_OSERR); 4460 } 4461 continue; 4462 } 4463 4464 #ifdef IPV6_V6ONLY 4465 if (next->ai_family == AF_INET6) { 4466 error = setsockopt(sfd, IPPROTO_IPV6, IPV6_V6ONLY, (char *) &flags, sizeof(flags)); 4467 if (error != 0) { 4468 perror("setsockopt"); 4469 close(sfd); 4470 continue; 4471 } 4472 } 4473 #endif 4474 4475 setsockopt(sfd, SOL_SOCKET, SO_REUSEADDR, (void *)&flags, sizeof(flags)); 4476 if (IS_UDP(transport)) { 4477 maximize_sndbuf(sfd); 4478 } else { 4479 error = setsockopt(sfd, SOL_SOCKET, SO_KEEPALIVE, (void *)&flags, sizeof(flags)); 4480 if (error != 0) 4481 perror("setsockopt"); 4482 4483 error = setsockopt(sfd, SOL_SOCKET, SO_LINGER, (void *)&ling, sizeof(ling)); 4484 if (error != 0) 4485 perror("setsockopt"); 4486 4487 error = setsockopt(sfd, IPPROTO_TCP, TCP_NODELAY, (void *)&flags, sizeof(flags)); 4488 if (error != 0) 4489 perror("setsockopt"); 4490 } 4491 4492 if (bind(sfd, next->ai_addr, next->ai_addrlen) == -1) { 4493 if (errno != EADDRINUSE) { 4494 perror("bind()"); 4495 close(sfd); 4496 freeaddrinfo(ai); 4497 return 1; 4498 } 4499 close(sfd); 4500 continue; 4501 } else { 4502 success++; 4503 if (!IS_UDP(transport) && listen(sfd, settings.backlog) == -1) { 4504 perror("listen()"); 4505 close(sfd); 4506 freeaddrinfo(ai); 4507 return 1; 4508 } 4509 if (portnumber_file != NULL && 4510 (next->ai_addr->sa_family == AF_INET || 4511 next->ai_addr->sa_family == AF_INET6)) { 4512 union { 4513 struct sockaddr_in in; 4514 struct sockaddr_in6 in6; 4515 } my_sockaddr; 4516 socklen_t len = sizeof(my_sockaddr); 4517 if (getsockname(sfd, (struct sockaddr*)&my_sockaddr, &len)==0) { 4518 if (next->ai_addr->sa_family == AF_INET) { 4519 fprintf(portnumber_file, "%s INET: %u\n", 4520 IS_UDP(transport) ? "UDP" : "TCP", 4521 ntohs(my_sockaddr.in.sin_port)); 4522 } else { 4523 fprintf(portnumber_file, "%s INET6: %u\n", 4524 IS_UDP(transport) ? "UDP" : "TCP", 4525 ntohs(my_sockaddr.in6.sin6_port)); 4526 } 4527 } 4528 } 4529 } 4530 4531 if (IS_UDP(transport)) { 4532 int c; 4533 4534 for (c = 0; c < settings.num_threads_per_udp; c++) { 4535 /* Allocate one UDP file descriptor per worker thread; 4536 * this allows "stats conns" to separately list multiple 4537 * parallel UDP requests in progress. 4538 * 4539 * The dispatch code round-robins new connection requests 4540 * among threads, so this is guaranteed to assign one 4541 * FD to each thread. 4542 */ 4543 int per_thread_fd = c ? dup(sfd) : sfd; 4544 dispatch_conn_new(per_thread_fd, conn_read, 4545 EV_READ | EV_PERSIST, 4546 UDP_READ_BUFFER_SIZE, transport); 4547 } 4548 } else { 4549 if (!(listen_conn_add = conn_new(sfd, conn_listening, 4550 EV_READ | EV_PERSIST, 1, 4551 transport, main_base))) { 4552 fprintf(stderr, "failed to create listening connection\n"); 4553 exit(EXIT_FAILURE); 4554 } 4555 listen_conn_add->next = listen_conn; 4556 listen_conn = listen_conn_add; 4557 } 4558 } 4559 4560 freeaddrinfo(ai); 4561 4562 /* Return zero iff we detected no errors in starting up connections */ 4563 return success == 0; 4564 } 4565 4566 static int server_sockets(int port, enum network_transport transport, 4567 FILE *portnumber_file) { 4568 if (settings.inter == NULL) { 4569 return server_socket(settings.inter, port, transport, portnumber_file); 4570 } else { 4571 // tokenize them and bind to each one of them.. 4572 char *b; 4573 int ret = 0; 4574 char *list = strdup(settings.inter); 4575 4576 if (list == NULL) { 4577 fprintf(stderr, "Failed to allocate memory for parsing server interface string\n"); 4578 return 1; 4579 } 4580 for (char *p = strtok_r(list, ";,", &b); 4581 p != NULL; 4582 p = strtok_r(NULL, ";,", &b)) { 4583 int the_port = port; 4584 char *s = strchr(p, ':'); 4585 if (s != NULL) { 4586 *s = '\0'; 4587 ++s; 4588 if (!safe_strtol(s, &the_port)) { 4589 fprintf(stderr, "Invalid port number: \"%s\"", s); 4590 return 1; 4591 } 4592 } 4593 if (strcmp(p, "*") == 0) { 4594 p = NULL; 4595 } 4596 ret |= server_socket(p, the_port, transport, portnumber_file); 4597 } 4598 free(list); 4599 return ret; 4600 } 4601 } 4602 4603 static int new_socket_unix(void) { 4604 int sfd; 4605 int flags; 4606 4607 if ((sfd = socket(AF_UNIX, SOCK_STREAM, 0)) == -1) { 4608 perror("socket()"); 4609 return -1; 4610 } 4611 4612 if ((flags = fcntl(sfd, F_GETFL, 0)) < 0 || 4613 fcntl(sfd, F_SETFL, flags | O_NONBLOCK) < 0) { 4614 perror("setting O_NONBLOCK"); 4615 close(sfd); 4616 return -1; 4617 } 4618 return sfd; 4619 } 4620 4621 static int server_socket_unix(const char *path, int access_mask) { 4622 int sfd; 4623 struct linger ling = {0, 0}; 4624 struct sockaddr_un addr; 4625 struct stat tstat; 4626 int flags =1; 4627 int old_umask; 4628 4629 if (!path) { 4630 return 1; 4631 } 4632 4633 if ((sfd = new_socket_unix()) == -1) { 4634 return 1; 4635 } 4636 4637 /* 4638 * Clean up a previous socket file if we left it around 4639 */ 4640 if (lstat(path, &tstat) == 0) { 4641 if (S_ISSOCK(tstat.st_mode)) 4642 unlink(path); 4643 } 4644 4645 setsockopt(sfd, SOL_SOCKET, SO_REUSEADDR, (void *)&flags, sizeof(flags)); 4646 setsockopt(sfd, SOL_SOCKET, SO_KEEPALIVE, (void *)&flags, sizeof(flags)); 4647 setsockopt(sfd, SOL_SOCKET, SO_LINGER, (void *)&ling, sizeof(ling)); 4648 4649 /* 4650 * the memset call clears nonstandard fields in some impementations 4651 * that otherwise mess things up. 4652 */ 4653 memset(&addr, 0, sizeof(addr)); 4654 4655 addr.sun_family = AF_UNIX; 4656 strncpy(addr.sun_path, path, sizeof(addr.sun_path) - 1); 4657 assert(strcmp(addr.sun_path, path) == 0); 4658 old_umask = umask( ~(access_mask&0777)); 4659 if (bind(sfd, (struct sockaddr *)&addr, sizeof(addr)) == -1) { 4660 perror("bind()"); 4661 close(sfd); 4662 umask(old_umask); 4663 return 1; 4664 } 4665 umask(old_umask); 4666 if (listen(sfd, settings.backlog) == -1) { 4667 perror("listen()"); 4668 close(sfd); 4669 return 1; 4670 } 4671 if (!(listen_conn = conn_new(sfd, conn_listening, 4672 EV_READ | EV_PERSIST, 1, 4673 local_transport, main_base))) { 4674 fprintf(stderr, "failed to create listening connection\n"); 4675 exit(EXIT_FAILURE); 4676 } 4677 4678 return 0; 4679 } 4680 4681 /* 4682 * We keep the current time of day in a global variable that's updated by a 4683 * timer event. This saves us a bunch of time() system calls (we really only 4684 * need to get the time once a second, whereas there can be tens of thousands 4685 * of requests a second) and allows us to use server-start-relative timestamps 4686 * rather than absolute UNIX timestamps, a space savings on systems where 4687 * sizeof(time_t) > sizeof(unsigned int). 4688 */ 4689 volatile rel_time_t current_time; 4690 static struct event clockevent; 4691 4692 /* libevent uses a monotonic clock when available for event scheduling. Aside 4693 * from jitter, simply ticking our internal timer here is accurate enough. 4694 * Note that users who are setting explicit dates for expiration times *must* 4695 * ensure their clocks are correct before starting memcached. */ 4696 static void clock_handler(const int fd, const short which, void *arg) { 4697 struct timeval t = {.tv_sec = 1, .tv_usec = 0}; 4698 static bool initialized = false; 4699 #if defined(HAVE_CLOCK_GETTIME) && defined(CLOCK_MONOTONIC) 4700 static bool monotonic = false; 4701 static time_t monotonic_start; 4702 #endif 4703 4704 if (initialized) { 4705 /* only delete the event if it's actually there. */ 4706 evtimer_del(&clockevent); 4707 } else { 4708 initialized = true; 4709 /* process_started is initialized to time() - 2. We initialize to 1 so 4710 * flush_all won't underflow during tests. */ 4711 #if defined(HAVE_CLOCK_GETTIME) && defined(CLOCK_MONOTONIC) 4712 struct timespec ts; 4713 if (clock_gettime(CLOCK_MONOTONIC, &ts) == 0) { 4714 monotonic = true; 4715 monotonic_start = ts.tv_sec - ITEM_UPDATE_INTERVAL - 2; 4716 } 4717 #endif 4718 } 4719 4720 evtimer_set(&clockevent, clock_handler, 0); 4721 event_base_set(main_base, &clockevent); 4722 evtimer_add(&clockevent, &t); 4723 4724 #if defined(HAVE_CLOCK_GETTIME) && defined(CLOCK_MONOTONIC) 4725 if (monotonic) { 4726 struct timespec ts; 4727 if (clock_gettime(CLOCK_MONOTONIC, &ts) == -1) 4728 return; 4729 current_time = (rel_time_t) (ts.tv_sec - monotonic_start); 4730 return; 4731 } 4732 #endif 4733 { 4734 struct timeval tv; 4735 gettimeofday(&tv, NULL); 4736 current_time = (rel_time_t) (tv.tv_sec - process_started); 4737 } 4738 } 4739 4740 static void usage(void) { 4741 printf(PACKAGE " " VERSION "\n"); 4742 printf("-p <num> TCP port number to listen on (default: 11211)\n" 4743 "-U <num> UDP port number to listen on (default: 11211, 0 is off)\n" 4744 "-s <file> UNIX socket path to listen on (disables network support)\n" 4745 "-A enable ascii \"shutdown\" command\n" 4746 "-a <mask> access mask for UNIX socket, in octal (default: 0700)\n" 4747 "-l <addr> interface to listen on (default: INADDR_ANY, all addresses)\n" 4748 " <addr> may be specified as host:port. If you don't specify\n" 4749 " a port number, the value you specified with -p or -U is\n" 4750 " used. You may specify multiple addresses separated by comma\n" 4751 " or by using -l multiple times\n" 4752 4753 "-d run as a daemon\n" 4754 "-r maximize core file limit\n" 4755 "-u <username> assume identity of <username> (only when run as root)\n" 4756 "-m <num> max memory to use for items in megabytes (default: 64 MB)\n" 4757 "-M return error on memory exhausted (rather than removing items)\n" 4758 "-c <num> max simultaneous connections (default: 1024)\n" 4759 "-k lock down all paged memory. Note that there is a\n" 4760 " limit on how much memory you may lock. Trying to\n" 4761 " allocate more than that would fail, so be sure you\n" 4762 " set the limit correctly for the user you started\n" 4763 " the daemon with (not for -u <username> user;\n" 4764 " under sh this is done with 'ulimit -S -l NUM_KB').\n" 4765 "-v verbose (print errors/warnings while in event loop)\n" 4766 "-vv very verbose (also print client commands/reponses)\n" 4767 "-vvv extremely verbose (also print internal state transitions)\n" 4768 "-h print this help and exit\n" 4769 "-i print memcached and libevent license\n" 4770 "-P <file> save PID in <file>, only used with -d option\n" 4771 "-f <factor> chunk size growth factor (default: 1.25)\n" 4772 "-n <bytes> minimum space allocated for key+value+flags (default: 48)\n"); 4773 printf("-L Try to use large memory pages (if available). Increasing\n" 4774 " the memory page size could reduce the number of TLB misses\n" 4775 " and improve the performance. In order to get large pages\n" 4776 " from the OS, memcached will allocate the total item-cache\n" 4777 " in one large chunk.\n"); 4778 printf("-D <char> Use <char> as the delimiter between key prefixes and IDs.\n" 4779 " This is used for per-prefix stats reporting. The default is\n" 4780 " \":\" (colon). If this option is specified, stats collection\n" 4781 " is turned on automatically; if not, then it may be turned on\n" 4782 " by sending the \"stats detail on\" command to the server.\n"); 4783 printf("-t <num> number of threads to use (default: 4)\n"); 4784 printf("-R Maximum number of requests per event, limits the number of\n" 4785 " requests process for a given connection to prevent \n" 4786 " starvation (default: 20)\n"); 4787 printf("-C Disable use of CAS\n"); 4788 printf("-b Set the backlog queue limit (default: 1024)\n"); 4789 printf("-B Binding protocol - one of ascii, binary, or auto (default)\n"); 4790 printf("-I Override the size of each slab page. Adjusts max item size\n" 4791 " (default: 1mb, min: 1k, max: 128m)\n"); 4792 #ifdef ENABLE_SASL 4793 printf("-S Turn on Sasl authentication\n"); 4794 #endif 4795 printf("-F Disable flush_all command\n"); 4796 printf("-o Comma separated list of extended or experimental options\n" 4797 " - (EXPERIMENTAL) maxconns_fast: immediately close new\n" 4798 " connections if over maxconns limit\n" 4799 " - hashpower: An integer multiplier for how large the hash\n" 4800 " table should be. Can be grown at runtime if not big enough.\n" 4801 " Set this based on \"STAT hash_power_level\" before a \n" 4802 " restart.\n" 4803 " - tail_repair_time: Time in seconds that indicates how long to wait before\n" 4804 " forcefully taking over the LRU tail item whose refcount has leaked.\n" 4805 " The default is 3 hours.\n" 4806 " - hash_algorithm: The hash table algorithm\n" 4807 " default is jenkins hash. options: jenkins, murmur3\n" 4808 " - lru_crawler: Enable LRU Crawler background thread\n" 4809 " - lru_crawler_sleep: Microseconds to sleep between items\n" 4810 " default is 100.\n" 4811 " - lru_crawler_tocrawl: Max items to crawl per slab per run\n" 4812 " default is 0 (unlimited)\n" 4813 ); 4814 return; 4815 } 4816 4817 static void usage_license(void) { 4818 printf(PACKAGE " " VERSION "\n\n"); 4819 printf( 4820 "Copyright (c) 2003, Danga Interactive, Inc. <http://www.danga.com/>\n" 4821 "All rights reserved.\n" 4822 "\n" 4823 "Redistribution and use in source and binary forms, with or without\n" 4824 "modification, are permitted provided that the following conditions are\n" 4825 "met:\n" 4826 "\n" 4827 " * Redistributions of source code must retain the above copyright\n" 4828 "notice, this list of conditions and the following disclaimer.\n" 4829 "\n" 4830 " * Redistributions in binary form must reproduce the above\n" 4831 "copyright notice, this list of conditions and the following disclaimer\n" 4832 "in the documentation and/or other materials provided with the\n" 4833 "distribution.\n" 4834 "\n" 4835 " * Neither the name of the Danga Interactive nor the names of its\n" 4836 "contributors may be used to endorse or promote products derived from\n" 4837 "this software without specific prior written permission.\n" 4838 "\n" 4839 "THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS\n" 4840 "\"AS IS\" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT\n" 4841 "LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR\n" 4842 "A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT\n" 4843 "OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,\n" 4844 "SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT\n" 4845 "LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,\n" 4846 "DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY\n" 4847 "THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT\n" 4848 "(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE\n" 4849 "OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.\n" 4850 "\n" 4851 "\n" 4852 "This product includes software developed by Niels Provos.\n" 4853 "\n" 4854 "[ libevent ]\n" 4855 "\n" 4856 "Copyright 2000-2003 Niels Provos <[email protected]>\n" 4857 "All rights reserved.\n" 4858 "\n" 4859 "Redistribution and use in source and binary forms, with or without\n" 4860 "modification, are permitted provided that the following conditions\n" 4861 "are met:\n" 4862 "1. Redistributions of source code must retain the above copyright\n" 4863 " notice, this list of conditions and the following disclaimer.\n" 4864 "2. Redistributions in binary form must reproduce the above copyright\n" 4865 " notice, this list of conditions and the following disclaimer in the\n" 4866 " documentation and/or other materials provided with the distribution.\n" 4867 "3. All advertising materials mentioning features or use of this software\n" 4868 " must display the following acknowledgement:\n" 4869 " This product includes software developed by Niels Provos.\n" 4870 "4. The name of the author may not be used to endorse or promote products\n" 4871 " derived from this software without specific prior written permission.\n" 4872 "\n" 4873 "THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR\n" 4874 "IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES\n" 4875 "OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.\n" 4876 "IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT,\n" 4877 "INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT\n" 4878 "NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,\n" 4879 "DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY\n" 4880 "THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT\n" 4881 "(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF\n" 4882 "THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.\n" 4883 ); 4884 4885 return; 4886 } 4887 4888 static void save_pid(const char *pid_file) { 4889 FILE *fp; 4890 if (access(pid_file, F_OK) == 0) { 4891 if ((fp = fopen(pid_file, "r")) != NULL) { 4892 char buffer[1024]; 4893 if (fgets(buffer, sizeof(buffer), fp) != NULL) { 4894 unsigned int pid; 4895 if (safe_strtoul(buffer, &pid) && kill((pid_t)pid, 0) == 0) { 4896 fprintf(stderr, "WARNING: The pid file contained the following (running) pid: %u\n", pid); 4897 } 4898 } 4899 fclose(fp); 4900 } 4901 } 4902 4903 /* Create the pid file first with a temporary name, then 4904 * atomically move the file to the real name to avoid a race with 4905 * another process opening the file to read the pid, but finding 4906 * it empty. 4907 */ 4908 char tmp_pid_file[1024]; 4909 snprintf(tmp_pid_file, sizeof(tmp_pid_file), "%s.tmp", pid_file); 4910 4911 if ((fp = fopen(tmp_pid_file, "w")) == NULL) { 4912 vperror("Could not open the pid file %s for writing", tmp_pid_file); 4913 return; 4914 } 4915 4916 fprintf(fp,"%ld\n", (long)getpid()); 4917 if (fclose(fp) == -1) { 4918 vperror("Could not close the pid file %s", tmp_pid_file); 4919 } 4920 4921 if (rename(tmp_pid_file, pid_file) != 0) { 4922 vperror("Could not rename the pid file from %s to %s", 4923 tmp_pid_file, pid_file); 4924 } 4925 } 4926 4927 static void remove_pidfile(const char *pid_file) { 4928 if (pid_file == NULL) 4929 return; 4930 4931 if (unlink(pid_file) != 0) { 4932 vperror("Could not remove the pid file %s", pid_file); 4933 } 4934 4935 } 4936 4937 static void sig_handler(const int sig) { 4938 printf("SIGINT handled.\n"); 4939 exit(EXIT_SUCCESS); 4940 } 4941 4942 #ifndef HAVE_SIGIGNORE 4943 static int sigignore(int sig) { 4944 struct sigaction sa = { .sa_handler = SIG_IGN, .sa_flags = 0 }; 4945 4946 if (sigemptyset(&sa.sa_mask) == -1 || sigaction(sig, &sa, 0) == -1) { 4947 return -1; 4948 } 4949 return 0; 4950 } 4951 #endif 4952 4953 4954 /* 4955 * On systems that supports multiple page sizes we may reduce the 4956 * number of TLB-misses by using the biggest available page size 4957 */ 4958 static int enable_large_pages(void) { 4959 #if defined(HAVE_GETPAGESIZES) && defined(HAVE_MEMCNTL) 4960 int ret = -1; 4961 size_t sizes[32]; 4962 int avail = getpagesizes(sizes, 32); 4963 if (avail != -1) { 4964 size_t max = sizes[0]; 4965 struct memcntl_mha arg = {0}; 4966 int ii; 4967 4968 for (ii = 1; ii < avail; ++ii) { 4969 if (max < sizes[ii]) { 4970 max = sizes[ii]; 4971 } 4972 } 4973 4974 arg.mha_flags = 0; 4975 arg.mha_pagesize = max; 4976 arg.mha_cmd = MHA_MAPSIZE_BSSBRK; 4977 4978 if (memcntl(0, 0, MC_HAT_ADVISE, (caddr_t)&arg, 0, 0) == -1) { 4979 fprintf(stderr, "Failed to set large pages: %s\n", 4980 strerror(errno)); 4981 fprintf(stderr, "Will use default page size\n"); 4982 } else { 4983 ret = 0; 4984 } 4985 } else { 4986 fprintf(stderr, "Failed to get supported pagesizes: %s\n", 4987 strerror(errno)); 4988 fprintf(stderr, "Will use default page size\n"); 4989 } 4990 4991 return ret; 4992 #else 4993 return -1; 4994 #endif 4995 } 4996 4997 /** 4998 * Do basic sanity check of the runtime environment 4999 * @return true if no errors found, false if we can't use this env 5000 */ 5001 static bool sanitycheck(void) { 5002 /* One of our biggest problems is old and bogus libevents */ 5003 const char *ever = event_get_version(); 5004 if (ever != NULL) { 5005 if (strncmp(ever, "1.", 2) == 0) { 5006 /* Require at least 1.3 (that's still a couple of years old) */ 5007 if ((ever[2] == '1' || ever[2] == '2') && !isdigit(ever[3])) { 5008 fprintf(stderr, "You are using libevent %s.\nPlease upgrade to" 5009 " a more recent version (1.3 or newer)\n", 5010 event_get_version()); 5011 return false; 5012 } 5013 } 5014 } 5015 5016 return true; 5017 } 5018 5019 int main (int argc, char **argv) { 5020 int c; 5021 bool lock_memory = false; 5022 bool do_daemonize = false; 5023 bool preallocate = false; 5024 int maxcore = 0; 5025 char *username = NULL; 5026 char *pid_file = NULL; 5027 struct passwd *pw; 5028 struct rlimit rlim; 5029 char *buf; 5030 char unit = '\0'; 5031 int size_max = 0; 5032 int retval = EXIT_SUCCESS; 5033 /* listening sockets */ 5034 static int *l_socket = NULL; 5035 5036 /* udp socket */ 5037 static int *u_socket = NULL; 5038 bool protocol_specified = false; 5039 bool tcp_specified = false; 5040 bool udp_specified = false; 5041 enum hashfunc_type hash_type = JENKINS_HASH; 5042 uint32_t tocrawl; 5043 5044 char *subopts; 5045 char *subopts_value; 5046 enum { 5047 MAXCONNS_FAST = 0, 5048 HASHPOWER_INIT, 5049 SLAB_REASSIGN, 5050 SLAB_AUTOMOVE, 5051 TAIL_REPAIR_TIME, 5052 HASH_ALGORITHM, 5053 LRU_CRAWLER, 5054 LRU_CRAWLER_SLEEP, 5055 LRU_CRAWLER_TOCRAWL 5056 }; 5057 char *const subopts_tokens[] = { 5058 [MAXCONNS_FAST] = "maxconns_fast", 5059 [HASHPOWER_INIT] = "hashpower", 5060 [SLAB_REASSIGN] = "slab_reassign", 5061 [SLAB_AUTOMOVE] = "slab_automove", 5062 [TAIL_REPAIR_TIME] = "tail_repair_time", 5063 [HASH_ALGORITHM] = "hash_algorithm", 5064 [LRU_CRAWLER] = "lru_crawler", 5065 [LRU_CRAWLER_SLEEP] = "lru_crawler_sleep", 5066 [LRU_CRAWLER_TOCRAWL] = "lru_crawler_tocrawl", 5067 NULL 5068 }; 5069 5070 if (!sanitycheck()) { 5071 return EX_OSERR; 5072 } 5073 5074 /* handle SIGINT */ 5075 signal(SIGINT, sig_handler); 5076 5077 /* init settings */ 5078 settings_init(); 5079 5080 /* set stderr non-buffering (for running under, say, daemontools) */ 5081 setbuf(stderr, NULL); 5082 5083 /* process arguments */ 5084 while (-1 != (c = getopt(argc, argv, 5085 "a:" /* access mask for unix socket */ 5086 "A" /* enable admin shutdown commannd */ 5087 "p:" /* TCP port number to listen on */ 5088 "s:" /* unix socket path to listen on */ 5089 "U:" /* UDP port number to listen on */ 5090 "m:" /* max memory to use for items in megabytes */ 5091 "M" /* return error on memory exhausted */ 5092 "c:" /* max simultaneous connections */ 5093 "k" /* lock down all paged memory */ 5094 "hi" /* help, licence info */ 5095 "r" /* maximize core file limit */ 5096 "v" /* verbose */ 5097 "d" /* daemon mode */ 5098 "l:" /* interface to listen on */ 5099 "u:" /* user identity to run as */ 5100 "P:" /* save PID in file */ 5101 "f:" /* factor? */ 5102 "n:" /* minimum space allocated for key+value+flags */ 5103 "t:" /* threads */ 5104 "D:" /* prefix delimiter? */ 5105 "L" /* Large memory pages */ 5106 "R:" /* max requests per event */ 5107 "C" /* Disable use of CAS */ 5108 "b:" /* backlog queue limit */ 5109 "B:" /* Binding protocol */ 5110 "I:" /* Max item size */ 5111 "S" /* Sasl ON */ 5112 "F" /* Disable flush_all */ 5113 "o:" /* Extended generic options */ 5114 ))) { 5115 switch (c) { 5116 case 'A': 5117 /* enables "shutdown" command */ 5118 settings.shutdown_command = true; 5119 break; 5120 5121 case 'a': 5122 /* access for unix domain socket, as octal mask (like chmod)*/ 5123 settings.access= strtol(optarg,NULL,8); 5124 break; 5125 5126 case 'U': 5127 settings.udpport = atoi(optarg); 5128 udp_specified = true; 5129 break; 5130 case 'p': 5131 settings.port = atoi(optarg); 5132 tcp_specified = true; 5133 break; 5134 case 's': 5135 settings.socketpath = optarg; 5136 break; 5137 case 'm': 5138 settings.maxbytes = ((size_t)atoi(optarg)) * 1024 * 1024; 5139 break; 5140 case 'M': 5141 settings.evict_to_free = 0; 5142 break; 5143 case 'c': 5144 settings.maxconns = atoi(optarg); 5145 break; 5146 case 'h': 5147 usage(); 5148 exit(EXIT_SUCCESS); 5149 case 'i': 5150 usage_license(); 5151 exit(EXIT_SUCCESS); 5152 case 'k': 5153 lock_memory = true; 5154 break; 5155 case 'v': 5156 settings.verbose++; 5157 break; 5158 case 'l': 5159 if (settings.inter != NULL) { 5160 size_t len = strlen(settings.inter) + strlen(optarg) + 2; 5161 char *p = malloc(len); 5162 if (p == NULL) { 5163 fprintf(stderr, "Failed to allocate memory\n"); 5164 return 1; 5165 } 5166 snprintf(p, len, "%s,%s", settings.inter, optarg); 5167 free(settings.inter); 5168 settings.inter = p; 5169 } else { 5170 settings.inter= strdup(optarg); 5171 } 5172 break; 5173 case 'd': 5174 do_daemonize = true; 5175 break; 5176 case 'r': 5177 maxcore = 1; 5178 break; 5179 case 'R': 5180 settings.reqs_per_event = atoi(optarg); 5181 if (settings.reqs_per_event == 0) { 5182 fprintf(stderr, "Number of requests per event must be greater than 0\n"); 5183 return 1; 5184 } 5185 break; 5186 case 'u': 5187 username = optarg; 5188 break; 5189 case 'P': 5190 pid_file = optarg; 5191 break; 5192 case 'f': 5193 settings.factor = atof(optarg); 5194 if (settings.factor <= 1.0) { 5195 fprintf(stderr, "Factor must be greater than 1\n"); 5196 return 1; 5197 } 5198 break; 5199 case 'n': 5200 settings.chunk_size = atoi(optarg); 5201 if (settings.chunk_size == 0) { 5202 fprintf(stderr, "Chunk size must be greater than 0\n"); 5203 return 1; 5204 } 5205 break; 5206 case 't': 5207 settings.num_threads = atoi(optarg); 5208 if (settings.num_threads <= 0) { 5209 fprintf(stderr, "Number of threads must be greater than 0\n"); 5210 return 1; 5211 } 5212 /* There're other problems when you get above 64 threads. 5213 * In the future we should portably detect # of cores for the 5214 * default. 5215 */ 5216 if (settings.num_threads > 64) { 5217 fprintf(stderr, "WARNING: Setting a high number of worker" 5218 "threads is not recommended.\n" 5219 " Set this value to the number of cores in" 5220 " your machine or less.\n"); 5221 } 5222 break; 5223 case 'D': 5224 if (! optarg || ! optarg[0]) { 5225 fprintf(stderr, "No delimiter specified\n"); 5226 return 1; 5227 } 5228 settings.prefix_delimiter = optarg[0]; 5229 settings.detail_enabled = 1; 5230 break; 5231 case 'L' : 5232 if (enable_large_pages() == 0) { 5233 preallocate = true; 5234 } else { 5235 fprintf(stderr, "Cannot enable large pages on this system\n" 5236 "(There is no Linux support as of this version)\n"); 5237 return 1; 5238 } 5239 break; 5240 case 'C' : 5241 settings.use_cas = false; 5242 break; 5243 case 'b' : 5244 settings.backlog = atoi(optarg); 5245 break; 5246 case 'B': 5247 protocol_specified = true; 5248 if (strcmp(optarg, "auto") == 0) { 5249 settings.binding_protocol = negotiating_prot; 5250 } else if (strcmp(optarg, "binary") == 0) { 5251 settings.binding_protocol = binary_prot; 5252 } else if (strcmp(optarg, "ascii") == 0) { 5253 settings.binding_protocol = ascii_prot; 5254 } else { 5255 fprintf(stderr, "Invalid value for binding protocol: %s\n" 5256 " -- should be one of auto, binary, or ascii\n", optarg); 5257 exit(EX_USAGE); 5258 } 5259 break; 5260 case 'I': 5261 buf = strdup(optarg); 5262 unit = buf[strlen(buf)-1]; 5263 if (unit == 'k' || unit == 'm' || 5264 unit == 'K' || unit == 'M') { 5265 buf[strlen(buf)-1] = '\0'; 5266 size_max = atoi(buf); 5267 if (unit == 'k' || unit == 'K') 5268 size_max *= 1024; 5269 if (unit == 'm' || unit == 'M') 5270 size_max *= 1024 * 1024; 5271 settings.item_size_max = size_max; 5272 } else { 5273 settings.item_size_max = atoi(buf); 5274 } 5275 if (settings.item_size_max < 1024) { 5276 fprintf(stderr, "Item max size cannot be less than 1024 bytes.\n"); 5277 return 1; 5278 } 5279 if (settings.item_size_max > 1024 * 1024 * 128) { 5280 fprintf(stderr, "Cannot set item size limit higher than 128 mb.\n"); 5281 return 1; 5282 } 5283 if (settings.item_size_max > 1024 * 1024) { 5284 fprintf(stderr, "WARNING: Setting item max size above 1MB is not" 5285 " recommended!\n" 5286 " Raising this limit increases the minimum memory requirements\n" 5287 " and will decrease your memory efficiency.\n" 5288 ); 5289 } 5290 free(buf); 5291 break; 5292 case 'S': /* set Sasl authentication to true. Default is false */ 5293 #ifndef ENABLE_SASL 5294 fprintf(stderr, "This server is not built with SASL support.\n"); 5295 exit(EX_USAGE); 5296 #endif 5297 settings.sasl = true; 5298 break; 5299 case 'F' : 5300 settings.flush_enabled = false; 5301 break; 5302 case 'o': /* It's sub-opts time! */ 5303 subopts = optarg; 5304 5305 while (*subopts != '\0') { 5306 5307 switch (getsubopt(&subopts, subopts_tokens, &subopts_value)) { 5308 case MAXCONNS_FAST: 5309 settings.maxconns_fast = true; 5310 break; 5311 case HASHPOWER_INIT: 5312 if (subopts_value == NULL) { 5313 fprintf(stderr, "Missing numeric argument for hashpower\n"); 5314 return 1; 5315 } 5316 settings.hashpower_init = atoi(subopts_value); 5317 if (settings.hashpower_init < 12) { 5318 fprintf(stderr, "Initial hashtable multiplier of %d is too low\n", 5319 settings.hashpower_init); 5320 return 1; 5321 } else if (settings.hashpower_init > 64) { 5322 fprintf(stderr, "Initial hashtable multiplier of %d is too high\n" 5323 "Choose a value based on \"STAT hash_power_level\" from a running instance\n", 5324 settings.hashpower_init); 5325 return 1; 5326 } 5327 break; 5328 case SLAB_REASSIGN: 5329 settings.slab_reassign = true; 5330 break; 5331 case SLAB_AUTOMOVE: 5332 if (subopts_value == NULL) { 5333 settings.slab_automove = 1; 5334 break; 5335 } 5336 settings.slab_automove = atoi(subopts_value); 5337 if (settings.slab_automove < 0 || settings.slab_automove > 2) { 5338 fprintf(stderr, "slab_automove must be between 0 and 2\n"); 5339 return 1; 5340 } 5341 break; 5342 case TAIL_REPAIR_TIME: 5343 if (subopts_value == NULL) { 5344 fprintf(stderr, "Missing numeric argument for tail_repair_time\n"); 5345 return 1; 5346 } 5347 settings.tail_repair_time = atoi(subopts_value); 5348 if (settings.tail_repair_time < 10) { 5349 fprintf(stderr, "Cannot set tail_repair_time to less than 10 seconds\n"); 5350 return 1; 5351 } 5352 break; 5353 case HASH_ALGORITHM: 5354 if (subopts_value == NULL) { 5355 fprintf(stderr, "Missing hash_algorithm argument\n"); 5356 return 1; 5357 }; 5358 if (strcmp(subopts_value, "jenkins") == 0) { 5359 hash_type = JENKINS_HASH; 5360 } else if (strcmp(subopts_value, "murmur3") == 0) { 5361 hash_type = MURMUR3_HASH; 5362 } else { 5363 fprintf(stderr, "Unknown hash_algorithm option (jenkins, murmur3)\n"); 5364 return 1; 5365 } 5366 break; 5367 case LRU_CRAWLER: 5368 if (start_item_crawler_thread() != 0) { 5369 fprintf(stderr, "Failed to enable LRU crawler thread\n"); 5370 return 1; 5371 } 5372 break; 5373 case LRU_CRAWLER_SLEEP: 5374 settings.lru_crawler_sleep = atoi(subopts_value); 5375 if (settings.lru_crawler_sleep > 1000000 || settings.lru_crawler_sleep < 0) { 5376 fprintf(stderr, "LRU crawler sleep must be between 0 and 1 second\n"); 5377 return 1; 5378 } 5379 break; 5380 case LRU_CRAWLER_TOCRAWL: 5381 if (!safe_strtoul(subopts_value, &tocrawl)) { 5382 fprintf(stderr, "lru_crawler_tocrawl takes a numeric 32bit value\n"); 5383 return 1; 5384 } 5385 settings.lru_crawler_tocrawl = tocrawl; 5386 break; 5387 default: 5388 printf("Illegal suboption \"%s\"\n", subopts_value); 5389 return 1; 5390 } 5391 5392 } 5393 break; 5394 default: 5395 fprintf(stderr, "Illegal argument \"%c\"\n", c); 5396 return 1; 5397 } 5398 } 5399 5400 if (hash_init(hash_type) != 0) { 5401 fprintf(stderr, "Failed to initialize hash_algorithm!\n"); 5402 exit(EX_USAGE); 5403 } 5404 5405 /* 5406 * Use one workerthread to serve each UDP port if the user specified 5407 * multiple ports 5408 */ 5409 if (settings.inter != NULL && strchr(settings.inter, ',')) { 5410 settings.num_threads_per_udp = 1; 5411 } else { 5412 settings.num_threads_per_udp = settings.num_threads; 5413 } 5414 5415 if (settings.sasl) { 5416 if (!protocol_specified) { 5417 settings.binding_protocol = binary_prot; 5418 } else { 5419 if (settings.binding_protocol != binary_prot) { 5420 fprintf(stderr, "ERROR: You cannot allow the ASCII protocol while using SASL.\n"); 5421 exit(EX_USAGE); 5422 } 5423 } 5424 } 5425 5426 if (tcp_specified && !udp_specified) { 5427 settings.udpport = settings.port; 5428 } else if (udp_specified && !tcp_specified) { 5429 settings.port = settings.udpport; 5430 } 5431 5432 if (maxcore != 0) { 5433 struct rlimit rlim_new; 5434 /* 5435 * First try raising to infinity; if that fails, try bringing 5436 * the soft limit to the hard. 5437 */ 5438 if (getrlimit(RLIMIT_CORE, &rlim) == 0) { 5439 rlim_new.rlim_cur = rlim_new.rlim_max = RLIM_INFINITY; 5440 if (setrlimit(RLIMIT_CORE, &rlim_new)!= 0) { 5441 /* failed. try raising just to the old max */ 5442 rlim_new.rlim_cur = rlim_new.rlim_max = rlim.rlim_max; 5443 (void)setrlimit(RLIMIT_CORE, &rlim_new); 5444 } 5445 } 5446 /* 5447 * getrlimit again to see what we ended up with. Only fail if 5448 * the soft limit ends up 0, because then no core files will be 5449 * created at all. 5450 */ 5451 5452 if ((getrlimit(RLIMIT_CORE, &rlim) != 0) || rlim.rlim_cur == 0) { 5453 fprintf(stderr, "failed to ensure corefile creation\n"); 5454 exit(EX_OSERR); 5455 } 5456 } 5457 5458 /* 5459 * If needed, increase rlimits to allow as many connections 5460 * as needed. 5461 */ 5462 5463 if (getrlimit(RLIMIT_NOFILE, &rlim) != 0) { 5464 fprintf(stderr, "failed to getrlimit number of files\n"); 5465 exit(EX_OSERR); 5466 } else { 5467 rlim.rlim_cur = settings.maxconns; 5468 rlim.rlim_max = settings.maxconns; 5469 if (setrlimit(RLIMIT_NOFILE, &rlim) != 0) { 5470 fprintf(stderr, "failed to set rlimit for open files. Try starting as root or requesting smaller maxconns value.\n"); 5471 exit(EX_OSERR); 5472 } 5473 } 5474 5475 /* lose root privileges if we have them */ 5476 if (getuid() == 0 || geteuid() == 0) { 5477 if (username == 0 || *username == '\0') { 5478 fprintf(stderr, "can't run as root without the -u switch\n"); 5479 exit(EX_USAGE); 5480 } 5481 if ((pw = getpwnam(username)) == 0) { 5482 fprintf(stderr, "can't find the user %s to switch to\n", username); 5483 exit(EX_NOUSER); 5484 } 5485 if (setgid(pw->pw_gid) < 0 || setuid(pw->pw_uid) < 0) { 5486 fprintf(stderr, "failed to assume identity of user %s\n", username); 5487 exit(EX_OSERR); 5488 } 5489 } 5490 5491 /* Initialize Sasl if -S was specified */ 5492 if (settings.sasl) { 5493 init_sasl(); 5494 } 5495 5496 /* daemonize if requested */ 5497 /* if we want to ensure our ability to dump core, don't chdir to / */ 5498 if (do_daemonize) { 5499 if (sigignore(SIGHUP) == -1) { 5500 perror("Failed to ignore SIGHUP"); 5501 } 5502 if (daemonize(maxcore, settings.verbose) == -1) { 5503 fprintf(stderr, "failed to daemon() in order to daemonize\n"); 5504 exit(EXIT_FAILURE); 5505 } 5506 } 5507 5508 /* lock paged memory if needed */ 5509 if (lock_memory) { 5510 #ifdef HAVE_MLOCKALL 5511 int res = mlockall(MCL_CURRENT | MCL_FUTURE); 5512 if (res != 0) { 5513 fprintf(stderr, "warning: -k invalid, mlockall() failed: %s\n", 5514 strerror(errno)); 5515 } 5516 #else 5517 fprintf(stderr, "warning: -k invalid, mlockall() not supported on this platform. proceeding without.\n"); 5518 #endif 5519 } 5520 5521 /* initialize main thread libevent instance */ 5522 main_base = event_init(); 5523 5524 /* initialize other stuff */ 5525 stats_init(); 5526 assoc_init(settings.hashpower_init); 5527 conn_init(); 5528 slabs_init(settings.maxbytes, settings.factor, preallocate); 5529 5530 /* 5531 * ignore SIGPIPE signals; we can use errno == EPIPE if we 5532 * need that information 5533 */ 5534 if (sigignore(SIGPIPE) == -1) { 5535 perror("failed to ignore SIGPIPE; sigaction"); 5536 exit(EX_OSERR); 5537 } 5538 /* start up worker threads if MT mode */ 5539 thread_init(settings.num_threads, main_base); 5540 5541 if (start_assoc_maintenance_thread() == -1) { 5542 exit(EXIT_FAILURE); 5543 } 5544 5545 if (settings.slab_reassign && 5546 start_slab_maintenance_thread() == -1) { 5547 exit(EXIT_FAILURE); 5548 } 5549 5550 /* Run regardless of initializing it later */ 5551 init_lru_crawler(); 5552 5553 /* initialise clock event */ 5554 clock_handler(0, 0, 0); 5555 5556 /* create unix mode sockets after dropping privileges */ 5557 if (settings.socketpath != NULL) { 5558 errno = 0; 5559 if (server_socket_unix(settings.socketpath,settings.access)) { 5560 vperror("failed to listen on UNIX socket: %s", settings.socketpath); 5561 exit(EX_OSERR); 5562 } 5563 } 5564 5565 /* create the listening socket, bind it, and init */ 5566 if (settings.socketpath == NULL) { 5567 const char *portnumber_filename = getenv("MEMCACHED_PORT_FILENAME"); 5568 char temp_portnumber_filename[PATH_MAX]; 5569 FILE *portnumber_file = NULL; 5570 5571 if (portnumber_filename != NULL) { 5572 snprintf(temp_portnumber_filename, 5573 sizeof(temp_portnumber_filename), 5574 "%s.lck", portnumber_filename); 5575 5576 portnumber_file = fopen(temp_portnumber_filename, "a"); 5577 if (portnumber_file == NULL) { 5578 fprintf(stderr, "Failed to open \"%s\": %s\n", 5579 temp_portnumber_filename, strerror(errno)); 5580 } 5581 } 5582 5583 errno = 0; 5584 if (settings.port && server_sockets(settings.port, tcp_transport, 5585 portnumber_file)) { 5586 vperror("failed to listen on TCP port %d", settings.port); 5587 exit(EX_OSERR); 5588 } 5589 5590 /* 5591 * initialization order: first create the listening sockets 5592 * (may need root on low ports), then drop root if needed, 5593 * then daemonise if needed, then init libevent (in some cases 5594 * descriptors created by libevent wouldn't survive forking). 5595 */ 5596 5597 /* create the UDP listening socket and bind it */ 5598 errno = 0; 5599 if (settings.udpport && server_sockets(settings.udpport, udp_transport, 5600 portnumber_file)) { 5601 vperror("failed to listen on UDP port %d", settings.udpport); 5602 exit(EX_OSERR); 5603 } 5604 5605 if (portnumber_file) { 5606 fclose(portnumber_file); 5607 rename(temp_portnumber_filename, portnumber_filename); 5608 } 5609 } 5610 5611 /* Give the sockets a moment to open. I know this is dumb, but the error 5612 * is only an advisory. 5613 */ 5614 usleep(1000); 5615 if (stats.curr_conns + stats.reserved_fds >= settings.maxconns - 1) { 5616 fprintf(stderr, "Maxconns setting is too low, use -c to increase.\n"); 5617 exit(EXIT_FAILURE); 5618 } 5619 5620 if (pid_file != NULL) { 5621 save_pid(pid_file); 5622 } 5623 5624 /* Drop privileges no longer needed */ 5625 drop_privileges(); 5626 5627 /* enter the event loop */ 5628 if (event_base_loop(main_base, 0) != 0) { 5629 retval = EXIT_FAILURE; 5630 } 5631 5632 stop_assoc_maintenance_thread(); 5633 5634 /* remove the PID file if we're a daemon */ 5635 if (do_daemonize) 5636 remove_pidfile(pid_file); 5637 /* Clean up strdup() call for bind() address */ 5638 if (settings.inter) 5639 free(settings.inter); 5640 if (l_socket) 5641 free(l_socket); 5642 if (u_socket) 5643 free(u_socket); 5644 5645 return retval; 5646 } 5647