共享内存的网络聊天室

引言

本篇参考书目《Linux高性能服务器编程》，这是我第三次实现网络聊天室，这一次是在内存的方面进行了极大的改进，并且本篇文章主要使用c++11来实现的，将书本里面的c语言重写了一遍。

代码已经上传到了github上面，fengyue05/Linux-: 本仓库里面会有关于《Linux高性能服务器编程》的一些主要代码实现

代码

在开始真正代码的实现前面，我们需要准备一些变量，首先是缓存的大小，用户的限制，事件的限制，进程数量的限制。我们这里主要是通过主进程来接受连接，广播消息，其实相当于我们的服务器端，而主进程创建出来的子进程作用是和客户端连接。所以我们客户端的数据结构体里面包含了其地址，对应的文件描述符，对应的进程，还有最重要的就是一个管道，我们的子进程只能和父进程通信，所以我们通过管道来通信，我们通过socketpair把管道两边封装socket，然后父进程使用pipefd[0]，子进程使用pipefd[1]

然后父进程作为服务器还需要处理信号的到来，这里我们也使用管道来通信处理。最后我们是通过共享内存的方式进行信息的交流，需要一个指针。

为了方便我们的代码，我们特意开辟了两个数组，一个数组是存用户的数据，一个数组是存进程，方便后面使用。

最后我们需要强调一个volatile，这个变量的值可能在程序“看不到的地方”被改变，所以每次使用它时，都应该重新从内存读取，不要过度优化。

sig_atomic_t 强调的是原子性，特别是不会被信号打断。

constexpr int USER_LIMIT = 5;
constexpr int BUFFER_SIZE = 1024;
constexpr int FD_LIMIT = 65535;
constexpr int MAX_EVENT_NUMBER = 1024;
constexpr int PROCESS_LIMIT = 65535;
constexpr const char* SHM_NAME = "/my_shm";

struct ClientData {
sockaddr_in address;
int connfd;
pid_t pid; // 处理这个连接的子进程的PID
std::array<int, 2> pipefd; // 和父进程通信的管道
};

std::array<int, 2> m_signalPipe{{-1, -1}};
int m_epollfd = -1;
int m_listened = -1;
char* m_sharedMemory = nullptr;

// 客户连接数组。进程用客户连接的编号来索引这个数组，即可取得相关的客户连接数据
std::array<ClientData, USER_LIMIT + 1> m_users;
std::array<int, PROCESS_LIMIT> m_subProcess;
int m_userCount = 0;

// 当程序正在正常执行时，即使突然被信号打断，读取或写入这个变量也不会出现“写到一半”的情况。
volatile sig_atomic_t m_stopChild = 0;

接下来是一个比较常规的操作：（如果有疑问，可以看一下我之前的博客）

int setNoBlocking(int fd) {
const int old_option = ::fcntl(fd, F_GETFL);
const int new_option = old_option | O_NONBLOCK;
fcntl(fd, F_SETFL, new_option);
return old_option;
}

void addFd (int epollfd, int fd) {
epoll_event event;
event.data.fd = fd;
event.events = EPOLLIN | EPOLLET;
assert(::epoll_ctl(epollfd, EPOLL_CTL_ADD, fd, &event) != -1);
setNoBlocking(fd);
}

void removeFd(int epollfd, int fd) {
if (fd < 0) {
return;
}
::epoll_ctl(epollfd, EPOLL_CTL_DEL, fd, nullptr);
::close(fd);
}

void signalHandler(int sig) {
const int saveErrno = errno;
const char message = static_cast<char>(sig);
::send(m_signalPipe[1], &message, 1, 0);
errno = saveErrno;
}

void addSignal (int sig, void (*handler)(int), bool restart = true) {
struct sigaction sa;
sa.sa_handler = handler;
if (restart) {
sa.sa_flags = SA_RESTART;
}
::sigfillset(&sa.sa_mask);
assert (::sigaction(sig, &sa, nullptr) != -1);
}

// 停止一个子进程
void childTermHandler(int) {
m_stopChild = 1;
}

void closeFd(int& fd) {
if (fd >= 0) {
::close(fd);
fd = -1;
}
}

这个函数是主进程里面最后销毁所有内存的函数，首先我们需要关闭所有存储的文件描述符，其实主要就是关闭[0]号管道，因为主进程只靠[0]号管道来和对应的进程通信，而[1]号管道是在进程创建的时候就在主进程里面关闭了。

还有一个重要的就是我们释放共享内存的函数munmap，这个和mmap是一对函数，一个是创建共享内存，一个是释放内存。

最后因为我们是通过shm_open打开POSIX共享内存的对象，所以我们需要关闭unlink

void cleanResouces() {
for (int i = 0; i < m_userCount; i++) {
// 是在父进程中执行的，而父进程保存的是每个 socketpair 的 pipefd[0]（与对应子进程通信）
closeFd(m_users[i].pipefd[0]);
}

closeFd(m_signalPipe[0]);
closeFd(m_signalPipe[1]);
closeFd(m_listened);
closeFd(m_epollfd);

if (m_sharedMemory != nullptr && m_sharedMemory != MAP_FAILED) {
::munmap(m_sharedMemory, USER_LIMIT * BUFFER_SIZE);
m_sharedMemory = nullptr;
}
::shm_unlink(SHM_NAME);
}

我们子进程的作用是和客户端建立联系并且通信，所以我们有一个子进程专门运行的函数。我们传入的参数是客户端对应的index，从而通过我们存储的数据索引index，找到对应的socket等。每一个进程都有一个epollfd，因为这个是专门和用户进行监听的，同时加入了一个退出的信号，以便后面终端的退出。

对于一个子进程来说，继承了父进程的资源，所以也会有users的数据

在while（）循环里面，当监听到了事件之后，循环处理每一个事件，我们对每一个事件有分类处理：

当事件的sockfd和存储的connfd相同的时候，就说明是该进程对应的客户来消息了，这个时候就要recv消息，不过我们是一个共享内存，所以我们不是开辟内存，而是直接利用共享内存，每个进程对应的共享内存的地方是确定的，地址是m_sharedMemory + index * BUFFER_SIZE，然后把接受到的数据放在这里面。在接受完信息了之后，发送消息给主进程（服务器端），send的是client的编号，这样子主进程（服务器端）就可以知道是哪一个进程（也就是对应的客户端）。可能有人会疑问，那为什么不发送接收到客户端的消息呢？原因就是这个是共享内存，我们只需要知道发消息对应进程所属的地址，这个样子直接可以取数据了，所以我们传递的都是index。

当事件的sockfd和父进程通信的管道一样的时候，说明是接收到了父进程的消息，也就是广播，这个时候我们接受到了client的index，然后就可以找到对应的共享内存位置，然后发送给客户端。

// 子进程运行的函数，参数index指出该子进程处理的客户连接的编号
void runChild(int index) {
std::array<epoll_event, MAX_EVENT_NUMBER> events;
const int childEpollfd = ::epoll_create(1);
assert(childEpollfd != -1);

const int connfd = m_users[index].connfd;
const int pipefd = m_users[index].pipefd[1]; // 子进程的socket

addSignal(SIGTERM, childTermHandler, false);

while(!m_stopChild) {
const int number = ::epoll_wait(childEpollfd, events.data(), MAX_EVENT_NUMBER, -1);

if (number < 0 && errno != EINTR) {
std::cerr << "child epoll_wait failed:" << std::strerror(errno) << std::endl;
break;
}

for (int i = 0; i < number; i++) {
const int sockfd = events[i].data.fd;
// 本子进程负责的客户连接有数据到达，写到本客户端对应的共享内存操作
// 再把客户编号通知给父进程
if (sockfd == connfd && (events[i].events & EPOLLIN)) {
char* buffer = m_sharedMemory + index * BUFFER_SIZE;
std::memset(buffer, 0, BUFFER_SIZE);

int received = ::recv(connfd, buffer, BUFFER_SIZE – 1, 0);
if (received < 0) {
if (errno != EAGAIN && errno != EWOULDBLOCK) {
m_stopChild = 1;
}
}
else if(received == 0) {
m_stopChild = 1;
}
else {
// 传递第几个客户
::send(pipefd, &index, sizeof(index), 0);
}
}
// 父进程通知本进程（通过管道）将第client个客户的数据发送到本进程负责的客户端
else if (sockfd == pipefd && (events[i].events & EPOLLIN)) {
int client = -1;
int received = ::recv(pipefd, &client, sizeof(client), 0);

if (received < 0) {
if (errno != EAGAIN && errno != EWOULDBLOCK) {
m_stopChild = 1;
}
}
else if(received == 0) {
m_stopChild = 1;
}
else if (received == static_cast<ssize_t>(sizeof(client)) && client >= 0 && client < USER_LIMIT) {
// 直接使用共享内存，发送给对应的客户端
::send(connfd, m_sharedMemory + client * BUFFER_SIZE, BUFFER_SIZE, 0);
}
}
}
}

::close(connfd);
::close(pipefd);
::close(childEpollfd);
}

当客户端关闭了之后，我们删除客户端的一切数据。这个是在父进程里面调用的，所以要移除与子进程的通信socket。并且我们的数组排列一定要是连续的，否则多次关闭连接之后会浪费很多的内存。所以当我们删除了中间一个index，那么我们要把最后一个填补到我们删除的地方，包括pid

void eraseUser (int index) {
if (index < 0 || index >= m_userCount) {
return;
}

// 移除父进程为每个子进程保留的通信管道
removeFd(m_epollfd, m_users[index].pipefd[0]);

const int last = m_userCount – 1;
if (index != last) {
m_users[index] = m_users[last];
const pid_t movedPid = m_users[index].pid;
if (movedPid >= 0 && movedPid < PROCESS_LIMIT) {
m_subProcess[static_cast<int>(movedPid)] = index;
}
}
m_userCount–;
}

对于共享内存，我们的一系列操作就是：

先shm_open打开共享内存的对象，其返回的文件描述符很关键，我们可以通过这个文件描述符重构内存的大小，通过ftruncate（）函数，最后通过mmap（）创建我们的共享内存并返回这块内存的指针。一切结束了之后就可以关闭这个打开共享内存的文件描述符

std::array<epoll_event, MAX_EVENT_NUMBER> events;
m_epollfd = epoll_create(1);
assert(m_epollfd == -1);
addFd(m_epollfd, m_listened);

assert (::socketpair(AF_UNIX, SOCK_STREAM, 0, m_signalPipe.data()));
setNoBlocking(m_signalPipe[1]);
addFd(m_epollfd, m_signalPipe[0]);

addSignal(SIGCHLD, signalHandler);
addSignal(SIGTERM, signalHandler);
addSignal(SIGINT, signalHandler);
addSignal(SIGPIPE, SIG_IGN);

bool stopServer = false;
bool terminating = false;

const int shmfd = ::shm_open(SHM_NAME, O_CREAT | O_RDWR, 0666);
assert(shmfd != -1);
// 通过文件描述符修改文件或共享内存对象的大小
assert(::ftruncate(shmfd, USER_LIMIT * BUFFER_SIZE) != -1);

m_sharedMemory = static_cast<char*>(::mmap(nullptr, USER_LIMIT * BUFFER_SIZE, PROT_READ | PROT_WRITE, MAP_SHARED, shmfd, 0));
::close(shmfd);

在while（）里面，我们解决的就是几个问题：

1、如果有连接发来，存储对应的数据，创建对应的管道，然后创建对应的子进程，在子进程里面只保留和父进程通信的socket和客户端通信的socket，其他全部关闭，如果结束了子进程，就要munmap其共享内存的部分。而父进程要做的事情就是存储子进程的信息，不过我们进程数量都是有限的，所以当我们创建了子进程之后要判断如果超过了这个上限，那么就要关闭。

2、当有信号到来的时候，如果是子进程退出的信号，父进程非阻塞回收，因为有信号来了一定是可以回收的，而非阻塞回收的特点就是高效。如果是父进程被关闭的信号来了，那么要关闭所有的子进程。

3、通过广播发送信息，其实就是传递对应的index，让子进程里面运行的函数接受index，自己读取共享内存里面的数据。

while (!stopServer) {
const int number = epoll_wait(m_epollfd, events.data(), MAX_EVENT_NUMBER, -1);
if (number < 0 && errno != EINTR) {
std::cerr << "epoll_wait failed: "
<< std::strerror(errno) << '\\n';
break;
}

for (int i = 0; i < number; i++) {
const int sockfd = events[i].data.fd;
if (sockfd == m_listened && (events[i].events & EPOLLIN)) {
sockaddr_in clientAddress;
socklen_t len = sizeof(clientAddress);
const int connfd = ::accept(m_listened, (sockaddr*)&clientAddress, &len);
if (connfd < 0) {
if (errno != EAGAIN && errno != EWOULDBLOCK) {
std::cerr << "accept failed:" << std::strerror(errno) << std::endl;
}
continue;
}

if (m_userCount >= USER_LIMIT) {
const std::string message = "too many users\\n";
std::cerr << message;
::send(connfd, message.data(), message.size(), 0);
::close(connfd);
continue;
}

const int index = m_userCount;
m_users[index].address = clientAddress;
m_users[index].connfd = connfd;
assert(::socketpair(AF_UNIX, SOCK_STREAM, 0, m_users[index].pipefd.data()) != -1);

const pid_t pid = ::fork();
// fork失败啦
if (pid < 0) {
::close(connfd);
::close(m_users[index].pipefd[0]);
::close(m_users[index].pipefd[1]);
continue;
}

if (pid == 0) {
::close(m_epollfd);
::close(m_listened);
::close(m_users[index].pipefd[0]);
::close(m_signalPipe[0]);
::close(m_signalPipe[1]);

runChild(index);
::munmap(m_sharedMemory, USER_LIMIT * BUFFER_SIZE);
_exit(EXIT_SUCCESS);
}

::close(connfd);
::close(m_users[index].pipefd[1]);
m_users[index].pipefd[1] = -1;
addFd (m_epollfd, m_users[index].pipefd[0]);

m_users[index].pid = pid;
if (pid >= PROCESS_LIMIT) {
std::cerr << "child PID exceeds PROCESS_LIMIT\\n";
::kill(pid, SIGTERM);
removeFd(m_epollfd, m_users[index].pipefd[0]);
continue;
}
m_subProcess[static_cast<int>(pid)] = index;
m_userCount++;
}
// 统一信号源，从信号管道读取信号编号
else if (sockfd == m_signalPipe[0] && (events[i].events & EPOLLIN)) {
std::array<char, 1024> signals;
int received = ::recv(m_signalPipe[0], signals.data(), signals.size(), 0);
if (received <= 0) {
continue;
}

for (int j = 0; j < received; j++) {
switch (signals[j])
{
case SIGCHLD:
while(true) {
int status = 0;
const pid_t pid = ::waitpid(-1, &status, WNOHANG);
if (pid <= 0) {
break;
}
if (pid < PROCESS_LIMIT) {
const int index = m_subProcess[pid];
m_subProcess[pid] = -1;
eraseUser(index);
}
}
if (terminating && m_userCount == 0) {
stopServer = true;
}
break;
case SIGTERM:
case SIGINT:
if (m_userCount == 0) {
stopServer = true;
}
else {
for (int k = 0; k < m_userCount; k++) {
::kill(m_users[k].pid, SIGTERM);
}
terminating = true;
}
break;
default:
break;
}
}
}
// 某个子进程写入了共享内存；通过广播广播给其他子进程
else if (events[i].events & EPOLLIN) {
int child = -1;
const int received = ::recv(sockfd, &child, sizeof(child), 0);
if (received <= 0) {
continue;
}

for (int j = 0; j < m_userCount; j++) {
if (m_users[j].pipefd[0] == sockfd) {
continue;
}
::send(m_users[j].pipefd[0], &child, sizeof(child), 0);
}
}
}
}
cleanResouces();

所以共享内存的聊天室最大的优点就是不需要send信息，减少了拷贝的开销，而是仅仅读取共享内存的数据，这个对于性能有很大的提高。

总结

本篇文章到这里就结束了！！！希望可以帮助大家理解~~~