基于读写模式的udp多线程池编程

/************************************************************** * SocketBase.cpp : implementation file * Copyright 2004-2005 TianMuLingHang * All rights reserved. * Author: Zeng Wenchuan * Date: Jan 9, 2005 * Modify: * Date: * Site: http://zeng_aven.go.nease.net/ * paid services: zeng_aven@163.com *************************************************************/ #include "SocketBase.h" #include "ThreadPool.h" #include "string.h" #include <iostream> #include <algorithm> #include <string.h> #include <errno.h> #include <fcntl.h> CSocketBase::t_mapSocket CSocketBase::m_mapSock; CSocketBase::t_mapSocket CSocketBase::m_mapAccept; CSocketBase::t_listSocket CSocketBase::m_listAccept; CSocketBase::t_listSocket CSocketBase::m_listSock; CThreadPool *CSocketBase::m_pThreadPool = NULL; fd_set CSocketBase::m_rfd; CSocketBase::CSocketBase() : m_sock(-1), m_bBusy(false) { m_dsize = getdtablesize(); } CSocketBase::~CSocketBase() { Close(); } bool CSocketBase::Create(const int domain, const int type) { m_sock = socket(domain, type, 0); AddSockInList(); return m_sock != SOCKET_ERROR; } bool CSocketBase::Bind(const int port, struct sockaddr *addr) { if (m_sock < 0) return false; if (port < 1024) return false; int ret = bind(m_sock, addr, sizeof(struct sockaddr)); return ret != SOCKET_ERROR; } bool CSocketBase::Bind(const int port) { //套接口使用的通信协议 m_addr.sin_family = AF_INET; //如果port=0, 随机选择一个没有使用的端口 m_addr.sin_port = htons(port); //本机所有IP m_addr.sin_addr.s_addr = htonl(INADDR_ANY); bzero(&(m_addr.sin_zero), 8); return bind(m_sock, (struct sockaddr *)&m_addr, sizeof(struct sockaddr)) != SOCKET_ERROR; } bool CSocketBase::Bind() { return Bind(0); } bool CSocketBase::Listen(int backlog) { if (!Is_valid()) return false; if (backlog <= 0 ) backlog = 5; int ret = listen(m_sock, backlog); return ret != SOCKET_ERROR; } bool CSocketBase::AddSockInList() { if (m_sock <= 0) return false; m_listSock.push_back(m_sock); m_mapSock[m_sock] = this; FD_SET(m_sock,&m_rfd); return true; } bool CSocketBase::AddAcceptInList() { if (m_sock <= 0) return false; m_listAccept.push_back(m_sock); m_mapAccept[m_sock] = this; FD_SET(m_sock,&m_rfd); return true; } bool CSocketBase::Attach(int &sockfd) { ERROR_TRY if (sockfd <=0 ) return false; m_sock = sockfd; ERROR_UNKNOW_RETURN(CSOCKETBAUSE, "Attach", true) } bool CSocketBase::Gethostname(char *hostname, int size) { return gethostname(hostname, size) != SOCKET_ERROR; } bool CSocketBase::GetPeerName(char *fromaddr, UNIT &port) { ERROR_TRY if (m_sock <= 0) return false; struct sockaddr_in addr; socklen_t len = sizeof(struct sockaddr_in); if (getpeername(m_sock, (sockaddr *)&addr, &len) == SOCKET_ERROR) return false; fromaddr = inet_ntoa(addr.sin_addr); port = ntohs(addr.sin_port); ERROR_UNKNOW_RETURN(CSOCKETBAUSE, "GetPeerName", true) } void CSocketBase::SetConnected(sockaddr_in &addr) { m_addr = addr; } void CSocketBase::Set_non_blocking(const bool b) { ERROR_TRY int opts; opts = fcntl (m_sock, F_GETFL); if (opts < 0) return; if (b) opts = (opts | O_NONBLOCK); else opts = (opts & ~O_NONBLOCK); fcntl(m_sock, F_SETFL,opts); ERROR_UNKNOW(CSOCKETBAUSE, "Set_non_blocking") } bool CSocketBase::Close() { ERROR_TRY if (Is_valid()) { FD_CLR(m_sock, &m_rfd); m_listSock.remove(m_sock); m_listAccept.remove(m_sock); return close(m_sock) != SOCKET_ERROR; } ERROR_UNKNOW_RETURN(CSOCKETBAUSE, "Close", false) } bool CSocketBase::ShutDown(const int how) { if (Is_valid()) return shutdown(m_sock, how) != SOCKET_ERROR; return false; } bool CSocketBase::InitThreadPool(int nThreadNum) { ERROR_TRY if (m_pThreadPool) return false; m_pThreadPool = new CThreadPool(nThreadNum); if (m_pThreadPool) BeginThread(); else return false; ERROR_UNKNOW_RETURN(CSOCKETBAUSE, "InitThreadPool", true) } bool CSocketBase::operator == (const CSocketBase &sock) const { return m_sock == sock.GetSocketId(); } void CSocketBase::BeginThread() { m_pThreadPool->ThreadProc(this); } void CSocketBase::StopThread() { //暂时没有，任务只要一进入线程池就无法退出，只能等到任务结束 } void CSocketBase::SelectMsg() { fd_set Readrfd; FD_ZERO(&Readrfd); Readrfd = m_rfd; t_iListSocket ilist; t_imapSocket imap; CSocketBase *pSocket; struct timeval timeOut; timeOut.tv_sec = 1; timeOut.tv_usec = 0; int ret = 0; ret = select(m_dsize, &Readrfd, NULL, NULL, &timeOut); //启动另一个线程 BeginThread(); if (ret <= 0) return; ilist = find_if(m_listSock.begin(), m_listSock.end(), SocketIsSet(Readrfd)); //TCP服务端的ACCEPT if (ilist != m_listSock.end()) { imap = m_mapSock.find(*ilist); pSocket = imap->second; if (pSocket->IsBusy()) return; pSocket->SetBusy(true); pSocket->OnAccept(); pSocket->SetBusy(false); return; } ilist = find_if(m_listAccept.begin(), m_listAccept.end(), SocketIsSet(Readrfd)); //接收 if (ilist != m_listAccept.end()) { imap = m_mapAccept.find(*ilist); pSocket = imap->second; if (pSocket->IsBusy()) return; pSocket->SetBusy(true); pSocket->OnReceive(); pSocket->SetBusy(false); } }