xuxiuxi

2017-05-11 109ffe9a777658936a38d0c146579a67c60a0d17

 FaceServer/ev_server.cpp

@@ -1,6 +1,6 @@
#include "ev_server.h" 

#include "ev_proto.h"

#include "logger.h"

#include <logger.h>



#include <sys/types.h>

#include <sys/socket.h>

@@ -8,6 +8,8 @@
#include <arpa/inet.h>



#include <sys/time.h>



#include <netinet/tcp.h> // for TCP_NODELAY



#include <stdlib.h>

#include <stdio.h>

@@ -32,12 +34,14 @@
   size_t recvbuff_end;

   size_t recvbuff_max;

   size_t read_times; // read times for a command

   bool toClose;

   

   evclient_proc_t proc;

   

   EVClient() : 

      fd(-1), buf_ev(nullptr), 

      recvbuff(nullptr), recvbuff_end(0), recvbuff_max(0), read_times(0), 

      toClose(false), 
      proc(nullptr)

   { }

};

@@ -47,6 +51,12 @@
#endif



// Set a socket to non-blocking mode.

static int setnodelay(int fd)

{

   int flag = 1; 
   return setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, (char *) &flag, sizeof(int));

}



static int setnonblock(int fd)

{

   int flags;

@@ -61,6 +71,15 @@
   return 0;

}



static int setlinger(int fd)

{

   struct linger so_linger;

   so_linger.l_onoff = 1;

   so_linger.l_linger = 0;

	
   return setsockopt(fd, SOL_SOCKET, SO_LINGER, &so_linger, sizeof(so_linger));

}



// Called by libevent when there is an error on the underlying socket descriptor.

static void buffered_on_error(struct bufferevent *bufev, short what, void *arg)

{

@@ -69,11 +88,11 @@
   if (what & EVBUFFER_EOF)

   {

      //Client disconnected, remove the read event and the free the client structure.

      LOG_INFO << "Client disconnected." << std::endl;

      LOG_INFO << "Client disconnected." << LOG_ENDL;

   }

   else

   {

      LOG_WARN << "Client socket error, disconnecting." << std::endl;

      LOG_WARN << "Client socket error, disconnecting." << LOG_ENDL;

   }

   bufferevent_free(client->buf_ev);

   close(client->fd);

@@ -89,57 +108,58 @@
   // Write back the read buffer. It is important to note that

   // bufferevent_write_buffer will drain the incoming data so it

   // is effectively gone after we call it.

   //LOG_DEBUG << (char*)bufev->input << std::endl;

   //LOG_DEBUG << (char*)bufev->input << LOG_ENDL;

   //bufferevent_write_buffer(bufev, bufev->input);

   

   //char buff[100] = {'\0'};

   //size_t readSize = bufferevent_read(bufev, buff, sizeof(buff));

   //LOG_DEBUG << "readSize=" << readSize << "\t" << buff << std::endl;

   //LOG_DEBUG << "readSize=" << readSize << "\t" << buff << LOG_ENDL;



   EVPHeader* evpHeader = (EVPHeader*)client->recvbuff;

   

   // read header if expected

   if (client->recvbuff_end == 0)

   {

      uint8_t headerBuff[sizeof(EVPHeader)];

      if (client->recvbuff_end == 0)



      size_t readSize = bufferevent_read(bufev, headerBuff, sizeof(headerBuff));

      client->read_times = 1;

      if (readSize != sizeof(headerBuff))

      {

         size_t readSize = bufferevent_read(bufev, headerBuff, sizeof(headerBuff));

         client->read_times = 1;

         if (readSize != sizeof(headerBuff))

         {

            LOG_WARN << "client send incomplete header" << std::endl;

            buffered_on_error(bufev, 0, arg);

            return;

         }

			
         evpHeader = (EVPHeader*)headerBuff;

			
         // check header

         if (evpHeader->cmd <= EVPCommand::EVPC__FIRST || evpHeader->cmd >= EVPCommand::EVPC__LAST || 
            evpHeader->size < sizeof(EVPHeader) || evpHeader->size > CLIENT_BUFFER_MAX)

         {

            LOG_WARN << "client send invalid header" << std::endl;

            buffered_on_error(bufev, 0, arg);

            return;

         }

			
         if (client->recvbuff_max < evpHeader->size)

         {

            delete[] client->recvbuff;

            client->recvbuff = nullptr;

            client->recvbuff_max = 0;

         }

			
         if (client->recvbuff == nullptr)

         {

            uint32_t _CLIENT_BUFFER_MAX = CLIENT_BUFFER_MAX;

            client->recvbuff_max = std::min(evpHeader->size, _CLIENT_BUFFER_MAX);

            client->recvbuff = new uint8_t[client->recvbuff_max];

         }

			
         memcpy(client->recvbuff, headerBuff, sizeof(headerBuff));

         client->recvbuff_end = sizeof(headerBuff);

         LOG_WARN << "client send incomplete header" << LOG_ENDL;

         buffered_on_error(bufev, 0, arg);

         return;

      }

		
      evpHeader = (EVPHeader*)headerBuff;

      evpHeader->ntoh();

		
      // check header

      if (evpHeader->proto <= EVPProto::EVPP__FIRST || evpHeader->proto >= EVPProto::EVPP__LAST || 
         evpHeader->cmd <= EVPCommand::EVPC__FIRST || evpHeader->cmd >= EVPCommand::EVPC__LAST || 
         evpHeader->size < sizeof(EVPHeader) || evpHeader->size > CLIENT_BUFFER_MAX)

      {

         LOG_WARN << "client send invalid header" << LOG_ENDL;

         buffered_on_error(bufev, 0, arg);

         return;

      }

		
      if (client->recvbuff_max < evpHeader->size)

      {

         delete[] client->recvbuff;

         client->recvbuff = nullptr;

         client->recvbuff_max = 0;

      }

		
      if (client->recvbuff == nullptr)

      {

         uint32_t _CLIENT_BUFFER_MAX = CLIENT_BUFFER_MAX;

         client->recvbuff_max = std::min(evpHeader->size, _CLIENT_BUFFER_MAX);

         client->recvbuff = new uint8_t[client->recvbuff_max];

      }

		
      memcpy(client->recvbuff, headerBuff, sizeof(headerBuff));

      client->recvbuff_end = sizeof(headerBuff);

   }

   

   // read sub command or data

@@ -147,9 +167,10 @@
   do

   {

      readSize = bufferevent_read(bufev, client->recvbuff + client->recvbuff_end, client->recvbuff_max - client->recvbuff_end);

      client->read_times++;

      if (readSize == 0)

         break;

      else

         client->read_times++;



      client->recvbuff_end += readSize;

   } while (readSize > 0);

@@ -158,39 +179,43 @@
   if (evpHeader->size == client->recvbuff_end)

   {

      // call client proc

      bool closeClient = true;

      if (client->proc != nullptr)

      {

         EVClientStub cs(client->recvbuff, client->recvbuff_end);

         cs.id = client->fd;

         closeClient = !(client->proc(cs));

         client->toClose = !(client->proc(cs));

         

         if (cs.sendBuff != nullptr && cs.sendBuffSize > 0)

         {

            size_t writeSize = bufferevent_write(bufev, cs.sendBuff, cs.sendBuffSize);

            if (writeSize != cs.sendBuffSize)

               LOG_WARN << "server send truncate " << (cs.sendBuffSize - writeSize) << " bytes" << std::endl;

				
            bufferevent_enable(client->buf_ev, EV_WRITE);

            if (writeSize != 0)

               LOG_WARN << "server send truncate" << LOG_ENDL;



            if (cs.deleteSendBuff)

            {

               delete[] cs.sendBuff;

               cs.sendBuff = nullptr;

            }

         }

      }

		
      if (closeClient)

      {

         LOG_DEBUG << "server initiative close" << std::endl;

         buffered_on_error(bufev, 0, arg);

      }



      // reset client

      client->recvbuff_end = 0;

      client->read_times = 0;

   }

   else

   {

      //LOG_WARN << "recvbuff incomplete, evpHeader.size=" << evpHeader->size 
      //   << ", recvbuff_end=" << client->recvbuff_end 
      //   << ", read_times=" << client->read_times 
      //   << LOG_ENDL;

   }

   

   // check read times

   if (client->read_times > CLIENT_READ_TIMES_MAX)

   {

      LOG_WARN << "client read times to max" << std::endl;

      LOG_WARN << "client read times to max" << LOG_ENDL;

      buffered_on_error(bufev, 0, arg);

   }

}

@@ -199,6 +224,23 @@
// We only provide this because libevent expects it, but we don't use it.

static void buffered_on_write(struct bufferevent *bufev, void *arg)

{

   struct EVClient *client = (struct EVClient *)arg;

   if (evbuffer_get_length(bufev->output) == 0)

   {

      bufferevent_disable(client->buf_ev, EV_WRITE);

      if (client->toClose)

      {

         LOG_DEBUG << "server initiative close" << LOG_ENDL;

         buffered_on_error(bufev, 0, arg);

      }

      else

      {

         //bufferevent_flush(bufev, EV_WRITE, BEV_FLUSH);//#todo not work

         //bufferevent_flush(bufev, EV_WRITE, BEV_FINISHED);

         // flush socket

         shutdown(client->fd, SHUT_WR);

      }

   }

}



// This function will be called by libevent when there is a connection ready to be accepted.

@@ -209,19 +251,25 @@
   int client_fd = accept(fd, (struct sockaddr *)&client_addr, &client_len);

   if (client_fd < 0)

   {

      LOG_WARN << "accept failed" << std::endl;

      LOG_WARN << "accept failed" << LOG_ENDL;

      return;

   }



   // Set the client socket to non-blocking mode.

   if (setnonblock(client_fd) < 0)

      LOG_WARN << "failed to set client socket non-blocking" << std::endl;

      LOG_WARN << "failed to set client socket non-blocking" << LOG_ENDL;

	
   if (setnodelay(client_fd) < 0)

      LOG_WARN << "failed to set client socket no-delay" << LOG_ENDL;

	
   //if (setlinger(client_fd) < 0)

   //   LOG_WARN << "failed to set client socket linger" << LOG_ENDL;



   // We've accepted a new client, create a client object.

   struct EVClient* client = new EVClient;

   if (client == NULL)

   {

      LOG_ERROR << "malloc failed" << std::endl;

      LOG_ERROR << "malloc failed" << LOG_ENDL;

   }

   client->fd = client_fd;

   client->proc = evclient_proc;

@@ -232,12 +280,12 @@
   // We have to enable it before our callbacks will be called.

   bufferevent_enable(client->buf_ev, EV_READ);



   LOG_INFO << "Accepted connection from " << inet_ntoa(client_addr.sin_addr) << std::endl;

   LOG_INFO << "Accepted connection from " << inet_ntoa(client_addr.sin_addr) << LOG_ENDL;

}



int server_main(int argc, char **argv)

{

   //initLogger(LV_DEBUG);

   LOG_NOTICE << "server_main" << LOG_ENDL;



   // Initialize libevent.

   event_init();

@@ -246,7 +294,7 @@
   int listen_fd = socket(AF_INET, SOCK_STREAM, 0);

   if (listen_fd < 0)

   {

      LOG_ERROR << "create socket failed" << std::endl;

      LOG_ERROR << "create socket failed" << LOG_ENDL;

      return EXIT_FAILURE;

   }

   

@@ -255,26 +303,28 @@
   listen_addr.sin_family = AF_INET;

   listen_addr.sin_addr.s_addr = INADDR_ANY;

   listen_addr.sin_port = htons(SERVER_PORT);

	
   int reuseaddr_on = REUSEADDR_ON;

   setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, &reuseaddr_on, sizeof(reuseaddr_on));

   if (setnonblock(listen_fd) < 0)

   {

      LOG_ERROR << "failed to set server socket to non-blocking" << LOG_ENDL;

      return EXIT_FAILURE;

   }



   if (bind(listen_fd, (struct sockaddr *)&listen_addr, sizeof(listen_addr)) < 0)

   {

      LOG_ERROR << "bind failed" << std::endl;

      LOG_ERROR << "bind failed" << LOG_ENDL;

      return EXIT_FAILURE;

   }

   

   if (listen(listen_fd, 5) < 0)

   {

      LOG_ERROR << "listen failed" << std::endl;

      LOG_ERROR << "listen failed" << LOG_ENDL;

      return EXIT_FAILURE;

   }

   

   // Set the socket to non-blocking, this is essential in event based programming with libevent.

   int reuseaddr_on = REUSEADDR_ON;

   setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, &reuseaddr_on, sizeof(reuseaddr_on));

   if (setnonblock(listen_fd) < 0)

   {

      LOG_ERROR << "failed to set server socket to non-blocking" << std::endl;

      return EXIT_FAILURE;

   }



   // We now have a listening socket, we create a read event to be notified when a client connects.

   struct event ev_accept;

@@ -284,7 +334,15 @@
   // Start the event loop.

   event_dispatch();



   close(listen_fd);

   listen_fd = 0;

   return EXIT_SUCCESS;

}



void server_stop()

{

   LOG_NOTICE << "server_stop" << LOG_ENDL;

   event_loopexit(NULL);

}



void ev_send_status_packet(EVClientStub& client, EVPStatus::EVPS status)

@@ -296,7 +354,9 @@
   EVPHeader* evpHeader = new (client.sendBuff) EVPHeader;

   evpHeader->cmd = EVPCommand::EVPC_STATUS;

   evpHeader->size = client.sendBuffSize;

   evpHeader->hton();

   

   EVP_Status* evpStatus = new (client.sendBuff + sizeof(EVPHeader)) EVP_Status;

   evpStatus->status = status;

   evpHeader->hton();

}