memcached（二）事件模型源码分析

/** Initializes the thread subsystem, creating various worker threads.** nthreads  Number of worker event handler threads to spawn* main_base Event base for main thread*/
void thread_init(int nthreads, struct event_base *main_base) {int         i;int         power;pthread_mutex_init(&cache_lock, NULL);pthread_mutex_init(&stats_lock, NULL);pthread_mutex_init(&init_lock, NULL);pthread_cond_init(&init_cond, NULL);pthread_mutex_init(&cqi_freelist_lock, NULL);cqi_freelist = NULL;/* Want a wide lock table, but don't waste memory */if (nthreads < 3) {power = 10;} else if (nthreads < 4) {power = 11;} else if (nthreads < 5) {power = 12;} else {/* 8192 buckets, and central locks don't scale much past 5 threads */power = 13;}item_lock_count = hashsize(power);item_lock_hashpower = power;item_locks = calloc(item_lock_count, sizeof(pthread_mutex_t));if (! item_locks) {perror("Can't allocate item locks");exit(1);}for (i = 0; i < item_lock_count; i++) {pthread_mutex_init(&item_locks[i], NULL);}pthread_key_create(&item_lock_type_key, NULL);pthread_mutex_init(&item_global_lock, NULL);threads = calloc(nthreads, sizeof(LIBEVENT_THREAD));if (! threads) {perror("Can't allocate thread descriptors");exit(1);}dispatcher_thread.base = main_base;dispatcher_thread.thread_id = pthread_self();for (i = 0; i < nthreads; i++) {int fds[2];if (pipe(fds)) {perror("Can't create notify pipe");exit(1);}threads[i].notify_receive_fd = fds[0];threads[i].notify_send_fd = fds[1];setup_thread(&threads[i]);/* Reserve three fds for the libevent base, and two for the pipe */stats.reserved_fds += 5;}/* Create threads after we've done all the libevent setup. */for (i = 0; i < nthreads; i++) {create_worker(worker_libevent, &threads[i]);}/* Wait for all the threads to set themselves up before returning. */pthread_mutex_lock(&init_lock);wait_for_thread_registration(nthreads);pthread_mutex_unlock(&init_lock);
}

/*** Create a socket and bind it to a specific port number* @param interface the interface to bind to* @param port the port number to bind to* @param transport the transport protocol (TCP / UDP)* @param portnumber_file A filepointer to write the port numbers to*        when they are successfully added to the list of ports we*        listen on.*/
static int server_socket(const char *interface,int port,enum network_transport transport,FILE *portnumber_file) {int sfd;struct linger ling = {0, 0};struct addrinfo *ai;struct addrinfo *next;struct addrinfo hints = { .ai_flags = AI_PASSIVE,.ai_family = AF_UNSPEC };char port_buf[NI_MAXSERV];int error;int success = 0;int flags =1;hints.ai_socktype = IS_UDP(transport) ? SOCK_DGRAM : SOCK_STREAM;if (port == -1) {port = 0;}snprintf(port_buf, sizeof(port_buf), "%d", port);error= getaddrinfo(interface, port_buf, &hints, &ai);if (error != 0) {if (error != EAI_SYSTEM)fprintf(stderr, "getaddrinfo(): %s\n", gai_strerror(error));elseperror("getaddrinfo()");return 1;}for (next= ai; next; next= next->ai_next) {conn *listen_conn_add;if ((sfd = new_socket(next)) == -1) {/* getaddrinfo can return "junk" addresses,* we make sure at least one works before erroring.*/if (errno == EMFILE) {/* ...unless we're out of fds */perror("server_socket");exit(EX_OSERR);}continue;}#ifdef IPV6_V6ONLYif (next->ai_family == AF_INET6) {error = setsockopt(sfd, IPPROTO_IPV6, IPV6_V6ONLY, (char *) &flags, sizeof(flags));if (error != 0) {perror("setsockopt");close(sfd);continue;}}
#endifsetsockopt(sfd, SOL_SOCKET, SO_REUSEADDR, (void *)&flags, sizeof(flags));if (IS_UDP(transport)) {maximize_sndbuf(sfd);} else {error = setsockopt(sfd, SOL_SOCKET, SO_KEEPALIVE, (void *)&flags, sizeof(flags));if (error != 0)perror("setsockopt");error = setsockopt(sfd, SOL_SOCKET, SO_LINGER, (void *)&ling, sizeof(ling));if (error != 0)perror("setsockopt");error = setsockopt(sfd, IPPROTO_TCP, TCP_NODELAY, (void *)&flags, sizeof(flags));if (error != 0)perror("setsockopt");}if (bind(sfd, next->ai_addr, next->ai_addrlen) == -1) {if (errno != EADDRINUSE) {perror("bind()");close(sfd);freeaddrinfo(ai);return 1;}close(sfd);continue;} else {success++;if (!IS_UDP(transport) && listen(sfd, settings.backlog) == -1) {perror("listen()");close(sfd);freeaddrinfo(ai);return 1;}if (portnumber_file != NULL &&(next->ai_addr->sa_family == AF_INET ||next->ai_addr->sa_family == AF_INET6)) {union {struct sockaddr_in in;struct sockaddr_in6 in6;} my_sockaddr;socklen_t len = sizeof(my_sockaddr);if (getsockname(sfd, (struct sockaddr*)&my_sockaddr, &len)==0) {if (next->ai_addr->sa_family == AF_INET) {fprintf(portnumber_file, "%s INET: %u\n",IS_UDP(transport) ? "UDP" : "TCP",ntohs(my_sockaddr.in.sin_port));} else {fprintf(portnumber_file, "%s INET6: %u\n",IS_UDP(transport) ? "UDP" : "TCP",ntohs(my_sockaddr.in6.sin6_port));}}}}if (IS_UDP(transport)) {int c;for (c = 0; c < settings.num_threads_per_udp; c++) {/* Allocate one UDP file descriptor per worker thread;* this allows "stats conns" to separately list multiple* parallel UDP requests in progress.** The dispatch code round-robins new connection requests* among threads, so this is guaranteed to assign one* FD to each thread.*/int per_thread_fd = c ? dup(sfd) : sfd;dispatch_conn_new(per_thread_fd, conn_read,EV_READ | EV_PERSIST,UDP_READ_BUFFER_SIZE, transport);}} else {if (!(listen_conn_add = conn_new(sfd, conn_listening,EV_READ | EV_PERSIST, 1,transport, main_base))) {fprintf(stderr, "failed to create listening connection\n");exit(EXIT_FAILURE);}listen_conn_add->next = listen_conn;listen_conn = listen_conn_add;}}freeaddrinfo(ai);/* Return zero iff we detected no errors in starting up connections */return success == 0;
}