6. 下面我们看看主线程dispatch_thread的事件处理设置。在memcached.c中的main函数中 /* create unix mode sockets after dropping privileges */ if ( settings . socketpath ! = null ) { errno = 0 ; if ( server_socket_unix ( settings . socketpath ,
6. 下面我们看看主线程dispatch_thread的事件处理设置。在memcached.c中的main函数中
/* create unix mode sockets after dropping privileges */
if (settings.socketpath != null) {
errno = 0;
if (server_socket_unix(settings.socketpath,settings.access)) {
vperror(failed to listen on unix socket: %s, settings.socketpath);
exit(ex_oserr);
}
}
/* create the listening socket, bind it, and init */
if (settings.socketpath == null) {
int udp_port;
const char *portnumber_filename = getenv(memcached_port_filename);
char temp_portnumber_filename[path_max];
file *portnumber_file = null;
if (portnumber_filename != null) {
snprintf(temp_portnumber_filename,
sizeof(temp_portnumber_filename),
%s.lck, portnumber_filename);
portnumber_file = fopen(temp_portnumber_filename, a);
if (portnumber_file == null) {
fprintf(stderr, failed to open \%s\: %s\n,
temp_portnumber_filename, strerror(errno));
}
}
errno = 0;
if (settings.port && server_socket(settings.port, tcp_transport,
portnumber_file)) {
vperror(failed to listen on tcp port %d, settings.port);
exit(ex_oserr);
}
/*
* initialization order: first create the listening sockets
* (may need root on low ports), then drop root if needed,
* then daemonise if needed, then init libevent (in some cases
* descriptors created by libevent wouldn't survive forking).
*/
udp_port = settings.udpport ? settings.udpport : settings.port;
/* create the udp listening socket and bind it */
errno = 0;
if (settings.udpport && server_socket(settings.udpport, udp_transport,
portnumber_file)) {
vperror(failed to listen on udp port %d, settings.udpport);
exit(ex_oserr);
}
if (portnumber_file) {
fclose(portnumber_file);
rename(temp_portnumber_filename, portnumber_filename);
}
}
/* drop privileges no longer needed */
drop_privileges();
/* enter the event loop */
event_base_loop(main_base, 0); //主线程(dispatcher_thread)的事件监听循环。。。
7. 继续跟踪server_socket函数（memcached.c中）
/**
* create a socket and bind it to a specific port number
* @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(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(settings.inter, port_buf, &hints, &ai);
if (error != 0) {
if (error != eai_system)
fprintf(stderr, getaddrinfo(): %s\n, gai_strerror(error));
else
perror(getaddrinfo());
return 1;
}
for (next= ai; next; next= next->ai_next) {
conn *listen_conn_add;
if ((sfd = new_socket(next)) == -1) { //creat socket
/* getaddrinfo can return junk addresses,
* we make sure at least one works before erroring.
*/
continue;
}
#ifdef ipv6_v6only
if (next->ai_family == af_inet6) {
error = setsockopt(sfd, ipproto_ipv6, ipv6_v6only, (char *) &flags, sizeof(flags));
if (error != 0) {
perror(setsockopt);
close(sfd);
continue;
}
}
#endif
//setsockopt
setsockopt(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);
}
//bind
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++;
//tcp listen
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)) {
//udp 的处理中不需要accept,所以直接派发connection到工作线程。
int c;
for (c = 0; c settings.num_threads; c++) {
/* this is guaranteed to hit all threads because we round-robin */
dispatch_conn_new(sfd, conn_read, ev_read | ev_persist,
udp_read_buffer_size, transport);
}
} else {
//tcp的处理(注意，这里dispatcher_thread同样调用了conn_new来绑定conn_event到其main_base. 并且此时conn的初始状态为conn_listening, 事件为持久可读， 而在conn_new中注册了conn_event的回调函数为event_handler,所以，dispatche_thread在当前listen的socket可读时就会调用event_handler,进而调用driver_machine(c) 进入状态机。而在driver_machine中如果是主线程(dispatcher_thread)则会在accept socket后调用dispatch_new_conn函数来给各worker_thread派发connection...)
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;
}
8. 看看udp和tcp模式下dispatcher_thread都会调用的dispatch_new_conn函数(在thread.c中)
/*
* dispatches a new connection to another thread. this is only ever called
* from the main thread, either during initialization (for udp) or because
* of an incoming connection.
*/
void dispatch_conn_new(int sfd, enum conn_states init_state, int event_flags,
int read_buffer_size, enum network_transport transport) {
cq_item *item = cqi_new();
int tid = (last_thread + 1) % settings.num_threads; //轮询的方式找worker_thread
libevent_thread *thread = threads + tid;
last_thread = tid;
item->sfd = sfd;
item->init_state = init_state;
item->event_flags = event_flags;
item->read_buffer_size = read_buffer_size;
item->transport = transport;
//push conn到worker_thread的cq中
cq_push(thread->new_conn_queue, item);
memcached_conn_dispatch(sfd, thread->thread_id);
if (write(thread->notify_send_fd, , 1) != 1) {
perror(writing to thread notify pipe);
}
}
9. 先看看unix domain socket模式下主线程的事件处理设置。（在上面的6中调用的 server_socket_unix函数）
static int server_socket_unix(const char *path, int access_mask) {
int sfd;
struct linger ling = {0, 0};
struct sockaddr_un addr;
struct stat tstat;
int flags =1;
int old_umask;
if (!path) {
return 1;
}
if ((sfd = new_socket_unix()) == -1) {
return 1;
}
/*
* clean up a previous socket file if we left it around
*/
if (lstat(path, &tstat) == 0) {
if (s_issock(tstat.st_mode))
unlink(path);
}
setsockopt(sfd, sol_socket, so_reuseaddr, (void *)&flags, sizeof(flags));
setsockopt(sfd, sol_socket, so_keepalive, (void *)&flags, sizeof(flags));
setsockopt(sfd, sol_socket, so_linger, (void *)&ling, sizeof(ling));
/*
* the memset call clears nonstandard fields in some impementations
* that otherwise mess things up.
*/
memset(&addr, 0, sizeof(addr));
addr.sun_family = af_unix;
strncpy(addr.sun_path, path, sizeof(addr.sun_path) - 1);
assert(strcmp(addr.sun_path, path) == 0);
old_umask = umask( ~(access_mask&0777));
if (bind(sfd, (struct sockaddr *)&addr, sizeof(addr)) == -1) {
perror(bind());
close(sfd);
umask(old_umask);
return 1;
}
umask(old_umask);
if (listen(sfd, settings.backlog) == -1) {
perror(listen());
close(sfd);
return 1;
}
if (!(listen_conn = conn_new(sfd, conn_listening, //同样是调用conn_new
ev_read | ev_persist, 1,
local_transport, main_base))) {
fprintf(stderr, failed to create listening connection\n);
exit(exit_failure);
}
return 0;
}