1 /* -*- Mode: C; tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*- */ 2 /* 3 * memcached - memory caching daemon 4 * 5 * http://www.danga.com/memcached/ 6 * 7 * Copyright 2003 Danga Interactive, Inc. All rights reserved. 8 * 9 * Use and distribution licensed under the BSD license. See 10 * the LICENSE file for full text. 11 * 12 * Authors: 13 * Anatoly Vorobey <[email protected]> 14 * Brad Fitzpatrick <[email protected]> 15 * 16 * $Id$ 17 */ 18 #include "config.h" 19 #include <sys/types.h> 20 #include <sys/stat.h> 21 #include <sys/time.h> 22 #include <sys/socket.h> 23 #include <sys/un.h> 24 #include <sys/signal.h> 25 #include <sys/resource.h> 26 #include <sys/uio.h> 27 28 /* some POSIX systems need the following definition 29 * to get mlockall flags out of sys/mman.h. */ 30 #ifndef _P1003_1B_VISIBLE 31 #define _P1003_1B_VISIBLE 32 #endif 33 /* need this to get IOV_MAX on some platforms. */ 34 #ifndef __need_IOV_MAX 35 #define __need_IOV_MAX 36 #endif 37 #include <pwd.h> 38 #include <sys/mman.h> 39 #include <fcntl.h> 40 #include <stdlib.h> 41 #include <stdio.h> 42 #include <string.h> 43 #include <unistd.h> 44 #include <netinet/in.h> 45 #include <netinet/tcp.h> 46 #include <arpa/inet.h> 47 #include <errno.h> 48 #include <time.h> 49 #include <event.h> 50 #include <assert.h> 51 #include <limits.h> 52 53 #ifdef HAVE_MALLOC_H 54 #include <malloc.h> 55 #endif 56 57 /* FreeBSD 4.x doesn't have IOV_MAX exposed. */ 58 #ifndef IOV_MAX 59 #if defined(__FreeBSD__) 60 # define IOV_MAX 1024 61 #endif 62 #endif 63 64 #include "memcached.h" 65 66 struct stats stats; 67 struct settings settings; 68 69 static item **todelete = 0; 70 static int delcurr; 71 static int deltotal; 72 73 #define TRANSMIT_COMPLETE 0 74 #define TRANSMIT_INCOMPLETE 1 75 #define TRANSMIT_SOFT_ERROR 2 76 #define TRANSMIT_HARD_ERROR 3 77 78 int *buckets = 0; /* bucket->generation array for a managed instance */ 79 80 #define REALTIME_MAXDELTA 60*60*24*30 81 rel_time_t realtime(time_t exptime) { 82 /* no. of seconds in 30 days - largest possible delta exptime */ 83 84 if (exptime == 0) return 0; /* 0 means never expire */ 85 86 if (exptime > REALTIME_MAXDELTA) 87 return (rel_time_t) (exptime - stats.started); 88 else { 89 return (rel_time_t) (exptime + current_time); 90 } 91 } 92 93 void stats_init(void) { 94 stats.curr_items = stats.total_items = stats.curr_conns = stats.total_conns = stats.conn_structs = 0; 95 stats.get_cmds = stats.set_cmds = stats.get_hits = stats.get_misses = 0; 96 stats.curr_bytes = stats.bytes_read = stats.bytes_written = 0; 97 98 /* make the time we started always be 1 second before we really 99 did, so time(0) - time.started is never zero. if so, things 100 like 'settings.oldest_live' which act as booleans as well as 101 values are now false in boolean context... */ 102 stats.started = time(0) - 1; 103 } 104 void stats_reset(void) { 105 stats.total_items = stats.total_conns = 0; 106 stats.get_cmds = stats.set_cmds = stats.get_hits = stats.get_misses = 0; 107 stats.bytes_read = stats.bytes_written = 0; 108 } 109 110 void settings_init(void) { 111 settings.port = 11211; 112 settings.udpport = 11211; 113 settings.interface.s_addr = htonl(INADDR_ANY); 114 settings.maxbytes = 64*1024*1024; /* default is 64MB */ 115 settings.maxconns = 1024; /* to limit connections-related memory to about 5MB */ 116 settings.verbose = 0; 117 settings.oldest_live = 0; 118 settings.evict_to_free = 1; /* push old items out of cache when memory runs out */ 119 settings.socketpath = NULL; /* by default, not using a unix socket */ 120 settings.managed = 0; 121 settings.factor = 1.25; 122 settings.chunk_size = 48; /* space for a modest key and value */ 123 } 124 125 /* returns true if a deleted item's delete-locked-time is over, and it 126 should be removed from the namespace */ 127 int item_delete_lock_over (item *it) { 128 assert(it->it_flags & ITEM_DELETED); 129 return (current_time >= it->exptime); 130 } 131 132 /* wrapper around assoc_find which does the lazy expiration/deletion logic */ 133 item *get_item_notedeleted(char *key, int *delete_locked) { 134 item *it = assoc_find(key); 135 if (delete_locked) *delete_locked = 0; 136 if (it && (it->it_flags & ITEM_DELETED)) { 137 /* it's flagged as delete-locked. let's see if that condition 138 is past due, and the 5-second delete_timer just hasn't 139 gotten to it yet... */ 140 if (! item_delete_lock_over(it)) { 141 if (delete_locked) *delete_locked = 1; 142 it = 0; 143 } 144 } 145 if (it && settings.oldest_live && settings.oldest_live <= current_time && 146 it->time <= settings.oldest_live) { 147 item_unlink(it); 148 it = 0; 149 } 150 if (it && it->exptime && it->exptime <= current_time) { 151 item_unlink(it); 152 it = 0; 153 } 154 return it; 155 } 156 157 item *get_item(char *key) { 158 return get_item_notedeleted(key, 0); 159 } 160 161 /* 162 * Adds a message header to a connection. 163 * 164 * Returns 0 on success, -1 on out-of-memory. 165 */ 166 int add_msghdr(conn *c) 167 { 168 struct msghdr *msg; 169 170 if (c->msgsize == c->msgused) { 171 msg = realloc(c->msglist, c->msgsize * 2 * sizeof(struct msghdr)); 172 if (! msg) 173 return -1; 174 c->msglist = msg; 175 c->msgsize *= 2; 176 } 177 178 msg = c->msglist + c->msgused; 179 180 /* this wipes msg_iovlen, msg_control, msg_controllen, and 181 msg_flags, the last 3 of which aren't defined on solaris: */ 182 memset(msg, 0, sizeof(struct msghdr)); 183 184 msg->msg_iov = &c->iov[c->iovused]; 185 msg->msg_name = &c->request_addr; 186 msg->msg_namelen = c->request_addr_size; 187 188 c->msgbytes = 0; 189 c->msgused++; 190 191 if (c->udp) { 192 /* Leave room for the UDP header, which we'll fill in later. */ 193 return add_iov(c, NULL, UDP_HEADER_SIZE); 194 } 195 196 return 0; 197 } 198 199 conn **freeconns; 200 int freetotal; 201 int freecurr; 202 203 void conn_init(void) { 204 freetotal = 200; 205 freecurr = 0; 206 freeconns = (conn **)malloc(sizeof (conn *)*freetotal); 207 return; 208 } 209 210 conn *conn_new(int sfd, int init_state, int event_flags, int read_buffer_size, 211 int is_udp) { 212 conn *c; 213 214 /* do we have a free conn structure from a previous close? */ 215 if (freecurr > 0) { 216 c = freeconns[--freecurr]; 217 } else { /* allocate a new one */ 218 if (!(c = (conn *)malloc(sizeof(conn)))) { 219 perror("malloc()"); 220 return 0; 221 } 222 c->rbuf = c->wbuf = 0; 223 c->ilist = 0; 224 c->iov = 0; 225 c->msglist = 0; 226 c->hdrbuf = 0; 227 228 c->rsize = read_buffer_size; 229 c->wsize = DATA_BUFFER_SIZE; 230 c->isize = ITEM_LIST_INITIAL; 231 c->iovsize = IOV_LIST_INITIAL; 232 c->msgsize = MSG_LIST_INITIAL; 233 c->hdrsize = 0; 234 235 c->rbuf = (char *) malloc(c->rsize); 236 c->wbuf = (char *) malloc(c->wsize); 237 c->ilist = (item **) malloc(sizeof(item *) * c->isize); 238 c->iov = (struct iovec *) malloc(sizeof(struct iovec) * c->iovsize); 239 c->msglist = (struct msghdr *) malloc(sizeof(struct msghdr) * c->msgsize); 240 241 if (c->rbuf == 0 || c->wbuf == 0 || c->ilist == 0 || c->iov == 0 || 242 c->msglist == 0) { 243 if (c->rbuf != 0) free(c->rbuf); 244 if (c->wbuf != 0) free(c->wbuf); 245 if (c->ilist !=0) free(c->ilist); 246 if (c->iov != 0) free(c->iov); 247 if (c->msglist != 0) free(c->msglist); 248 free(c); 249 perror("malloc()"); 250 return 0; 251 } 252 253 stats.conn_structs++; 254 } 255 256 if (settings.verbose > 1) { 257 if (init_state == conn_listening) 258 fprintf(stderr, "<%d server listening\n", sfd); 259 else if (is_udp) 260 fprintf(stderr, "<%d server listening (udp)\n", sfd); 261 else 262 fprintf(stderr, "<%d new client connection\n", sfd); 263 } 264 265 c->sfd = sfd; 266 c->udp = is_udp; 267 c->state = init_state; 268 c->rlbytes = 0; 269 c->rbytes = c->wbytes = 0; 270 c->wcurr = c->wbuf; 271 c->rcurr = c->rbuf; 272 c->ritem = 0; 273 c->icurr = c->ilist; 274 c->ileft = 0; 275 c->iovused = 0; 276 c->msgcurr = 0; 277 c->msgused = 0; 278 279 c->write_and_go = conn_read; 280 c->write_and_free = 0; 281 c->item = 0; 282 c->bucket = -1; 283 c->gen = 0; 284 285 event_set(&c->event, sfd, event_flags, event_handler, (void *)c); 286 c->ev_flags = event_flags; 287 288 if (event_add(&c->event, 0) == -1) { 289 if (freecurr < freetotal) { 290 freeconns[freecurr++] = c; 291 } else { 292 if (c->hdrbuf) 293 free (c->hdrbuf); 294 free (c->msglist); 295 free (c->rbuf); 296 free (c->wbuf); 297 free (c->ilist); 298 free (c->iov); 299 free (c); 300 } 301 return 0; 302 } 303 304 stats.curr_conns++; 305 stats.total_conns++; 306 307 return c; 308 } 309 310 void conn_cleanup(conn *c) { 311 if (c->item) { 312 item_free(c->item); 313 c->item = 0; 314 } 315 316 if (c->ileft) { 317 for (; c->ileft > 0; c->ileft--,c->icurr++) { 318 item_remove(*(c->icurr)); 319 } 320 } 321 322 if (c->write_and_free) { 323 free(c->write_and_free); 324 c->write_and_free = 0; 325 } 326 } 327 328 /* 329 * Frees a connection. 330 */ 331 static void conn_free(conn *c) { 332 if (c) { 333 if (c->hdrbuf) 334 free(c->hdrbuf); 335 if (c->msglist) 336 free(c->msglist); 337 if (c->rbuf) 338 free(c->rbuf); 339 if (c->wbuf) 340 free(c->wbuf); 341 if (c->ilist) 342 free(c->ilist); 343 if (c->iov) 344 free(c->iov); 345 free(c); 346 } 347 } 348 349 void conn_close(conn *c) { 350 /* delete the event, the socket and the conn */ 351 event_del(&c->event); 352 353 if (settings.verbose > 1) 354 fprintf(stderr, "<%d connection closed.\n", c->sfd); 355 356 close(c->sfd); 357 conn_cleanup(c); 358 359 /* if the connection has big buffers, just free it */ 360 if (c->rsize > READ_BUFFER_HIGHWAT) { 361 conn_free(c); 362 } else if (freecurr < freetotal) { 363 /* if we have enough space in the free connections array, put the structure there */ 364 freeconns[freecurr++] = c; 365 } else { 366 /* try to enlarge free connections array */ 367 conn **new_freeconns = realloc(freeconns, sizeof(conn *)*freetotal*2); 368 if (new_freeconns) { 369 freetotal *= 2; 370 freeconns = new_freeconns; 371 freeconns[freecurr++] = c; 372 } else { 373 conn_free(c); 374 } 375 } 376 377 stats.curr_conns--; 378 379 return; 380 } 381 382 /* 383 * Reallocates memory and updates a buffer size if successful. 384 */ 385 int do_realloc(void **orig, int newsize, int bytes_per_item, int *size) { 386 void *newbuf = realloc(*orig, newsize * bytes_per_item); 387 if (newbuf) { 388 *orig = newbuf; 389 *size = newsize; 390 return 1; 391 } 392 return 0; 393 } 394 395 /* 396 * Shrinks a connection's buffers if they're too big. This prevents 397 * periodic large "get" requests from permanently chewing lots of server 398 * memory. 399 * 400 * This should only be called in between requests since it can wipe output 401 * buffers! 402 */ 403 void conn_shrink(conn *c) { 404 if (c->udp) 405 return; 406 407 if (c->rsize > READ_BUFFER_HIGHWAT && c->rbytes < DATA_BUFFER_SIZE) { 408 do_realloc((void **)&c->rbuf, DATA_BUFFER_SIZE, 1, &c->rsize); 409 } 410 411 if (c->isize > ITEM_LIST_HIGHWAT) { 412 do_realloc((void **)&c->ilist, ITEM_LIST_INITIAL, sizeof(c->ilist[0]), &c->isize); 413 } 414 415 if (c->msgsize > MSG_LIST_HIGHWAT) { 416 do_realloc((void **)&c->msglist, MSG_LIST_INITIAL, sizeof(c->msglist[0]), &c->msgsize); 417 } 418 419 if (c->iovsize > IOV_LIST_HIGHWAT) { 420 do_realloc((void **)&c->iov, IOV_LIST_INITIAL, sizeof(c->iov[0]), &c->iovsize); 421 } 422 } 423 424 /* 425 * Sets a connection's current state in the state machine. Any special 426 * processing that needs to happen on certain state transitions can 427 * happen here. 428 */ 429 void conn_set_state(conn *c, int state) { 430 if (state != c->state) { 431 if (state == conn_read) { 432 conn_shrink(c); 433 } 434 c->state = state; 435 } 436 } 437 438 439 /* 440 * Ensures that there is room for another struct iovec in a connection's 441 * iov list. 442 * 443 * Returns 0 on success, -1 on out-of-memory. 444 */ 445 int ensure_iov_space(conn *c) { 446 if (c->iovused >= c->iovsize) { 447 int i, iovnum; 448 struct iovec *new_iov = (struct iovec *) realloc(c->iov, 449 (c->iovsize * 2) * sizeof(struct iovec)); 450 if (! new_iov) 451 return -1; 452 c->iov = new_iov; 453 c->iovsize *= 2; 454 455 /* Point all the msghdr structures at the new list. */ 456 for (i = 0, iovnum = 0; i < c->msgused; i++) { 457 c->msglist[i].msg_iov = &c->iov[iovnum]; 458 iovnum += c->msglist[i].msg_iovlen; 459 } 460 } 461 462 return 0; 463 } 464 465 466 /* 467 * Adds data to the list of pending data that will be written out to a 468 * connection. 469 * 470 * Returns 0 on success, -1 on out-of-memory. 471 */ 472 473 int add_iov(conn *c, const void *buf, int len) { 474 struct msghdr *m; 475 int i; 476 int leftover; 477 int limit_to_mtu; 478 479 do { 480 m = &c->msglist[c->msgused - 1]; 481 482 /* 483 * Limit UDP packets, and the first payloads of TCP replies, to 484 * UDP_MAX_PAYLOAD_SIZE bytes. 485 */ 486 limit_to_mtu = c->udp || (1 == c->msgused); 487 488 /* We may need to start a new msghdr if this one is full. */ 489 if (m->msg_iovlen == IOV_MAX || 490 limit_to_mtu && c->msgbytes >= UDP_MAX_PAYLOAD_SIZE) { 491 add_msghdr(c); 492 m = &c->msglist[c->msgused - 1]; 493 } 494 495 if (ensure_iov_space(c)) 496 return -1; 497 498 /* If the fragment is too big to fit in the datagram, split it up */ 499 if (limit_to_mtu && len + c->msgbytes > UDP_MAX_PAYLOAD_SIZE) { 500 leftover = len + c->msgbytes - UDP_MAX_PAYLOAD_SIZE; 501 len -= leftover; 502 } else { 503 leftover = 0; 504 } 505 506 m = &c->msglist[c->msgused - 1]; 507 m->msg_iov[m->msg_iovlen].iov_base = (void*) buf; 508 m->msg_iov[m->msg_iovlen].iov_len = len; 509 510 c->msgbytes += len; 511 c->iovused++; 512 m->msg_iovlen++; 513 514 buf = ((char *)buf) + len; 515 len = leftover; 516 } while (leftover > 0); 517 518 return 0; 519 } 520 521 522 /* 523 * Constructs a set of UDP headers and attaches them to the outgoing messages. 524 */ 525 int build_udp_headers(conn *c) { 526 int i; 527 unsigned char *hdr; 528 529 if (c->msgused > c->hdrsize) { 530 void *new_hdrbuf; 531 if (c->hdrbuf) 532 new_hdrbuf = realloc(c->hdrbuf, c->msgused * 2 * UDP_HEADER_SIZE); 533 else 534 new_hdrbuf = malloc(c->msgused * 2 * UDP_HEADER_SIZE); 535 if (! new_hdrbuf) 536 return -1; 537 c->hdrbuf = (unsigned char *) new_hdrbuf; 538 c->hdrsize = c->msgused * 2; 539 } 540 541 hdr = c->hdrbuf; 542 for (i = 0; i < c->msgused; i++) { 543 c->msglist[i].msg_iov[0].iov_base = hdr; 544 c->msglist[i].msg_iov[0].iov_len = UDP_HEADER_SIZE; 545 *hdr++ = c->request_id / 256; 546 *hdr++ = c->request_id % 256; 547 *hdr++ = i / 256; 548 *hdr++ = i % 256; 549 *hdr++ = c->msgused / 256; 550 *hdr++ = c->msgused % 256; 551 *hdr++ = 0; 552 *hdr++ = 0; 553 assert((void*) hdr == (void*) c->msglist[i].msg_iov[0].iov_base + UDP_HEADER_SIZE); 554 } 555 556 return 0; 557 } 558 559 560 void out_string(conn *c, char *str) { 561 int len; 562 563 if (settings.verbose > 1) 564 fprintf(stderr, ">%d %s\n", c->sfd, str); 565 566 len = strlen(str); 567 if (len + 2 > c->wsize) { 568 /* ought to be always enough. just fail for simplicity */ 569 str = "SERVER_ERROR output line too long"; 570 len = strlen(str); 571 } 572 573 strcpy(c->wbuf, str); 574 strcpy(c->wbuf + len, "\r\n"); 575 c->wbytes = len + 2; 576 c->wcurr = c->wbuf; 577 578 conn_set_state(c, conn_write); 579 c->write_and_go = conn_read; 580 return; 581 } 582 583 /* 584 * we get here after reading the value in set/add/replace commands. The command 585 * has been stored in c->item_comm, and the item is ready in c->item. 586 */ 587 588 void complete_nread(conn *c) { 589 item *it = c->item; 590 int comm = c->item_comm; 591 item *old_it; 592 int delete_locked = 0; 593 char *key = ITEM_key(it); 594 595 stats.set_cmds++; 596 597 if (strncmp(ITEM_data(it) + it->nbytes - 2, "\r\n", 2) != 0) { 598 out_string(c, "CLIENT_ERROR bad data chunk"); 599 goto err; 600 } 601 602 old_it = get_item_notedeleted(key, &delete_locked); 603 604 if (old_it && comm == NREAD_ADD) { 605 item_update(old_it); /* touches item, promotes to head of LRU */ 606 out_string(c, "NOT_STORED"); 607 goto err; 608 } 609 610 if (!old_it && comm == NREAD_REPLACE) { 611 out_string(c, "NOT_STORED"); 612 goto err; 613 } 614 615 if (delete_locked) { 616 if (comm == NREAD_REPLACE || comm == NREAD_ADD) { 617 out_string(c, "NOT_STORED"); 618 goto err; 619 } 620 621 /* but "set" commands can override the delete lock 622 window... in which case we have to find the old hidden item 623 that's in the namespace/LRU but wasn't returned by 624 get_item.... because we need to replace it (below) */ 625 old_it = assoc_find(key); 626 } 627 628 if (old_it) 629 item_replace(old_it, it); 630 else 631 item_link(it); 632 633 c->item = 0; 634 out_string(c, "STORED"); 635 return; 636 637 err: 638 item_free(it); 639 c->item = 0; 640 return; 641 } 642 643 void process_stat(conn *c, char *command) { 644 rel_time_t now = current_time; 645 646 if (strcmp(command, "stats") == 0) { 647 char temp[1024]; 648 pid_t pid = getpid(); 649 char *pos = temp; 650 struct rusage usage; 651 652 getrusage(RUSAGE_SELF, &usage); 653 654 pos += sprintf(pos, "STAT pid %u\r\n", pid); 655 pos += sprintf(pos, "STAT uptime %u\r\n", now); 656 pos += sprintf(pos, "STAT time %ld\r\n", now + stats.started); 657 pos += sprintf(pos, "STAT version " VERSION "\r\n"); 658 pos += sprintf(pos, "STAT rusage_user %ld.%06ld\r\n", usage.ru_utime.tv_sec, usage.ru_utime.tv_usec); 659 pos += sprintf(pos, "STAT rusage_system %ld.%06ld\r\n", usage.ru_stime.tv_sec, usage.ru_stime.tv_usec); 660 pos += sprintf(pos, "STAT curr_items %u\r\n", stats.curr_items); 661 pos += sprintf(pos, "STAT total_items %u\r\n", stats.total_items); 662 pos += sprintf(pos, "STAT bytes %llu\r\n", stats.curr_bytes); 663 pos += sprintf(pos, "STAT curr_connections %u\r\n", stats.curr_conns - 1); /* ignore listening conn */ 664 pos += sprintf(pos, "STAT total_connections %u\r\n", stats.total_conns); 665 pos += sprintf(pos, "STAT connection_structures %u\r\n", stats.conn_structs); 666 pos += sprintf(pos, "STAT cmd_get %llu\r\n", stats.get_cmds); 667 pos += sprintf(pos, "STAT cmd_set %llu\r\n", stats.set_cmds); 668 pos += sprintf(pos, "STAT get_hits %llu\r\n", stats.get_hits); 669 pos += sprintf(pos, "STAT get_misses %llu\r\n", stats.get_misses); 670 pos += sprintf(pos, "STAT bytes_read %llu\r\n", stats.bytes_read); 671 pos += sprintf(pos, "STAT bytes_written %llu\r\n", stats.bytes_written); 672 pos += sprintf(pos, "STAT limit_maxbytes %llu\r\n", (unsigned long long) settings.maxbytes); 673 pos += sprintf(pos, "END"); 674 out_string(c, temp); 675 return; 676 } 677 678 if (strcmp(command, "stats reset") == 0) { 679 stats_reset(); 680 out_string(c, "RESET"); 681 return; 682 } 683 684 #ifdef HAVE_MALLOC_H 685 #ifdef HAVE_STRUCT_MALLINFO 686 if (strcmp(command, "stats malloc") == 0) { 687 char temp[512]; 688 struct mallinfo info; 689 char *pos = temp; 690 691 info = mallinfo(); 692 pos += sprintf(pos, "STAT arena_size %d\r\n", info.arena); 693 pos += sprintf(pos, "STAT free_chunks %d\r\n", info.ordblks); 694 pos += sprintf(pos, "STAT fastbin_blocks %d\r\n", info.smblks); 695 pos += sprintf(pos, "STAT mmapped_regions %d\r\n", info.hblks); 696 pos += sprintf(pos, "STAT mmapped_space %d\r\n", info.hblkhd); 697 pos += sprintf(pos, "STAT max_total_alloc %d\r\n", info.usmblks); 698 pos += sprintf(pos, "STAT fastbin_space %d\r\n", info.fsmblks); 699 pos += sprintf(pos, "STAT total_alloc %d\r\n", info.uordblks); 700 pos += sprintf(pos, "STAT total_free %d\r\n", info.fordblks); 701 pos += sprintf(pos, "STAT releasable_space %d\r\nEND", info.keepcost); 702 out_string(c, temp); 703 return; 704 } 705 #endif /* HAVE_STRUCT_MALLINFO */ 706 #endif /* HAVE_MALLOC_H */ 707 708 if (strcmp(command, "stats maps") == 0) { 709 char *wbuf; 710 int wsize = 8192; /* should be enough */ 711 int fd; 712 int res; 713 714 wbuf = (char *)malloc(wsize); 715 if (wbuf == 0) { 716 out_string(c, "SERVER_ERROR out of memory"); 717 return; 718 } 719 720 fd = open("/proc/self/maps", O_RDONLY); 721 if (fd == -1) { 722 out_string(c, "SERVER_ERROR cannot open the maps file"); 723 free(wbuf); 724 return; 725 } 726 727 res = read(fd, wbuf, wsize - 6); /* 6 = END\r\n\0 */ 728 if (res == wsize - 6) { 729 out_string(c, "SERVER_ERROR buffer overflow"); 730 free(wbuf); close(fd); 731 return; 732 } 733 if (res == 0 || res == -1) { 734 out_string(c, "SERVER_ERROR can't read the maps file"); 735 free(wbuf); close(fd); 736 return; 737 } 738 strcpy(wbuf + res, "END\r\n"); 739 c->write_and_free=wbuf; 740 c->wcurr=wbuf; 741 c->wbytes = res + 6; 742 conn_set_state(c, conn_write); 743 c->write_and_go = conn_read; 744 close(fd); 745 return; 746 } 747 748 if (strncmp(command, "stats cachedump", 15) == 0) { 749 char *buf; 750 unsigned int bytes, id, limit = 0; 751 char *start = command + 15; 752 if (sscanf(start, "%u %u\r\n", &id, &limit) < 1) { 753 out_string(c, "CLIENT_ERROR bad command line"); 754 return; 755 } 756 757 buf = item_cachedump(id, limit, &bytes); 758 if (buf == 0) { 759 out_string(c, "SERVER_ERROR out of memory"); 760 return; 761 } 762 763 c->write_and_free = buf; 764 c->wcurr = buf; 765 c->wbytes = bytes; 766 conn_set_state(c, conn_write); 767 c->write_and_go = conn_read; 768 return; 769 } 770 771 if (strcmp(command, "stats slabs")==0) { 772 int bytes = 0; 773 char *buf = slabs_stats(&bytes); 774 if (!buf) { 775 out_string(c, "SERVER_ERROR out of memory"); 776 return; 777 } 778 c->write_and_free = buf; 779 c->wcurr = buf; 780 c->wbytes = bytes; 781 conn_set_state(c, conn_write); 782 c->write_and_go = conn_read; 783 return; 784 } 785 786 if (strcmp(command, "stats items")==0) { 787 char buffer[4096]; 788 item_stats(buffer, 4096); 789 out_string(c, buffer); 790 return; 791 } 792 793 if (strcmp(command, "stats sizes")==0) { 794 int bytes = 0; 795 char *buf = item_stats_sizes(&bytes); 796 if (! buf) { 797 out_string(c, "SERVER_ERROR out of memory"); 798 return; 799 } 800 801 c->write_and_free = buf; 802 c->wcurr = buf; 803 c->wbytes = bytes; 804 conn_set_state(c, conn_write); 805 c->write_and_go = conn_read; 806 return; 807 } 808 809 out_string(c, "ERROR"); 810 } 811 812 void process_command(conn *c, char *command) { 813 814 int comm = 0; 815 int incr = 0; 816 817 /* 818 * for commands set/add/replace, we build an item and read the data 819 * directly into it, then continue in nread_complete(). 820 */ 821 822 if (settings.verbose > 1) 823 fprintf(stderr, "<%d %s\n", c->sfd, command); 824 825 c->msgcurr = 0; 826 c->msgused = 0; 827 c->iovused = 0; 828 if (add_msghdr(c)) { 829 out_string(c, "SERVER_ERROR out of memory"); 830 return; 831 } 832 833 if ((strncmp(command, "add ", 4) == 0 && (comm = NREAD_ADD)) || 834 (strncmp(command, "set ", 4) == 0 && (comm = NREAD_SET)) || 835 (strncmp(command, "replace ", 8) == 0 && (comm = NREAD_REPLACE))) { 836 837 char key[251]; 838 int flags; 839 time_t expire; 840 int len, res; 841 item *it; 842 843 res = sscanf(command, "%*s %250s %u %ld %d\n", key, &flags, &expire, &len); 844 if (res!=4 || strlen(key)==0 ) { 845 out_string(c, "CLIENT_ERROR bad command line format"); 846 return; 847 } 848 849 if (settings.managed) { 850 int bucket = c->bucket; 851 if (bucket == -1) { 852 out_string(c, "CLIENT_ERROR no BG data in managed mode"); 853 return; 854 } 855 c->bucket = -1; 856 if (buckets[bucket] != c->gen) { 857 out_string(c, "ERROR_NOT_OWNER"); 858 return; 859 } 860 } 861 862 expire = realtime(expire); 863 it = item_alloc(key, flags, expire, len+2); 864 865 if (it == 0) { 866 if (! item_size_ok(key, flags, len + 2)) 867 out_string(c, "SERVER_ERROR object too large for cache"); 868 else 869 out_string(c, "SERVER_ERROR out of memory"); 870 /* swallow the data line */ 871 c->write_and_go = conn_swallow; 872 c->sbytes = len+2; 873 return; 874 } 875 876 c->item_comm = comm; 877 c->item = it; 878 c->ritem = ITEM_data(it); 879 c->rlbytes = it->nbytes; 880 conn_set_state(c, conn_nread); 881 return; 882 } 883 884 if ((strncmp(command, "incr ", 5) == 0 && (incr = 1)) || 885 (strncmp(command, "decr ", 5) == 0)) { 886 char temp[32]; 887 unsigned int value; 888 item *it; 889 unsigned int delta; 890 char key[251]; 891 int res; 892 char *ptr; 893 res = sscanf(command, "%*s %250s %u\n", key, &delta); 894 if (res!=2 || strlen(key)==0 ) { 895 out_string(c, "CLIENT_ERROR bad command line format"); 896 return; 897 } 898 if (settings.managed) { 899 int bucket = c->bucket; 900 if (bucket == -1) { 901 out_string(c, "CLIENT_ERROR no BG data in managed mode"); 902 return; 903 } 904 c->bucket = -1; 905 if (buckets[bucket] != c->gen) { 906 out_string(c, "ERROR_NOT_OWNER"); 907 return; 908 } 909 } 910 it = get_item(key); 911 if (!it) { 912 out_string(c, "NOT_FOUND"); 913 return; 914 } 915 ptr = ITEM_data(it); 916 while (*ptr && (*ptr<'0' && *ptr>'9')) ptr++; // BUG: can't be true 917 value = atoi(ptr); 918 if (incr) 919 value+=delta; 920 else { 921 if (delta >= value) value = 0; 922 else value-=delta; 923 } 924 sprintf(temp, "%u", value); 925 res = strlen(temp); 926 if (res + 2 > it->nbytes) { /* need to realloc */ 927 item *new_it; 928 new_it = item_alloc(ITEM_key(it), atoi(ITEM_suffix(it) + 1), it->exptime, res + 2 ); 929 if (new_it == 0) { 930 out_string(c, "SERVER_ERROR out of memory"); 931 return; 932 } 933 memcpy(ITEM_data(new_it), temp, res); 934 memcpy(ITEM_data(new_it) + res, "\r\n", 2); 935 item_replace(it, new_it); 936 } else { /* replace in-place */ 937 memcpy(ITEM_data(it), temp, res); 938 memset(ITEM_data(it) + res, ' ', it->nbytes-res-2); 939 } 940 out_string(c, temp); 941 return; 942 } 943 944 if (strncmp(command, "bget ", 5) == 0) { 945 c->binary = 1; 946 goto get; 947 } 948 if (strncmp(command, "get ", 4) == 0) { 949 950 char *start = command + 4; 951 char key[251]; 952 int next; 953 int i; 954 item *it; 955 rel_time_t now; 956 get: 957 now = current_time; 958 i = 0; 959 960 if (settings.managed) { 961 int bucket = c->bucket; 962 if (bucket == -1) { 963 out_string(c, "CLIENT_ERROR no BG data in managed mode"); 964 return; 965 } 966 c->bucket = -1; 967 if (buckets[bucket] != c->gen) { 968 out_string(c, "ERROR_NOT_OWNER"); 969 return; 970 } 971 } 972 973 while(sscanf(start, " %250s%n", key, &next) >= 1) { 974 start+=next; 975 stats.get_cmds++; 976 it = get_item(key); 977 if (it) { 978 if (i >= c->isize) { 979 item **new_list = realloc(c->ilist, sizeof(item *)*c->isize*2); 980 if (new_list) { 981 c->isize *= 2; 982 c->ilist = new_list; 983 } else break; 984 } 985 986 /* 987 * Construct the response. Each hit adds three elements to the 988 * outgoing data list: 989 * "VALUE " 990 * key 991 * " " + flags + " " + data length + "\r\n" + data (with \r\n) 992 */ 993 /* TODO: can we avoid the strlen() func call and cache that in wasted byte in item struct? */ 994 if (add_iov(c, "VALUE ", 6) || 995 add_iov(c, ITEM_key(it), strlen(ITEM_key(it))) || 996 add_iov(c, ITEM_suffix(it), it->nsuffix + it->nbytes)) 997 { 998 break; 999 } 1000 if (settings.verbose > 1) 1001 fprintf(stderr, ">%d sending key %s\n", c->sfd, ITEM_key(it)); 1002 1003 stats.get_hits++; 1004 it->refcount++; 1005 item_update(it); 1006 *(c->ilist + i) = it; 1007 i++; 1008 } else stats.get_misses++; 1009 } 1010 1011 c->icurr = c->ilist; 1012 c->ileft = i; 1013 1014 if (settings.verbose > 1) 1015 fprintf(stderr, ">%d END\n", c->sfd); 1016 add_iov(c, "END\r\n", 5); 1017 1018 if (c->udp && build_udp_headers(c)) { 1019 out_string(c, "SERVER_ERROR out of memory"); 1020 } 1021 else { 1022 conn_set_state(c, conn_mwrite); 1023 c->msgcurr = 0; 1024 } 1025 return; 1026 } 1027 1028 if (strncmp(command, "delete ", 7) == 0) { 1029 char key[251]; 1030 item *it; 1031 int res; 1032 time_t exptime = 0; 1033 1034 if (settings.managed) { 1035 int bucket = c->bucket; 1036 if (bucket == -1) { 1037 out_string(c, "CLIENT_ERROR no BG data in managed mode"); 1038 return; 1039 } 1040 c->bucket = -1; 1041 if (buckets[bucket] != c->gen) { 1042 out_string(c, "ERROR_NOT_OWNER"); 1043 return; 1044 } 1045 } 1046 res = sscanf(command, "%*s %250s %ld", key, &exptime); 1047 it = get_item(key); 1048 if (!it) { 1049 out_string(c, "NOT_FOUND"); 1050 return; 1051 } 1052 if (exptime == 0) { 1053 item_unlink(it); 1054 out_string(c, "DELETED"); 1055 return; 1056 } 1057 if (delcurr >= deltotal) { 1058 item **new_delete = realloc(todelete, sizeof(item *) * deltotal * 2); 1059 if (new_delete) { 1060 todelete = new_delete; 1061 deltotal *= 2; 1062 } else { 1063 /* 1064 * can't delete it immediately, user wants a delay, 1065 * but we ran out of memory for the delete queue 1066 */ 1067 out_string(c, "SERVER_ERROR out of memory"); 1068 return; 1069 } 1070 } 1071 1072 it->refcount++; 1073 /* use its expiration time as its deletion time now */ 1074 it->exptime = realtime(exptime); 1075 it->it_flags |= ITEM_DELETED; 1076 todelete[delcurr++] = it; 1077 out_string(c, "DELETED"); 1078 return; 1079 } 1080 1081 if (strncmp(command, "own ", 4) == 0) { 1082 int bucket, gen; 1083 char *start = command+4; 1084 if (!settings.managed) { 1085 out_string(c, "CLIENT_ERROR not a managed instance"); 1086 return; 1087 } 1088 if (sscanf(start, "%u:%u\r\n", &bucket,&gen) == 2) { 1089 if ((bucket < 0) || (bucket >= MAX_BUCKETS)) { 1090 out_string(c, "CLIENT_ERROR bucket number out of range"); 1091 return; 1092 } 1093 buckets[bucket] = gen; 1094 out_string(c, "OWNED"); 1095 return; 1096 } else { 1097 out_string(c, "CLIENT_ERROR bad format"); 1098 return; 1099 } 1100 } 1101 1102 if (strncmp(command, "disown ", 7) == 0) { 1103 int bucket; 1104 char *start = command+7; 1105 if (!settings.managed) { 1106 out_string(c, "CLIENT_ERROR not a managed instance"); 1107 return; 1108 } 1109 if (sscanf(start, "%u\r\n", &bucket) == 1) { 1110 if ((bucket < 0) || (bucket >= MAX_BUCKETS)) { 1111 out_string(c, "CLIENT_ERROR bucket number out of range"); 1112 return; 1113 } 1114 buckets[bucket] = 0; 1115 out_string(c, "DISOWNED"); 1116 return; 1117 } else { 1118 out_string(c, "CLIENT_ERROR bad format"); 1119 return; 1120 } 1121 } 1122 1123 if (strncmp(command, "bg ", 3) == 0) { 1124 int bucket, gen; 1125 char *start = command+3; 1126 if (!settings.managed) { 1127 out_string(c, "CLIENT_ERROR not a managed instance"); 1128 return; 1129 } 1130 if (sscanf(start, "%u:%u\r\n", &bucket,&gen) == 2) { 1131 /* we never write anything back, even if input's wrong */ 1132 if ((bucket < 0) || (bucket >= MAX_BUCKETS) || (gen<=0)) { 1133 /* do nothing, bad input */ 1134 } else { 1135 c->bucket = bucket; 1136 c->gen = gen; 1137 } 1138 conn_set_state(c, conn_read); 1139 return; 1140 } else { 1141 out_string(c, "CLIENT_ERROR bad format"); 1142 return; 1143 } 1144 } 1145 1146 if (strncmp(command, "stats", 5) == 0) { 1147 process_stat(c, command); 1148 return; 1149 } 1150 1151 if (strncmp(command, "flush_all", 9) == 0) { 1152 time_t exptime = 0; 1153 int res; 1154 set_current_time(); 1155 1156 if (strcmp(command, "flush_all") == 0) { 1157 settings.oldest_live = current_time; 1158 out_string(c, "OK"); 1159 return; 1160 } 1161 1162 res = sscanf(command, "%*s %ld", &exptime); 1163 if (res != 1) { 1164 out_string(c, "ERROR"); 1165 return; 1166 } 1167 1168 settings.oldest_live = realtime(exptime); 1169 out_string(c, "OK"); 1170 return; 1171 } 1172 1173 if (strcmp(command, "version") == 0) { 1174 out_string(c, "VERSION " VERSION); 1175 return; 1176 } 1177 1178 if (strcmp(command, "quit") == 0) { 1179 conn_set_state(c, conn_closing); 1180 return; 1181 } 1182 1183 if (strncmp(command, "slabs reassign ", 15) == 0) { 1184 #ifdef ALLOW_SLABS_REASSIGN 1185 int src, dst; 1186 char *start = command+15; 1187 if (sscanf(start, "%u %u\r\n", &src, &dst) == 2) { 1188 int rv = slabs_reassign(src, dst); 1189 if (rv == 1) { 1190 out_string(c, "DONE"); 1191 return; 1192 } 1193 if (rv == 0) { 1194 out_string(c, "CANT"); 1195 return; 1196 } 1197 if (rv == -1) { 1198 out_string(c, "BUSY"); 1199 return; 1200 } 1201 } 1202 out_string(c, "CLIENT_ERROR bogus command"); 1203 #else 1204 out_string(c, "CLIENT_ERROR Slab reassignment not supported"); 1205 #endif 1206 return; 1207 } 1208 1209 out_string(c, "ERROR"); 1210 return; 1211 } 1212 1213 /* 1214 * if we have a complete line in the buffer, process it. 1215 */ 1216 int try_read_command(conn *c) { 1217 char *el, *cont; 1218 1219 if (!c->rbytes) 1220 return 0; 1221 el = memchr(c->rcurr, '\n', c->rbytes); 1222 if (!el) 1223 return 0; 1224 cont = el + 1; 1225 if (el - c->rcurr > 1 && *(el - 1) == '\r') { 1226 el--; 1227 } 1228 *el = '\0'; 1229 1230 process_command(c, c->rcurr); 1231 1232 c->rbytes -= (cont - c->rcurr); 1233 c->rcurr = cont; 1234 1235 return 1; 1236 } 1237 1238 /* 1239 * read a UDP request. 1240 * return 0 if there's nothing to read. 1241 */ 1242 int try_read_udp(conn *c) { 1243 int res; 1244 1245 c->request_addr_size = sizeof(c->request_addr); 1246 res = recvfrom(c->sfd, c->rbuf, c->rsize, 1247 0, &c->request_addr, &c->request_addr_size); 1248 if (res > 8) { 1249 unsigned char *buf = (unsigned char *)c->rbuf; 1250 stats.bytes_read += res; 1251 1252 /* Beginning of UDP packet is the request ID; save it. */ 1253 c->request_id = buf[0] * 256 + buf[1]; 1254 1255 /* If this is a multi-packet request, drop it. */ 1256 if (buf[4] != 0 || buf[5] != 1) { 1257 out_string(c, "SERVER_ERROR multi-packet request not supported"); 1258 return 0; 1259 } 1260 1261 /* Don't care about any of the rest of the header. */ 1262 res -= 8; 1263 memmove(c->rbuf, c->rbuf + 8, res); 1264 1265 c->rbytes += res; 1266 return 1; 1267 } 1268 return 0; 1269 } 1270 1271 /* 1272 * read from network as much as we can, handle buffer overflow and connection 1273 * close. 1274 * before reading, move the remaining incomplete fragment of a command 1275 * (if any) to the beginning of the buffer. 1276 * return 0 if there's nothing to read on the first read. 1277 */ 1278 int try_read_network(conn *c) { 1279 int gotdata = 0; 1280 int res; 1281 1282 if (c->rcurr != c->rbuf) { 1283 if (c->rbytes != 0) /* otherwise there's nothing to copy */ 1284 memmove(c->rbuf, c->rcurr, c->rbytes); 1285 c->rcurr = c->rbuf; 1286 } 1287 1288 while (1) { 1289 if (c->rbytes >= c->rsize) { 1290 char *new_rbuf = realloc(c->rbuf, c->rsize*2); 1291 if (!new_rbuf) { 1292 if (settings.verbose > 0) 1293 fprintf(stderr, "Couldn't realloc input buffer\n"); 1294 c->rbytes = 0; /* ignore what we read */ 1295 out_string(c, "SERVER_ERROR out of memory"); 1296 c->write_and_go = conn_closing; 1297 return 1; 1298 } 1299 c->rcurr = c->rbuf = new_rbuf; 1300 c->rsize *= 2; 1301 } 1302 1303 /* unix socket mode doesn't need this, so zeroed out. but why 1304 * is this done for every command? presumably for UDP 1305 * mode. */ 1306 if (!settings.socketpath) { 1307 c->request_addr_size = sizeof(c->request_addr); 1308 } else { 1309 c->request_addr_size = 0; 1310 } 1311 1312 res = read(c->sfd, c->rbuf + c->rbytes, c->rsize - c->rbytes); 1313 if (res > 0) { 1314 stats.bytes_read += res; 1315 gotdata = 1; 1316 c->rbytes += res; 1317 continue; 1318 } 1319 if (res == 0) { 1320 /* connection closed */ 1321 conn_set_state(c, conn_closing); 1322 return 1; 1323 } 1324 if (res == -1) { 1325 if (errno == EAGAIN || errno == EWOULDBLOCK) break; 1326 else return 0; 1327 } 1328 } 1329 return gotdata; 1330 } 1331 1332 int update_event(conn *c, int new_flags) { 1333 if (c->ev_flags == new_flags) 1334 return 1; 1335 if (event_del(&c->event) == -1) return 0; 1336 event_set(&c->event, c->sfd, new_flags, event_handler, (void *)c); 1337 c->ev_flags = new_flags; 1338 if (event_add(&c->event, 0) == -1) return 0; 1339 return 1; 1340 } 1341 1342 /* 1343 * Transmit the next chunk of data from our list of msgbuf structures. 1344 * 1345 * Returns: 1346 * TRANSMIT_COMPLETE All done writing. 1347 * TRANSMIT_INCOMPLETE More data remaining to write. 1348 * TRANSMIT_SOFT_ERROR Can't write any more right now. 1349 * TRANSMIT_HARD_ERROR Can't write (c->state is set to conn_closing) 1350 */ 1351 int transmit(conn *c) { 1352 int res; 1353 1354 if (c->msgcurr < c->msgused && 1355 c->msglist[c->msgcurr].msg_iovlen == 0) { 1356 /* Finished writing the current msg; advance to the next. */ 1357 c->msgcurr++; 1358 } 1359 if (c->msgcurr < c->msgused) { 1360 struct msghdr *m = &c->msglist[c->msgcurr]; 1361 res = sendmsg(c->sfd, m, 0); 1362 if (res > 0) { 1363 stats.bytes_written += res; 1364 1365 /* We've written some of the data. Remove the completed 1366 iovec entries from the list of pending writes. */ 1367 while (m->msg_iovlen > 0 && res >= m->msg_iov->iov_len) { 1368 res -= m->msg_iov->iov_len; 1369 m->msg_iovlen--; 1370 m->msg_iov++; 1371 } 1372 1373 /* Might have written just part of the last iovec entry; 1374 adjust it so the next write will do the rest. */ 1375 if (res > 0) { 1376 m->msg_iov->iov_base += res; 1377 m->msg_iov->iov_len -= res; 1378 } 1379 return TRANSMIT_INCOMPLETE; 1380 } 1381 if (res == -1 && (errno == EAGAIN || errno == EWOULDBLOCK)) { 1382 if (!update_event(c, EV_WRITE | EV_PERSIST)) { 1383 if (settings.verbose > 0) 1384 fprintf(stderr, "Couldn't update event\n"); 1385 conn_set_state(c, conn_closing); 1386 return TRANSMIT_HARD_ERROR; 1387 } 1388 return TRANSMIT_SOFT_ERROR; 1389 } 1390 /* if res==0 or res==-1 and error is not EAGAIN or EWOULDBLOCK, 1391 we have a real error, on which we close the connection */ 1392 if (settings.verbose > 0) 1393 perror("Failed to write, and not due to blocking"); 1394 1395 if (c->udp) 1396 conn_set_state(c, conn_read); 1397 else 1398 conn_set_state(c, conn_closing); 1399 return TRANSMIT_HARD_ERROR; 1400 } else { 1401 return TRANSMIT_COMPLETE; 1402 } 1403 } 1404 1405 void drive_machine(conn *c) { 1406 1407 int exit = 0; 1408 int sfd, flags = 1; 1409 socklen_t addrlen; 1410 struct sockaddr addr; 1411 conn *newc; 1412 int res; 1413 1414 while (!exit) { 1415 switch(c->state) { 1416 case conn_listening: 1417 addrlen = sizeof(addr); 1418 if ((sfd = accept(c->sfd, &addr, &addrlen)) == -1) { 1419 if (errno == EAGAIN || errno == EWOULDBLOCK) { 1420 exit = 1; 1421 break; 1422 } else { 1423 perror("accept()"); 1424 } 1425 break; 1426 } 1427 if ((flags = fcntl(sfd, F_GETFL, 0)) < 0 || 1428 fcntl(sfd, F_SETFL, flags | O_NONBLOCK) < 0) { 1429 perror("setting O_NONBLOCK"); 1430 close(sfd); 1431 break; 1432 } 1433 newc = conn_new(sfd, conn_read, EV_READ | EV_PERSIST, 1434 DATA_BUFFER_SIZE, 0); 1435 if (!newc) { 1436 if (settings.verbose > 0) 1437 fprintf(stderr, "couldn't create new connection\n"); 1438 close(sfd); 1439 break; 1440 } 1441 1442 break; 1443 1444 case conn_read: 1445 if (try_read_command(c)) { 1446 continue; 1447 } 1448 if (c->udp ? try_read_udp(c) : try_read_network(c)) { 1449 continue; 1450 } 1451 /* we have no command line and no data to read from network */ 1452 if (!update_event(c, EV_READ | EV_PERSIST)) { 1453 if (settings.verbose > 0) 1454 fprintf(stderr, "Couldn't update event\n"); 1455 conn_set_state(c, conn_closing); 1456 break; 1457 } 1458 exit = 1; 1459 break; 1460 1461 case conn_nread: 1462 /* we are reading rlbytes into ritem; */ 1463 if (c->rlbytes == 0) { 1464 complete_nread(c); 1465 break; 1466 } 1467 /* first check if we have leftovers in the conn_read buffer */ 1468 if (c->rbytes > 0) { 1469 int tocopy = c->rbytes > c->rlbytes ? c->rlbytes : c->rbytes; 1470 memcpy(c->ritem, c->rcurr, tocopy); 1471 c->ritem += tocopy; 1472 c->rlbytes -= tocopy; 1473 c->rcurr += tocopy; 1474 c->rbytes -= tocopy; 1475 break; 1476 } 1477 1478 /* now try reading from the socket */ 1479 res = read(c->sfd, c->ritem, c->rlbytes); 1480 if (res > 0) { 1481 stats.bytes_read += res; 1482 c->ritem += res; 1483 c->rlbytes -= res; 1484 break; 1485 } 1486 if (res == 0) { /* end of stream */ 1487 conn_set_state(c, conn_closing); 1488 break; 1489 } 1490 if (res == -1 && (errno == EAGAIN || errno == EWOULDBLOCK)) { 1491 if (!update_event(c, EV_READ | EV_PERSIST)) { 1492 if (settings.verbose > 0) 1493 fprintf(stderr, "Couldn't update event\n"); 1494 conn_set_state(c, conn_closing); 1495 break; 1496 } 1497 exit = 1; 1498 break; 1499 } 1500 /* otherwise we have a real error, on which we close the connection */ 1501 if (settings.verbose > 0) 1502 fprintf(stderr, "Failed to read, and not due to blocking\n"); 1503 conn_set_state(c, conn_closing); 1504 break; 1505 1506 case conn_swallow: 1507 /* we are reading sbytes and throwing them away */ 1508 if (c->sbytes == 0) { 1509 conn_set_state(c, conn_read); 1510 break; 1511 } 1512 1513 /* first check if we have leftovers in the conn_read buffer */ 1514 if (c->rbytes > 0) { 1515 int tocopy = c->rbytes > c->sbytes ? c->sbytes : c->rbytes; 1516 c->sbytes -= tocopy; 1517 c->rcurr += tocopy; 1518 c->rbytes -= tocopy; 1519 break; 1520 } 1521 1522 /* now try reading from the socket */ 1523 res = read(c->sfd, c->rbuf, c->rsize > c->sbytes ? c->sbytes : c->rsize); 1524 if (res > 0) { 1525 stats.bytes_read += res; 1526 c->sbytes -= res; 1527 break; 1528 } 1529 if (res == 0) { /* end of stream */ 1530 conn_set_state(c, conn_closing); 1531 break; 1532 } 1533 if (res == -1 && (errno == EAGAIN || errno == EWOULDBLOCK)) { 1534 if (!update_event(c, EV_READ | EV_PERSIST)) { 1535 if (settings.verbose > 0) 1536 fprintf(stderr, "Couldn't update event\n"); 1537 conn_set_state(c, conn_closing); 1538 break; 1539 } 1540 exit = 1; 1541 break; 1542 } 1543 /* otherwise we have a real error, on which we close the connection */ 1544 if (settings.verbose > 0) 1545 fprintf(stderr, "Failed to read, and not due to blocking\n"); 1546 conn_set_state(c, conn_closing); 1547 break; 1548 1549 case conn_write: 1550 /* 1551 * We want to write out a simple response. If we haven't already, 1552 * assemble it into a msgbuf list (this will be a single-entry 1553 * list for TCP or a two-entry list for UDP). 1554 */ 1555 if (c->iovused == 0) { 1556 if (add_iov(c, c->wcurr, c->wbytes) || 1557 c->udp && build_udp_headers(c)) { 1558 if (settings.verbose > 0) 1559 fprintf(stderr, "Couldn't build response\n"); 1560 conn_set_state(c, conn_closing); 1561 break; 1562 } 1563 } 1564 1565 /* fall through... */ 1566 1567 case conn_mwrite: 1568 switch (transmit(c)) { 1569 case TRANSMIT_COMPLETE: 1570 if (c->state == conn_mwrite) { 1571 while (c->ileft > 0) { 1572 item *it = *(c->icurr); 1573 assert((it->it_flags & ITEM_SLABBED) == 0); 1574 item_remove(it); 1575 c->icurr++; 1576 c->ileft--; 1577 } 1578 conn_set_state(c, conn_read); 1579 } else if (c->state == conn_write) { 1580 if (c->write_and_free) { 1581 free(c->write_and_free); 1582 c->write_and_free = 0; 1583 } 1584 conn_set_state(c, c->write_and_go); 1585 } else { 1586 if (settings.verbose > 0) 1587 fprintf(stderr, "Unexpected state %d\n", c->state); 1588 conn_set_state(c, conn_closing); 1589 } 1590 break; 1591 1592 case TRANSMIT_INCOMPLETE: 1593 case TRANSMIT_HARD_ERROR: 1594 break; /* Continue in state machine. */ 1595 1596 case TRANSMIT_SOFT_ERROR: 1597 exit = 1; 1598 break; 1599 } 1600 break; 1601 1602 case conn_closing: 1603 if (c->udp) 1604 conn_cleanup(c); 1605 else 1606 conn_close(c); 1607 exit = 1; 1608 break; 1609 } 1610 1611 } 1612 1613 return; 1614 } 1615 1616 void event_handler(int fd, short which, void *arg) { 1617 conn *c; 1618 1619 c = (conn *)arg; 1620 c->which = which; 1621 1622 /* sanity */ 1623 if (fd != c->sfd) { 1624 if (settings.verbose > 0) 1625 fprintf(stderr, "Catastrophic: event fd doesn't match conn fd!\n"); 1626 conn_close(c); 1627 return; 1628 } 1629 1630 /* do as much I/O as possible until we block */ 1631 drive_machine(c); 1632 1633 /* wait for next event */ 1634 return; 1635 } 1636 1637 int new_socket(int is_udp) { 1638 int sfd; 1639 int flags; 1640 1641 if ((sfd = socket(AF_INET, is_udp ? SOCK_DGRAM : SOCK_STREAM, 0)) == -1) { 1642 perror("socket()"); 1643 return -1; 1644 } 1645 1646 if ((flags = fcntl(sfd, F_GETFL, 0)) < 0 || 1647 fcntl(sfd, F_SETFL, flags | O_NONBLOCK) < 0) { 1648 perror("setting O_NONBLOCK"); 1649 close(sfd); 1650 return -1; 1651 } 1652 return sfd; 1653 } 1654 1655 1656 /* 1657 * Sets a socket's send buffer size to the maximum allowed by the system. 1658 */ 1659 void maximize_sndbuf(int sfd) { 1660 socklen_t intsize = sizeof(int); 1661 int last_good; 1662 int min, max, avg; 1663 int old_size; 1664 1665 /* Start with the default size. */ 1666 if (getsockopt(sfd, SOL_SOCKET, SO_SNDBUF, &old_size, &intsize)) { 1667 if (settings.verbose > 0) 1668 perror("getsockopt(SO_SNDBUF)"); 1669 return; 1670 } 1671 1672 /* Binary-search for the real maximum. */ 1673 min = old_size; 1674 max = MAX_SENDBUF_SIZE; 1675 1676 while (min <= max) { 1677 avg = ((unsigned int) min + max) / 2; 1678 if (setsockopt(sfd, SOL_SOCKET, SO_SNDBUF, &avg, intsize) == 0) { 1679 last_good = avg; 1680 min = avg + 1; 1681 } else { 1682 max = avg - 1; 1683 } 1684 } 1685 1686 if (settings.verbose > 1) 1687 fprintf(stderr, "<%d send buffer was %d, now %d\n", sfd, old_size, last_good); 1688 } 1689 1690 1691 int server_socket(int port, int is_udp) { 1692 int sfd; 1693 struct linger ling = {0, 0}; 1694 struct sockaddr_in addr; 1695 int flags =1; 1696 1697 if ((sfd = new_socket(is_udp)) == -1) { 1698 return -1; 1699 } 1700 1701 setsockopt(sfd, SOL_SOCKET, SO_REUSEADDR, &flags, sizeof(flags)); 1702 if (is_udp) { 1703 maximize_sndbuf(sfd); 1704 } else { 1705 setsockopt(sfd, SOL_SOCKET, SO_KEEPALIVE, &flags, sizeof(flags)); 1706 setsockopt(sfd, SOL_SOCKET, SO_LINGER, &ling, sizeof(ling)); 1707 setsockopt(sfd, IPPROTO_TCP, TCP_NODELAY, &flags, sizeof(flags)); 1708 } 1709 1710 /* 1711 * the memset call clears nonstandard fields in some impementations 1712 * that otherwise mess things up. 1713 */ 1714 memset(&addr, 0, sizeof(addr)); 1715 1716 addr.sin_family = AF_INET; 1717 addr.sin_port = htons(port); 1718 addr.sin_addr = settings.interface; 1719 if (bind(sfd, (struct sockaddr *) &addr, sizeof(addr)) == -1) { 1720 perror("bind()"); 1721 close(sfd); 1722 return -1; 1723 } 1724 if (! is_udp && listen(sfd, 1024) == -1) { 1725 perror("listen()"); 1726 close(sfd); 1727 return -1; 1728 } 1729 return sfd; 1730 } 1731 1732 int new_socket_unix(void) { 1733 int sfd; 1734 int flags; 1735 1736 if ((sfd = socket(AF_UNIX, SOCK_STREAM, 0)) == -1) { 1737 perror("socket()"); 1738 return -1; 1739 } 1740 1741 if ((flags = fcntl(sfd, F_GETFL, 0)) < 0 || 1742 fcntl(sfd, F_SETFL, flags | O_NONBLOCK) < 0) { 1743 perror("setting O_NONBLOCK"); 1744 close(sfd); 1745 return -1; 1746 } 1747 return sfd; 1748 } 1749 1750 int server_socket_unix(char *path) { 1751 int sfd; 1752 struct linger ling = {0, 0}; 1753 struct sockaddr_un addr; 1754 struct stat tstat; 1755 int flags =1; 1756 1757 if (!path) { 1758 return -1; 1759 } 1760 1761 if ((sfd = new_socket_unix()) == -1) { 1762 return -1; 1763 } 1764 1765 /* 1766 * Clean up a previous socket file if we left it around 1767 */ 1768 if (!lstat(path, &tstat)) { 1769 if (S_ISSOCK(tstat.st_mode)) 1770 unlink(path); 1771 } 1772 1773 setsockopt(sfd, SOL_SOCKET, SO_REUSEADDR, &flags, sizeof(flags)); 1774 setsockopt(sfd, SOL_SOCKET, SO_KEEPALIVE, &flags, sizeof(flags)); 1775 setsockopt(sfd, SOL_SOCKET, SO_LINGER, &ling, sizeof(ling)); 1776 1777 /* 1778 * the memset call clears nonstandard fields in some impementations 1779 * that otherwise mess things up. 1780 */ 1781 memset(&addr, 0, sizeof(addr)); 1782 1783 addr.sun_family = AF_UNIX; 1784 strcpy(addr.sun_path, path); 1785 if (bind(sfd, (struct sockaddr *) &addr, sizeof(addr)) == -1) { 1786 perror("bind()"); 1787 close(sfd); 1788 return -1; 1789 } 1790 if (listen(sfd, 1024) == -1) { 1791 perror("listen()"); 1792 close(sfd); 1793 return -1; 1794 } 1795 return sfd; 1796 } 1797 1798 1799 /* invoke right before gdb is called, on assert */ 1800 void pre_gdb () { 1801 int i = 0; 1802 if(l_socket) close(l_socket); 1803 if(u_socket > -1) close(u_socket); 1804 for (i=3; i<=500; i++) close(i); /* so lame */ 1805 kill(getpid(), SIGABRT); 1806 } 1807 1808 /* 1809 * We keep the current time of day in a global variable that's updated by a 1810 * timer event. This saves us a bunch of time() system calls (we really only 1811 * need to get the time once a second, whereas there can be tens of thousands 1812 * of requests a second) and allows us to use server-start-relative timestamps 1813 * rather than absolute UNIX timestamps, a space savings on systems where 1814 * sizeof(time_t) > sizeof(unsigned int). 1815 */ 1816 volatile rel_time_t current_time; 1817 struct event clockevent; 1818 1819 /* time-sensitive callers can call it by hand with this, outside the normal ever-1-second timer */ 1820 void set_current_time () { 1821 current_time = (rel_time_t) (time(0) - stats.started); 1822 } 1823 1824 void clock_handler(int fd, short which, void *arg) { 1825 struct timeval t; 1826 static int initialized = 0; 1827 1828 if (initialized) { 1829 /* only delete the event if it's actually there. */ 1830 evtimer_del(&clockevent); 1831 } else { 1832 initialized = 1; 1833 } 1834 1835 evtimer_set(&clockevent, clock_handler, 0); 1836 t.tv_sec = 1; 1837 t.tv_usec = 0; 1838 evtimer_add(&clockevent, &t); 1839 1840 set_current_time(); 1841 } 1842 1843 struct event deleteevent; 1844 1845 void delete_handler(int fd, short which, void *arg) { 1846 struct timeval t; 1847 static int initialized = 0; 1848 1849 if (initialized) { 1850 /* some versions of libevent don't like deleting events that don't exist, 1851 so only delete once we know this event has been added. */ 1852 evtimer_del(&deleteevent); 1853 } else { 1854 initialized = 1; 1855 } 1856 1857 evtimer_set(&deleteevent, delete_handler, 0); 1858 t.tv_sec = 5; t.tv_usec=0; 1859 evtimer_add(&deleteevent, &t); 1860 1861 { 1862 int i, j=0; 1863 rel_time_t now = current_time; 1864 for (i=0; i<delcurr; i++) { 1865 item *it = todelete[i]; 1866 if (item_delete_lock_over(it)) { 1867 assert(it->refcount > 0); 1868 it->it_flags &= ~ITEM_DELETED; 1869 item_unlink(it); 1870 item_remove(it); 1871 } else { 1872 todelete[j++] = it; 1873 } 1874 } 1875 delcurr = j; 1876 } 1877 } 1878 1879 void usage(void) { 1880 printf(PACKAGE " " VERSION "\n"); 1881 printf("-p <num> port number to listen on\n"); 1882 printf("-s <file> unix socket path to listen on (disables network support)\n"); 1883 printf("-l <ip_addr> interface to listen on, default is INDRR_ANY\n"); 1884 printf("-d run as a daemon\n"); 1885 printf("-r maximize core file limit\n"); 1886 printf("-u <username> assume identity of <username> (only when run as root)\n"); 1887 printf("-m <num> max memory to use for items in megabytes, default is 64 MB\n"); 1888 printf("-M return error on memory exhausted (rather than removing items)\n"); 1889 printf("-c <num> max simultaneous connections, default is 1024\n"); 1890 printf("-k lock down all paged memory\n"); 1891 printf("-v verbose (print errors/warnings while in event loop)\n"); 1892 printf("-vv very verbose (also print client commands/reponses)\n"); 1893 printf("-h print this help and exit\n"); 1894 printf("-i print memcached and libevent license\n"); 1895 printf("-b run a managed instanced (mnemonic: buckets)\n"); 1896 printf("-P <file> save PID in <file>, only used with -d option\n"); 1897 printf("-f <factor> chunk size growth factor, default 1.25\n"); 1898 printf("-n <bytes> minimum space allocated for key+value+flags, default 48\n"); 1899 return; 1900 } 1901 1902 void usage_license(void) { 1903 printf(PACKAGE " " VERSION "\n\n"); 1904 printf( 1905 "Copyright (c) 2003, Danga Interactive, Inc. <http://www.danga.com/>\n" 1906 "All rights reserved.\n" 1907 "\n" 1908 "Redistribution and use in source and binary forms, with or without\n" 1909 "modification, are permitted provided that the following conditions are\n" 1910 "met:\n" 1911 "\n" 1912 " * Redistributions of source code must retain the above copyright\n" 1913 "notice, this list of conditions and the following disclaimer.\n" 1914 "\n" 1915 " * Redistributions in binary form must reproduce the above\n" 1916 "copyright notice, this list of conditions and the following disclaimer\n" 1917 "in the documentation and/or other materials provided with the\n" 1918 "distribution.\n" 1919 "\n" 1920 " * Neither the name of the Danga Interactive nor the names of its\n" 1921 "contributors may be used to endorse or promote products derived from\n" 1922 "this software without specific prior written permission.\n" 1923 "\n" 1924 "THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS\n" 1925 "\"AS IS\" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT\n" 1926 "LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR\n" 1927 "A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT\n" 1928 "OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,\n" 1929 "SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT\n" 1930 "LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,\n" 1931 "DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY\n" 1932 "THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT\n" 1933 "(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE\n" 1934 "OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.\n" 1935 "\n" 1936 "\n" 1937 "This product includes software developed by Niels Provos.\n" 1938 "\n" 1939 "[ libevent ]\n" 1940 "\n" 1941 "Copyright 2000-2003 Niels Provos <[email protected]>\n" 1942 "All rights reserved.\n" 1943 "\n" 1944 "Redistribution and use in source and binary forms, with or without\n" 1945 "modification, are permitted provided that the following conditions\n" 1946 "are met:\n" 1947 "1. Redistributions of source code must retain the above copyright\n" 1948 " notice, this list of conditions and the following disclaimer.\n" 1949 "2. Redistributions in binary form must reproduce the above copyright\n" 1950 " notice, this list of conditions and the following disclaimer in the\n" 1951 " documentation and/or other materials provided with the distribution.\n" 1952 "3. All advertising materials mentioning features or use of this software\n" 1953 " must display the following acknowledgement:\n" 1954 " This product includes software developed by Niels Provos.\n" 1955 "4. The name of the author may not be used to endorse or promote products\n" 1956 " derived from this software without specific prior written permission.\n" 1957 "\n" 1958 "THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR\n" 1959 "IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES\n" 1960 "OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.\n" 1961 "IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT,\n" 1962 "INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT\n" 1963 "NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,\n" 1964 "DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY\n" 1965 "THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT\n" 1966 "(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF\n" 1967 "THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.\n" 1968 ); 1969 1970 return; 1971 } 1972 1973 void save_pid(pid_t pid,char *pid_file) { 1974 FILE *fp; 1975 if (!pid_file) 1976 return; 1977 1978 if (!(fp = fopen(pid_file,"w"))) { 1979 fprintf(stderr,"Could not open the pid file %s for writing\n",pid_file); 1980 return; 1981 } 1982 1983 fprintf(fp,"%ld\n",(long) pid); 1984 if (fclose(fp) == -1) { 1985 fprintf(stderr,"Could not close the pid file %s.\n",pid_file); 1986 return; 1987 } 1988 } 1989 1990 void remove_pidfile(char *pid_file) { 1991 if (!pid_file) 1992 return; 1993 1994 if (unlink(pid_file)) { 1995 fprintf(stderr,"Could not remove the pid file %s.\n",pid_file); 1996 } 1997 1998 } 1999 2000 int l_socket=0; 2001 int u_socket=-1; 2002 2003 void sig_handler(int sig) { 2004 printf("SIGINT handled.\n"); 2005 exit(0); 2006 } 2007 2008 int main (int argc, char **argv) { 2009 int c; 2010 conn *l_conn; 2011 conn *u_conn; 2012 struct in_addr addr; 2013 int lock_memory = 0; 2014 int daemonize = 0; 2015 int maxcore = 0; 2016 char *username = 0; 2017 struct passwd *pw; 2018 struct sigaction sa; 2019 struct rlimit rlim; 2020 char *pid_file = NULL; 2021 2022 /* handle SIGINT */ 2023 signal(SIGINT, sig_handler); 2024 2025 /* init settings */ 2026 settings_init(); 2027 2028 /* set stderr non-buffering (for running under, say, daemontools) */ 2029 setbuf(stderr, NULL); 2030 2031 /* process arguments */ 2032 while ((c = getopt(argc, argv, "bp:s:U:m:Mc:khirvdl:u:P:f:s:")) != -1) { 2033 switch (c) { 2034 case 'U': 2035 settings.udpport = atoi(optarg); 2036 break; 2037 case 'b': 2038 settings.managed = 1; 2039 break; 2040 case 'p': 2041 settings.port = atoi(optarg); 2042 break; 2043 case 's': 2044 settings.socketpath = optarg; 2045 break; 2046 case 'm': 2047 settings.maxbytes = ((size_t)atoi(optarg))*1024*1024; 2048 break; 2049 case 'M': 2050 settings.evict_to_free = 0; 2051 break; 2052 case 'c': 2053 settings.maxconns = atoi(optarg); 2054 break; 2055 case 'h': 2056 usage(); 2057 exit(0); 2058 case 'i': 2059 usage_license(); 2060 exit(0); 2061 case 'k': 2062 lock_memory = 1; 2063 break; 2064 case 'v': 2065 settings.verbose++; 2066 break; 2067 case 'l': 2068 if (!inet_pton(AF_INET, optarg, &addr)) { 2069 fprintf(stderr, "Illegal address: %s\n", optarg); 2070 return 1; 2071 } else { 2072 settings.interface = addr; 2073 } 2074 break; 2075 case 'd': 2076 daemonize = 1; 2077 break; 2078 case 'r': 2079 maxcore = 1; 2080 break; 2081 case 'u': 2082 username = optarg; 2083 break; 2084 case 'P': 2085 pid_file = optarg; 2086 break; 2087 case 'f': 2088 settings.factor = atof(optarg); 2089 if (settings.factor <= 1.0) { 2090 fprintf(stderr, "Factor must be greater than 1\n"); 2091 return 1; 2092 } 2093 break; 2094 case 'n': 2095 settings.chunk_size = atoi(optarg); 2096 if (settings.chunk_size == 0) { 2097 fprintf(stderr, "Chunk size must be greater than 0\n"); 2098 return 1; 2099 } 2100 break; 2101 default: 2102 fprintf(stderr, "Illegal argument \"%c\"\n", c); 2103 return 1; 2104 } 2105 } 2106 2107 if (maxcore) { 2108 struct rlimit rlim_new; 2109 /* 2110 * First try raising to infinity; if that fails, try bringing 2111 * the soft limit to the hard. 2112 */ 2113 if (getrlimit(RLIMIT_CORE, &rlim)==0) { 2114 rlim_new.rlim_cur = rlim_new.rlim_max = RLIM_INFINITY; 2115 if (setrlimit(RLIMIT_CORE, &rlim_new)!=0) { 2116 /* failed. try raising just to the old max */ 2117 rlim_new.rlim_cur = rlim_new.rlim_max = 2118 rlim.rlim_max; 2119 (void) setrlimit(RLIMIT_CORE, &rlim_new); 2120 } 2121 } 2122 /* 2123 * getrlimit again to see what we ended up with. Only fail if 2124 * the soft limit ends up 0, because then no core files will be 2125 * created at all. 2126 */ 2127 2128 if ((getrlimit(RLIMIT_CORE, &rlim)!=0) || rlim.rlim_cur==0) { 2129 fprintf(stderr, "failed to ensure corefile creation\n"); 2130 exit(1); 2131 } 2132 } 2133 2134 /* 2135 * If needed, increase rlimits to allow as many connections 2136 * as needed. 2137 */ 2138 2139 if (getrlimit(RLIMIT_NOFILE, &rlim) != 0) { 2140 fprintf(stderr, "failed to getrlimit number of files\n"); 2141 exit(1); 2142 } else { 2143 int maxfiles = settings.maxconns; 2144 if (rlim.rlim_cur < maxfiles) 2145 rlim.rlim_cur = maxfiles + 3; 2146 if (rlim.rlim_max < rlim.rlim_cur) 2147 rlim.rlim_max = rlim.rlim_cur; 2148 if (setrlimit(RLIMIT_NOFILE, &rlim) != 0) { 2149 fprintf(stderr, "failed to set rlimit for open files. Try running as root or requesting smaller maxconns value.\n"); 2150 exit(1); 2151 } 2152 } 2153 2154 /* 2155 * initialization order: first create the listening sockets 2156 * (may need root on low ports), then drop root if needed, 2157 * then daemonise if needed, then init libevent (in some cases 2158 * descriptors created by libevent wouldn't survive forking). 2159 */ 2160 2161 /* create the listening socket and bind it */ 2162 if (!settings.socketpath) { 2163 l_socket = server_socket(settings.port, 0); 2164 if (l_socket == -1) { 2165 fprintf(stderr, "failed to listen\n"); 2166 exit(1); 2167 } 2168 } 2169 2170 if (settings.udpport > 0 && ! settings.socketpath) { 2171 /* create the UDP listening socket and bind it */ 2172 u_socket = server_socket(settings.udpport, 1); 2173 if (u_socket == -1) { 2174 fprintf(stderr, "failed to listen\n"); 2175 exit(1); 2176 } 2177 } 2178 2179 /* lose root privileges if we have them */ 2180 if (getuid()== 0 || geteuid()==0) { 2181 if (username==0 || *username=='\0') { 2182 fprintf(stderr, "can't run as root without the -u switch\n"); 2183 return 1; 2184 } 2185 if ((pw = getpwnam(username)) == 0) { 2186 fprintf(stderr, "can't find the user %s to switch to\n", username); 2187 return 1; 2188 } 2189 if (setgid(pw->pw_gid)<0 || setuid(pw->pw_uid)<0) { 2190 fprintf(stderr, "failed to assume identity of user %s\n", username); 2191 return 1; 2192 } 2193 } 2194 2195 /* create unix mode sockets after dropping privileges */ 2196 if (settings.socketpath) { 2197 l_socket = server_socket_unix(settings.socketpath); 2198 if (l_socket == -1) { 2199 fprintf(stderr, "failed to listen\n"); 2200 exit(1); 2201 } 2202 } 2203 2204 /* daemonize if requested */ 2205 /* if we want to ensure our ability to dump core, don't chdir to / */ 2206 if (daemonize) { 2207 int res; 2208 res = daemon(maxcore, settings.verbose); 2209 if (res == -1) { 2210 fprintf(stderr, "failed to daemon() in order to daemonize\n"); 2211 return 1; 2212 } 2213 } 2214 2215 2216 /* initialize other stuff */ 2217 item_init(); 2218 event_init(); 2219 stats_init(); 2220 assoc_init(); 2221 conn_init(); 2222 slabs_init(settings.maxbytes, settings.factor); 2223 2224 /* managed instance? alloc and zero a bucket array */ 2225 if (settings.managed) { 2226 buckets = malloc(sizeof(int)*MAX_BUCKETS); 2227 if (buckets == 0) { 2228 fprintf(stderr, "failed to allocate the bucket array"); 2229 exit(1); 2230 } 2231 memset(buckets, 0, sizeof(int)*MAX_BUCKETS); 2232 } 2233 2234 /* lock paged memory if needed */ 2235 if (lock_memory) { 2236 #ifdef HAVE_MLOCKALL 2237 mlockall(MCL_CURRENT | MCL_FUTURE); 2238 #else 2239 fprintf(stderr, "warning: mlockall() not supported on this platform. proceeding without.\n"); 2240 #endif 2241 } 2242 2243 /* 2244 * ignore SIGPIPE signals; we can use errno==EPIPE if we 2245 * need that information 2246 */ 2247 sa.sa_handler = SIG_IGN; 2248 sa.sa_flags = 0; 2249 if (sigemptyset(&sa.sa_mask) == -1 || 2250 sigaction(SIGPIPE, &sa, 0) == -1) { 2251 perror("failed to ignore SIGPIPE; sigaction"); 2252 exit(1); 2253 } 2254 /* create the initial listening connection */ 2255 if (!(l_conn = conn_new(l_socket, conn_listening, EV_READ | EV_PERSIST, 1, 0))) { 2256 fprintf(stderr, "failed to create listening connection"); 2257 exit(1); 2258 } 2259 /* create the initial listening udp connection */ 2260 if (u_socket > -1 && 2261 !(u_conn = conn_new(u_socket, conn_read, EV_READ | EV_PERSIST, UDP_READ_BUFFER_SIZE, 1))) { 2262 fprintf(stderr, "failed to create udp connection"); 2263 exit(1); 2264 } 2265 /* initialise clock event */ 2266 clock_handler(0,0,0); 2267 /* initialise deletion array and timer event */ 2268 deltotal = 200; delcurr = 0; 2269 todelete = malloc(sizeof(item *)*deltotal); 2270 delete_handler(0,0,0); /* sets up the event */ 2271 /* save the PID in if we're a daemon */ 2272 if (daemonize) 2273 save_pid(getpid(),pid_file); 2274 /* enter the loop */ 2275 event_loop(0); 2276 /* remove the PID file if we're a daemon */ 2277 if (daemonize) 2278 remove_pidfile(pid_file); 2279 return 0; 2280 } 2281