Lines Matching refs:c
66 static void drive_machine(conn *c);
79 static int try_read_command_negotiate(conn *c);
80 static int try_read_command_udp(conn *c);
82 static enum try_read_result try_read_network(conn *c);
83 static enum try_read_result try_read_udp(conn *c);
89 static void conn_to_str(const conn *c, char *addr, char *svr_addr);
96 static void conn_close(conn *c);
98 static bool update_event(conn *c, const int new_flags);
99 static void complete_nread(conn *c);
101 static void conn_free(conn *c);
131 ssize_t tcp_read(conn *c, void *buf, size_t count) { in tcp_read() argument
132 assert (c != NULL); in tcp_read()
133 return read(c->sfd, buf, count); in tcp_read()
136 ssize_t tcp_sendmsg(conn *c, struct msghdr *msg, int flags) { in tcp_sendmsg() argument
137 assert (c != NULL); in tcp_sendmsg()
138 return sendmsg(c->sfd, msg, flags); in tcp_sendmsg()
141 ssize_t tcp_write(conn *c, void *buf, size_t count) { in tcp_write() argument
142 assert (c != NULL); in tcp_write()
143 return write(c->sfd, buf, count); in tcp_write()
146 static enum transmit_result transmit(conn *c);
294 conn *c; in conn_timeout_thread() local
321 c = conns[i]; in conn_timeout_thread()
323 if (!IS_TCP(c->transport)) in conn_timeout_thread()
326 if (c->state != conn_new_cmd && c->state != conn_read) in conn_timeout_thread()
329 if ((current_time - c->last_cmd_time) > settings.idle_timeout) { in conn_timeout_thread()
330 timeout_conn(c); in conn_timeout_thread()
332 if (c->last_cmd_time < oldest_last_cmd) in conn_timeout_thread()
333 oldest_last_cmd = c->last_cmd_time; in conn_timeout_thread()
392 static void rbuf_release(conn *c) { in rbuf_release() argument
393 if (c->rbuf != NULL && c->rbytes == 0 && !IS_UDP(c->transport)) { in rbuf_release()
394 if (c->rbuf_malloced) { in rbuf_release()
395 free(c->rbuf); in rbuf_release()
396 c->rbuf_malloced = false; in rbuf_release()
398 do_cache_free(c->thread->rbuf_cache, c->rbuf); in rbuf_release()
400 c->rsize = 0; in rbuf_release()
401 c->rbuf = NULL; in rbuf_release()
402 c->rcurr = NULL; in rbuf_release()
406 static bool rbuf_alloc(conn *c) { in rbuf_alloc() argument
407 if (c->rbuf == NULL) { in rbuf_alloc()
408 c->rbuf = do_cache_alloc(c->thread->rbuf_cache); in rbuf_alloc()
409 if (!c->rbuf) { in rbuf_alloc()
410 THR_STATS_LOCK(c->thread); in rbuf_alloc()
411 c->thread->stats.read_buf_oom++; in rbuf_alloc()
412 THR_STATS_UNLOCK(c->thread); in rbuf_alloc()
415 c->rsize = READ_BUFFER_SIZE; in rbuf_alloc()
416 c->rcurr = c->rbuf; in rbuf_alloc()
424 bool rbuf_switch_to_malloc(conn *c) { in rbuf_switch_to_malloc() argument
426 size_t size = c->rsize * 2; in rbuf_switch_to_malloc()
431 memcpy(tmp, c->rcurr, c->rbytes); in rbuf_switch_to_malloc()
432 do_cache_free(c->thread->rbuf_cache, c->rbuf); in rbuf_switch_to_malloc()
434 c->rcurr = c->rbuf = tmp; in rbuf_switch_to_malloc()
435 c->rsize = size; in rbuf_switch_to_malloc()
436 c->rbuf_malloced = true; in rbuf_switch_to_malloc()
500 void conn_close_idle(conn *c) { in conn_close_idle() argument
502 (current_time - c->last_cmd_time) > settings.idle_timeout) { in conn_close_idle()
503 if (c->state != conn_new_cmd && c->state != conn_read) { in conn_close_idle()
506 "fd %d wants to timeout, but isn't in read state", c->sfd); in conn_close_idle()
511 fprintf(stderr, "Closing idle fd %d\n", c->sfd); in conn_close_idle()
513 pthread_mutex_lock(&c->thread->stats.mutex); in conn_close_idle()
514 c->thread->stats.idle_kicks++; in conn_close_idle()
515 pthread_mutex_unlock(&c->thread->stats.mutex); in conn_close_idle()
517 c->close_reason = IDLE_TIMEOUT_CLOSE; in conn_close_idle()
519 conn_set_state(c, conn_closing); in conn_close_idle()
520 drive_machine(c); in conn_close_idle()
524 static void _conn_event_readd(conn *c) { in _conn_event_readd() argument
525 c->ev_flags = EV_READ | EV_PERSIST; in _conn_event_readd()
526 event_set(&c->event, c->sfd, c->ev_flags, event_handler, (void *)c); in _conn_event_readd()
527 event_base_set(c->thread->base, &c->event); in _conn_event_readd()
530 if (event_add(&c->event, 0) == -1) { in _conn_event_readd()
536 void conn_worker_readd(conn *c) { in conn_worker_readd() argument
537 if (c->io_queues_submitted) { // TODO: ensure this is safe? in conn_worker_readd()
538 c->io_queues_submitted--; in conn_worker_readd()
541 if (c->io_queues_submitted != 0) { in conn_worker_readd()
546 switch (c->state) { in conn_worker_readd()
548 drive_machine(c); in conn_worker_readd()
553 _conn_event_readd(c); in conn_worker_readd()
556 conn_set_state(c, conn_io_resume); in conn_worker_readd()
558 drive_machine(c); in conn_worker_readd()
561 event_del(&c->event); in conn_worker_readd()
562 _conn_event_readd(c); in conn_worker_readd()
563 conn_set_state(c, conn_new_cmd); in conn_worker_readd()
579 void conn_io_queue_setup(conn *c) { in conn_io_queue_setup() argument
580 io_queue_cb_t *qcb = c->thread->io_queues; in conn_io_queue_setup()
581 io_queue_t *q = c->io_queues; in conn_io_queue_setup()
593 static void conn_io_queue_reset(conn *c) { in conn_io_queue_reset() argument
594 for (io_queue_t *q = c->io_queues; q->type != IO_QUEUE_NONE; q++) { in conn_io_queue_reset()
611 io_queue_t *conn_io_queue_get(conn *c, int type) { in conn_io_queue_get() argument
612 io_queue_t *q = c->io_queues; in conn_io_queue_get()
632 conn *c; in conn_new() local
635 c = conns[sfd]; in conn_new()
637 if (NULL == c) { in conn_new()
638 if (!(c = (conn *)calloc(1, sizeof(conn)))) { in conn_new()
645 MEMCACHED_CONN_CREATE(c); in conn_new()
646 c->read = NULL; in conn_new()
647 c->sendmsg = NULL; in conn_new()
648 c->write = NULL; in conn_new()
649 c->rbuf = NULL; in conn_new()
651 c->rsize = read_buffer_size; in conn_new()
654 if (c->rsize) { in conn_new()
655 c->rbuf = (char *)malloc((size_t)c->rsize); in conn_new()
658 if (c->rsize && c->rbuf == NULL) { in conn_new()
659 conn_free(c); in conn_new()
672 c->sfd = sfd; in conn_new()
673 conns[sfd] = c; in conn_new()
676 c->transport = transport; in conn_new()
677 c->protocol = bproto; in conn_new()
678 c->tag = conntag; in conn_new()
684 c->request_addr_size = sizeof(c->request_addr); in conn_new()
686 c->request_addr_size = 0; in conn_new()
690 if (getpeername(sfd, (struct sockaddr *) &c->request_addr, in conn_new()
691 &c->request_addr_size)) { in conn_new()
693 memset(&c->request_addr, 0, sizeof(c->request_addr)); in conn_new()
699 &c->request_addr, c->request_addr_size, c->transport, 0, sfd); in conn_new()
705 prot_text(c->protocol)); in conn_new()
708 } else if (c->protocol == negotiating_prot) { in conn_new()
711 } else if (c->protocol == ascii_prot) { in conn_new()
713 } else if (c->protocol == binary_prot) { in conn_new()
716 } else if (c->protocol == proxy_prot) { in conn_new()
721 sfd, c->protocol); in conn_new()
726 c->state = init_state; in conn_new()
727 c->rlbytes = 0; in conn_new()
728 c->cmd = -1; in conn_new()
729 c->rbytes = 0; in conn_new()
730 c->rcurr = c->rbuf; in conn_new()
731 c->ritem = 0; in conn_new()
732 c->rbuf_malloced = false; in conn_new()
733 c->item_malloced = false; in conn_new()
734 c->sasl_started = false; in conn_new()
735 c->set_stale = false; in conn_new()
736 c->mset_res = false; in conn_new()
737 c->close_after_write = false; in conn_new()
738 c->last_cmd_time = current_time; /* initialize for idle kicker */ in conn_new()
740 memset(c->io_queues, 0, sizeof(c->io_queues)); in conn_new()
741 c->io_queues_submitted = 0; in conn_new()
743 c->item = 0; in conn_new()
744 c->ssl = NULL; in conn_new()
746 c->ssl_wbuf = NULL; in conn_new()
749 c->noreply = false; in conn_new()
754 ssl_init_conn(c, ssl); in conn_new()
755 c->ssl_enabled = true; in conn_new()
757 c->read = tcp_read; in conn_new()
758 c->sendmsg = tcp_sendmsg; in conn_new()
759 c->write = tcp_write; in conn_new()
760 c->ssl_enabled = false; in conn_new()
764 c->try_read_command = try_read_command_udp; in conn_new()
766 switch (c->protocol) { in conn_new()
769 c->authenticated = true; in conn_new()
770 c->try_read_command = try_read_command_ascii; in conn_new()
772 c->authenticated = false; in conn_new()
773 c->try_read_command = try_read_command_asciiauth; in conn_new()
778 c->authenticated = false; in conn_new()
779 c->try_read_command = try_read_command_binary; in conn_new()
782 c->try_read_command = try_read_command_negotiate; in conn_new()
786 c->try_read_command = try_read_command_proxy; in conn_new()
792 event_set(&c->event, sfd, event_flags, event_handler, (void *)c); in conn_new()
793 event_base_set(base, &c->event); in conn_new()
794 c->ev_flags = event_flags; in conn_new()
796 if (event_add(&c->event, 0) == -1) { in conn_new()
806 MEMCACHED_CONN_ALLOCATE(c->sfd); in conn_new()
808 return c; in conn_new()
811 void conn_release_items(conn *c) { in conn_release_items() argument
812 assert(c != NULL); in conn_release_items()
814 if (c->item) { in conn_release_items()
815 if (c->item_malloced) { in conn_release_items()
816 free(c->item); in conn_release_items()
817 c->item_malloced = false; in conn_release_items()
819 item_remove(c->item); in conn_release_items()
821 c->item = 0; in conn_release_items()
825 if (c->resp_head) { in conn_release_items()
826 mc_resp *resp = c->resp_head; in conn_release_items()
834 c->sfd, c->protocol == binary_prot ? "binary" : "ascii"); in conn_release_items()
837 c->resp_head = NULL; in conn_release_items()
838 c->resp = NULL; in conn_release_items()
841 resp = resp_finish(c, resp); in conn_release_items()
843 conn_io_queue_reset(c); in conn_release_items()
847 static void conn_cleanup(conn *c) { in conn_cleanup() argument
848 assert(c != NULL); in conn_cleanup()
850 conn_release_items(c); in conn_cleanup()
852 if (c->proxy_rctx) { in conn_cleanup()
853 proxy_cleanup_conn(c); in conn_cleanup()
856 if (c->sasl_conn) { in conn_cleanup()
858 sasl_dispose(&c->sasl_conn); in conn_cleanup()
859 c->sasl_conn = NULL; in conn_cleanup()
862 if (IS_UDP(c->transport)) { in conn_cleanup()
863 conn_set_state(c, conn_read); in conn_cleanup()
870 void conn_free(conn *c) { in conn_free() argument
871 if (c) { in conn_free()
872 assert(c != NULL); in conn_free()
873 assert(c->sfd >= 0 && c->sfd < max_fds); in conn_free()
875 MEMCACHED_CONN_DESTROY(c); in conn_free()
876 conns[c->sfd] = NULL; in conn_free()
877 if (c->rbuf) in conn_free()
878 free(c->rbuf); in conn_free()
880 if (c->ssl_wbuf) in conn_free()
881 c->ssl_wbuf = NULL; in conn_free()
884 free(c); in conn_free()
888 static void conn_close(conn *c) { in conn_close() argument
889 assert(c != NULL); in conn_close()
891 if (c->thread) { in conn_close()
892 LOGGER_LOG(c->thread->l, LOG_CONNEVENTS, LOGGER_CONNECTION_CLOSE, NULL, in conn_close()
893 &c->request_addr, c->request_addr_size, c->transport, in conn_close()
894 c->close_reason, c->sfd); in conn_close()
898 event_del(&c->event); in conn_close()
901 fprintf(stderr, "<%d connection closed.\n", c->sfd); in conn_close()
903 conn_cleanup(c); in conn_close()
906 if (c->thread) { in conn_close()
907 c->rbytes = 0; in conn_close()
908 rbuf_release(c); in conn_close()
911 MEMCACHED_CONN_RELEASE(c->sfd); in conn_close()
912 conn_set_state(c, conn_closed); in conn_close()
913 if (c->ssl_enabled) { in conn_close()
914 ssl_conn_close(c->ssl); in conn_close()
916 close(c->sfd); in conn_close()
917 c->close_reason = 0; in conn_close()
968 void conn_set_state(conn *c, enum conn_states state) { in conn_set_state() argument
969 assert(c != NULL); in conn_set_state()
972 if (state != c->state) { in conn_set_state()
975 c->sfd, state_text(c->state), in conn_set_state()
980 MEMCACHED_PROCESS_COMMAND_END(c->sfd, c->resp->wbuf, c->resp->wbytes); in conn_set_state()
982 c->state = state; in conn_set_state()
1037 static mc_resp* resp_allocate(conn *c) { in resp_allocate() argument
1038 LIBEVENT_THREAD *th = c->thread; in resp_allocate()
1147 bool resp_start(conn *c) { in resp_start() argument
1148 mc_resp *resp = resp_allocate(c); in resp_start()
1150 THR_STATS_LOCK(c->thread); in resp_start()
1151 c->thread->stats.response_obj_oom++; in resp_start()
1152 THR_STATS_UNLOCK(c->thread); in resp_start()
1157 THR_STATS_LOCK(c->thread); in resp_start()
1158 c->thread->stats.response_obj_count++; in resp_start()
1159 THR_STATS_UNLOCK(c->thread); in resp_start()
1161 if (!c->resp_head) { in resp_start()
1162 c->resp_head = resp; in resp_start()
1164 if (!c->resp) { in resp_start()
1165 c->resp = resp; in resp_start()
1167 c->resp->next = resp; in resp_start()
1168 c->resp = resp; in resp_start()
1170 if (IS_UDP(c->transport)) { in resp_start()
1172 c->resp->request_id = c->request_id; in resp_start()
1173 c->resp->request_addr = c->request_addr; in resp_start()
1174 c->resp->request_addr_size = c->request_addr_size; in resp_start()
1179 mc_resp *resp_start_unlinked(conn *c) { in resp_start_unlinked() argument
1180 mc_resp *resp = resp_allocate(c); in resp_start_unlinked()
1182 THR_STATS_LOCK(c->thread); in resp_start_unlinked()
1183 c->thread->stats.response_obj_oom++; in resp_start_unlinked()
1184 THR_STATS_UNLOCK(c->thread); in resp_start_unlinked()
1189 THR_STATS_LOCK(c->thread); in resp_start_unlinked()
1190 c->thread->stats.response_obj_count++; in resp_start_unlinked()
1191 THR_STATS_UNLOCK(c->thread); in resp_start_unlinked()
1193 if (IS_UDP(c->transport)) { in resp_start_unlinked()
1195 c->resp->request_id = c->request_id; in resp_start_unlinked()
1196 c->resp->request_addr = c->request_addr; in resp_start_unlinked()
1197 c->resp->request_addr_size = c->request_addr_size; in resp_start_unlinked()
1204 mc_resp* resp_finish(conn *c, mc_resp *resp) { in resp_finish() argument
1227 do_cache_free(c->thread->io_cache, io); in resp_finish()
1230 if (c->resp_head == resp) { in resp_finish()
1231 c->resp_head = next; in resp_finish()
1233 if (c->resp == resp) { in resp_finish()
1234 c->resp = NULL; in resp_finish()
1236 resp_free(c->thread, resp); in resp_finish()
1241 bool resp_has_stack(conn *c) { in resp_has_stack() argument
1242 return c->resp_head->next != NULL ? true : false; in resp_has_stack()
1245 void out_string(conn *c, const char *str) { in out_string() argument
1247 assert(c != NULL); in out_string()
1248 mc_resp *resp = c->resp; in out_string()
1257 if (c->noreply) { in out_string()
1262 fprintf(stderr, ">%d NOREPLY %s\n", c->sfd, str); in out_string()
1263 conn_set_state(c, conn_new_cmd); in out_string()
1268 fprintf(stderr, ">%d %s\n", c->sfd, str); in out_string()
1283 conn_set_state(c, conn_new_cmd); in out_string()
1289 void out_errstring(conn *c, const char *str) { in out_errstring() argument
1290 c->noreply = false; in out_errstring()
1291 out_string(c, str); in out_errstring()
1298 void out_of_memory(conn *c, char *ascii_error) { in out_of_memory() argument
1302 if (c->protocol == binary_prot) { in out_of_memory()
1307 write_bin_error(c, PROTOCOL_BINARY_RESPONSE_ENOMEM, ascii_error, 0); in out_of_memory()
1309 out_string(c, ascii_error); in out_of_memory()
1315 conn *c) { in append_bin_stats() argument
1316 char *buf = c->stats.buffer + c->stats.offset; in append_bin_stats()
1324 .response.opaque = c->opaque in append_bin_stats()
1339 c->stats.offset += sizeof(header.response) + bodylen; in append_bin_stats()
1344 conn *c) { in append_ascii_stats() argument
1345 char *pos = c->stats.buffer + c->stats.offset; in append_ascii_stats()
1347 int remaining = c->stats.size - c->stats.offset; in append_ascii_stats()
1358 c->stats.offset += nbytes; in append_ascii_stats()
1361 static bool grow_stats_buf(conn *c, size_t needed) { in grow_stats_buf() argument
1362 size_t nsize = c->stats.size; in grow_stats_buf()
1363 size_t available = nsize - c->stats.offset; in grow_stats_buf()
1367 if (c->stats.buffer == NULL) { in grow_stats_buf()
1369 available = c->stats.size = c->stats.offset = 0; in grow_stats_buf()
1375 available = nsize - c->stats.offset; in grow_stats_buf()
1378 if (nsize != c->stats.size) { in grow_stats_buf()
1379 char *ptr = realloc(c->stats.buffer, nsize); in grow_stats_buf()
1381 c->stats.buffer = ptr; in grow_stats_buf()
1382 c->stats.size = nsize; in grow_stats_buf()
1403 conn *c = (conn*)cookie; in append_stats() local
1405 if (c->protocol == binary_prot) { in append_stats()
1407 if (!grow_stats_buf(c, needed)) { in append_stats()
1410 append_bin_stats(key, klen, val, vlen, c); in append_stats()
1413 if (!grow_stats_buf(c, needed)) { in append_stats()
1416 append_ascii_stats(key, klen, val, vlen, c); in append_stats()
1419 assert(c->stats.offset <= c->stats.size); in append_stats()
1422 static void reset_cmd_handler(conn *c) { in reset_cmd_handler() argument
1423 c->cmd = -1; in reset_cmd_handler()
1424 c->substate = bin_no_state; in reset_cmd_handler()
1425 if (c->item != NULL) { in reset_cmd_handler()
1428 if (c->item_malloced) { in reset_cmd_handler()
1429 free(c->item); in reset_cmd_handler()
1430 c->item_malloced = false; in reset_cmd_handler()
1432 item_remove(c->item); in reset_cmd_handler()
1434 c->item = NULL; in reset_cmd_handler()
1436 if (c->rbytes > 0) { in reset_cmd_handler()
1437 conn_set_state(c, conn_parse_cmd); in reset_cmd_handler()
1438 } else if (c->resp_head) { in reset_cmd_handler()
1439 conn_set_state(c, conn_mwrite); in reset_cmd_handler()
1441 conn_set_state(c, conn_waiting); in reset_cmd_handler()
1445 static void complete_nread(conn *c) { in complete_nread() argument
1446 assert(c != NULL); in complete_nread()
1448 assert(c->protocol == ascii_prot in complete_nread()
1449 || c->protocol == binary_prot in complete_nread()
1450 || c->protocol == proxy_prot); in complete_nread()
1452 assert(c->protocol == ascii_prot in complete_nread()
1453 || c->protocol == binary_prot); in complete_nread()
1455 if (c->protocol == ascii_prot) { in complete_nread()
1456 complete_nread_ascii(c); in complete_nread()
1457 } else if (c->protocol == binary_prot) { in complete_nread()
1458 complete_nread_binary(c); in complete_nread()
1460 } else if (c->protocol == proxy_prot) { in complete_nread()
1461 complete_nread_proxy(c); in complete_nread()
1740 void write_and_free(conn *c, char *buf, int bytes) { in write_and_free() argument
1742 mc_resp *resp = c->resp; in write_and_free()
1745 conn_set_state(c, conn_new_cmd); in write_and_free()
1747 out_of_memory(c, "SERVER_ERROR out of memory writing stats"); in write_and_free()
1751 void append_stat(const char *name, ADD_STAT add_stats, conn *c, in append_stat() argument
1759 assert(c); in append_stat()
1766 add_stats(name, strlen(name), val_str, vlen, c); in append_stat()
1770 void server_stats(ADD_STAT add_stats, void *c) { in server_stats() argument
1793 append_stat("rusage_user", add_stats, c, "%ld.%06ld", in server_stats()
1796 append_stat("rusage_system", add_stats, c, "%ld.%06ld", in server_stats()
1898 storage_stats(add_stats, c); in server_stats()
1901 proxy_stats(settings.proxy_ctx, add_stats, c); in server_stats()
1917 void process_stat_settings(ADD_STAT add_stats, void *c) { in process_stat_settings() argument
2018 bool get_stats(const char *stat_type, int nkey, ADD_STAT add_stats, void *c) { in get_stats() argument
2030 item_stats_totals(add_stats, c); in get_stats()
2032 item_stats(add_stats, c); in get_stats()
2034 slabs_stats(add_stats, c); in get_stats()
2036 item_stats_sizes(add_stats, c); in get_stats()
2047 static inline void get_conn_text(const conn *c, const int af, in get_conn_text() argument
2061 protoname = IS_UDP(c->transport) ? "udp" : "tcp"; in get_conn_text()
2074 protoname = IS_UDP(c->transport) ? "udp6" : "tcp6"; in get_conn_text()
2123 static void conn_to_str(const conn *c, char *addr, char *svr_addr) { in conn_to_str() argument
2124 if (!c) { in conn_to_str()
2126 } else if (c->state == conn_closed) { in conn_to_str()
2130 struct sockaddr *sock_addr = (void *)&c->request_addr; in conn_to_str()
2133 if (c->state == conn_listening || in conn_to_str()
2134 (IS_UDP(c->transport) && in conn_to_str()
2135 c->state == conn_read)) { in conn_to_str()
2139 if (getsockname(c->sfd, in conn_to_str()
2145 get_conn_text(c, sock_addr->sa_family, addr, sock_addr); in conn_to_str()
2147 if (c->state != conn_listening && !(IS_UDP(c->transport) && in conn_to_str()
2148 c->state == conn_read)) { in conn_to_str()
2152 getsockname(c->sfd, (struct sockaddr *)&svr_sock_addr, &svr_addr_len); in conn_to_str()
2153 get_conn_text(c, svr_sock_addr.ss_family, svr_addr, (struct sockaddr *)&svr_sock_addr); in conn_to_str()
2158 static char *conn_queue_to_str(const conn *c, io_queue_t *q) { in conn_queue_to_str() argument
2168 void process_stats_conns(ADD_STAT add_stats, void *c) { in process_stats_conns() argument
2378 static int try_read_command_negotiate(conn *c) { argument
2379 assert(c != NULL);
2380 assert(c->protocol == negotiating_prot);
2381 assert(c->rcurr <= (c->rbuf + c->rsize));
2382 assert(c->rbytes > 0);
2384 if ((unsigned char)c->rbuf[0] == (unsigned char)PROTOCOL_BINARY_REQ) {
2385 c->protocol = binary_prot;
2386 c->try_read_command = try_read_command_binary;
2389 c->protocol = ascii_prot;
2390 c->try_read_command = try_read_command_ascii;
2394 fprintf(stderr, "%d: Client using the %s protocol\n", c->sfd,
2395 prot_text(c->protocol));
2398 return c->try_read_command(c);
2401 static int try_read_command_udp(conn *c) { argument
2402 assert(c != NULL);
2403 assert(c->rcurr <= (c->rbuf + c->rsize));
2404 assert(c->rbytes > 0);
2406 if ((unsigned char)c->rbuf[0] == (unsigned char)PROTOCOL_BINARY_REQ) {
2407 c->protocol = binary_prot;
2408 return try_read_command_binary(c);
2410 c->protocol = ascii_prot;
2411 return try_read_command_ascii(c);
2418 static enum try_read_result try_read_udp(conn *c) { argument
2421 assert(c != NULL);
2423 c->request_addr_size = sizeof(c->request_addr);
2424 res = recvfrom(c->sfd, c->rbuf, c->rsize,
2425 0, (struct sockaddr *)&c->request_addr,
2426 &c->request_addr_size);
2428 unsigned char *buf = (unsigned char *)c->rbuf;
2429 pthread_mutex_lock(&c->thread->stats.mutex);
2430 c->thread->stats.bytes_read += res;
2431 pthread_mutex_unlock(&c->thread->stats.mutex);
2434 c->request_id = buf[0] * 256 + buf[1];
2443 memmove(c->rbuf, c->rbuf + 8, res);
2445 c->rbytes = res;
2446 c->rcurr = c->rbuf;
2464 static enum try_read_result try_read_network(conn *c) { argument
2468 assert(c != NULL);
2470 if (c->rcurr != c->rbuf) {
2471 if (c->rbytes > 0) /* otherwise there's nothing to copy */
2472 memmove(c->rbuf, c->rcurr, c->rbytes);
2473 c->rcurr = c->rbuf;
2478 if (c->rbytes >= c->rsize && c->rbuf_malloced) {
2483 char *new_rbuf = realloc(c->rbuf, c->rsize * 2);
2491 c->rbytes = 0; /* ignore what we read */
2492 out_of_memory(c, "SERVER_ERROR out of memory reading request");
2493 c->close_after_write = true;
2496 c->rcurr = c->rbuf = new_rbuf;
2497 c->rsize *= 2;
2500 int avail = c->rsize - c->rbytes;
2501 res = c->read(c, c->rbuf + c->rbytes, avail);
2503 pthread_mutex_lock(&c->thread->stats.mutex);
2504 c->thread->stats.bytes_read += res;
2505 pthread_mutex_unlock(&c->thread->stats.mutex);
2507 c->rbytes += res;
2508 if (res == avail && c->rbuf_malloced) {
2516 c->close_reason = NORMAL_CLOSE;
2529 static bool update_event(conn *c, const int new_flags) { argument
2530 assert(c != NULL);
2532 struct event_base *base = c->event.ev_base;
2533 if (c->ev_flags == new_flags)
2535 if (event_del(&c->event) == -1) return false;
2536 event_set(&c->event, c->sfd, new_flags, event_handler, (void *)c);
2537 event_base_set(base, &c->event);
2538 c->ev_flags = new_flags;
2539 if (event_add(&c->event, 0) == -1) return false;
2588 static int _transmit_pre(conn *c, struct iovec *iovs, int iovused, bool one_resp) { argument
2589 mc_resp *resp = c->resp_head;
2657 static void _transmit_post(conn *c, ssize_t res) { argument
2660 mc_resp *resp = c->resp_head;
2664 resp = resp_finish(c, resp);
2671 resp = resp_finish(c, resp);
2699 resp = resp_finish(c, resp);
2716 static enum transmit_result transmit(conn *c) { argument
2717 assert(c != NULL);
2726 iovused = _transmit_pre(c, iovs, iovused, TRANSMIT_ALL_RESP);
2730 _transmit_post(c, 0);
2737 res = c->sendmsg(c, &msg, 0);
2739 pthread_mutex_lock(&c->thread->stats.mutex);
2740 c->thread->stats.bytes_written += res;
2741 pthread_mutex_unlock(&c->thread->stats.mutex);
2744 _transmit_post(c, res);
2746 if (c->resp_head) {
2754 if (!update_event(c, EV_WRITE | EV_PERSIST)) {
2757 conn_set_state(c, conn_closing);
2767 conn_set_state(c, conn_closing);
2813 static enum transmit_result transmit_udp(conn *c) { argument
2814 assert(c != NULL);
2823 resp = c->resp_head;
2830 resp = resp_finish(c, resp);
2851 iovused = _transmit_pre(c, iovs, iovused, TRANSMIT_ONE_RESP);
2873 res = sendmsg(c->sfd, &msg, 0);
2875 pthread_mutex_lock(&c->thread->stats.mutex);
2876 c->thread->stats.bytes_written += res;
2877 pthread_mutex_unlock(&c->thread->stats.mutex);
2883 _transmit_post(c, res);
2885 if (c->resp_head) {
2893 if (!update_event(c, EV_WRITE | EV_PERSIST)) {
2896 conn_set_state(c, conn_closing);
2906 conn_set_state(c, conn_read);
2915 static int read_into_chunked_item(conn *c) { argument
2918 assert(c->rcurr != c->ritem);
2920 while (c->rlbytes > 0) {
2921 item_chunk *ch = (item_chunk *)c->ritem;
2925 c->ritem = (char *) ch->next;
2928 c->ritem = (char *) do_item_alloc_chunk(ch, c->rlbytes +
2929 ((c->protocol == binary_prot) ? 2 : 0));
2930 if (!c->ritem) {
2943 if (c->rbytes > 0) {
2945 int tocopy = c->rbytes > c->rlbytes ? c->rlbytes : c->rbytes;
2947 if (c->ritem != c->rcurr) {
2948 memmove(ch->data + ch->used, c->rcurr, tocopy);
2951 c->rlbytes -= tocopy;
2952 c->rcurr += tocopy;
2953 c->rbytes -= tocopy;
2955 if (c->rlbytes == 0) {
2960 res = c->read(c, ch->data + ch->used,
2961 (unused > c->rlbytes ? c->rlbytes : unused));
2963 pthread_mutex_lock(&c->thread->stats.mutex);
2964 c->thread->stats.bytes_read += res;
2965 pthread_mutex_unlock(&c->thread->stats.mutex);
2968 c->rlbytes -= res;
2982 if (c->rlbytes == 0 && c->protocol == binary_prot && total >= 0) {
2983 item_chunk *ch = (item_chunk *)c->ritem;
2985 c->ritem = (char *) do_item_alloc_chunk(ch, 2);
2986 if (!c->ritem) {
2994 static void drive_machine(conn *c) { argument
3008 assert(c != NULL);
3012 switch(c->state) {
3017 sfd = accept4(c->sfd, (struct sockaddr *)&addr, &addrlen, SOCK_NONBLOCK);
3019 sfd = accept(c->sfd, (struct sockaddr *)&addr, &addrlen);
3022 sfd = accept(c->sfd, (struct sockaddr *)&addr, &addrlen);
3071 void *ssl_v = ssl_accept(c, sfd, &fail);
3078 READ_BUFFER_CACHED, c->transport, ssl_v, c->tag, c->protocol);
3085 rbuf_release(c);
3086 if (!update_event(c, EV_READ | EV_PERSIST)) {
3089 conn_set_state(c, conn_closing);
3093 conn_set_state(c, conn_read);
3098 if (!IS_UDP(c->transport)) {
3100 if (!rbuf_alloc(c)) {
3102 conn_set_state(c, conn_closing);
3105 res = try_read_network(c);
3108 res = try_read_udp(c);
3113 conn_set_state(c, conn_waiting);
3116 conn_set_state(c, conn_parse_cmd);
3119 conn_set_state(c, conn_closing);
3128 c->noreply = false;
3129 if (c->try_read_command(c) == 0) {
3131 if (c->resp_head) {
3133 conn_set_state(c, conn_mwrite);
3135 conn_set_state(c, conn_waiting);
3147 reset_cmd_handler(c);
3148 } else if (c->resp_head) {
3150 conn_set_state(c, conn_mwrite);
3152 pthread_mutex_lock(&c->thread->stats.mutex);
3153 c->thread->stats.conn_yields++;
3154 pthread_mutex_unlock(&c->thread->stats.mutex);
3155 if (c->rbytes > 0) {
3162 if (!update_event(c, EV_WRITE | EV_PERSIST)) {
3165 conn_set_state(c, conn_closing);
3174 if (c->rlbytes == 0) {
3175 complete_nread(c);
3180 if (c->rlbytes < 0) {
3182 fprintf(stderr, "Invalid rlbytes to read: len %d\n", c->rlbytes);
3184 conn_set_state(c, conn_closing);
3188 if (c->item_malloced || ((((item *)c->item)->it_flags & ITEM_CHUNKED) == 0) ) {
3190 if (c->rbytes > 0) {
3191 int tocopy = c->rbytes > c->rlbytes ? c->rlbytes : c->rbytes;
3192 memmove(c->ritem, c->rcurr, tocopy);
3193 c->ritem += tocopy;
3194 c->rlbytes -= tocopy;
3195 c->rcurr += tocopy;
3196 c->rbytes -= tocopy;
3197 if (c->rlbytes == 0) {
3203 res = c->read(c, c->ritem, c->rlbytes);
3205 pthread_mutex_lock(&c->thread->stats.mutex);
3206 c->thread->stats.bytes_read += res;
3207 pthread_mutex_unlock(&c->thread->stats.mutex);
3208 if (c->rcurr == c->ritem) {
3209 c->rcurr += res;
3211 c->ritem += res;
3212 c->rlbytes -= res;
3216 res = read_into_chunked_item(c);
3222 c->close_reason = NORMAL_CLOSE;
3223 conn_set_state(c, conn_closing);
3228 if (!update_event(c, EV_READ | EV_PERSIST)) {
3231 conn_set_state(c, conn_closing);
3240 out_of_memory(c, "SERVER_ERROR Out of memory during read");
3241 c->sbytes = c->rlbytes;
3242 conn_set_state(c, conn_swallow);
3246 c->set_stale = false;
3247 c->mset_res = false;
3256 (void *)c->rcurr, (void *)c->ritem, (void *)c->rbuf,
3257 (int)c->rlbytes, (int)c->rsize);
3259 conn_set_state(c, conn_closing);
3264 if (c->sbytes <= 0) {
3265 conn_set_state(c, conn_new_cmd);
3270 if (c->rbytes > 0) {
3271 int tocopy = c->rbytes > c->sbytes ? c->sbytes : c->rbytes;
3272 c->sbytes -= tocopy;
3273 c->rcurr += tocopy;
3274 c->rbytes -= tocopy;
3279 res = c->read(c, c->rbuf, c->rsize > c->sbytes ? c->sbytes : c->rsize);
3281 pthread_mutex_lock(&c->thread->stats.mutex);
3282 c->thread->stats.bytes_read += res;
3283 pthread_mutex_unlock(&c->thread->stats.mutex);
3284 c->sbytes -= res;
3288 c->close_reason = NORMAL_CLOSE;
3289 conn_set_state(c, conn_closing);
3293 if (!update_event(c, EV_READ | EV_PERSIST)) {
3296 conn_set_state(c, conn_closing);
3305 conn_set_state(c, conn_closing);
3314 assert(c->io_queues_submitted == 0);
3316 for (io_queue_t *q = c->io_queues; q->type != IO_QUEUE_NONE; q++) {
3318 io_queue_cb_t *qcb = thread_io_queue_get(c->thread, q->type);
3320 c->io_queues_submitted++;
3323 if (c->io_queues_submitted != 0) {
3324 conn_set_state(c, conn_io_queue);
3330 switch (!IS_UDP(c->transport) ? transmit(c) : transmit_udp(c)) {
3332 if (c->state == conn_mwrite) {
3334 conn_release_items(c);
3335 conn_set_state(c, conn_new_cmd);
3336 if (c->close_after_write) {
3337 conn_set_state(c, conn_closing);
3341 fprintf(stderr, "Unexpected state %d\n", c->state);
3342 conn_set_state(c, conn_closing);
3357 if (IS_UDP(c->transport))
3358 conn_cleanup(c);
3360 conn_close(c);
3375 event_del(&c->event);
3376 conn_set_state(c, conn_io_pending);
3385 conn_set_state(c, conn_mwrite);
3397 conn *c; local
3399 c = (conn *)arg;
3400 assert(c != NULL);
3402 c->which = which;
3405 if (fd != c->sfd) {
3408 conn_close(c);
3412 drive_machine(c);
3614 int c; local
3616 for (c = 0; c < settings.num_threads_per_udp; c++) {
3626 if (c == 0) {
4716 int c; local
4984 while (-1 != (c = getopt_long(argc, argv, shortopts,
4987 while (-1 != (c = getopt(argc, argv, shortopts))) {
4994 switch (c) {
5650 fprintf(stderr, "Illegal argument \"%c\"\n", c);