作为进程间通信及网络通信的一种重要技术，在实际的开发中，socket编程是经常被用到的。关于socket编程的一般步骤，这里不再赘述，相关资料和文章很多，google/baidu即可。
本文主要是探讨如何更好地进行socket读写处理，以及如何检测连接断开。
首先，有以下几点需要注意：
- 对于全双工的socket，同时读写是没问题的。比如，一个socket程序有两个线程，一个线程对socket进行读操作(recv/read)，一个线程对socket进行写操作(send/write)，这里是不需要进行互斥操作或做临界区保护的。
- 在Unix系统下， 对一个对端已经关闭的socket调用两次write，第二次将会生成SIGPIPE信号，该信号的默认处理动作是终止进程。为了防止在这种情况下导致进程退出，我们需要屏蔽该信号的默认处理动作。有两种方法，a) 在程序开头调用signal(SIGPIPE, SIG_IGN)，忽略SIGPIPE信号的默认动作；b) 采用send函数的MSG_NOSIGNAL标志位，忽略SIGPIPE信号。当然，虽然我们忽略了SIGPIPE，但errno还是会被设置为EPIPE的。因此，我们可以根据这点，按照我们的情况来进行相应的处理了。
- 对文件描述符进行select/poll操作，a) 如果select/poll返回值大于0，此时进行recv/read，recv/read返回0，表明连接关闭; b) recv/read返回-1，并且errno为ECONNRESET、EBADF、EPIPE、ENOTSOCK之一时，也表明连接关闭。
- 对文件描述符进行send/write操作，如果send/write返回-1，并且errno为ECONNRESET、EBADF、EPIPE、ENOTSOCK之一时，也表明连接关闭。
下面是一个demo程序，server端代码(server.c)为：
#define _GNU_SOURCE #include <stdio.h> #include <sys/socket.h> #include <sys/un.h> #include <sys/types.h> #include <unistd.h> #include <string.h> #include <poll.h> #include <pthread.h> #include <errno.h> #define UNIX_PATH_MAX 108 #define SOCK_PATH "/tmp/test.sock" int sock_write(int fd, char* buf, int len) { int err_num, res; char err_str[128]; int disconnect = 0; while(!disconnect){ /* Use send with a MSG_NOSIGNAL to ignore a SIGPIPE signal which * would cause the process exit. */ res = send(fd, buf, len, MSG_NOSIGNAL | MSG_DONTWAIT); if( -1 == res ){ err_num = errno; printf("send error:%s! ", strerror_r(err_num, err_str, sizeof(err_str))); switch(err_num){ case ECONNRESET: case EBADF: case EPIPE: case ENOTSOCK: disconnect = 1; break; //case EWOULDBLOCK: case EAGAIN: usleep(10000); break; case EINTR: break; default: break; } }else if( res > 0 ){ if( res < len ){ /* Incomplete information. */ buf += res; len -= res; }else if( res == len ){ /* Message has sended successfully. */ break; } } } return disconnect; } void* write_handle(void* fd) { int len, connection_fd; char buffer[256]; char* end_flag = " "; connection_fd = *(int*)fd; len = snprintf(buffer, 256, "buffer data ends with a end flag%s", end_flag); buffer[len] = 0; while(0 == sock_write(connection_fd, buffer, len)){ sleep(1); } } int main(void) { struct sockaddr_un address; int socket_fd, connection_fd; socklen_t address_length; socket_fd = socket(PF_UNIX, SOCK_STREAM, 0); if(socket_fd < 0) { printf("%s:socket() failed ", SOCK_PATH); return -1; } unlink(SOCK_PATH); memset(&address, 0, sizeof(struct sockaddr_un)); address.sun_family = AF_UNIX; snprintf(address.sun_path, UNIX_PATH_MAX, SOCK_PATH); if(bind(socket_fd, (struct sockaddr *) &address, sizeof(struct sockaddr_un)) != 0) { printf("%s:bind() failed ", SOCK_PATH); return -1; } if(listen(socket_fd, 0) != 0) { printf("%s:listen() failed ", SOCK_PATH); return -1; } while((connection_fd = accept(socket_fd, (struct sockaddr *) &address,&address_length)) > -1) { pthread_t w_thread; struct pollfd pfd; char buffer[512]; int nbytes; int cnt; int res; int err_num; int disconnect=0; #ifdef DEBUG char str[128]; #endif printf(" %s:accept a connection! ", SOCK_PATH); pthread_create(&w_thread, NULL, write_handle, &connection_fd); pfd.fd = connection_fd; pfd.events = POLLIN | POLLHUP | POLLRDNORM; pfd.revents = 0; while(1){ res = poll(&pfd, 1, 500); if(res >= 0){ // if result > 0, this means that there is either data available on the // socket, or the socket has been closed cnt = recv(connection_fd, buffer, sizeof(buffer), MSG_PEEK | MSG_DONTWAIT); if( 0 == cnt ){ if(res > 0){ // if recv returns zero, that means the connection has been closed. #ifdef DEBUG printf("connection disconnect 1! "); #endif disconnect = 1; break; } }else if( -1 == cnt ){ err_num = errno; #ifdef DEBUG printf("recv error:%s! ", strerror_r(errno, str, sizeof(str))); #endif switch(err_num){ case ECONNRESET: case EBADF: case EPIPE: case ENOTSOCK: disconnect = 1; break; default: break; } if( disconnect ){ #ifdef DEBUG printf("connection disconnect 2! "); #endif break; } }else if( cnt > 0 ){ /* discard everything received from client.*/ while((nbytes = recv(connection_fd, buffer, sizeof(buffer)-1, MSG_DONTWAIT)) > 0){ buffer[nbytes] = 0; #ifdef DEBUG printf("buffer:%s ", buffer); #endif } #ifdef DEBUG if( 0 == nbytes ){ printf("All received! "); } #endif } } #ifdef DEBUG else if(res == -1){ /* This case shouldn't happen, we sleep 5 seconds here in case and retry it. */ printf("Error: poll return -1!!! "); sleep(5); } #endif } close(connection_fd); pthread_cancel(w_thread); } close(socket_fd); unlink(SOCK_PATH); return 0; }
client端代码(client.c)为:
#define _GNU_SOURCE #include <stdio.h> #include <sys/socket.h> #include <sys/un.h> #include <unistd.h> #include <string.h> #include <pthread.h> #include <poll.h> #include <errno.h> #define UNIX_PATH_MAX 108 #define SOCK_PATH "/tmp/test.sock" int main(void) { struct sockaddr_un address; int socket_fd, res; char buffer[256]; pthread_t w_thread; struct pollfd pfd; int err_num; int disconnect; while(1){ socket_fd = socket(PF_UNIX, SOCK_STREAM, 0); if(socket_fd < 0) { printf("%s:socket() failed ", SOCK_PATH); sleep(5); continue; } memset(&address, 0, sizeof(struct sockaddr_un)); address.sun_family = AF_UNIX; snprintf(address.sun_path, UNIX_PATH_MAX, SOCK_PATH); if(connect(socket_fd, (struct sockaddr *) &address, sizeof(struct sockaddr_un)) != 0) { printf("%s:connect() failed ", SOCK_PATH); close(socket_fd); socket_fd = -1; sleep(5); continue; } #ifdef DEBUG printf("connect success! "); #endif #if 1 pfd.fd = socket_fd; pfd.events = POLLIN | POLLHUP | POLLRDNORM; pfd.revents = 0; disconnect = 0; while(1){ // use the poll system call to be notified about socket status changes res = poll(&pfd, 1, 60000); if(res >= 0){ // if result > 0, this means that there is either data available on the // socket, or the socket has been closed char buffer[512]; int cnt, nbytes; cnt = recv(socket_fd, buffer, sizeof(buffer)-1, MSG_PEEK | MSG_DONTWAIT); if( -1 == cnt){ err_num = errno; switch(err_num){ case ECONNRESET: case EBADF: case EPIPE: case ENOTSOCK: disconnect = 1; break; default: break; } if(disconnect){ break; } } else if( 0 == cnt){ if(res > 0){ // if recv returns zero, that means the connection has been closed: disconnect = 1; break; } } else if( cnt > 0 ){ while((nbytes = recv(socket_fd, buffer, sizeof(buffer)-1, MSG_DONTWAIT)) > 0){ buffer[nbytes] = 0; printf("buffer:%s ", buffer); } } } } #endif close(socket_fd); socket_fd = -1; #ifdef DEBUG printf("server disconnect! "); #endif /* here sleep 5 seconds and re-connect. */ sleep(5); } return 0; }