Dive Deep Memcached ~ SETコマンド実行時の動作 ~
hayashierRedis本を執筆しました。Memcachedとの詳細な比較も付録で取り上げていますので、こちらも是非合わせてご覧ください。
---------------------------------------------------------------------------------------------------------------------------------------------------------
Memcached 1.5.14をソースからビルドしてデフォルト設定時、SETコマンド実行時の動作をソースコードで確認します。 https://github.com/memcached/memcached
概要
最初の初期化
- 1 main at memcached.c
- 2 sanitycheck at memcached.c
- 2 settings_init at memcached.c
- 2 init_lru_maintainer at items.c
- 2 hash_init at hash.c
- 2 logger_init at logger.c
- 3 start_logger_thread at logger.c
- 3 STATS_LOCK, STATS_UNLOCK at thread.c
- 2 stats_init at memcached.c
- 3 stats_prefix_init at stats.c
- 3 assoc_init at assoc.c
- 2 conn_init at memcached.c
- 2 slabs_init at slabs.c
- 2 memcached_thread_init at thread.c
- 3 setup_thread at thread.c
- 4 cq_init at thread.c
- 4 cache_create at cache.c
- 3 create_worker at thread.c
- 3 setup_thread at thread.c
- 2 init_lru_crawler at crawler.c
- 2 start_assoc_maintenance_thread at assoc.c
- 2 start_item_crawler_thread at crawler.c
- 2 start_lru_maintainer_thread at items.c
- 2 start_slab_maintenance_thread at slabs.c
- 2 clock_handler at memcached.c
- 3 assoc_start_expand at assoc.c
- 2 server_sockets at memcached.c
- 3 server_socket at memcached.c
- 4 new_socket at memcached.c
- 4 conn_new at memcached.c
- 3 server_socket at memcached.c
- 2 uriencode_init at util.c
- 2 event_base_loop at memcached.c : main_baseに対するイベント処理のループ
SETコマンドの挙動
- 1 event_handler at memcached.c
- 2 drive_machine at memcached.c
- 3 dispatch_conn_new at thread.c
- 4 cqi_new at thread.c
- 4 cq_push at thread.c
- 3 reset_cmd_handler at memcached.c
- 4 conn_shrink at memcached.c
- 4 conn_set_state at memcached.c
- 3 update_event at memcached.c
- 3 try_read_network at memcached.c
- 4 tcp_read at memcached.c
- 3 try_read_command at memcached.c
- 4 process_command at memcached.c
- 5 add_msghdr at memcached.c
- 5 tokenize_command at memcached.c
- 5 process_update_command at memcached.c
- 6 set_noreply_maybe at memcached.c
- 6 realtime at memcached.c
- 6 item_alloc at thread.c
- 7 do_item_alloc at items.c
- 8 item_make_header at items.c
- 8 slabs_clsid at slabs.c
- 8 slabs_alloc at slabs.c
- 9 do_slabs_alloc at slabs.c
- 10 do_slabs_newslab at slabs.c
- 11 grow_slab_list at slabs.c
- 11 memory_allocate at slabs.c
- 11 split_slab_page_into_freelist at slabs.c
- 12 do_slabs_free at slabs.c
- 10 do_slabs_newslab at slabs.c
- 9 do_slabs_alloc at slabs.c
- 7 do_item_alloc at items.c
- 4 process_command at memcached.c
- 3 complete_nread at memcached.c
- 4 complete_nread_ascii at memcached.c
- 5 store_item at thread.c
- 6 hash(MurmurHash3_x86_32)
- 7 getblock32 at murmur3_hash.c
- 7 rotl32 at murmur3_hash.c
- 7 fmix32 at murmur3_hash.c
- 6 item_lock, item_unlock at thread.c
- 6 do_store_item at memcached.c
- 7 do_item_get at items.c
- 7 do_item_link at items.c
- 8 get_cas_id at items.c
- 8 assoc_insert at assoc.c
- 8 item_link_q at items.c
- 9 do_item_link_q at items.c
- 8 item_stats_sizes_add at items.c
- 6 hash(MurmurHash3_x86_32)
- 5 out_string at memcached.c
- 5 item_remove at thread.c
- 6 do_item_remove at items.c
- 5 store_item at thread.c
- 4 complete_nread_ascii at memcached.c
- 3 add_iov at memcached.c
- 4 ensure_iov_space at memcached.c
- 3 transmit at memcached.c
- 4 tcp_sendmsg at memcached.c
- 3 reset_cmd_handler at memcached.c
- 3 dispatch_conn_new at thread.c
- 2 drive_machine at memcached.c
詳細
最初の初期化
memcached -u ec2-user -vvv コマンドを実行した際の挙動
1 main at memcached.c
6762│ int main (int argc, char **argv) {
6763│ int c;
6764│ bool lock_memory = false;
6765│ bool do_daemonize = false;
6766│ bool preallocate = false;
6767│ int maxcore = 0;
6768│ char *username = NULL;
6769│ char *pid_file = NULL;
6770│ struct passwd *pw;
6771│ struct rlimit rlim;
6772│ char *buf;
6773│ char unit = '\0';
6774│ int size_max = 0;
6775├───> int retval = EXIT_SUCCESS;
6776│ bool protocol_specified = false;
6777│ bool tcp_specified = false;
6778│ bool udp_specified = false;
6779│ bool start_lru_maintainer = true;
6780│ bool start_lru_crawler = true;
6781│ bool start_assoc_maint = true;
6782│ enum hashfunc_type hash_type = MURMUR3_HASH;
6783│ uint32_t tocrawl;
6784│ uint32_t slab_sizes[MAX_NUMBER_OF_SLAB_CLASSES];
6785│ bool use_slab_sizes = false;
6786│ char *slab_sizes_unparsed = NULL;
6787│ bool slab_chunk_size_changed = false;
sanitycheck関数後に、signal関数。その後、settings_init
6929├───> if (!sanitycheck()) {
6930│ return EX_OSERR;
6931│ }
6932│
6933│ /* handle SIGINT, SIGTERM */
6934│ signal(SIGINT, sig_handler);
6935│ signal(SIGTERM, sig_handler);
6936│
6937│ /* init settings */
6938│ settings_init();
6939│ #ifdef EXTSTORE
6940│ settings.ext_item_size = 512;
6941│ settings.ext_item_age = UINT_MAX;
2 sanitycheck at memcached.c
event_get_versionはlibeventの関数。ここでは、"2.0.21-stable"という文字列を取得しeverに格納されている。
6702│ /**
6703│ * Do basic sanity check of the runtime environment
6704│ * @return true if no errors found, false if we can't use this env
6705│ */
6706│ static bool sanitycheck(void) {
6707│ /* One of our biggest problems is old and bogus libevents */
6708├───> const char *ever = event_get_version();
6709│ if (ever != NULL) {
6710│ if (strncmp(ever, "1.", 2) == 0) {
6711│ /* Require at least 1.3 (that's still a couple of years old) */
6712│ if (('0' <= ever[2] && ever[2] < '3') && !isdigit(ever[3])) {
6713│ fprintf(stderr, "You are using libevent %s.\nPlease upgrade to"
6714│ " a more recent version (1.3 or newer)\n",
6715│ event_get_version());
6716│ return false;
6717│ }
6718│ }
6719│ }
6720│
2 settings_init at memcached.c
パラメータのデフォルト値が設定されている
247│ static void settings_init(void) {
248│ settings.use_cas = true;
249│ settings.access = 0700;
250│ settings.port = 11211;
251├───> settings.udpport = 0;
252│ #ifdef TLS
253│ settings.ssl_enabled = false;
254│ settings.ssl_ctx = NULL;
255│ settings.ssl_chain_cert = NULL;
256│ settings.ssl_key = NULL;
257│ settings.ssl_verify_mode = SSL_VERIFY_NONE;
258│ settings.ssl_keyformat = SSL_FILETYPE_PEM;
259│ settings.ssl_ciphers = NULL;
260│ settings.ssl_ca_cert = NULL;
261│ settings.ssl_last_cert_refresh_time = current_time;
262│ settings.ssl_wbuf_size = 16 * 1024; // default is 16KB (SSL max frame size is 17KB)
263│ #endif
264│ /* By default this string should be NULL for getaddrinfo() */
265│ settings.inter = NULL;
266│ settings.maxbytes = 64 * 1024 * 1024; /* default is 64MB */
267│ settings.maxconns = 1024; /* to limit connections-related memory to about 5MB */
268│ settings.verbose = 0;
269│ settings.oldest_live = 0;
270│ settings.oldest_cas = 0; /* supplements accuracy of oldest_live */
271│ settings.evict_to_free = 1; /* push old items out of cache when memory runs out */
272│ settings.socketpath = NULL; /* by default, not using a unix socket */
273│ settings.factor = 1.25;
274│ settings.chunk_size = 48; /* space for a modest key and value */
275│ settings.num_threads = 4; /* N workers */
276│ settings.num_threads_per_udp = 0;
277├───> settings.prefix_delimiter = ':';
278│ settings.detail_enabled = 0;
279│ settings.reqs_per_event = 20;
280│ settings.backlog = 1024;
281│ settings.binding_protocol = negotiating_prot;
282│ settings.item_size_max = 1024 * 1024; /* The famous 1MB upper limit. */
283│ settings.slab_page_size = 1024 * 1024; /* chunks are split from 1MB pages. */
284│ settings.slab_chunk_size_max = settings.slab_page_size / 2;
285│ settings.sasl = false;
286│ settings.maxconns_fast = true;
287│ settings.lru_crawler = false;
288│ settings.lru_crawler_sleep = 100;
289│ settings.lru_crawler_tocrawl = 0;
290│ settings.lru_maintainer_thread = false;
291│ settings.lru_segmented = true;
292│ settings.hot_lru_pct = 20;
293│ settings.warm_lru_pct = 40;
294│ settings.hot_max_factor = 0.2;
295│ settings.warm_max_factor = 2.0;
296│ settings.inline_ascii_response = false;
297│ settings.temp_lru = false;
298│ settings.temporary_ttl = 61;
299│ settings.idle_timeout = 0; /* disabled */
300│ settings.hashpower_init = 0;
301│ settings.slab_reassign = true;
302│ settings.slab_automove = 1;
303├───> settings.slab_automove_ratio = 0.8;
304│ settings.slab_automove_window = 30;
305│ settings.shutdown_command = false;
306│ settings.tail_repair_time = TAIL_REPAIR_TIME_DEFAULT;
307│ settings.flush_enabled = true;
308│ settings.dump_enabled = true;
309│ settings.crawls_persleep = 1000;
310│ settings.logger_watcher_buf_size = LOGGER_WATCHER_BUF_SIZE;
311│ settings.logger_buf_size = LOGGER_BUF_SIZE;
312│ settings.drop_privileges = false;
313│ #ifdef MEMCACHED_DEBUG
314│ settings.relaxed_privileges = false;
315│ #endif
1 main at memcached.c
6957│
6958│ /* Run regardless of initializing it later */
6959├───> init_lru_maintainer();
6960│
6961│ /* set stderr non-buffering (for running under, say, daemontools) */
6962│ setbuf(stderr, NULL);
6963│
6964│ char *shortopts =
6965│ "a:" /* access mask for unix socket */
6966│ "A" /* enable admin shutdown command */
6967│ "Z" /* enable SSL */
6968│ "p:" /* TCP port number to listen on */
6969│ "s:" /* unix socket path to listen on */
6970│ "U:" /* UDP port number to listen on */
6971│ "m:" /* max memory to use for items in megabytes */
2 init_lru_maintainer at items.c
1704│ int init_lru_maintainer(void) {
1705├───> if (lru_maintainer_initialized == 0) {
1706│ pthread_mutex_init(&lru_maintainer_lock, NULL);
1707│ lru_maintainer_initialized = 1;
1708│ }
1709│ return 0;
1710│ }
1 main at memcached.c
6998│ /* process arguments */
6999│ #ifdef HAVE_GETOPT_LONG
7000├───> const struct option longopts[] = {
7001│ {"unix-mask", required_argument, 0, 'a'},
7002│ {"enable-shutdown", no_argument, 0, 'A'},
7003│ {"enable-ssl", no_argument, 0, 'Z'},
7004│ {"port", required_argument, 0, 'p'},
7005│ {"unix-socket", required_argument, 0, 's'},
7006│ {"udp-port", required_argument, 0, 'U'},
7007│ {"memory-limit", required_argument, 0, 'm'},
7008│ {"disable-evictions", no_argument, 0, 'M'},
7009│ {"conn-limit", required_argument, 0, 'c'},
7010│ {"lock-memory", no_argument, 0, 'k'},
7011│ {"help", no_argument, 0, 'h'},
7012│ {"license", no_argument, 0, 'i'},
:
:
7028│ {"max-item-size", required_argument, 0, 'I'},
7029│ {"enable-sasl", no_argument, 0, 'S'},
7030│ {"disable-flush-all", no_argument, 0, 'F'},
7031│ {"disable-dumping", no_argument, 0, 'X'},
7032│ {"extended", required_argument, 0, 'o'},
7033│ {0, 0, 0, 0}
7034│ };
7035│ int optindex;
7036│ while (-1 != (c = getopt_long(argc, argv, shortopts,
7037│ longopts, &optindex))) {
7038│ #else
7039│ while (-1 != (c = getopt(argc, argv, shortopts))) {
7040│ #endif
7041├───────> switch (c) {
7042│ case 'A':
7043│ /* enables "shutdown" command */
7044│ settings.shutdown_command = true;
7045│ break;
7046│ case 'Z':
7047│ /* enable secure communication*/
7048│ #ifdef TLS
7049│ settings.ssl_enabled = true;
7050│ #else
7051│ fprintf(stderr, "This server is not built with TLS support.\n");
7052│ exit(EX_USAGE);
7053│ #endif
-u ec2-user のようにユーザー名を与えているので、username 変数に"ec2-user"が設定される。
7119│ case 'r':
7120│ maxcore = 1;
7121│ break;
7122│ case 'R':
7123│ settings.reqs_per_event = atoi(optarg);
7124│ if (settings.reqs_per_event == 0) {
7125│ fprintf(stderr, "Number of requests per event must be greater than 0\n");
7126│ return 1;
7127│ }
7128│ break;
7129│ case 'u':
7130├───────────> username = optarg;
7131│ break;
7132│ case 'P':
7133│ pid_file = optarg;
7134│ break;
7135│ case 'f':
7136│ settings.factor = atof(optarg);
7137│ if (settings.factor <= 1.0) {
7138│ fprintf(stderr, "Factor must be greater than 1\n");
7139│ return 1;
7140│ }
7141│ break;
-vvv と指定しているので、getopt_longの取得結果のvの処理はwhile文で3回繰り返される。
7083│ case 'h':
7084│ usage();
7085│ exit(EXIT_SUCCESS);
7086│ case 'i':
7087│ usage_license();
7088│ exit(EXIT_SUCCESS);
7089│ case 'V':
7090│ printf(PACKAGE " " VERSION "\n");
7091│ exit(EXIT_SUCCESS);
7092│ case 'k':
7093│ lock_memory = true;
7094│ break;
7095│ case 'v':
7096├───────────> settings.verbose++;
7097│ break;
7098│ case 'l':
7099│ if (settings.inter != NULL) {
7100│ if (strstr(settings.inter, optarg) != NULL) {
7101│ break;
7102│ }
7103│ size_t len = strlen(settings.inter) + strlen(optarg) + 2;
7104│ char *p = malloc(len);
7105│ if (p == NULL) {
7106│ fprintf(stderr, "Failed to allocate memory\n");
7107│ return 1;
7108│ }
その後、パラメータの値が、制限値外のものが指定されていたらエラーで終了する。
7746│ if (settings.item_size_max < 1024) {
7747│ fprintf(stderr, "Item max size cannot be less than 1024 bytes.\n");
7748│ exit(EX_USAGE);
7749│ }
7750│ if (settings.item_size_max > (settings.maxbytes / 2)) {
7751│ fprintf(stderr, "Cannot set item size limit higher than 1/2 of memory max.\n");
7752│ exit(EX_USAGE);
7753│ }
7754│ if (settings.item_size_max > (1024 * 1024 * 1024)) {
7755│ fprintf(stderr, "Cannot set item size limit higher than a gigabyte.\n");
7756│ exit(EX_USAGE);
7757│ }
7758├───> if (settings.item_size_max > 1024 * 1024) {
7759│ if (!slab_chunk_size_changed) {
7760│ // Ideal new default is 16k, but needs stitching.
7761│ settings.slab_chunk_size_max = settings.slab_page_size / 2;
7762│ }
7763│ }
7764│
7765│ if (settings.slab_chunk_size_max > settings.item_size_max) {
7766│ fprintf(stderr, "slab_chunk_max (bytes: %d) cannot be larger than -I (item_size_max %d)\n",
7767│ settings.slab_chunk_size_max, settings.item_size_max);
7768│ exit(EX_USAGE);
7769│ }
7770│
7771├───> if (settings.item_size_max % settings.slab_chunk_size_max != 0) {
7772│ fprintf(stderr, "-I (item_size_max: %d) must be evenly divisible by slab_chunk_max (bytes: %d)\n",
7773│ settings.item_size_max, settings.slab_chunk_size_max);
7774│ exit(EX_USAGE);
7775│ }
7776│
7777│ if (settings.slab_page_size % settings.slab_chunk_size_max != 0) {
7778│ fprintf(stderr, "slab_chunk_max (bytes: %d) must divide evenly into %d (slab_page_size)\n",
7779│ settings.slab_chunk_size_max, settings.slab_page_size);
7780│ exit(EX_USAGE);
7781│ }
:
:
7811│ if (slab_sizes_unparsed != NULL) {
7812│ if (_parse_slab_sizes(slab_sizes_unparsed, slab_sizes)) {
7813│ use_slab_sizes = true;
7814│ } else {
7815│ exit(EX_USAGE);
7816│ }
7817│ }
7818│
7819│ if (settings.hot_lru_pct + settings.warm_lru_pct > 80) {
7820│ fprintf(stderr, "hot_lru_pct + warm_lru_pct cannot be more than 80%% combined\n");
7821│ exit(EX_USAGE);
7822│ }
7823│
7824├───> if (settings.temp_lru && !start_lru_maintainer) {
7825│ fprintf(stderr, "temporary_ttl requires lru_maintainer to be enabled\n");
7826│ exit(EX_USAGE);
7827│ }
7828│
7829│ if (hash_init(hash_type) != 0) {
7830│ fprintf(stderr, "Failed to initialize hash_algorithm!\n");
7831│ exit(EX_USAGE);
7832│ }
hash_init関数が実行されている
2 hash_init at hash.c
typeでMURMUR3_HASHが指定されている。
7│ int hash_init(enum hashfunc_type type) {
8├───> switch(type) {
9│ case JENKINS_HASH:
10│ hash = jenkins_hash;
11│ settings.hash_algorithm = "jenkins";
12│ break;
13│ case MURMUR3_HASH:
14│ hash = MurmurHash3_x86_32;
15│ settings.hash_algorithm = "murmur3";
16│ break;
17│ default:
18│ return -1;
19│ }
20│ return 0;
21│ }
1 main at memcacached.c
必要に応じて多くのコネクション数を許可するためにgetrlimit
7903│ /*
7904│ * If needed, increase rlimits to allow as many connections
7905│ * as needed.
7906│ */
7907│
7908├───> if (getrlimit(RLIMIT_NOFILE, &rlim) != 0) {
7909│ fprintf(stderr, "failed to getrlimit number of files\n");
7910│ exit(EX_OSERR);
7911│ } else {
7912│ rlim.rlim_cur = settings.maxconns;
7913│ rlim.rlim_max = settings.maxconns;
7914│ if (setrlimit(RLIMIT_NOFILE, &rlim) != 0) {
7915│ fprintf(stderr, "failed to set rlimit for open files. Try starting as root or requesting smaller
7916│ exit(EX_OSERR);
7917│ }
7918│ }
mainスレッドのlibeventインスタンスの初期化
7970│ /* initialize main thread libevent instance */
7971│ #if defined(LIBEVENT_VERSION_NUMBER) && LIBEVENT_VERSION_NUMBER >= 0x02000101
7972│ /* If libevent version is larger/equal to 2.0.2-alpha, use newer version */
7973│ struct event_config *ev_config;
7974├───> ev_config = event_config_new();
7975│ event_config_set_flag(ev_config, EVENT_BASE_FLAG_NOLOCK);
7976│ main_base = event_base_new_with_config(ev_config);
7977│ event_config_free(ev_config);
7978│ #else
7979│ /* Otherwise, use older API */
7980│ main_base = event_init();
7981│ #endif
各種スレッド等の初期化
7983│ /* initialize other stuff */
7984├───> logger_init();
7985│ stats_init();
7986│ assoc_init(settings.hashpower_init);
7987│ conn_init();
7988│ slabs_init(settings.maxbytes, settings.factor, preallocate,
7989│ use_slab_sizes ? slab_sizes : NULL);
2 logger_init at logger.c
563│ /*************************
564│ * Public functions for submitting logs and starting loggers from workers.
565│ *************************/
566│
567│ /* Global logger thread start/init */
568│ void logger_init(void) {
569│ /* TODO: auto destructor when threads exit */
570│ /* TODO: error handling */
571│
572│ /* init stack for iterating loggers */
573│ logger_stack_head = 0;
574│ logger_stack_tail = 0;
575├───> pthread_key_create(&logger_key, NULL);
576│
577│ if (start_logger_thread() != 0) {
578│ abort();
579│ }
580│
581│ /* This can be removed once the global stats initializer is improved */
582│ STATS_LOCK();
583│ stats.log_worker_dropped = 0;
584│ stats.log_worker_written = 0;
585│ stats.log_watcher_skipped = 0;
586│ stats.log_watcher_sent = 0;
587│ STATS_UNLOCK();
588│ /* This is what adding a STDERR watcher looks like. should replace old
589│ * "verbose" settings. */
590│ //logger_add_watcher(NULL, 0);
591│ return;
592│ }
3 start_logger_thread at logger.c
545│ static int start_logger_thread(void) {
546│ int ret;
547├───> do_run_logger_thread = 1;
548│ if ((ret = pthread_create(&logger_tid, NULL,
549│ logger_thread, NULL)) != 0) {
550│ fprintf(stderr, "Can't start logger thread: %s\n", strerror(ret));
551│ return -1;
552│ }
553│ return 0;
554│ }
3 STATS_LOCK, STATS_UNLOCK at thread.c
696│ /******************************* GLOBAL STATS ******************************/
697│
698│ void STATS_LOCK() {
699│ pthread_mutex_lock(&stats_lock);
700│ }
701│
702│ void STATS_UNLOCK() {
703├───> pthread_mutex_unlock(&stats_lock);
704│ }
2 stats_init at memcached.c
225│ static void stats_init(void) {
226├───> memset(&stats, 0, sizeof(struct stats));
227│ memset(&stats_state, 0, sizeof(struct stats_state));
228│ stats_state.accepting_conns = true; /* assuming we start in this state. */
229│
230│ /* make the time we started always be 2 seconds before we really
231│ did, so time(0) - time.started is never zero. if so, things
232│ like 'settings.oldest_live' which act as booleans as well as
233│ values are now false in boolean context... */
234│ process_started = time(0) - ITEM_UPDATE_INTERVAL - 2;
235│ stats_prefix_init();
236│ }
3 stats_prefix_init at stats.c
38│ void stats_prefix_init() {
39├───> memset(prefix_stats, 0, sizeof(prefix_stats));
40│ }
3 assoc_init at assoc.c
59│ void assoc_init(const int hashtable_init) {
60├───> if (hashtable_init) {
61│ hashpower = hashtable_init;
62│ }
63│ primary_hashtable = calloc(hashsize(hashpower), sizeof(void *));
64│ if (! primary_hashtable) {
65│ fprintf(stderr, "Failed to init hashtable.\n");
66│ exit(EXIT_FAILURE);
67│ }
68│ STATS_LOCK();
69│ stats_state.hash_power_level = hashpower;
70│ stats_state.hash_bytes = hashsize(hashpower) * sizeof(void *);
71│ STATS_UNLOCK();
72│ }
2 conn_init at memcached.c
448│ /*
449│ * Initializes the connections array. We don't actually allocate connection
450│ * structures until they're needed, so as to avoid wasting memory when the
451│ * maximum connection count is much higher than the actual number of
452│ * connections.
453│ *
454│ * This does end up wasting a few pointers' worth of memory for FDs that are
455│ * used for things other than connections, but that's worth it in exchange for
456│ * being able to directly index the conns array by FD.
457│ */
458│ static void conn_init(void) {
459│ /* We're unlikely to see an FD much higher than maxconns. */
460│ int next_fd = dup(1);
461├───> int headroom = 10; /* account for extra unexpected open FDs */
462│ struct rlimit rl;
463│
464│ max_fds = settings.maxconns + headroom + next_fd;
465│
466│ /* But if possible, get the actual highest FD we can possibly ever see. */
467│ if (getrlimit(RLIMIT_NOFILE, &rl) == 0) {
468│ max_fds = rl.rlim_max;
469│ } else {
470│ fprintf(stderr, "Failed to query maximum file descriptor; "
471│ "falling back to maxconns\n");
472│ }
473│
2 slabs_init (limit=67108864, factor=1.25, prealloc=false, slab_sizes=0x0) at slabs.c:158
Slabの初期化 ★5
153│ /**
154│ * Determines the chunk sizes and initializes the slab class descriptors
155│ * accordingly.
156│ */
157│ void slabs_init(const size_t limit, const double factor, const bool prealloc, const uint32_t *slab_sizes) {
158│ int i = POWER_SMALLEST - 1;
159│ unsigned int size = sizeof(item) + settings.chunk_size;
160│
161│ /* Some platforms use runtime transparent hugepages. If for any reason
162│ * the initial allocation fails, the required settings do not persist
163│ * for remaining allocations. As such it makes little sense to do slab
164│ * preallocation. */
165├───> bool __attribute__ ((unused)) do_slab_prealloc = false;
166│
167│ mem_limit = limit;
168│
169│ if (prealloc) {
170│ #if defined(__linux__) && defined(MADV_HUGEPAGE)
171│ mem_base = alloc_large_chunk_linux(mem_limit);
172│ if (mem_base)
173│ do_slab_prealloc = true;
174│ #else
175│ /* Allocate everything in a big chunk with malloc */
176│ mem_base = malloc(mem_limit);
177│ do_slab_prealloc = true;
slabclass配列の初期化
190│ while (++i < MAX_NUMBER_OF_SLAB_CLASSES-1) {
191│ if (slab_sizes != NULL) {
192│ if (slab_sizes[i-1] == 0)
193│ break;
194│ size = slab_sizes[i-1];
195│ } else if (size >= settings.slab_chunk_size_max / factor) {
196│ break;
197│ }
198│ /* Make sure items are always n-byte aligned */
199│ if (size % CHUNK_ALIGN_BYTES)
200│ size += CHUNK_ALIGN_BYTES - (size % CHUNK_ALIGN_BYTES);
201│
202├───────> slabclass[i].size = size;
203│ slabclass[i].perslab = settings.slab_page_size / slabclass[i].size;
204│ if (slab_sizes == NULL)
205│ size *= factor;
206│ if (settings.verbose > 1) {
207│ fprintf(stderr, "slab class %3d: chunk size %9u perslab %7u\n",
208│ i, slabclass[i].size, slabclass[i].perslab);
209│ }
210│ }
220│ /* for the test suite: faking of how much we've already malloc'd */
221│ {
222│ char *t_initial_malloc = getenv("T_MEMD_INITIAL_MALLOC");
223│ if (t_initial_malloc) {
224│ mem_malloced = (size_t)atol(t_initial_malloc);
225│ }
226│
227│ }
228│
229│ if (prealloc && do_slab_prealloc) {
230│ slabs_preallocate(power_largest);
231│ }
232├>}
1 main at memcacached.c
workerスレッドの設定
8025│ /* start up worker threads if MT mode */
8026│ #ifdef EXTSTORE
8027│ slabs_set_storage(storage);
8028│ memcached_thread_init(settings.num_threads, storage);
8029│ init_lru_crawler(storage);
8030│ #else
8031├───> memcached_thread_init(settings.num_threads, NULL);
8032│ init_lru_crawler(NULL);
8033│ #endif
8034│
8035│ if (start_assoc_maint && start_assoc_maintenance_thread() == -1) {
8036│ exit(EXIT_FAILURE);
8037│ }
8038│ if (start_lru_crawler && start_item_crawler_thread() != 0) {
8039│ fprintf(stderr, "Failed to enable LRU crawler thread\n");
8040│ exit(EXIT_FAILURE);
8041│ }
8042│ #ifdef EXTSTORE
8043│ if (storage && start_storage_compact_thread(storage) != 0) {
8044│ fprintf(stderr, "Failed to start storage compaction thread\n");
8045│ exit(EXIT_FAILURE);
8046│ }
8047│ if (storage && start_storage_write_thread(storage) != 0) {
8048│ fprintf(stderr, "Failed to start storage writer thread\n");
8049│ exit(EXIT_FAILURE);
8050│ }
8051│
8052│ if (start_lru_maintainer && start_lru_maintainer_thread(storage) != 0) {
8053│ #else
8054├───> if (start_lru_maintainer && start_lru_maintainer_thread(NULL) != 0) {
8055│ #endif
8056│ fprintf(stderr, "Failed to enable LRU maintainer thread\n");
8057│ return 1;
8058│ }
8059│
8060│ if (settings.slab_reassign &&
8061├───────> start_slab_maintenance_thread() == -1) {
8062│ exit(EXIT_FAILURE);
8063│ }
8064│
8065│ if (settings.idle_timeout && start_conn_timeout_thread() == -1) {
8066│ exit(EXIT_FAILURE);
8067│ }
8068│
2 memcached_thread_init at thread.c
772│ /*
773│ * Initializes the thread subsystem, creating various worker threads.
774│ *
775│ * nthreads Number of worker event handler threads to spawn
776│ */
777│ void memcached_thread_init(int nthreads, void *arg) {
778│ int i;
779│ int power;
780│
781│ for (i = 0; i < POWER_LARGEST; i++) {
782├───────> pthread_mutex_init(&lru_locks[i], NULL);
783│ }
784│ pthread_mutex_init(&worker_hang_lock, NULL);
785│
786│ pthread_mutex_init(&init_lock, NULL);
787│ pthread_cond_init(&init_cond, NULL);
788│
789│ pthread_mutex_init(&cqi_freelist_lock, NULL);
790│ cqi_freelist = NULL;
791│
792│ /* Want a wide lock table, but don't waste memory */
793│ if (nthreads < 3) {
794│ power = 10;
815│ item_lock_count = hashsize(power);
816│ item_lock_hashpower = power;
817│
818│ item_locks = calloc(item_lock_count, sizeof(pthread_mutex_t));
819│ if (! item_locks) {
820│ perror("Can't allocate item locks");
821│ exit(1);
822│ }
823│ for (i = 0; i < item_lock_count; i++) {
824│ pthread_mutex_init(&item_locks[i], NULL);
825│ }
826│
827├───> threads = calloc(nthreads, sizeof(LIBEVENT_THREAD));
828│ if (! threads) {
829│ perror("Can't allocate thread descriptors");
830│ exit(1);
831│ }
832│
833│ for (i = 0; i < nthreads; i++) {
834│ int fds[2];
835│ if (pipe(fds)) {
836│ perror("Can't create notify pipe");
837│ exit(1);
838│ }
839│
:
:
833│ for (i = 0; i < nthreads; i++) {
834│ int fds[2];
835│ if (pipe(fds)) {
836│ perror("Can't create notify pipe");
837│ exit(1);
838│ }
839│
840│ threads[i].notify_receive_fd = fds[0];
841│ threads[i].notify_send_fd = fds[1];
842│ #ifdef EXTSTORE
843│ threads[i].storage = arg;
844│ #endif
845├───────> setup_thread(&threads[i]);
846│ /* Reserve three fds for the libevent base, and two for the pipe */
847│ stats_state.reserved_fds += 5;
848│ }
849│
850│ /* Create threads after we've done all the libevent setup. */
851│ for (i = 0; i < nthreads; i++) {
852│ create_worker(worker_libevent, &threads[i]);
853│ }
854│
855│ /* Wait for all the threads to set themselves up before returning. */
856│ pthread_mutex_lock(&init_lock);
857│ wait_for_thread_registration(nthreads);
858│ pthread_mutex_unlock(&init_lock);
859│ }
3 setup_thread (me=0x6486f0) at thread.c:321
313│ /****************************** LIBEVENT THREADS *****************************/
314│
315│ /*
316│ * Set up a thread's information.
317│ */
318│ static void setup_thread(LIBEVENT_THREAD *me) {
319│ #if defined(LIBEVENT_VERSION_NUMBER) && LIBEVENT_VERSION_NUMBER >= 0x02000101
320│ struct event_config *ev_config;
321│ ev_config = event_config_new();
322│ event_config_set_flag(ev_config, EVENT_BASE_FLAG_NOLOCK);
323│ me->base = event_base_new_with_config(ev_config);
324├───> event_config_free(ev_config);
325│ #else
326│ me->base = event_init();
327│ #endif
328│
329│ if (! me->base) {
330│ fprintf(stderr, "Can't allocate event base\n");
331│ exit(1);
332│ }
333│
334│ /* Listen for notifications from other threads */
335│ event_set(&me->notify_event, me->notify_receive_fd,
336│ EV_READ | EV_PERSIST, thread_libevent_process, me);
337│ event_base_set(me->base, &me->notify_event);
338│
339│ if (event_add(&me->notify_event, 0) == -1) {
340│ fprintf(stderr, "Can't monitor libevent notify pipe\n");
341│ exit(1);
342│ }
343│
344│ me->new_conn_queue = malloc(sizeof(struct conn_queue));
345├───> if (me->new_conn_queue == NULL) {
346│ perror("Failed to allocate memory for connection queue");
347│ exit(EXIT_FAILURE);
348│ }
349│ cq_init(me->new_conn_queue);
350│
351│ if (pthread_mutex_init(&me->stats.mutex, NULL) != 0) {
352│ perror("Failed to initialize mutex");
353│ exit(EXIT_FAILURE);
354│ }
355│
356│ me->suffix_cache = cache_create("suffix", SUFFIX_SIZE, sizeof(char*),
357│ NULL, NULL);
4 cq_init (cq=0x64fab0) at thread.c:196
192│ /*
193│ * Initializes a connection queue.
194│ */
195│ static void cq_init(CQ *cq) {
196├───> pthread_mutex_init(&cq->lock, NULL);
197│ cq->head = NULL;
198│ cq->tail = NULL;
199│ }
4 cache_create (name=0x42b442 "suffix", bufsize=50, align=8, constructor=0x0, destructor=0x0) at cache.c:22
19│ cache_t* cache_create(const char *name, size_t bufsize, size_t align,
20│ cache_constructor_t* constructor,
21│ cache_destructor_t* destructor) {
22│ cache_t* ret = calloc(1, sizeof(cache_t));
23│ char* nm = strdup(name);
24│ void** ptr = calloc(initial_pool_size, sizeof(void*));
25├───> if (ret == NULL || nm == NULL || ptr == NULL ||
26│ pthread_mutex_init(&ret->mutex, NULL) == -1) {
27│ free(ret);
28│ free(nm);
29│ free(ptr);
30│ return NULL;
31│ }
32│
33│ ret->name = nm;
34│ ret->ptr = ptr;
35│ ret->freetotal = initial_pool_size;
36│ ret->constructor = constructor;
37│ ret->destructor = destructor;
38│
39│ #ifndef NDEBUG
40│ ret->bufsize = bufsize + 2 * sizeof(redzone_pattern);
41│ #else
42│ ret->bufsize = bufsize;
43│ #endif
44│
45├───> return ret;
46│ }
3 create_worker (func=0x41dddf <worker_libevent>, arg=0x6486f0) at thread.c:296
289│ /*
290│ * Creates a worker thread.
291│ */
292│ static void create_worker(void *(*func)(void *), void *arg) {
293│ pthread_attr_t attr;
294│ int ret;
295│
296├───> pthread_attr_init(&attr);
297│
298│ if ((ret = pthread_create(&((LIBEVENT_THREAD*)arg)->thread_id, &attr, func, arg)) != 0) {
299│ fprintf(stderr, "Can't create thread: %s\n",
300│ strerror(ret));
301│ exit(1);
302│ }
303│ }
2 init_lru_crawler (arg=0x0) at crawler.c:686
685│ int init_lru_crawler(void *arg) {
686│ if (lru_crawler_initialized == 0) {
687│ #ifdef EXTSTORE
688│ storage = arg;
689│ #endif
690├───────> if (pthread_cond_init(&lru_crawler_cond, NULL) != 0) {
691│ fprintf(stderr, "Can't initialize lru crawler condition\n");
692│ return -1;
693│ }
694│ pthread_mutex_init(&lru_crawler_lock, NULL);
695│ active_crawler_mod.c.c = NULL;
696│ active_crawler_mod.mod = NULL;
697│ active_crawler_mod.data = NULL;
698│ lru_crawler_initialized = 1;
699│ }
700│ return 0;
701│ }
2 start_assoc_maintenance_thread () at assoc.c:270
268│ int start_assoc_maintenance_thread() {
269│ int ret;
270│ char *env = getenv("MEMCACHED_HASH_BULK_MOVE");
271│ if (env != NULL) {
272│ hash_bulk_move = atoi(env);
273│ if (hash_bulk_move == 0) {
274│ hash_bulk_move = DEFAULT_HASH_BULK_MOVE;
275│ }
276│ }
277├───> pthread_mutex_init(&maintenance_lock, NULL);
278│ if ((ret = pthread_create(&maintenance_tid, NULL,
279│ assoc_maintenance_thread, NULL)) != 0) {
280│ fprintf(stderr, "Can't create thread: %s\n", strerror(ret));
281│ return -1;
282│ }
283│ return 0;
284│ }
2 start_item_crawler_thread () at crawler.c:497
484│ * caller locks mtx. caller spawns thread.
485│ * thread blocks on mutex.
486│ * caller waits on condition, releases lock.
487│ * thread gets lock, sends signal.
488│ * caller can't wait, as thread has lock.
489│ * thread waits on condition, releases lock
490│ * caller wakes on condition, gets lock.
491│ * caller immediately releases lock.
492│ * thread is now safely waiting on condition before the caller returns.
493│ */
494│ int start_item_crawler_thread(void) {
495│ int ret;
496│
497├───> if (settings.lru_crawler)
498│ return -1;
499│ pthread_mutex_lock(&lru_crawler_lock);
500│ do_run_lru_crawler_thread = 1;
501│ if ((ret = pthread_create(&item_crawler_tid, NULL,
502│ item_crawler_thread, NULL)) != 0) {
503│ fprintf(stderr, "Can't create LRU crawler thread: %s\n",
504│ strerror(ret));
505│ pthread_mutex_unlock(&lru_crawler_lock);
506│ return -1;
507│ }
508│ /* Avoid returning until the crawler has actually started */
509│ pthread_cond_wait(&lru_crawler_cond, &lru_crawler_lock);
510│ pthread_mutex_unlock(&lru_crawler_lock);
511│
512│ return 0;
513│ }
2 start_lru_maintainer_thread (arg=0x0) at items.c:1680
1677│ int start_lru_maintainer_thread(void *arg) {
1678│ int ret;
1679│
1680│ pthread_mutex_lock(&lru_maintainer_lock);
1681├───> do_run_lru_maintainer_thread = 1;
1682│ settings.lru_maintainer_thread = true;
1683│ if ((ret = pthread_create(&lru_maintainer_tid, NULL,
1684│ lru_maintainer_thread, arg)) != 0) {
1685│ fprintf(stderr, "Can't create LRU maintainer thread: %s\n",
1686│ strerror(ret));
1687│ pthread_mutex_unlock(&lru_maintainer_lock);
1688│ return -1;
1689│ }
1690│ pthread_mutex_unlock(&lru_maintainer_lock);
1691│
1692│ return 0;
1693│ }
2 start_slab_maintenance_thread () at slabs.c:1311
1309│ int start_slab_maintenance_thread(void) {
1310│ int ret;
1311│ slab_rebalance_signal = 0;
1312│ slab_rebal.slab_start = NULL;
1313│ char *env = getenv("MEMCACHED_SLAB_BULK_CHECK");
1314│ if (env != NULL) {
1315│ slab_bulk_check = atoi(env);
1316│ if (slab_bulk_check == 0) {
1317│ slab_bulk_check = DEFAULT_SLAB_BULK_CHECK;
1318│ }
1319│ }
1320│
1321├───> if (pthread_cond_init(&slab_rebalance_cond, NULL) != 0) {
1322│ fprintf(stderr, "Can't initialize rebalance condition\n");
1323│ return -1;
1324│ }
1325│ pthread_mutex_init(&slabs_rebalance_lock, NULL);
1326│
1327│ if ((ret = pthread_create(&rebalance_tid, NULL,
1328│ slab_rebalance_thread, NULL)) != 0) {
1329│ fprintf(stderr, "Can't create rebal thread: %s\n", strerror(ret));
1330│ return -1;
1331│ }
1332│ return 0;
1333│ }
2 clock_handler (fd=0, which=0, arg=0x0) at memcached.c:6321
6316│ /* libevent uses a monotonic clock when available for event scheduling. Aside
6317│ * from jitter, simply ticking our internal timer here is accurate enough.
6318│ * Note that users who are setting explicit dates for expiration times *must*
6319│ * ensure their clocks are correct before starting memcached. */
6320│ static void clock_handler(const int fd, const short which, void *arg) {
6321├───> struct timeval t = {.tv_sec = 1, .tv_usec = 0};
6322│ static bool initialized = false;
6323│ #if defined(HAVE_CLOCK_GETTIME) && defined(CLOCK_MONOTONIC)
6324│ static bool monotonic = false;
6325│ static time_t monotonic_start;
6326│ #endif
6327│
6328│ if (initialized) {
6329│ /* only delete the event if it's actually there. */
6330│ evtimer_del(&clockevent);
6331│ } else {
6332│ initialized = true;
6333│ /* process_started is initialized to time() - 2. We initialize to 1 so
6334│ * flush_all won't underflow during tests. */
6335│ #if defined(HAVE_CLOCK_GETTIME) && defined(CLOCK_MONOTONIC)
6336│ struct timespec ts;
6337│ if (clock_gettime(CLOCK_MONOTONIC, &ts) == 0) {
6338│ monotonic = true;
6339│ monotonic_start = ts.tv_sec - ITEM_UPDATE_INTERVAL - 2;
6340│ }
6341│ #endif
6342│ }
6343│
6344│ // While we're here, check for hash table expansion.
6345│ // This function should be quick to avoid delaying the timer.
6346├───> assoc_start_expand(stats_state.curr_items);
6347│
6348│ evtimer_set(&clockevent, clock_handler, 0);
6349│ event_base_set(main_base, &clockevent);
6350│ evtimer_add(&clockevent, &t);
6351│
6352│ #if defined(HAVE_CLOCK_GETTIME) && defined(CLOCK_MONOTONIC)
6353│ if (monotonic) {
6354│ struct timespec ts;
6355│ if (clock_gettime(CLOCK_MONOTONIC, &ts) == -1)
6356│ return;
6357│ current_time = (rel_time_t) (ts.tv_sec - monotonic_start);
6358│ return;
6359│ }
6360│ #endif
6361│ {
6362│ struct timeval tv;
6363│ gettimeofday(&tv, NULL);
6364│ current_time = (rel_time_t) (tv.tv_sec - process_started);
6365│ }
6366│ }
6358行目で関数がreturnされる。
3 assoc_start_expand (curr_items=0) at assoc.c:144
143│ void assoc_start_expand(uint64_t curr_items) {
144├───> if (started_expanding)
145│ return;
146│
147│ if (curr_items > (hashsize(hashpower) * 3) / 2 &&
148│ hashpower < HASHPOWER_MAX) {
149│ started_expanding = true;
150│ pthread_cond_signal(&maintenance_cond);
151│ }
152│ }
1 main at memcached.c
8072│ /* create unix mode sockets after dropping privileges */
8073│ if (settings.socketpath != NULL) {
8074│ errno = 0;
8075│ if (server_socket_unix(settings.socketpath,settings.access)) {
8076│ vperror("failed to listen on UNIX socket: %s", settings.socketpath);
8077│ exit(EX_OSERR);
8078│ }
8079│ }
8080│
8081│ /* create the listening socket, bind it, and init */
8082├───> if (settings.socketpath == NULL) {
8083│ const char *portnumber_filename = getenv("MEMCACHED_PORT_FILENAME");
8084│ char *temp_portnumber_filename = NULL;
8085│ size_t len;
8086│ FILE *portnumber_file = NULL;
8087│
8088│ if (portnumber_filename != NULL) {
8089│ len = strlen(portnumber_filename)+4+1;
8090│ temp_portnumber_filename = malloc(len);
8091│ snprintf(temp_portnumber_filename,
8092│ len,
8093│ "%s.lck", portnumber_filename);
8094│
8095│ portnumber_file = fopen(temp_portnumber_filename, "a");
8096│ if (portnumber_file == NULL) {
8097│ fprintf(stderr, "Failed to open \"%s\": %s\n",
8098│ temp_portnumber_filename, strerror(errno));
8099│ }
8100│ }
8101│
8102├───────> errno = 0;
8103│ if (settings.port && server_sockets(settings.port, tcp_transport,
8104│ portnumber_file)) {
8105│ vperror("failed to listen on TCP port %d", settings.port);
8106│ exit(EX_OSERR);
8107│ }
8108│
8109│ /*
8110│ * initialization order: first create the listening sockets
8111│ * (may need root on low ports), then drop root if needed,
8112│ * then daemonize if needed, then init libevent (in some cases
8113│ * descriptors created by libevent wouldn't survive forking).
8114│ */
8115│
8116│ /* create the UDP listening socket and bind it */
8117│ errno = 0;
8118├───────> if (settings.udpport && server_sockets(settings.udpport, udp_transport,
8119│ portnumber_file)) {
8120│ vperror("failed to listen on UDP port %d", settings.udpport);
8121│ exit(EX_OSERR);
8122│ }
8123│
8124│ if (portnumber_file) {
8125│ fclose(portnumber_file);
8126│ rename(temp_portnumber_filename, portnumber_filename);
8127│ }
8128│ if (temp_portnumber_filename)
8129│ free(temp_portnumber_filename);
8130│ }
2 server_sockets (port=11211, transport=tcp_transport, portnumber_file=0x0) at memcached.c:6146
6144│ static int server_sockets(int port, enum network_transport transport,
6145│ FILE *portnumber_file) {
6146├───> bool ssl_enabled = false;
6147│
6148│ #ifdef TLS
6149│ const char *notls = "notls";
6150│ ssl_enabled = settings.ssl_enabled;
6151│ #endif
6152│
6153│ if (settings.inter == NULL) {
6154│ return server_socket(settings.inter, port, transport, portnumber_file, ssl_enabled);
6155│ } else {
6156│ // tokenize them and bind to each one of them..
6157│ char *b;
6158│ int ret = 0;
3 server_socket (interface=0x0, port=11211, transport=tcp_transport, portnumber_file=0x0, ssl_enabled=false) at memcached.c:5998
5988│ * @param transport the transport protocol (TCP / UDP)
5989│ * @param portnumber_file A filepointer to write the port numbers to
5990│ * when they are successfully added to the list of ports we
5991│ * listen on.
5992│ */
5993│ static int server_socket(const char *interface,
5994│ int port,
5995│ enum network_transport transport,
5996│ FILE *portnumber_file, bool ssl_enabled) {
5997│ int sfd;
5998│ struct linger ling = {0, 0};
5999│ struct addrinfo *ai;
6000│ struct addrinfo *next;
6001├───> struct addrinfo hints = { .ai_flags = AI_PASSIVE,
6002│ .ai_family = AF_UNSPEC };
6003│ char port_buf[NI_MAXSERV];
6004│ int error;
6005│ int success = 0;
6006│ int flags =1;
6007│
6008│ hints.ai_socktype = IS_UDP(transport) ? SOCK_DGRAM : SOCK_STREAM;
6009│
6010│ if (port == -1) {
6011│ port = 0;
6012│ }
6013│ snprintf(port_buf, sizeof(port_buf), "%d", port);
6014│ error= getaddrinfo(interface, port_buf, &hints, &ai);
6015│ if (error != 0) {
6016│ if (error != EAI_SYSTEM)
6017│ fprintf(stderr, "getaddrinfo(): %s\n", gai_strerror(error));
6018│ else
6019│ perror("getaddrinfo()");
6020│ return 1;
6021│ }
6022│
6023├───> for (next= ai; next; next= next->ai_next) {
6024│ conn *listen_conn_add;
6025│ if ((sfd = new_socket(next)) == -1) {
6026│ /* getaddrinfo can return "junk" addresses,
6027│ * we make sure at least one works before erroring.
6028│ */
6029│ if (errno == EMFILE) {
6030│ /* ...unless we're out of fds */
6031│ perror("server_socket");
6032│ exit(EX_OSERR);
6033│ }
6034│ continue;
6035│ }
4 new_socket (ai=0x6520f0) at memcached.c:5936
5932│ static int new_socket(struct addrinfo *ai) {
5933│ int sfd;
5934│ int flags;
5935│
5936├───> if ((sfd = socket(ai->ai_family, ai->ai_socktype, ai->ai_protocol)) == -1) {
5937│ return -1;
5938│ }
5939│
5940│ if ((flags = fcntl(sfd, F_GETFL, 0)) < 0 ||
5941│ fcntl(sfd, F_SETFL, flags | O_NONBLOCK) < 0) {
5942│ perror("setting O_NONBLOCK");
5943│ close(sfd);
5944│ return -1;
5945│ }
5946│ return sfd;
5947│ }
3
ソケットオプションの設定
6048│ setsockopt(sfd, SOL_SOCKET, SO_REUSEADDR, (void *)&flags, sizeof(flags));
6049│ if (IS_UDP(transport)) {
6050│ maximize_sndbuf(sfd);
6051│ } else {
6052├───────────> error = setsockopt(sfd, SOL_SOCKET, SO_KEEPALIVE, (void *)&flags, sizeof(flags));
6053│ if (error != 0)
6054│ perror("setsockopt");
6055│
6056│ error = setsockopt(sfd, SOL_SOCKET, SO_LINGER, (void *)&ling, sizeof(ling));
6057│ if (error != 0)
6058│ perror("setsockopt");
6059│
6060│ error = setsockopt(sfd, IPPROTO_TCP, TCP_NODELAY, (void *)&flags, sizeof(flags));
6061│ if (error != 0)
6062│ perror("setsockopt");
6063│ }
6064│
bind処理
6065│ if (bind(sfd, next->ai_addr, next->ai_addrlen) == -1) {
6066│ if (errno != EADDRINUSE) {
6067│ perror("bind()");
6068│ close(sfd);
6069│ freeaddrinfo(ai);
6070│ return 1;
6071│ }
6072│ close(sfd);
6073│ continue;
6074│ } else {
6075├───────────> success++;
6076│ if (!IS_UDP(transport) && listen(sfd, settings.backlog) == -1) {
6077│ perror("listen()");
6078│ close(sfd);
6079│ freeaddrinfo(ai);
6080│ return 1;
6081│ }
6082│ if (portnumber_file != NULL &&
6083│ (next->ai_addr->sa_family == AF_INET ||
6084│ next->ai_addr->sa_family == AF_INET6)) {
6085│ union {
6086│ struct sockaddr_in in;
6087│ struct sockaddr_in6 in6;
conn_new関数が呼ばれる
6109│ * this allows "stats conns" to separately list multiple
6110│ * parallel UDP requests in progress.
6111│ *
6112│ * The dispatch code round-robins new connection requests
6113│ * among threads, so this is guaranteed to assign one
6114│ * FD to each thread.
6115│ */
6116│ int per_thread_fd = c ? dup(sfd) : sfd;
6117│ dispatch_conn_new(per_thread_fd, conn_read,
6118│ EV_READ | EV_PERSIST,
6119│ UDP_READ_BUFFER_SIZE, transport, NULL);
6120│ }
6121│ } else {
6122├───────────> if (!(listen_conn_add = conn_new(sfd, conn_listening,
6123│ EV_READ | EV_PERSIST, 1,
6124│ transport, main_base, NULL))) {
6125│ fprintf(stderr, "failed to create listening connection\n");
6126│ exit(EXIT_FAILURE);
6127│ }
6128│ #ifdef TLS
6129│ listen_conn_add->ssl_enabled = ssl_enabled;
6130│ #else
6131│ assert(ssl_enabled == false);
6132│ #endif
6133│ listen_conn_add->next = listen_conn;
6134│ listen_conn = listen_conn_add;
6135│ }
6136│ }
6137│
6138│ freeaddrinfo(ai);
6139│
6140│ /* Return zero iff we detected no errors in starting up connections */
6141│ return success == 0;
6142│ }
4 conn_new (sfd=35, init_state=conn_listening, event_flags=18, read_buffer_size=1, transport=tcp_transport, base=0x646040, ssl=0x0) at memcached.c:547
540│ conn *conn_new(const int sfd, enum conn_states init_state,
541│ const int event_flags,
542│ const int read_buffer_size, enum network_transport transport,
543│ struct event_base *base, void *ssl) {
544│ conn *c;
545│
546│ assert(sfd >= 0 && sfd < max_fds);
547├───> c = conns[sfd];
548│
549│ if (NULL == c) {
550│ if (!(c = (conn *)calloc(1, sizeof(conn)))) {
551│ STATS_LOCK();
552│ stats.malloc_fails++;
553│ STATS_UNLOCK();
554│ fprintf(stderr, "Failed to allocate connection object\n");
555│ return NULL;
556│ }
557│ MEMCACHED_CONN_CREATE(c);
558│ c->read = NULL;
559│ c->sendmsg = NULL;
560│ c->write = NULL;
561│ c->rbuf = c->wbuf = 0;
562│ c->ilist = 0;
563│ c->suffixlist = 0;
564│ c->iov = 0;
565│ c->msglist = 0;
566│ c->hdrbuf = 0;
567│
568│ c->rsize = read_buffer_size;
569│ c->wsize = DATA_BUFFER_SIZE;
570│ c->isize = ITEM_LIST_INITIAL;
571│ c->suffixsize = SUFFIX_LIST_INITIAL;
572│ c->iovsize = IOV_LIST_INITIAL;
573├───────> c->msgsize = MSG_LIST_INITIAL;
574│ c->hdrsize = 0;
575│
576│ c->rbuf = (char *)malloc((size_t)c->rsize);
577│ c->wbuf = (char *)malloc((size_t)c->wsize);
578│ c->ilist = (item **)malloc(sizeof(item *) * c->isize);
579│ c->suffixlist = (char **)malloc(sizeof(char *) * c->suffixsize);
580│ c->iov = (struct iovec *)malloc(sizeof(struct iovec) * c->iovsize);
581│ c->msglist = (struct msghdr *)malloc(sizeof(struct msghdr) * c->msgsize);
582│
583│ if (c->rbuf == 0 || c->wbuf == 0 || c->ilist == 0 || c->iov == 0 ||
584│ c->msglist == 0 || c->suffixlist == 0) {
585│ conn_free(c);
586│ STATS_LOCK();
587│ stats.malloc_fails++;
588│ STATS_UNLOCK();
589│ fprintf(stderr, "Failed to allocate buffers for connection\n");
590│ return NULL;
591│ }
592│
593│ STATS_LOCK();
594│ stats_state.conn_structs++;
595│ STATS_UNLOCK();
596│
597│ c->sfd = sfd;
598│ conns[sfd] = c;
599│ }
600│
601│ c->transport = transport;
602│ c->protocol = settings.binding_protocol;
603│
604│ /* unix socket mode doesn't need this, so zeroed out. but why
605│ * is this done for every command? presumably for UDP
606│ * mode. */
607│ if (!settings.socketpath) {
608│ c->request_addr_size = sizeof(c->request_addr);
609│ } else {
610│ c->request_addr_size = 0;
611│ }
612│
613├───> if (transport == tcp_transport && init_state == conn_new_cmd) {
614│ if (getpeername(sfd, (struct sockaddr *) &c->request_addr,
615│ &c->request_addr_size)) {
616│ perror("getpeername");
617│ memset(&c->request_addr, 0, sizeof(c->request_addr));
618│ }
619│ }
620│
621│ if (settings.verbose > 1) {
622│ if (init_state == conn_listening) {
623│ fprintf(stderr, "<%d server listening (%s)\n", sfd,
624│ prot_text(c->protocol));
625│ } else if (IS_UDP(transport)) {
ここで event_set で event_handler が設定されている。この event_handler関数が、後ほどコマンド実行される度に呼び出される関数。
691│
692├───> event_set(&c->event, sfd, event_flags, event_handler, (void *)c);
693│ event_base_set(base, &c->event);
694│ c->ev_flags = event_flags;
695│
696│ if (event_add(&c->event, 0) == -1) {
697│ perror("event_add");
698│ return NULL;
699│ }
700│
701│ STATS_LOCK();
702│ stats_state.curr_conns++;
703│ stats.total_conns++;
704│ STATS_UNLOCK();
705│
706│ MEMCACHED_CONN_ALLOCATE(c->sfd);
707│
708│ return c;
709│ }
1
8132│ /* Give the sockets a moment to open. I know this is dumb, but the error
8133│ * is only an advisory.
8134│ */
8135│ usleep(1000);
8136├───> if (stats_state.curr_conns + stats_state.reserved_fds >= settings.maxconns - 1) {
8137│ fprintf(stderr, "Maxconns setting is too low, use -c to increase.\n");
8138│ exit(EXIT_FAILURE);
8139│ }
8140│
8141│ if (pid_file != NULL) {
8142│ save_pid(pid_file);
8143│ }
8144│
8145│ /* Drop privileges no longer needed */
8146│ if (settings.drop_privileges) {
8147│ drop_privileges();
8148│ }
8149│
8150│ /* Initialize the uriencode lookup table. */
8151│ uriencode_init();
8152│
8153│ /* enter the event loop */
8154├───> if (event_base_loop(main_base, 0) != 0) {
8155│ retval = EXIT_FAILURE;
8156│ }
8157│
8158│ stop_assoc_maintenance_thread();
8159│
8160│ /* remove the PID file if we're a daemon */
8161│ if (do_daemonize)
8162│ remove_pidfile(pid_file);
8163│ /* Clean up strdup() call for bind() address */
8164│ if (settings.inter)
8165│ free(settings.inter);
8166│
最後にmain_baseがevent_base_loopによりイベントのループ処理が始まる。
2 uriencode_init () at util.c:16
14│ void uriencode_init(void) {
15│ int x;
16├───> char *str = uriencode_str;
17│ for (x = 0; x < 256; x++) {
18│ if (isalnum(x) || x == '-' || x == '.' || x == '_' || x == '~') {
19│ uriencode_map[x] = NULL;
20│ } else {
21│ snprintf(str, 4, "%%%02hhX", (unsigned char)x);
22│ uriencode_map[x] = str;
23│ str += 3; /* lobbing off the \0 is fine */
24│ }
25│ }
26│ }
SETコマンドの挙動
1 event_handler (fd=35, which=2, arg=0x6521e0) at memcached.c:5913
5910│ void event_handler(const int fd, const short which, void *arg) {
5911│ conn *c;
5912│
5913├───> c = (conn *)arg;
5914│ assert(c != NULL);
5915│
5916│ c->which = which;
5917│
5918│ /* sanity */
5919│ if (fd != c->sfd) {
5920│ if (settings.verbose > 0)
5921│ fprintf(stderr, "Catastrophic: event fd doesn't match conn fd!\n");
5922│ conn_close(c);
5923│ return;
5924│ }
5925│
5926│ drive_machine(c);
5927│
5928│ /* wait for next event */
5929│ return;
5930│ }
2 drive_machine (c=0x6521e0) at memcached.c:5500
5499│ static void drive_machine(conn *c) {
5500│ bool stop = false;
5501│ int sfd;
5502│ socklen_t addrlen;
5503│ struct sockaddr_storage addr;
5504│ int nreqs = settings.reqs_per_event;
5505│ int res;
5506│ const char *str;
5507│ #ifdef HAVE_ACCEPT4
5508│ static int use_accept4 = 1;
5509│ #else
5510│ static int use_accept4 = 0;
5511│ #endif
5512│
5513│ assert(c != NULL);
5514│
c->stateがconn_listening
5515├───> while (!stop) {
5517│ switch(c->state) {
5518│ case conn_listening:
5519│ addrlen = sizeof(addr);
5520│ #ifdef HAVE_ACCEPT4
5521│ if (use_accept4) {
5522│ sfd = accept4(c->sfd, (struct sockaddr *)&addr, &addrlen, SOCK_NONBLOCK);
5523│ } else {
5524│ sfd = accept(c->sfd, (struct sockaddr *)&addr, &addrlen);
5525│ }
5526│ #else
5527│ sfd = accept(c->sfd, (struct sockaddr *)&addr, &addrlen);
5528│ #endif
5529├───────────> if (sfd == -1) {
5530│ if (use_accept4 && errno == ENOSYS) {
5531│ use_accept4 = 0;
5532│ continue;
5533│ }
5534│ perror(use_accept4 ? "accept4()" : "accept()");
5535│ if (errno == EAGAIN || errno == EWOULDBLOCK) {
5536│ /* these are transient, so don't log anything */
5537│ stop = true;
5538│ } else if (errno == EMFILE) {
5539│ if (settings.verbose > 0)
5540│ fprintf(stderr, "Too many open connections\n");
5541│ accept_new_conns(false);
5605├───────────────> dispatch_conn_new(sfd, conn_new_cmd, EV_READ | EV_PERSIST,
5606│ DATA_BUFFER_SIZE, c->transport, ssl_v);
3 dispatch_conn_new (sfd=37, init_state=conn_new_cmd, event_flags=18, read_buffer_size=2048, transport=tcp_transport
, ssl=0x0) at thread.c:491
484│ /*
485│ * Dispatches a new connection to another thread. This is only ever called
486│ * from the main thread, either during initialization (for UDP) or because
487│ * of an incoming connection.
488│ */
489│ void dispatch_conn_new(int sfd, enum conn_states init_state, int event_flags,
490│ int read_buffer_size, enum network_transport transport, void *ssl) {
491├───> CQ_ITEM *item = cqi_new();
492│ char buf[1];
493│ if (item == NULL) {
494│ close(sfd);
495│ /* given that malloc failed this may also fail, but let's try */
496│ fprintf(stderr, "Failed to allocate memory for connection object\n");
497│ return ;
498│ }
499│
500│ int tid = (last_thread + 1) % settings.num_threads;
501│
502│ LIBEVENT_THREAD *thread = threads + tid;
503│
503│
504│ last_thread = tid;
505│
506│ item->sfd = sfd;
507│ item->init_state = init_state;
508│ item->event_flags = event_flags;
509│ item->read_buffer_size = read_buffer_size;
510│ item->transport = transport;
511│ item->mode = queue_new_conn;
512├───> item->ssl = ssl;
513│
514│ cq_push(thread->new_conn_queue, item);
515│
516│ MEMCACHED_CONN_DISPATCH(sfd, thread->thread_id);
517│ buf[0] = 'c';
518│ if (write(thread->notify_send_fd, buf, 1) != 1) {
519│ perror("Writing to thread notify pipe");
520│ }
521│ }
thread->notify_send_fdにはbufに格納された"c"をwriteする。
4 cqi_new () at thread.c:240
236│ /*
237│ * Returns a fresh connection queue item.
238│ */
239│ static CQ_ITEM *cqi_new(void) {
240├───> CQ_ITEM *item = NULL;
241│ pthread_mutex_lock(&cqi_freelist_lock);
242│ if (cqi_freelist) {
243│ item = cqi_freelist;
244│ cqi_freelist = item->next;
245│ }
246│ pthread_mutex_unlock(&cqi_freelist_lock);
247│
248│ if (NULL == item) {
249│ int i;
250│
251│ /* Allocate a bunch of items at once to reduce fragmentation */
252│ item = malloc(sizeof(CQ_ITEM) * ITEMS_PER_ALLOC);
253├───────> if (NULL == item) {
254│ STATS_LOCK();
255│ stats.malloc_fails++;
256│ STATS_UNLOCK();
257│ return NULL;
258│ }
259│
260│ /*
261│ * Link together all the new items except the first one
262│ * (which we'll return to the caller) for placement on
263│ * the freelist.
264│ */
265│ for (i = 2; i < ITEMS_PER_ALLOC; i++)
266│ item[i - 1].next = &item[i];
267│
268│ pthread_mutex_lock(&cqi_freelist_lock);
269│ item[ITEMS_PER_ALLOC - 1].next = cqi_freelist;
270│ cqi_freelist = &item[1];
271│ pthread_mutex_unlock(&cqi_freelist_lock);
272│ }
273│
274│ return item;
275│ }
4 cq_push (cq=0x64f250, item=0x658270) at thread.c:225
221│ /*
222│ * Adds an item to a connection queue.
223│ */
224│ static void cq_push(CQ *cq, CQ_ITEM *item) {
225├───> item->next = NULL;
226│
227│ pthread_mutex_lock(&cq->lock);
228│ if (NULL == cq->tail)
229│ cq->head = item;
230│ else
231│ cq->tail->next = item;
232│ cq->tail = item;
233│ pthread_mutex_unlock(&cq->lock);
234│ }
1 event_handler
2 drive_machine
その後一旦libeventの処理に戻り、event_handlerからdrive_machine
c->stateがconn_new_cmdとなっている。
5651│ case conn_new_cmd:
5652│ /* Only process nreqs at a time to avoid starving other
5653│ connections */
5654│
5655│ --nreqs;
5656├───────────> if (nreqs >= 0) {
5657│ reset_cmd_handler(c);
5658│ } else {
5659│ pthread_mutex_lock(&c->thread->stats.mutex);
5660│ c->thread->stats.conn_yields++;
5661│ pthread_mutex_unlock(&c->thread->stats.mutex);
5662│ if (c->rbytes > 0) {
5663│ /* We have already read in data into the input buffer,
5664│ so libevent will most likely not signal read events
5665│ on the socket (unless more data is available. As a
5666│ hack we should just put in a request to write data,
5667│ because that should be possible ;-)
5668│ */
3 reset_cmd_handler (c=0x7ffff00109a0) at memcached.c:2766
最後にconn_set_state(c, conn_waiting)が実行される。
2765│ static void reset_cmd_handler(conn *c) {
2766├───> c->cmd = -1;
2767│ c->substate = bin_no_state;
2768│ if(c->item != NULL) {
2769│ item_remove(c->item);
2770│ c->item = NULL;
2771│ }
2772│ conn_shrink(c);
2773│ if (c->rbytes > 0) {
2774│ conn_set_state(c, conn_parse_cmd);
2775│ } else {
2776│ conn_set_state(c, conn_waiting);
2777│ }
2778│ }
4 conn_shrink (c=0x7ffff00109a0) at memcached.c:910
899│ /*
900│ * Shrinks a connection's buffers if they're too big. This prevents
901│ * periodic large "get" requests from permanently chewing lots of server
902│ * memory.
903│ *
904│ * This should only be called in between requests since it can wipe output
905│ * buffers!
906│ */
907│ static void conn_shrink(conn *c) {
908│ assert(c != NULL);
909│
910├───> if (IS_UDP(c->transport))
911│ return;
912│
913│ if (c->rsize > READ_BUFFER_HIGHWAT && c->rbytes < DATA_BUFFER_SIZE) {
914│ char *newbuf;
915│
916│ if (c->rcurr != c->rbuf)
917│ memmove(c->rbuf, c->rcurr, (size_t)c->rbytes);
918│
919│ newbuf = (char *)realloc((void *)c->rbuf, DATA_BUFFER_SIZE);
920│
921│ if (newbuf) {
922│ c->rbuf = newbuf;
923│ c->rsize = DATA_BUFFER_SIZE;
924│ }
925│ /* TODO check other branch... */
926│ c->rcurr = c->rbuf;
927│ }
928│
929├───> if (c->isize > ITEM_LIST_HIGHWAT) {
930│ item **newbuf = (item**) realloc((void *)c->ilist, ITEM_LIST_INITIAL * sizeof(c->ilist[0]));
931│ if (newbuf) {
932│ c->ilist = newbuf;
933│ c->isize = ITEM_LIST_INITIAL;
934│ }
935│ /* TODO check error condition? */
936│ }
937│
938│ if (c->msgsize > MSG_LIST_HIGHWAT) {
939│ struct msghdr *newbuf = (struct msghdr *) realloc((void *)c->msglist, MSG_LIST_INITIAL * sizeof(c->m
940│ if (newbuf) {
941│ c->msglist = newbuf;
942│ c->msgsize = MSG_LIST_INITIAL;
943│ }
944│ /* TODO check error condition? */
945│ }
946│
947├───> if (c->iovsize > IOV_LIST_HIGHWAT) {
948│ struct iovec *newbuf = (struct iovec *) realloc((void *)c->iov, IOV_LIST_INITIAL * sizeof(c->iov[0])
949│ if (newbuf) {
950│ c->iov = newbuf;
951│ c->iovsize = IOV_LIST_INITIAL;
952│ }
953│ /* TODO check return value */
954│ }
955│ }
4 conn_set_state (c=0x7ffff00109a0, state=conn_waiting) at memcached.c:985
976│ /*
977│ * Sets a connection's current state in the state machine. Any special
978│ * processing that needs to happen on certain state transitions can
979│ * happen here.
980│ */
981│ static void conn_set_state(conn *c, enum conn_states state) {
982│ assert(c != NULL);
983│ assert(state >= conn_listening && state < conn_max_state);
984│
985├───> if (state != c->state) {
986│ if (settings.verbose > 2) {
987│ fprintf(stderr, "%d: going from %s to %s\n",
988│ c->sfd, state_text(c->state),
989│ state_text(state));
990│ }
991│
992│ if (state == conn_write || state == conn_mwrite) {
993│ MEMCACHED_PROCESS_COMMAND_END(c->sfd, c->wbuf, c->wbytes);
994│ }
995│ c->state = state;
996│ }
997│ }
1 event_handler
2 drive_machine
c->stateがconn_waitingになっている。最後にconn_set_state関数でconn_readに設定している。
5612│ case conn_waiting:
5613├───────────> if (!update_event(c, EV_READ | EV_PERSIST)) {
5614│ if (settings.verbose > 0)
5615│ fprintf(stderr, "Couldn't update event\n");
5616│ conn_set_state(c, conn_closing);
5617│ break;
5618│ }
5619│
5620│ conn_set_state(c, conn_read);
5621│ stop = true;
5622│ break;
5623│
3 update_event (c=0x7ffff00109a0, new_flags=18) at memcached.c:5295
5292│ static bool update_event(conn *c, const int new_flags) {
5293│ assert(c != NULL);
5294│
5295├───> struct event_base *base = c->event.ev_base;
5296│ if (c->ev_flags == new_flags)
5297│ return true;
5298│ if (event_del(&c->event) == -1) return false;
5299│ event_set(&c->event, c->sfd, new_flags, event_handler, (void *)c);
5300│ event_base_set(base, &c->event);
5301│ c->ev_flags = new_flags;
5302│ if (event_add(&c->event, 0) == -1) return false;
5303│ return true;
5304│ }
1 event_handler
2 drive_machine
再びlibeventの処理に戻る。event.cのevent_process_active関数中でevent_process_activeが実行されると再びevent_handlerに戻ってくる c->stateがconn_readに設定されている。
resがREAD_DATA_RECEIVEDより、conn_parse_cmdに設定される。
5624│ case conn_read:
5625├───────────> res = IS_UDP(c->transport) ? try_read_udp(c) : try_read_network(c);
5626│
5627│ switch (res) {
5628│ case READ_NO_DATA_RECEIVED:
5629│ conn_set_state(c, conn_waiting);
5630│ break;
5631│ case READ_DATA_RECEIVED:
5632│ conn_set_state(c, conn_parse_cmd);
5633│ break;
5634│ case READ_ERROR:
5635│ conn_set_state(c, conn_closing);
5636│ break;
5637│ case READ_MEMORY_ERROR: /* Failed to allocate more memory */
5638│ /* State already set by try_read_network */
5639│ break;
5640│ }
5641│ break;
5642│
3 try_read_network (c=0x7ffff00109a0) at memcached.c:5231
5218│ /*
5219│ * read from network as much as we can, handle buffer overflow and connection
5220│ * close.
5221│ * before reading, move the remaining incomplete fragment of a command
5222│ * (if any) to the beginning of the buffer.
5223│ *
5224│ * To protect us from someone flooding a connection with bogus data causing
5225│ * the connection to eat up all available memory, break out and start looking
5226│ * at the data I've got after a number of reallocs...
5227│ *
5228│ * @return enum try_read_result
5229│ */
5230│ static enum try_read_result try_read_network(conn *c) {
5231├───> enum try_read_result gotdata = READ_NO_DATA_RECEIVED;
5232│ int res;
5233│ int num_allocs = 0;
5234│ assert(c != NULL);
5235│
5236│ if (c->rcurr != c->rbuf) {
5237│ if (c->rbytes != 0) /* otherwise there's nothing to copy */
5238│ memmove(c->rbuf, c->rcurr, c->rbytes);
5239│ c->rcurr = c->rbuf;
5240│ }
5241│
5242│ while (1) {
5243│ if (c->rbytes >= c->rsize) {
5276行目でbreak。その後、関数もretrunされる。
5265│ int avail = c->rsize - c->rbytes;
5266├───────> res = c->read(c, c->rbuf + c->rbytes, avail);
5267│ if (res > 0) {
5268│ pthread_mutex_lock(&c->thread->stats.mutex);
5269│ c->thread->stats.bytes_read += res;
5270│ pthread_mutex_unlock(&c->thread->stats.mutex);
5271│ gotdata = READ_DATA_RECEIVED;
5272│ c->rbytes += res;
5273│ if (res == avail) {
5274│ continue;
5275│ } else {
5276│ break;
5277│ }
5278│ }
5279│ if (res == 0) {
5280│ return READ_ERROR;
5281│ }
5282│ if (res == -1) {
5283│ if (errno == EAGAIN || errno == EWOULDBLOCK) {
5284│ break;
5285│ }
5286│ return READ_ERROR;
5287│ }
5288│ }
4 tcp_read (c=0x7ffff00109a0, buf=0x7ffff0010bd0, count=2048) at memcached.c:162
159│ /* Default methods to read from/ write to a socket */
160│ ssize_t tcp_read(conn *c, void *buf, size_t count) {
161│ assert (c != NULL);
162├───> return read(c->sfd, buf, count);
163│ }
2
5643│ case conn_parse_cmd :
5644├───────────> if (try_read_command(c) == 0) {
5645│ /* wee need more data! */
5646│ conn_set_state(c, conn_waiting);
5647│ }
5648│
5649│ break;
5650│
3 try_read_command (c=0x7ffff00109a0) at memcached.c:5053
実際に入力した内容を読み込む処理。
5045│ /*
5046│ * if we have a complete line in the buffer, process it.
5047│ */
5048│ static int try_read_command(conn *c) {
5049│ assert(c != NULL);
5050│ assert(c->rcurr <= (c->rbuf + c->rsize));
5051│ assert(c->rbytes > 0);
5052│
5053├───> if (c->protocol == negotiating_prot || c->transport == udp_transport) {
5054│ if ((unsigned char)c->rbuf[0] == (unsigned char)PROTOCOL_BINARY_REQ) {
5055│ c->protocol = binary_prot;
5056│ } else {
5057│ c->protocol = ascii_prot;
5058│ }
5059│
5060│ if (settings.verbose > 1) {
5061│ fprintf(stderr, "%d: Client using the %s protocol\n", c->sfd,
5062│ prot_text(c->protocol));
5063│ }
5064│ }
5065│
MemcachedではASCIIモードとバイナリーモードが利用できるが、ここではASCIIモードを利用しているので、そちらの処理を実行。
5132│ } else {
5133│ char *el, *cont;
5134│
5135│ if (c->rbytes == 0)
5136│ return 0;
5137│
5138│ el = memchr(c->rcurr, '\n', c->rbytes);
5139├───────> if (!el) {
5140│ if (c->rbytes > 1024) {
5141│ /*
5142│ * We didn't have a '\n' in the first k. This _has_ to be a
5143│ * large multiget, if not we should just nuke the connection.
5144│ */
5145│ char *ptr = c->rcurr;
5146│ while (*ptr == ' ') { /* ignore leading whitespaces */
5147│ ++ptr;
5148│ }
5149│
5150│ if (ptr - c->rcurr > 100 ||
5151│ (strncmp(ptr, "get ", 4) && strncmp(ptr, "gets ", 5))) {
5152│
5153│ conn_set_state(c, conn_closing);
5154│ return 1;
5155│ }
5156│ }
5157│
5158│ return 0;
5159│ }
5160├───────> cont = el + 1;
5161│ if ((el - c->rcurr) > 1 && *(el - 1) == '\r') {
5162│ el--;
5163│ }
5164│ *el = '\0';
5165│
5166│ assert(cont <= (c->rcurr + c->rbytes));
5167│
5168│ c->last_cmd_time = current_time;
5169│ process_command(c, c->rcurr);
5170│
5171│ c->rbytes -= (cont - c->rcurr);
5172│ c->rcurr = cont;
4
483│ static const char *prot_text(enum protocol prot) {
484│ char *rv = "unknown";
485├───> switch(prot) {
486│ case ascii_prot:
487│ rv = "ascii";
488│ break;
489│ case binary_prot:
490│ rv = "binary";
491│ break;
492│ case negotiating_prot:
493│ rv = "auto-negotiate";
494│ break;
495│ }
496│ return rv;
497│ }
4 process_command (c=0x7ffff00109a0, command=0x7ffff0010bd0 "set test1 0 0 7") at memcached.c:4731
4721│ static void process_command(conn *c, char *command) {
4722│
4723│ token_t tokens[MAX_TOKENS];
4724│ size_t ntokens;
4725│ int comm;
4726│
4727│ assert(c != NULL);
4728│
4729│ MEMCACHED_PROCESS_COMMAND_START(c->sfd, c->rcurr, c->rbytes);
4730│
4731├───> if (settings.verbose > 1)
4732│ fprintf(stderr, "<%d %s\n", c->sfd, command);
4733│
4734│ /*
4735│ * for commands set/add/replace, we build an item and read the data
4736│ * directly into it, then continue in nread_complete().
4737│ */
4738│
4739│ c->msgcurr = 0;
4740│ c->msgused = 0;
4741│ c->iovused = 0;
4742│ if (add_msghdr(c) != 0) {
4743│ out_of_memory(c, "SERVER_ERROR out of memory preparing response");
4744│ return;
4745│ }
4746│
4747│ ntokens = tokenize_command(command, tokens, MAX_TOKENS);
5 add_msghdr (c=0x7ffff00109a0) at memcached.c:329
318│ /*
319│ * Adds a message header to a connection.
320│ *
321│ * Returns 0 on success, -1 on out-of-memory.
322│ */
323│ static int add_msghdr(conn *c)
324│ {
325│ struct msghdr *msg;
326│
327│ assert(c != NULL);
328│
329├───> if (c->msgsize == c->msgused) {
330│ msg = realloc(c->msglist, c->msgsize * 2 * sizeof(struct msghdr));
331│ if (! msg) {
332│ STATS_LOCK();
333│ stats.malloc_fails++;
334│ STATS_UNLOCK();
335│ return -1;
336│ }
337│ c->msglist = msg;
338│ c->msgsize *= 2;
339│ }
340│
341│ msg = c->msglist + c->msgused;
342│
343│ /* this wipes msg_iovlen, msg_control, msg_controllen, and
344│ msg_flags, the last 3 of which aren't defined on solaris: */
345│ memset(msg, 0, sizeof(struct msghdr));
346│
347│ msg->msg_iov = &c->iov[c->iovused];
348│
349│ if (IS_UDP(c->transport) && c->request_addr_size > 0) {
350│ msg->msg_name = &c->request_addr;
351│ msg->msg_namelen = c->request_addr_size;
352│ }
353│
354├───> c->msgbytes = 0;
355│ c->msgused++;
356│
357│ if (IS_UDP(c->transport)) {
358│ /* Leave room for the UDP header, which we'll fill in later. */
359│ return add_iov(c, NULL, UDP_HEADER_SIZE);
360│ }
361│
362│ return 0;
363│ }
5 tokenize_command (command=0x7ffff0010bd0 "set test1 0 0 7", tokens=0x7ffff6da8bd0, max_tokens=8) at memcached.c:3045
ntokensでは、"set test1 0 0 7 sample1"の要素数である6が返される。
3032│ *
3033│ * Usage example:
3034│ *
3035│ * while(tokenize_command(command, ncommand, tokens, max_tokens) > 0) {
3036│ * for(int ix = 0; tokens[ix].length != 0; ix++) {
3037│ * ...
3038│ * }
3039│ * ncommand = tokens[ix].value - command;
3040│ * command = tokens[ix].value;
3041│ * }
3042│ */
3043│ static size_t tokenize_command(char *command, token_t *tokens, const size_t max_tokens) {
3044│ char *s, *e;
3045├───> size_t ntokens = 0;
3046│ size_t len = strlen(command);
3047│ unsigned int i = 0;
3048│
3049│ assert(command != NULL && tokens != NULL && max_tokens > 1);
3050│
3051│ s = e = command;
3052│ for (i = 0; i < len; i++) {
3053│ if (*e == ' ') {
3054│ if (s != e) {
3055│ tokens[ntokens].value = s;
3056│ tokens[ntokens].length = e - s;
3057│ ntokens++;
3058│ *e = '\0';
3059│ if (ntokens == max_tokens - 1) {
3060│ e++;
3061│ s = e; /* so we don't add an extra token */
3062│ break;
3063│ }
3064│ }
3065│ s = e + 1;
3066│ }
3067├───────> e++;
3068│ }
3069│
3070│ if (s != e) {
3071│ tokens[ntokens].value = s;
3072│ tokens[ntokens].length = e - s;
3073├───────> ntokens++;
3074│ }
3075│
3076│ /*
3077│ * If we scanned the whole string, the terminal value pointer is null,
3078│ * otherwise it is the first unprocessed character.
3079│ */
3080│ tokens[ntokens].value = *e == '\0' ? NULL : e;
3081│ tokens[ntokens].length = 0;
3082│ ntokens++;
3083│
3084│ return ntokens;
3085│ }
4
tokenize_command関数の実行結果を元に処理が実行される。ここではSETコマンドなのでprocess_update_command関数の実行
4747│ ntokens = tokenize_command(command, tokens, MAX_TOKENS);
4748│ if (ntokens >= 3 &&
4749│ ((strcmp(tokens[COMMAND_TOKEN].value, "get") == 0) ||
4750│ (strcmp(tokens[COMMAND_TOKEN].value, "bget") == 0))) {
4751│
4752│ process_get_command(c, tokens, ntokens, false, false);
4753│
4754├───> } else if ((ntokens == 6 || ntokens == 7) &&
4755│ ((strcmp(tokens[COMMAND_TOKEN].value, "add") == 0 && (comm = NREAD_ADD)) ||
4756│ (strcmp(tokens[COMMAND_TOKEN].value, "set") == 0 && (comm = NREAD_SET)) ||
4757│ (strcmp(tokens[COMMAND_TOKEN].value, "replace") == 0 && (comm = NREAD_REPLACE)) ||
4758│ (strcmp(tokens[COMMAND_TOKEN].value, "prepend") == 0 && (comm = NREAD_PREPEND)) ||
4759│ (strcmp(tokens[COMMAND_TOKEN].value, "append") == 0 && (comm = NREAD_APPEND)) )) {
4760│
4761│ process_update_command(c, tokens, ntokens, comm, false);
4762│
4763│ } else if ((ntokens == 7 || ntokens == 8) && (strcmp(tokens[COMMAND_TOKEN].value, "cas") == 0 && (comm =
4764│
4765│ process_update_command(c, tokens, ntokens, comm, true);
4766│
5 process_update_command (c=0x7ffff00109a0, tokens=0x7ffff6da8bd0, ntokens=6, comm=2, handle_cas=false) at memcached.c:4107
4103│ static void process_update_command(conn *c, token_t *tokens, const size_t ntokens, int comm, bool handle_cas
4104│ char *key;
4105│ size_t nkey;
4106│ unsigned int flags;
4107│ int32_t exptime_int = 0;
4108│ time_t exptime;
4109│ int vlen;
4110├───> uint64_t req_cas_id=0;
4111│ item *it;
4112│
4113│ assert(c != NULL);
4114│
4115│ set_noreply_maybe(c, tokens, ntokens);
4116│
4117│ if (tokens[KEY_TOKEN].length > KEY_MAX_LENGTH) {
4118│ out_string(c, "CLIENT_ERROR bad command line format");
4119│ return;
4120│ }
4121│
6 set_noreply_maybe (c=0x7ffff00109a0, tokens=0x7ffff6da8bd0, ntokens=6) at memcached.c:3102
ここではSETコマンド実行時にnoreplyオプションを指定していないので何も実行されない。
3100│ static inline bool set_noreply_maybe(conn *c, token_t *tokens, size_t ntokens)
3101│ {
3102├───> int noreply_index = ntokens - 2;
3103│
3104│ /*
3105│ NOTE: this function is not the first place where we are going to
3106│ send the reply. We could send it instead from process_command()
3107│ if the request line has wrong number of tokens. However parsing
3108│ malformed line for "noreply" option is not reliable anyway, so
3109│ it can't be helped.
3110│ */
3111│ if (tokens[noreply_index].value
3112│ && strcmp(tokens[noreply_index].value, "noreply") == 0) {
3113│ c->noreply = true;
3114│ }
3115│ return c->noreply;
3116│ }
5
item_allocが実行される
4122│ key = tokens[KEY_TOKEN].value;
4123│ nkey = tokens[KEY_TOKEN].length;
4124│
4125│ if (! (safe_strtoul(tokens[2].value, (uint32_t *)&flags)
4126│ && safe_strtol(tokens[3].value, &exptime_int)
4127│ && safe_strtol(tokens[4].value, (int32_t *)&vlen))) {
4128│ out_string(c, "CLIENT_ERROR bad command line format");
4129│ return;
4130│ }
4131│
4132│ /* Ubuntu 8.04 breaks when I pass exptime to safe_strtol */
4133├───> exptime = exptime_int;
4134│
4135│ /* Negative exptimes can underflow and end up immortal. realtime() will
4136│ immediately expire values that are greater than REALTIME_MAXDELTA, but less
4137│ than process_started, so lets aim for that. */
4138│ if (exptime < 0)
4139│ exptime = REALTIME_MAXDELTA + 1;
4140│
4141│ // does cas value exist?
4142│ if (handle_cas) {
4143│ if (!safe_strtoull(tokens[5].value, &req_cas_id)) {
4144│ out_string(c, "CLIENT_ERROR bad command line format");
4145│ return;
4146│ }
4147│ }
4148│
4149│ if (vlen < 0 || vlen > (INT_MAX - 2)) {
4150│ out_string(c, "CLIENT_ERROR bad command line format");
4151│ return;
4152│ }
4153│ vlen += 2;
4154│
4155├───> if (settings.detail_enabled) {
4156│ stats_prefix_record_set(key, nkey);
4157│ }
4158│
4159│ it = item_alloc(key, nkey, flags, realtime(exptime), vlen);
4160│
4161│ if (it == 0) {
4162│ enum store_item_type status;
4163│ if (! item_size_ok(nkey, flags, vlen)) {
4164│ out_string(c, "SERVER_ERROR object too large for cache");
4165│ status = TOO_LARGE;
4166│ } else {
4167│ out_of_memory(c, "SERVER_ERROR out of memory storing object");
6 realtime (exptime=0) at memcached.c:208
200│ /*
201│ * given time value that's either unix time or delta from current unix time, return
202│ * unix time. Use the fact that delta can't exceed one month (and real time value can't
203│ * be that low).
204│ */
205│ static rel_time_t realtime(const time_t exptime) {
206│ /* no. of seconds in 30 days - largest possible delta exptime */
207│
208│ if (exptime == 0) return 0; /* 0 means never expire */
209│
210│ if (exptime > REALTIME_MAXDELTA) {
211│ /* if item expiration is at/before the server started, give it an
212│ expiration time of 1 second after the server started.
213│ (because 0 means don't expire). without this, we'd
214│ underflow and wrap around to some large value way in the
215│ future, effectively making items expiring in the past
216│ really expiring never */
217│ if (exptime <= process_started)
218│ return (rel_time_t)1;
219│ return (rel_time_t)(exptime - process_started);
220│ } else {
221│ return (rel_time_t)(exptime + current_time);
222│ }
223├>}
6 item_alloc (key=0x7ffff0010bd4 "test1", nkey=5, flags=0, exptime=0, nbytes=9) at thread.c:579
571│ /********************************* ITEM ACCESS *******************************/
572│
573│ /*
574│ * Allocates a new item.
575│ */
576│ item *item_alloc(char *key, size_t nkey, int flags, rel_time_t exptime, int nbytes) {
577│ item *it;
578│ /* do_item_alloc handles its own locks */
579├───> it = do_item_alloc(key, nkey, flags, exptime, nbytes);
580│ return it;
581│ }
7 do_item_alloc (key=0x7ffff0010bd4 "test1", nkey=5, flags=0, exptime=0, nbytes=9) at items.c:260
CASを有効にしていると、uint64_tのサイズ分の8バイトが全体サイズに追加されている処理も行われている。
257│ item *do_item_alloc(char *key, const size_t nkey, const unsigned int flags,
258│ const rel_time_t exptime, const int nbytes) {
259│ uint8_t nsuffix;
260│ item *it = NULL;
261│ char suffix[40];
262│ // Avoid potential underflows.
263├───> if (nbytes < 2)
264│ return 0;
265│
266│ size_t ntotal = item_make_header(nkey + 1, flags, nbytes, suffix, &nsuffix);
267│ if (settings.use_cas) {
268│ ntotal += sizeof(uint64_t);
269│ }
270│
271│ unsigned int id = slabs_clsid(ntotal);
272│ unsigned int hdr_id = 0;
273│ if (id == 0)
274│ return 0;
275│
8 item_make_header (nkey=6 '\006', flags=0, nbytes=9, suffix=0x7ffff6da8a70 "", nsuffix=0x7ffff6da8aa7 "") at items.c:171
158│ * Generates the variable-sized part of the header for an object.
159│ *
160│ * key - The key
161│ * nkey - The length of the key
162│ * flags - key flags
163│ * nbytes - Number of bytes to hold value and addition CRLF terminator
164│ * suffix - Buffer for the "VALUE" line suffix (flags, size).
165│ * nsuffix - The length of the suffix is stored here.
166│ *
167│ * Returns the total size of the header.
168│ */
169│ static size_t item_make_header(const uint8_t nkey, const unsigned int flags, const int nbytes,
170│ char *suffix, uint8_t *nsuffix) {
171├───> if (settings.inline_ascii_response) {
172│ /* suffix is defined at 40 chars elsewhere.. */
173│ *nsuffix = (uint8_t) snprintf(suffix, 40, " %u %d\r\n", flags, nbytes - 2);
174│ } else {
175│ if (flags == 0) {
176│ *nsuffix = 0;
177│ } else {
178│ *nsuffix = sizeof(flags);
179│ }
180│ }
181│ return sizeof(item) + nkey + *nsuffix + nbytes;
182│ }
8 slabs_clsid (size=71) at slabs.c:92
83│ /*
84│ * Figures out which slab class (chunk size) is required to store an item of
85│ * a given size.
86│ *
87│ * Given object size, return id to use when allocating/freeing memory for object
88│ * 0 means error: can't store such a large object
89│ */
90│
91│ unsigned int slabs_clsid(const size_t size) {
92├───> int res = POWER_SMALLEST;
93│
94│ if (size == 0 || size > settings.item_size_max)
95│ return 0;
96│ while (size > slabclass[res].size)
97│ if (res++ == power_largest) /* won't fit in the biggest slab */
98│ return power_largest;
99│ return res;
100│ }
7
elseのdo_item_alloc_pullが実行される
276│ /* This is a large item. Allocate a header object now, lazily allocate
277│ * chunks while reading the upload.
278│ */
279├───> if (ntotal > settings.slab_chunk_size_max) {
280│ /* We still link this item into the LRU for the larger slab class, but
281│ * we're pulling a header from an entirely different slab class. The
282│ * free routines handle large items specifically.
283│ */
284│ int htotal = nkey + 1 + nsuffix + sizeof(item) + sizeof(item_chunk);
285│ if (settings.use_cas) {
286│ htotal += sizeof(uint64_t);
287│ }
288│ #ifdef NEED_ALIGN
289│ // header chunk needs to be padded on some systems
290│ int remain = htotal % 8;
291│ if (remain != 0) {
292│ htotal += 8 - remain;
293│ }
294│ #endif
295│ hdr_id = slabs_clsid(htotal);
296│ it = do_item_alloc_pull(htotal, hdr_id);
297│ /* setting ITEM_CHUNKED is fine here because we aren't LINKED yet. */
298│ if (it != NULL)
299│ it->it_flags |= ITEM_CHUNKED;
300│ } else {
301├───────> it = do_item_alloc_pull(ntotal, id);
302│ }
303│
8 do_item_alloc (key=0x7ffff0010bd4 "test1", nkey=5, flags=0, exptime=0, nbytes=9) at items.c:272
slabs_allocが実行される。
184│ item *do_item_alloc_pull(const size_t ntotal, const unsigned int id) {
185│ item *it = NULL;
186│ int i;
187│ /* If no memory is available, attempt a direct LRU juggle/eviction */
188│ /* This is a race in order to simplify lru_pull_tail; in cases where
189│ * locked items are on the tail, you want them to fall out and cause
190│ * occasional OOM's, rather than internally work around them.
191│ * This also gives one fewer code path for slab alloc/free
192│ */
193├───> for (i = 0; i < 10; i++) {
194│ uint64_t total_bytes;
195│ /* Try to reclaim memory first */
196│ if (!settings.lru_segmented) {
197│ lru_pull_tail(id, COLD_LRU, 0, 0, 0, NULL);
198│ }
199│ it = slabs_alloc(ntotal, id, &total_bytes, 0);
200│
201│ if (settings.temp_lru)
202│ total_bytes -= temp_lru_size(id);
203│
204│ if (it == NULL) {
205│ if (lru_pull_tail(id, COLD_LRU, total_bytes, LRU_PULL_EVICT, 0, NULL) <= 0) {
206│ if (settings.lru_segmented) {
207│ lru_pull_tail(id, HOT_LRU, total_bytes, 0, 0, NULL);
208│ } else {
209│ break;
210│ }
211│ }
9 slabs_alloc (size=71, id=1, total_bytes=0x7ffff6da8a28, flags=0) at slabs.c:658
654│ void *slabs_alloc(size_t size, unsigned int id, uint64_t *total_bytes,
655│ unsigned int flags) {
656│ void *ret;
657│
658├───> pthread_mutex_lock(&slabs_lock);
659│ ret = do_slabs_alloc(size, id, total_bytes, flags);
660│ pthread_mutex_unlock(&slabs_lock);
661│ return ret;
662│ }
10 do_slabs_alloc (size=71, id=1, total_bytes=0x7ffff6da8a28, flags=0) at slabs.c:337
do_slabs_newslabが実行される。
333│ /*@null@*/
334│ static void *do_slabs_alloc(const size_t size, unsigned int id, uint64_t *total_bytes,
335│ unsigned int flags) {
336│ slabclass_t *p;
337│ void *ret = NULL;
338│ item *it = NULL;
339│
340├───> if (id < POWER_SMALLEST || id > power_largest) {
341│ MEMCACHED_SLABS_ALLOCATE_FAILED(size, 0);
342│ return NULL;
343│ }
344│ p = &slabclass[id];
345│ assert(p->sl_curr == 0 || ((item *)p->slots)->slabs_clsid == 0);
346│ if (total_bytes != NULL) {
347│ *total_bytes = p->requested;
348│ }
349│
350│ assert(size <= p->size);
351│ /* fail unless we have space at the end of a recently allocated page,
352│ we have something on our freelist, or we could allocate a new page */
353│ if (p->sl_curr == 0 && flags != SLABS_ALLOC_NO_NEWPAGE) {
354├───────> do_slabs_newslab(id);
355│ }
356│
357│ if (p->sl_curr != 0) {
358│ /* return off our freelist */
359│ it = (item *)p->slots;
360│ p->slots = it->next;
361│ if (it->next) it->next->prev = 0;
362│ /* Kill flag and initialize refcount here for lock safety in slab
363│ * mover's freeness detection. */
364│ it->it_flags &= ~ITEM_SLABBED;
365│ it->refcount = 1;
366│ p->sl_curr--;
:
:
372│ if (ret) {
373│ p->requested += size;
374│ MEMCACHED_SLABS_ALLOCATE(size, id, p->size, ret);
375│ } else {
376│ MEMCACHED_SLABS_ALLOCATE_FAILED(size, id);
377│ }
11 do_slabs_newslab (id=1) at slabs.c:302
301│ static int do_slabs_newslab(const unsigned int id) {
302│ slabclass_t *p = &slabclass[id];
303│ slabclass_t *g = &slabclass[SLAB_GLOBAL_PAGE_POOL];
304│ int len = (settings.slab_reassign || settings.slab_chunk_size_max != settings.slab_page_size)
305│ ? settings.slab_page_size
306│ : p->size * p->perslab;
307│ char *ptr;
308│
309├───> if ((mem_limit && mem_malloced + len > mem_limit && p->slabs > 0
310│ && g->slabs == 0)) {
311│ mem_limit_reached = true;
312│ MEMCACHED_SLABS_SLABCLASS_ALLOCATE_FAILED(id);
313│ return 0;
314│ }
315│
316│ if ((grow_slab_list(id) == 0) ||
317│ (((ptr = get_page_from_global_pool()) == NULL) &&
318│ ((ptr = memory_allocate((size_t)len)) == 0))) {
319│
320│ MEMCACHED_SLABS_SLABCLASS_ALLOCATE_FAILED(id);
321│ return 0;
322│ }
323│
324│ memset(ptr, 0, (size_t)len);
325│ split_slab_page_into_freelist(ptr, id);
326│
327│ p->slab_list[p->slabs++] = ptr;
328│ MEMCACHED_SLABS_SLABCLASS_ALLOCATE(id);
12 grow_slab_list (id=1) at slabs.c:270
269│ static int grow_slab_list (const unsigned int id) {
270├───> slabclass_t *p = &slabclass[id];
271│ if (p->slabs == p->list_size) {
272│ size_t new_size = (p->list_size != 0) ? p->list_size * 2 : 16;
273│ void *new_list = realloc(p->slab_list, new_size * sizeof(void *));
274│ if (new_list == 0) return 0;
275│ p->list_size = new_size;
276│ p->slab_list = new_list;
277│ }
278│ return 1;
279│ }
12 memory_allocate (size=1048576) at slabs.c:611
608│ static void *memory_allocate(size_t size) {
609│ void *ret;
610│
611├───> if (mem_base == NULL) {
612│ /* We are not using a preallocated large memory chunk */
613│ ret = malloc(size);
614│ } else {
615│ ret = mem_current;
616│
617│ if (size > mem_avail) {
618│ return NULL;
619│ }
620│
621│ /* mem_current pointer _must_ be aligned!!! */
622│ if (size % CHUNK_ALIGN_BYTES) {
623│ size += CHUNK_ALIGN_BYTES - (size % CHUNK_ALIGN_BYTES);
12 split_slab_page_into_freelist (ptr=0x7ffff44a4010 "", id=1) at slabs.c:282
281│ static void split_slab_page_into_freelist(char *ptr, const unsigned int id) {
282├───> slabclass_t *p = &slabclass[id];
283│ int x;
284│ for (x = 0; x < p->perslab; x++) {
285│ do_slabs_free(ptr, 0, id);
286│ ptr += p->size;
287│ }
288│ }
13 do_slabs_free (ptr=0x7ffff44a4010, size=0, id=1) at slabs.c:447
442│ static void do_slabs_free(void *ptr, const size_t size, unsigned int id) {
443│ slabclass_t *p;
444│ item *it;
445│
446│ assert(id >= POWER_SMALLEST && id <= power_largest);
447├───> if (id < POWER_SMALLEST || id > power_largest)
448│ return;
449│
450│ MEMCACHED_SLABS_FREE(size, id, ptr);
451│ p = &slabclass[id];
452│
453│ it = (item *)ptr;
454│ if ((it->it_flags & ITEM_CHUNKED) == 0) {
455│ #ifdef EXTSTORE
456│ bool is_hdr = it->it_flags & ITEM_HDR;
457│ #endif
458│ it->it_flags = ITEM_SLABBED;
459│ it->slabs_clsid = 0;
460│ it->prev = 0;
461│ it->next = p->slots;
462│ if (it->next) it->next->prev = it;
463│ p->slots = it;
464│
465├───────> p->sl_curr++;
466│ #ifdef EXTSTORE
467│ if (!is_hdr) {
468│ p->requested -= size;
469│ } else {
470│ p->requested -= (size - it->nbytes) + sizeof(item_hdr);
471│ }
472│ #else
473│ p->requested -= size;
474│ #endif
475│ } else {
476│ do_slabs_free_chunked(it, size);
477│ }
478│ return;
479│ }
8
317│ /* Items are initially loaded into the HOT_LRU. This is '0' but I want at
318│ * least a note here. Compiler (hopefully?) optimizes this out.
319│ */
320│ if (settings.temp_lru &&
321│ exptime - current_time <= settings.temporary_ttl) {
322│ id |= TEMP_LRU;
323│ } else if (settings.lru_segmented) {
324│ id |= HOT_LRU;
325│ } else {
326│ /* There is only COLD in compat-mode */
327│ id |= COLD_LRU;
328│ }
329├───> it->slabs_clsid = id;
330│
331│ DEBUG_REFCNT(it, '*');
332│ it->it_flags |= settings.use_cas ? ITEM_CAS : 0;
333│ it->nkey = nkey;
334│ it->nbytes = nbytes;
335│ memcpy(ITEM_key(it), key, nkey);
336│ it->exptime = exptime;
337│ if (settings.inline_ascii_response) {
338│ memcpy(ITEM_suffix(it), suffix, (size_t)nsuffix);
339│ } else if (nsuffix > 0) {
340│ memcpy(ITEM_suffix(it), &flags, sizeof(flags));
341│ }
342│ it->nsuffix = nsuffix;
343│
344│ /* Initialize internal chunk. */
345│ if (it->it_flags & ITEM_CHUNKED) {
346│ item_chunk *chunk = (item_chunk *) ITEM_schunk(it);
347│
348│ chunk->next = 0;
349│ chunk->prev = 0;
350│ chunk->used = 0;
351│ chunk->size = 0;
352│ chunk->head = it;
353│ chunk->orig_clsid = hdr_id;
354│ }
355├───> it->h_next = 0;
356│
357│ return it;
358│ }
5
conn_nreadの状態に設定される。
4189├───> ITEM_set_cas(it, req_cas_id);
4190│
4191│ c->item = it;
4192│ #ifdef NEED_ALIGN
4193│ if (it->it_flags & ITEM_CHUNKED) {
4194│ c->ritem = ITEM_schunk(it);
4195│ } else {
4196│ c->ritem = ITEM_data(it);
4197│ }
4198│ #else
4199│ c->ritem = ITEM_data(it);
4200│ #endif
4201│ c->rlbytes = it->nbytes;
4202│ c->cmd = comm;
4203│ conn_set_state(c, conn_nread);
4204│ }
2
drive_machineに再度戻ってきて、conn_nreadの処理が実行される。1週目はc->rlbytes > 0で特に何も実行されないが、2週目で再度case文が実行されて、c->rlbytesが0なので、complete_nreadが実行される
5680│ case conn_nread:
5681├───────────> if (c->rlbytes == 0) {
5682│ complete_nread(c);
5683│ break;
5684│ }
5685│
5686│ /* Check if rbytes < 0, to prevent crash */
5687│ if (c->rlbytes < 0) {
5688│ if (settings.verbose) {
5689│ fprintf(stderr, "Invalid rlbytes to read: len %d\n", c->rlbytes);
5690│ }
5691│ conn_set_state(c, conn_closing);
5692│ break;
5693│ }
3 complete_nread (c=0x7ffff00109a0) at memcached.c:2785
ASCIIモードなので、 complete_nread_asciiが実行される。
2780│ static void complete_nread(conn *c) {
2781│ assert(c != NULL);
2782│ assert(c->protocol == ascii_prot
2783│ || c->protocol == binary_prot);
2784│
2785├───> if (c->protocol == ascii_prot) {
2786│ complete_nread_ascii(c);
2787│ } else if (c->protocol == binary_prot) {
2788│ complete_nread_binary(c);
2789│ }
2790│ }
4 complete_nread_ascii (c=0x7ffff00109a0) at memcached.c:1231
1224│ /*
1225│ * we get here after reading the value in set/add/replace commands. The command
1226│ * has been stored in c->cmd, and the item is ready in c->item.
1227│ */
1228│ static void complete_nread_ascii(conn *c) {
1229│ assert(c != NULL);
1230│
1231│ item *it = c->item;
1232│ int comm = c->cmd;
1233│ enum store_item_type ret;
1234│ bool is_valid = false;
1235│
1236├───> pthread_mutex_lock(&c->thread->stats.mutex);
1237│ c->thread->stats.slab_stats[ITEM_clsid(it)].set_cmds++;
1238│ pthread_mutex_unlock(&c->thread->stats.mutex);
1239│
1240│ if ((it->it_flags & ITEM_CHUNKED) == 0) {
1241│ if (strncmp(ITEM_data(it) + it->nbytes - 2, "\r\n", 2) == 0) {
1242│ is_valid = true;
1243│ }
1244│ } else {
1245│ char buf[2];
1246│ /* should point to the final item chunk */
1247│ item_chunk *ch = (item_chunk *) c->ritem;
1248│ assert(ch->used != 0);
対象のデータをキャッシュに保存するstore_item関数が実行される
1268│ if (!is_valid) {
1269│ out_string(c, "CLIENT_ERROR bad data chunk");
1270│ } else {
1271├─────> ret = store_item(it, comm, c);
1272│
5 store_item (item=0x7ffff45a3f70, comm=2, c=0x7ffff00109a0) at thread.c:689
682│ /*
683│ * Stores an item in the cache (high level, obeys set/add/replace semantics)
684│ */
685│ enum store_item_type store_item(item *item, int comm, conn* c) {
686│ enum store_item_type ret;
687│ uint32_t hv;
688│
689├───> hv = hash(ITEM_key(item), item->nkey);
690│ item_lock(hv);
691│ ret = do_store_item(item, comm, c, hv);
692│ item_unlock(hv);
693│ return ret;
694│ }
6 hash
MurmurHash3_x86_32関数が実行される。
69│ /* Definition modified slightly from the public domain interface (no seed +
70│ * return value */
71│ uint32_t MurmurHash3_x86_32 ( const void * key, size_t length)
72│ {
73├─> const uint8_t * data = (const uint8_t*)key;
74│ const int nblocks = length / 4;
75│
76│ uint32_t h1 = 0;
77│
78│ uint32_t c1 = 0xcc9e2d51;
79│ uint32_t c2 = 0x1b873593;
80│
81│ //----------
82│ // body
83│
84│ const uint32_t * blocks = (const uint32_t *)(data + nblocks*4);
85│
86│ for(int i = -nblocks; i; i++)
87│ {
88│ uint32_t k1 = getblock32(blocks,i);
89│
90│ k1 *= c1;
91│ k1 = ROTL32(k1,15);
92│ k1 *= c2;
93│
94│ h1 ^= k1;
95│ h1 = ROTL32(h1,13);
96│ h1 = h1*5+0xe6546b64;
97│ }
98│
99│ //----------
100│ // tail
101│
102│ const uint8_t * tail = (const uint8_t*)(data + nblocks*4);
103│
104│ uint32_t k1 = 0;
105│
106│ switch(length & 3)
107│ {
108│ case 3: k1 ^= tail[2] << 16;
109│ case 2: k1 ^= tail[1] << 8;
110├─> case 1: k1 ^= tail[0];
111│ k1 *= c1; k1 = ROTL32(k1,15); k1 *= c2; h1 ^= k1;
112│ };
113│
114│ //----------
115│ // finalization
116│
117│ h1 ^= length;
118│
119│ h1 = fmix32(h1);
120│
121│ //*(uint32_t*)out = h1;
122│ return h1;
123│ }
7 getblock32 (i=-1, p=0x7ffff45a3fac) at murmur3_hash.c:50
44│ //-----------------------------------------------------------------------------
45│ // Block read - if your platform needs to do endian-swapping or can only
46│ // handle aligned reads, do the conversion here
47│
48│ static FORCE_INLINE uint32_t getblock32 ( const uint32_t * p, int i )
49│ {
50├─> return p[i];
51│ }
52│
53│ //-----------------------------------------------------------------------------
7 rotl32 (x=787512756, r=15 '\017') at murmur3_hash.c:35
33│ static inline uint32_t rotl32 ( uint32_t x, int8_t r )
34│ {
35├─> return (x << r) | (x >> (32 - r));
36│ }
37│
38│ #define ROTL32(x,y) rotl32(x,y)
7 fmix32 (h=2252073768) at murmur3_hash.c:58
54│ // Finalization mix - force all bits of a hash block to avalanche
55│
56│ static FORCE_INLINE uint32_t fmix32 ( uint32_t h )
57│ {
58├─> h ^= h >> 16;
59│ h *= 0x85ebca6b;
60│ h ^= h >> 13;
61│ h *= 0xc2b2ae35;
62│ h ^= h >> 16;
63│
64│ return h;
65│ }
66│
67│ //-----------------------------------------------------------------------------
68│
6 item_lock (hv=3155333867) at thread.c:104
6 item_unlock (hv=3155333867) at thread.c:120
95│ /* item_lock() must be held for an item before any modifications to either its
96│ * associated hash bucket, or the structure itself.
97│ * LRU modifications must hold the item lock, and the LRU lock.
98│ * LRU's accessing items must item_trylock() before modifying an item.
99│ * Items accessible from an LRU must not be freed or modified
100│ * without first locking and removing from the LRU.
101│ */
102│
103│ void item_lock(uint32_t hv) {
104├───> mutex_lock(&item_locks[hv & hashmask(item_lock_hashpower)]);
105│ }
119│ void item_unlock(uint32_t hv) {
120├───> mutex_unlock(&item_locks[hv & hashmask(item_lock_hashpower)]);
121│ }
6 do_store_item (it=0x7ffff45a3f70, comm=2, c=0x7ffff00109a0, hv=3155333867) at memcached.c:2897
2890│ /*
2891│ * Stores an item in the cache according to the semantics of one of the set
2892│ * commands. In threaded mode, this is protected by the cache lock.
2893│ *
2894│ * Returns the state of storage.
2895│ */
2896│ enum store_item_type do_store_item(item *it, int comm, conn *c, const uint32_t hv) {
2897│ char *key = ITEM_key(it);
2898├───> item *old_it = do_item_get(key, it->nkey, hv, c, DONT_UPDATE);
2899│ enum store_item_type stored = NOT_STORED;
2900│
2901│ item *new_it = NULL;
2902│ uint32_t flags;
2903│
2904│ if (old_it != NULL && comm == NREAD_ADD) {
2905│ /* add only adds a nonexistent item, but promote to head of LRU */
2906│ do_item_update(old_it);
2907│ } else if (!old_it && (comm == NREAD_REPLACE
2908│ || comm == NREAD_APPEND || comm == NREAD_PREPEND))
2909│ {
2910│ /* replace only replaces an existing value; don't store */
7 do_item_get (key=0x7ffff45a3fa8 "test1", nkey=5, hv=3155333867, c=0x7ffff00109a0, do_update=false) at items.c:955
953│ /** wrapper around assoc_find which does the lazy expiration logic */
954│ item *do_item_get(const char *key, const size_t nkey, const uint32_t hv, conn *c, const bool do_update) {
955│ item *it = assoc_find(key, nkey, hv);
956├───> if (it != NULL) {
957│ refcount_incr(it);
958│ /* Optimization for slab reassignment. prevents popular items from
959│ * jamming in busy wait. Can only do this here to satisfy lock order
960│ * of item_lock, slabs_lock. */
961│ /* This was made unsafe by removal of the cache_lock:
962│ * slab_rebalance_signal and slab_rebal.* are modified in a separate
963│ * thread under slabs_lock. If slab_rebalance_signal = 1, slab_start =
964│ * NULL (0), but slab_end is still equal to some value, this would end
965│ * up unlinking every item fetched.
966│ * This is either an acceptable loss, or if slab_rebalance_signal is
967│ * true, slab_start/slab_end should be put behind the slabs_lock.
968│ * Which would cause a huge potential slowdown.
969│ * Could also use a specific lock for slab_rebal.* and
970│ * slab_rebalance_signal (shorter lock?)
971│ */
972│ /*if (slab_rebalance_signal &&
973│ ((void *)it >= slab_rebal.slab_start && (void *)it < slab_rebal.slab_end)) {
974│ do_item_unlink(it, hv);
975│ do_item_remove(it);
976│ it = NULL;
977│ }*/
978│ }
Lazy expirationですでにあるキー中のメモリは上書きされる。ここでは特に保存したこともないので、"> NOT FOUND ")が表示される。
979├───> int was_found = 0;
980│
981│ if (settings.verbose > 2) {
982│ int ii;
983│ if (it == NULL) {
984│ fprintf(stderr, "> NOT FOUND ");
985│ } else {
986│ fprintf(stderr, "> FOUND KEY ");
987│ }
988│ for (ii = 0; ii < nkey; ++ii) {
989│ fprintf(stderr, "%c", key[ii]);
990│ }
991│ }
6
2943│ } else {
2944│ int failed_alloc = 0;
2945│ /*
2946│ * Append - combine new and old record into single one. Here it's
2947│ * atomic and thread-safe.
2948│ */
2949├───────> if (comm == NREAD_APPEND || comm == NREAD_PREPEND) {
2950│ /*
2951│ * Validate CAS
2952│ */
2953│ if (ITEM_get_cas(it) != 0) {
2954│ // CAS much be equal
2955│ if (ITEM_get_cas(it) != ITEM_get_cas(old_it)) {
2956│ stored = EXISTS;
2957│ }
2958│ }
do_item_linkが実行される。
2987├───────> if (stored == NOT_STORED && failed_alloc == 0) {
2988│ if (old_it != NULL) {
2989│ STORAGE_delete(c->thread->storage, old_it);
2990│ item_replace(old_it, it, hv);
2991│ } else {
2992│ do_item_link(it, hv);
2993│ }
2994│
2995│ c->cas = ITEM_get_cas(it);
2996│
2997│ stored = STORED;
2998│ }
2999│ }
7 do_item_link (it=0x7ffff45a3f70, hv=3155333867) at items.c:474
471│ int do_item_link(item *it, const uint32_t hv) {
472│ MEMCACHED_ITEM_LINK(ITEM_key(it), it->nkey, it->nbytes);
473│ assert((it->it_flags & (ITEM_LINKED|ITEM_SLABBED)) == 0);
474│ it->it_flags |= ITEM_LINKED;
475│ it->time = current_time;
476│
477│ STATS_LOCK();
478├───> stats_state.curr_bytes += ITEM_ntotal(it);
479│ stats_state.curr_items += 1;
480│ stats.total_items += 1;
481│ STATS_UNLOCK();
482│
483│ /* Allocate a new CAS ID on link. */
484│ ITEM_set_cas(it, (settings.use_cas) ? get_cas_id() : 0);
485│ assoc_insert(it, hv);
486│ item_link_q(it);
487│ refcount_incr(it);
488│ item_stats_sizes_add(it);
489│
490│ return 1;
491│ }
8 get_cas_id () at items.c:112
108│ /* Get the next CAS id for a new item. */
109│ /* TODO: refactor some atomics for this. */
110│ uint64_t get_cas_id(void) {
111│ static uint64_t cas_id = 0;
112├───> pthread_mutex_lock(&cas_id_lock);
113│ uint64_t next_id = ++cas_id;
114│ pthread_mutex_unlock(&cas_id_lock);
115│ return next_id;
116│ }
8 assoc_insert (it=0x7ffff45a3f70, hv=3155333867) at assoc.c:160
154│ /* Note: this isn't an assoc_update. The key must not already exist to call this */
155│ int assoc_insert(item *it, const uint32_t hv) {
156│ unsigned int oldbucket;
157│
158│ // assert(assoc_find(ITEM_key(it), it->nkey) == 0); /* shouldn't have duplicately named things defined */
159│
160├───> if (expanding &&
161│ (oldbucket = (hv & hashmask(hashpower - 1))) >= expand_bucket)
162│ {
163│ it->h_next = old_hashtable[oldbucket];
164│ old_hashtable[oldbucket] = it;
165│ } else {
166│ it->h_next = primary_hashtable[hv & hashmask(hashpower)];
167│ primary_hashtable[hv & hashmask(hashpower)] = it;
168│ }
169│
170│ MEMCACHED_ASSOC_INSERT(ITEM_key(it), it->nkey);
171│ return 1;
172│ }
8 item_link_q (it=0x7ffff45a3f70) at items.c:421
420│ static void item_link_q(item *it) {
421├───> pthread_mutex_lock(&lru_locks[it->slabs_clsid]);
422│ do_item_link_q(it);
423│ pthread_mutex_unlock(&lru_locks[it->slabs_clsid]);
424│ }
9 do_item_link_q (it=0x7ffff45a3f70) at items.c:397
393│ static void do_item_link_q(item *it) { /* item is the new head */
394│ item **head, **tail;
395│ assert((it->it_flags & ITEM_SLABBED) == 0);
396│
397│ head = &heads[it->slabs_clsid];
398│ tail = &tails[it->slabs_clsid];
399│ assert(it != *head);
400│ assert((*head && *tail) || (*head == 0 && *tail == 0));
401│ it->prev = 0;
402│ it->next = *head;
403├───> if (it->next) it->next->prev = it;
404│ *head = it;
405│ if (*tail == 0) *tail = it;
406│ sizes[it->slabs_clsid]++;
407│ #ifdef EXTSTORE
408│ if (it->it_flags & ITEM_HDR) {
409│ sizes_bytes[it->slabs_clsid] += (ITEM_ntotal(it) - it->nbytes) + sizeof(item_hdr);
410│ } else {
411│ sizes_bytes[it->slabs_clsid] += ITEM_ntotal(it);
412│ }
413│ #else
414│ sizes_bytes[it->slabs_clsid] += ITEM_ntotal(it);
415│ #endif
416│
417│ return;
418│ }
8 item_stats_sizes_add (it=0x7ffff45a3f70) at items.c:907
906│ void item_stats_sizes_add(item *it) {
907├───> if (stats_sizes_hist == NULL || stats_sizes_cas_min > ITEM_get_cas(it))
908│ return;
909│ int ntotal = ITEM_ntotal(it);
910│ int bucket = ntotal / 32;
911│ if ((ntotal % 32) != 0) bucket++;
912│ if (bucket < stats_sizes_buckets) stats_sizes_hist[bucket]++;
913│ }
6
3001├───> if (old_it != NULL)
3002│ do_item_remove(old_it); /* release our reference */
3003│ if (new_it != NULL)
3004│ do_item_remove(new_it);
3005│
3006│ if (stored == STORED) {
3007│ c->cas = ITEM_get_cas(it);
3008│ }
3009│ LOGGER_LOG(c->thread->l, LOG_MUTATIONS, LOGGER_ITEM_STORE, NULL,
3010│ stored, comm, ITEM_key(it), it->nkey, it->exptime, ITEM_clsid(it));
3011│
3012│ return stored;
3013│ }
4
retがSTOREDとなっている。
1303│ switch (ret) {
1304│ case STORED:
1305├─────────> out_string(c, "STORED");
1306│ break;
1307│ case EXISTS:
1308│ out_string(c, "EXISTS");
1309│ break;
1310│ case NOT_FOUND:
1311│ out_string(c, "NOT_FOUND");
1312│ break;
1313│ case NOT_STORED:
1314│ out_string(c, "NOT_STORED");
1315│ break;
1316│ default:
1317│ out_string(c, "SERVER_ERROR Unhandled storage type.");
5 out_string (c=0x7ffff00109a0, str=0x424f99 "STORED") at memcached.c:1171
conn_writeに設定される。
1166│ static void out_string(conn *c, const char *str) {
1167│ size_t len;
1168│
1169│ assert(c != NULL);
1170│
1171├───> if (c->noreply) {
1172│ if (settings.verbose > 1)
1173│ fprintf(stderr, ">%d NOREPLY %s\n", c->sfd, str);
1174│ c->noreply = false;
1175│ conn_set_state(c, conn_new_cmd);
1176│ return;
1177│ }
1178│
1179│ if (settings.verbose > 1)
1180│ fprintf(stderr, ">%d %s\n", c->sfd, str);
1181│
1182│ /* Nuke a partial output... */
1183│ c->msgcurr = 0;
1184│ c->msgused = 0;
1185│ c->iovused = 0;
1186├───> add_msghdr(c);
1187│
1188│ len = strlen(str);
1189│ if ((len + 2) > c->wsize) {
1190│ /* ought to be always enough. just fail for simplicity */
1191│ str = "SERVER_ERROR output line too long";
1192│ len = strlen(str);
1193│ }
1194│
1195│ memcpy(c->wbuf, str, len);
1196│ memcpy(c->wbuf + len, "\r\n", 2);
1197│ c->wbytes = len + 2;
1198│ c->wcurr = c->wbuf;
1199│
1200│ conn_set_state(c, conn_write);
1201│ c->write_and_go = conn_new_cmd;
1202│ return;
1203│ }
5 item_remove (item=0x7ffff45a3f70) at thread.c:638
hash関数は、先程と同様MurmurHash3_x86_32関数が実行される。
632│ /*
633│ * Decrements the reference count on an item and adds it to the freelist if
634│ * needed.
635│ */
636│ void item_remove(item *item) {
637│ uint32_t hv;
638├───> hv = hash(ITEM_key(item), item->nkey);
639│
640│ item_lock(hv);
641│ do_item_remove(item);
642│ item_unlock(hv);
643│ }
644│
6 do_item_remove (it=0x7ffff45a3f70) at items.c:529
524│ void do_item_remove(item *it) {
525│ MEMCACHED_ITEM_REMOVE(ITEM_key(it), it->nkey, it->nbytes);
526│ assert((it->it_flags & ITEM_SLABBED) == 0);
527│ assert(it->refcount > 0);
528│
529├───> if (refcount_decr(it) == 0) {
530│ item_free(it);
531│ }
532│ }
2
drive_machineに戻ってくる。c->stateはconn_write
5810│ case conn_write:
5811│ /*
5812│ * We want to write out a simple response. If we haven't already,
5813│ * assemble it into a msgbuf list (this will be a single-entry
5814│ * list for TCP or a two-entry list for UDP).
5815│ */
5816├───────────> if (c->iovused == 0 || (IS_UDP(c->transport) && c->iovused == 1)) {
5817│ if (add_iov(c, c->wcurr, c->wbytes) != 0) {
5818│ if (settings.verbose > 0)
5819│ fprintf(stderr, "Couldn't build response\n");
5820│ conn_set_state(c, conn_closing);
5821│ break;
5822│ }
5823│ }
5824│
5825│ /* fall through... */
5826│
:
:
5845├─────────> if (IS_UDP(c->transport) && c->msgcurr == 0 && build_udp_headers(c) != 0) {
5846│ if (settings.verbose > 0)
5847│ fprintf(stderr, "Failed to build UDP headers\n");
5848│ conn_set_state(c, conn_closing);
5849│ break;
5850│ }
3 add_iov (c=0x7ffff00109a0, buf=0x7ffff00113e0, len=8) at memcached.c:1048
1035│ *
1036│ * Returns 0 on success, -1 on out-of-memory.
1037│ * Note: This is a hot path for at least ASCII protocol. While there is
1038│ * redundant code in splitting TCP/UDP handling, any reduction in steps has a
1039│ * large impact for TCP connections.
1040│ */
1041│
1042│ static int add_iov(conn *c, const void *buf, int len) {
1043│ struct msghdr *m;
1044│ int leftover;
1045│
1046│ assert(c != NULL);
1047│
1048├───> if (IS_UDP(c->transport)) {
1049│ do {
1050│ m = &c->msglist[c->msgused - 1];
1051│
1052│ /*
1053│ * Limit UDP packets to UDP_MAX_PAYLOAD_SIZE bytes.
1054│ */
1055│
1056│ /* We may need to start a new msghdr if this one is full. */
1057│ if (m->msg_iovlen == IOV_MAX ||
1058│ (c->msgbytes >= UDP_MAX_PAYLOAD_SIZE)) {
1059│ add_msghdr(c);
1060│ m = &c->msglist[c->msgused - 1];
その後、ensure_iov_space関数が呼ばれる。ここでは特に何もしない。
1085│ } else {
1086│ /* Optimized path for TCP connections */
1087│ m = &c->msglist[c->msgused - 1];
1088│ if (m->msg_iovlen == IOV_MAX) {
1089│ add_msghdr(c);
1090│ m = &c->msglist[c->msgused - 1];
1091│ }
1092│
1093│ if (ensure_iov_space(c) != 0)
1094│ return -1;
1095│
1096├───────> m->msg_iov[m->msg_iovlen].iov_base = (void *)buf;
1097│ m->msg_iov[m->msg_iovlen].iov_len = len;
1098│ c->msgbytes += len;
1099│ c->iovused++;
1100│ m->msg_iovlen++;
1101│ }
1102│
1103│ return 0;
1104│ }
4 ensure_iov_space (c=0x7ffff00109a0) at memcached.c:1008
999│ /*
1000│ * Ensures that there is room for another struct iovec in a connection's
1001│ * iov list.
1002│ *
1003│ * Returns 0 on success, -1 on out-of-memory.
1004│ */
1005│ static int ensure_iov_space(conn *c) {
1006│ assert(c != NULL);
1007│
1008├───> if (c->iovused >= c->iovsize) {
1009│ int i, iovnum;
1010│ struct iovec *new_iov = (struct iovec *)realloc(c->iov,
1011│ (c->iovsize * 2) * sizeof(struct iovec));
1012│ if (! new_iov) {
1013│ STATS_LOCK();
1014│ stats.malloc_fails++;
1015│ STATS_UNLOCK();
1016│ return -1;
1017│ }
1018│ c->iov = new_iov;
1019│ c->iovsize *= 2;
1020│
3 transmit (c=0x7ffff00109a0) at memcached.c:5361
TRANSMIT_INCOMPLETEをreturnする。その後、抜けて、drive_machineからc->stateはconn_writeで再度この関数が実行される。その時は、TRANSMIT_COMPLETEをreturnする。
5349│ /*
5350│ * Transmit the next chunk of data from our list of msgbuf structures.
5351│ *
5352│ * Returns:
5353│ * TRANSMIT_COMPLETE All done writing.
5354│ * TRANSMIT_INCOMPLETE More data remaining to write.
5355│ * TRANSMIT_SOFT_ERROR Can't write any more right now.
5356│ * TRANSMIT_HARD_ERROR Can't write (c->state is set to conn_closing)
5357│ */
5358│ static enum transmit_result transmit(conn *c) {
5359│ assert(c != NULL);
5360│
5361├───> if (c->msgcurr < c->msgused &&
5362│ c->msglist[c->msgcurr].msg_iovlen == 0) {
5363│ /* Finished writing the current msg; advance to the next. */
5364│ c->msgcurr++;
5365│ }
5366│ if (c->msgcurr < c->msgused) {
5367│ ssize_t res;
5368│ struct msghdr *m = &c->msglist[c->msgcurr];
5369│
5370│ res = c->sendmsg(c, m, 0);
5371│ if (res > 0) {
5372│ pthread_mutex_lock(&c->thread->stats.mutex);
5373│ c->thread->stats.bytes_written += res;
5374│ pthread_mutex_unlock(&c->thread->stats.mutex);
5375│
5376│ /* We've written some of the data. Remove the completed
5377│ iovec entries from the list of pending writes. */
5378│ while (m->msg_iovlen > 0 && res >= m->msg_iov->iov_len) {
5379│ res -= m->msg_iov->iov_len;
5380│ m->msg_iovlen--;
5381│ m->msg_iov++;
5382│ }
5383│
5384│ /* Might have written just part of the last iovec entry;
5385│ adjust it so the next write will do the rest. */
5386│ if (res > 0) {
5387│ m->msg_iov->iov_base = (caddr_t)m->msg_iov->iov_base + res;
5388│ m->msg_iov->iov_len -= res;
5389│ }
5390├───────────> return TRANSMIT_INCOMPLETE;
5391│ }
5392│ if (res == -1 && (errno == EAGAIN || errno == EWOULDBLOCK)) {
5393│ if (!update_event(c, EV_WRITE | EV_PERSIST)) {
5394│ if (settings.verbose > 0)
5395│ fprintf(stderr, "Couldn't update event\n");
5396│ conn_set_state(c, conn_closing);
5397│ return TRANSMIT_HARD_ERROR;
5398│ }
5399│ return TRANSMIT_SOFT_ERROR;
5400│ }
5401│ /* if res == 0 or res == -1 and error is not EAGAIN or EWOULDBLOCK,
5402│ we have a real error, on which we close the connection */
4 tcp_sendmsg (c=0x7ffff00109a0, msg=0x7ffff0013e80, flags=0) at memcached.c:167
165│ ssize_t tcp_sendmsg(conn *c, struct msghdr *msg, int flags) {
166│ assert (c != NULL);
167├───> return sendmsg(c->sfd, msg, flags);
168│ }
2
c->write_and_goにはconn_new_cmdが設定されており、c->stateはこの状態に設定される。
5845│ if (IS_UDP(c->transport) && c->msgcurr == 0 && build_udp_headers(c) != 0) {
5846│ if (settings.verbose > 0)
5847│ fprintf(stderr, "Failed to build UDP headers\n");
5848│ conn_set_state(c, conn_closing);
5849│ break;
5850│ }
5851│ switch (transmit(c)) {
5852│ case TRANSMIT_COMPLETE:
5853├───────────────> if (c->state == conn_mwrite) {
5854│ conn_release_items(c);
5855│ /* XXX: I don't know why this wasn't the general case */
5856│ if(c->protocol == binary_prot) {
5857│ conn_set_state(c, c->write_and_go);
5858│ } else {
5859│ conn_set_state(c, conn_new_cmd);
5860│ }
5861│ } else if (c->state == conn_write) {
5862│ if (c->write_and_free) {
5863│ free(c->write_and_free);
5864│ c->write_and_free = 0;
5865│ }
5866├───────────────────> conn_set_state(c, c->write_and_go);
5867│ } else {
5868│ if (settings.verbose > 0)
5869│ fprintf(stderr, "Unexpected state %d\n", c->state);
5870│ conn_set_state(c, conn_closing);
5871│ }
5872│ break;
5873│
再びループでc->stateの状態に基づいた処理が実行され、ここでは、conn_new_cmd
5651│ case conn_new_cmd:
5652│ /* Only process nreqs at a time to avoid starving other
5653│ connections */
5654│
5655│ --nreqs;
5656├───────────> if (nreqs >= 0) {
5657│ reset_cmd_handler(c);
5658│ } else {
5659│ pthread_mutex_lock(&c->thread->stats.mutex);
5660│ c->thread->stats.conn_yields++;
5661│ pthread_mutex_unlock(&c->thread->stats.mutex);
5662│ if (c->rbytes > 0) {
5663│ /* We have already read in data into the input buffer,
5664│ so libevent will most likely not signal read events
5665│ on the socket (unless more data is available. As a
5666│ hack we should just put in a request to write data,
5667│ because that should be possible ;-)
5668│ */
3 reset_cmd_handler (c=0x7ffff00109a0) at memcached.c:2766
conn_waitingの状態に設定。
2765│ static void reset_cmd_handler(conn *c) {
2766├───> c->cmd = -1;
2767│ c->substate = bin_no_state;
2768│ if(c->item != NULL) {
2769│ item_remove(c->item);
2770│ c->item = NULL;
2771│ }
2772│ conn_shrink(c);
2773│ if (c->rbytes > 0) {
2774│ conn_set_state(c, conn_parse_cmd);
2775│ } else {
2776│ conn_set_state(c, conn_waiting);
2777│ }
2778│ }
2
update_event関数ではtrueが返され、conn_readが実行される。stop変数のフラグにtrueが設定されたことによりwhile文のループは終了する。
5612│ case conn_waiting:
5613├───────────> if (!update_event(c, EV_READ | EV_PERSIST)) {
5614│ if (settings.verbose > 0)
5615│ fprintf(stderr, "Couldn't update event\n");
5616│ conn_set_state(c, conn_closing);
5617│ break;
5618│ }
5619│
5620│ conn_set_state(c, conn_read);
5621│ stop = true;
5622│ break;
5623│
事前準備
Memcached
デバッグ用にビルドしておく。また、-O2で最適化しているとgdbの際に順にコードを追っていくと困難になるので-O0に変更しておく。そうでないと、デバッグの際に、順番に実行してもソースコード中の動きがばらばらになる。
$ git clone https://github.com/memcached/memcached.git
$ cd memcached
$ ./autogen.sh && ./configure && make CFLAGS="-g -O0" && sudo make install
cgdb
$ git clone https://github.com/cgdb/cgdb.git
$ cd cgdb
$ ./autogen.sh
$ CXXFLAGS='-std=c++11' ./configure --prefix=/usr/local
$ make
$ sudo make install
-std=c++11をつけないと、make時に"kui.cpp:310:15: error: ‘it’ does not name a type"のエラー https://github.com/cgdb/cgdb/issues/184
デバッグ
MemcachedはLibeventを利用しており、コマンド実行時に呼ばれる関数はevent_handlerを起点とする。そのため、ここにブレークポイントを打っておくと良い。
$ cgdb memcached
(gdb) b main
(gdb) b event_handler
(gdb) run -u ec2-user -vvv
event_base_loop の前まで実行
SETコマンド実行時、ここで別ターミナルから $ telnet localhost 11211 を実行。以下のコマンドを実行。
set test1 0 0 7
sample1
スレッド3に切り替えて、nextを続けていると、先程ブレークポイントを設定しておいたevent_handler関数で止まる。
(gdb) info threads
(gdb) thread 3
(gdb) next
:
:
繰り返し処理が多く、ループを飛ばすには以下を実行
(gdb) b linenum
(gdb) c
(gdb) del breaknum
定数確認にはシェルでgrep
(gdb) shell grep -r ITEMS_PER_ALLOC /tmp/memcached
