加载中…
个人资料
微博
  • 博客等级：
  • 博客积分：
  • 博客访问：
  • 关注人气：
  • 获赠金笔：0支
  • 赠出金笔：0支
  • 荣誉徽章：
正文 字体大小：大 中 小

LINUX网络编程（fork、select、epoll三种模式）

(2014-06-06 12:54:36)
标签：

股票

 分类： 开发相关

使用了fork、select、epoll三种socket服务器工作模式，客户端向服务端发送任何数据，服务端再原样返回给客户端，本文的目的只为加深偶的记忆。
fork:每accept到一个socket之后，开启一个子进程来负责收发处理工作。
select:监控文件描述符事件
epoll:监控文件描述符事件，比select性能优异，可最大支持2W个连接，有死连接时处理能力高
文末附注了SOCKET的一些常见错误标识

-------------------------- fork.cpp --------------------------
#include
#include
#include
#include
#include
#include
#include
#include
#include  
#include  
#include
#include
#include
#include

#define SERVER_PORT 8888 
#define BUFFER_LENGTH 1024

using namespace std;

class SocketException : public std::logic_error
{
public:
SocketException(const std::string& what) throw ()
: std::logic_error(what) {};
};

int SocketRead(int& fd, std::string& strValue)
{
// 每次只读取一个字节
int nRead = 1;

auto_ptr apBuf(new char[nRead]);
int r = read(fd, apBuf.get(), nRead) ;
if (0 < r)
{
   strValue.assign(apBuf.get(), r);
}

return r;
}

int SocketWrite(int& fd, const char* pBuf, int nBufLen)
{
int nWrite = 0;
int nError = -1;


// SIG_IGN:忽略信号
sighandler_t sh = signal(SIGPIPE, SIG_IGN);

while (nWrite < nBufLen)
{
   nError = send(fd, pBuf + nWrite, nBufLen - nWrite, 0); 
   if (0 > nError)
   {
    perror(strerror(errno));

    // 见文末
    if (nError == EINTR) continue;

    if (nError == EPIPE) 
    {
     close(fd); 
    }
    // 恢复信号
    signal(SIGPIPE, sh);
    throw SocketException(string("SocketWrite error, ") + strerror(errno));
   }
   else if (0 == nError)
   {
    signal(SIGPIPE, sh);
    throw SocketException(string("SocketWrite error, ") + strerror(errno));
   }
   else
   {
    nWrite += nError;
   }
}

signal(SIGPIPE, sh);
return nError;
}

void SetOption(int s, int level, int optname, const void *optval, socklen_t optlen)
{
if (setsockopt(s, level, optname, optval, optlen) < 0)
{
   perror(strerror(errno));
}
}

int main(int argc, char*argv[])
{
int fdServer = -1;
int   fdClient = -1;
int   nStatus = -1;
pid_t pid = 0;

int nSockAddrLen = sizeof(struct sockaddr_in);
struct sockaddr_in addrServer;
struct sockaddr_in addrSocket;
bzero(&addrServer, sizeof(struct sockaddr_in)); 
bzero(&addrSocket, sizeof(struct sockaddr_in));

if(-1 == (fdServer = socket(AF_INET, SOCK_STREAM, 0)))
{
   perror(strerror(errno));
   exit(-1);
}

int nReuseAddr = 1;
SetOption(fdServer, SOL_SOCKET, SO_REUSEADDR, &nReuseAddr, sizeof(int));

addrServer.sin_family = AF_INET;
addrServer.sin_port = htons(SERVER_PORT);
addrServer.sin_addr.s_addr = INADDR_ANY;

if (-1 == bind(fdServer, (struct sockaddr *)&addrServer, sizeof(struct sockaddr))) 
{ 
   perror(strerror(errno));
   exit(-1);
}

if (-1 == listen(fdServer, 128))
{
   perror(strerror(errno));
   exit(-1);
}

while (true)
{
   if ((fdClient = accept(fdServer, (struct sockaddr *)&addrSocket, (socklen_t*)&nSockAddrLen)) < 0)
   {
    perror(strerror(errno));
    exit(-1);
   }

   printf("client address:%s\t port:%d\r\n", inet_ntoa(addrSocket.sin_addr), ntohs(addrSocket.sin_port));
   if ((pid = fork()) < 0)
   {
    perror(strerror(errno));
    exit(-1);
   }
   else if (0 == pid)
   {
    while (true)
    {
     try
     {
      std::string strValue;
      if(0 < SocketRead(fdClient, strValue))
      {
       SocketWrite(fdClient, strValue.c_str(), strValue.length());
      }
      else
      {
       close(fdClient); 
      }
      // 要有延时，不然CPU使用率很高
      usleep(10);
     }
     catch(const SocketException& e)
     {
      cerr << e.what() << endl; 
      close(fdClient);
      break;
     }
    }
   }
   else
   {
    close(fdClient); 
   }
}

return 0;
}

-------------------------- select.cpp --------------------------
#include
#include
#include
#include
#include
#include
#include
#include

#include
#include
#include

#define SERVER_PORT 8888 
#define NUM_MON   5 
#define NUM_BUFFER 1024

#include  
using namespace std;

void SetOption(int s, int level, int optname, const void *optval, socklen_t optlen)
{
if (setsockopt(s, level, optname, optval, optlen) < 0)
{
   perror(strerror(errno));
}
}

int main(int argc, char*argv[])
{
int fdServer = -1;
int fdClient = -1;
int fdMax = -1;
int nRead = 0;
char buf[NUM_BUFFER];

int nSockAddrLen = sizeof(struct sockaddr_in);
struct sockaddr_in addrServer;
struct sockaddr_in addrSocket;
bzero(&addrServer, sizeof(struct sockaddr_in)); 
bzero(&addrSocket, sizeof(struct sockaddr_in));

int fdBuf[NUM_MON];
memset(fdBuf, -1, sizeof(fdBuf));

// 被监控的描述符集合
fd_set fsMon;
// 事件的描述符集合
fd_set fsRead;
FD_ZERO(&fsMon);
FD_ZERO(&fsRead);

if(-1 == (fdServer = socket(AF_INET, SOCK_STREAM, 0))) 
{
   perror(strerror(errno));
   exit(-1);
}

int nReuseAddr = 1;
SetOption(fdServer, SOL_SOCKET, SO_REUSEADDR, &nReuseAddr, sizeof(int));

addrServer.sin_family = AF_INET;
addrServer.sin_port = htons(SERVER_PORT);
addrServer.sin_addr.s_addr = INADDR_ANY;

if (-1 == bind(fdServer, (struct sockaddr *)&addrServer, sizeof(struct sockaddr))) 
{ 
   perror(strerror(errno));
   exit(-1);
}

if (-1 == listen(fdServer, 128))
{
   perror(strerror(errno));
   exit(-1);
}

// 服务端是被监控对象之一
FD_SET(fdServer, &fsMon);
fdMax = fdServer;

while (true)
{
   fsRead = fsMon;
   switch(select(fdMax + 1, &fsRead, NULL, NULL, NULL))
   {
    case -1:
     perror(strerror(errno));
     exit(-1);
    case 0:
     perror("timeout");
     continue;
    default:
     break;
   }

   // 是否为fdServer的可读事件
   if (FD_ISSET(fdServer, &fsRead)) 
   {
    if (0 > (fdClient = accept(fdServer, (struct sockaddr *)&addrSocket, (socklen_t*)&nSockAddrLen)))
    {
     perror(strerror(errno));
     exit(-1);
    }

    // socket加入fdBuf, fdBuf中是保存的所有socket的集合
    for (int i = 0; i < NUM_MON; ++i)
    {
     if(-1 == fdBuf[i]) 
     { 
      fdBuf[i] = fdClient;
      break;
     } 
    } 
    // 把socket加入监控集合
    FD_SET(fdClient, &fsMon) ;
    fdMax = max(fdMax, fdClient);
   
    continue; 
   }

   for (int i = 0; i < NUM_MON; ++i) 
   {
    if (-1 == fdBuf[i]) continue;
    if (!FD_ISSET(fdBuf[i], &fsMon)) continue;

    // 缓冲区中，是否有数据可读
    ioctl(fdBuf[i], FIONREAD, &nRead);
    if (0 >= nRead) 
    {
     // 非可读
     FD_CLR(fdBuf[i], &fsMon); 
     continue; 
    }
   
    read(fdBuf[i], buf, nRead) ;
    write(fdBuf[i], buf, nRead) ;
   }
} 
}

-------------------------- epoll.cpp --------------------------
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include

#include

using namespace std;

#define SERVER_PORT 8888

class SocketException : public std::logic_error
{
public:
SocketException(const std::string& what) throw ()
: std::logic_error(what) {};
};

int SocketRead(int& fd, std::string& strValue)
{
// 缓冲区中，是否有数据可读
int nRead = 0;
ioctl(fd, FIONREAD, &nRead);

if (nRead > 0) 
{
   auto_ptr apBuf(new char[nRead]);
   read(fd, apBuf.get(), nRead) ;
   strValue.assign(apBuf.get(), nRead);
}
else
{
   cout << nRead << endl; 
   close(fd);
}
return nRead;
}

int SocketWrite(int& fd, const char* pBuf, int nBufLen)
{
// 参见fork.cpp
int nWrite = 0;
int nError = -1;

sighandler_t sh = signal(SIGPIPE, SIG_IGN);

while (nWrite < nBufLen)
{
   nError = send(fd, pBuf + nWrite, nBufLen - nWrite, 0); 
   if (0 > nError)
   {
    perror(strerror(errno));

    if (nError == EINTR) continue;

    if (nError == EPIPE) 
    {
     close(fd); 
    }

    signal(SIGPIPE, sh);
    throw SocketException(string("SocketWrite error, ") + strerror(errno));
   }
   else if (0 == nError)
   {
    // out pipe is closed
    signal(SIGPIPE, sh);
    throw SocketException(string("SocketWrite error, ") + strerror(errno));
   }
   else
   {
    nWrite += nError;
   }
}

signal(SIGPIPE, sh);
return nError;
}

void SetOption(int s, int level, int optname, const void *optval, socklen_t optlen)
{
if (setsockopt(s, level, optname, optval, optlen) < 0)
{
   perror(strerror(errno));
}
}

// 设置socket为非阻塞式
void SetNonBlocking(int fd)
{
int flags = fcntl(fd, F_GETFL);
flags |= O_NONBLOCK;
fcntl(fd, F_SETFL, flags); 
}

int main(int argc, char*argv[])
{
int nfds, epfd, fdServer, fdClient;
int maxevents = 1000;
struct epoll_event ev, *events;

epfd = epoll_create(128);

int nSockAddrLen = sizeof(struct sockaddr_in);
struct sockaddr_in addrServer;
struct sockaddr_in addrSocket;
bzero(&addrServer, sizeof(struct sockaddr_in)); 
bzero(&addrSocket, sizeof(struct sockaddr_in));

if(-1 == (fdServer = socket(AF_INET, SOCK_STREAM, 0))) 
{
   perror(strerror(errno));
   exit(-1);
}

int nReuseAddr = 1;
SetOption(fdServer, SOL_SOCKET, SO_REUSEADDR, &nReuseAddr, sizeof(int));

addrServer.sin_family = AF_INET;
addrServer.sin_port = htons(SERVER_PORT);
addrServer.sin_addr.s_addr = INADDR_ANY;

if (-1 == bind(fdServer, (struct sockaddr *)&addrServer, sizeof(struct sockaddr))) 
{ 
   perror(strerror(errno));
   exit(-1);
}

// 监控对象fdServer
ev.data.fd = fdServer;
// 事件：边沿触发，IN事件
ev.events = EPOLLIN | EPOLLET;

if (epoll_ctl(epfd, EPOLL_CTL_ADD, fdServer, &ev) < 0)
{
   perror(strerror(errno));
   exit(-1);
}
  
if (-1 == listen(fdServer, 64))
{
   perror(strerror(errno));
   exit(-1);
}

while(true)
{
   nfds = epoll_wait(epfd, events, maxevents, -1);
   if (0 > nfds)
   {
    cerr << "epoll_wait:" << strerror(errno) << endl;
   }

   for(int n = 0; n < nfds; ++n) 
   {
    // 事件主体为服务端？
    if(events[n].data.fd == fdServer) 
    {
     fdClient = accept(fdServer, (struct sockaddr *)&addrSocket, (socklen_t*)&nSockAddrLen);
     if(0 > fdClient)
     {
      perror(strerror(errno)); 
      continue;
     } 
    
     // 非阻塞式，咱们目标是，快点处理完来下一轮
     SetNonBlocking(fdClient);

     // socket加入监控列表
     ev.events = EPOLLIN | EPOLLET; 
     ev.data.fd = fdClient; 
     if (epoll_ctl(epfd, EPOLL_CTL_ADD, fdClient, &ev) < 0) 
     { 
      cerr << "epoll set insertion error: fd=" << fdClient << endl;
      return -1; 
     } 
    } 
    else 
    {
     int nError = 0;
     char buf[1024];

     while(true)
     {
      nError = read(events[n].data.fd, buf, sizeof(buf)); 
      cout << " read bytes:" << nError << endl;
      if (-1 == nError)
      {
       perror(strerror(errno));
       // 这些见文末说明
       if (nError == EINTR) continue;

       if (nError == EPIPE) 
       {
        // 从监控列表中移除
        epoll_ctl(epfd, EPOLL_CTL_DEL, events[n].data.fd, &ev);
        close(events[n].data.fd); 
        break;
       }      
       break;
      }
      else if (0 == nError)
      {
       // 从监控列表中移除
       epoll_ctl(epfd, EPOLL_CTL_DEL, events[n].data.fd, &ev);
       close(events[n].data.fd); 
       break; 
      }

      SocketWrite(events[n].data.fd, buf, nError);
     }
    }
   }
}

return 0;
}
---------------------- 文末 ------------------


from http://hi.baidu.com/yuanhuiyong/item/448df958861b3019da1635e7

喜欢

0

阅读┊ 收藏 ┊ 喜欢▼ ┊打印┊举报/Report
  

新浪BLOG意见反馈留言板　欢迎批评指正

新浪简介 | About Sina | 广告服务 | 联系我们 | 招聘信息 | 网站律师 | SINA English | 产品答疑

Copyright © 1996 - 2022 SINA Corporation, All Rights Reserved

新浪公司 版权所有