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