BLOCKING

C

muduo/blob/master/examples/ace/ttcp/ttcp_blocking.cc
muduo/blob/master/examples/ace/ttcp/common.h

  • 服务端 ```c

    pragma once

include

include

struct Options { uint16_t port; int length; int number; bool transmit, receive, nodelay; std::string host; Options() : port(0), length(0), number(0), transmit(false), receive(false), nodelay(false) { } };

bool parseCommandLine(int argc, char argv[], Options opt); struct sockaddr_in resolveOrDie(const char* host, uint16_t port);

struct SessionMessage { int32t number; int32t length; } attribute ((__packed)); // 取消字节对齐, 紧凑排列

struct PayloadMessage { int32_t length; char data[0]; // C语言中: 在结构体的最后一个元素的数组大小是运行时决定的 };

void transmit(const Options& opt);

void receive(const Options& opt);

  1. ```c
  2. #include "examples/ace/ttcp/common.h"
  3. #include "muduo/base/Timestamp.h"
  4. #include "muduo/base/Types.h"
  6. #undef NDEBUG
  8. #include <assert.h>
  9. #include <errno.h>
  10. #include <stdio.h>
  11. #include <unistd.h>
  13. #include <netinet/in.h>
  14. #include <arpa/inet.h>
  16. static int acceptOrDie(uint16_t port)
  17. {
  18.     int listenfd = ::socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
  19.     assert(listenfd >= 0);
  21.     int yes = 1;
  22.     if (setsockopt(listenfd, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(yes)))
  23.     {
  24.         perror("setsockopt");
  25.         exit(1);
  26.     }
  28.     struct sockaddr_in addr;
  29.     muduo::memZero(&addr, sizeof(addr));
  30.     addr.sin_family = AF_INET;
  31.     addr.sin_port = htons(port);
  32.     addr.sin_addr.s_addr = INADDR_ANY;
  33.     if (bind(listenfd, reinterpret_cast<struct sockaddr*>(&addr), sizeof(addr)))
  34.     {
  35.         perror("bind");
  36.         exit(1);
  37.     }
  39.     if (listen(listenfd, 5))
  40.     {
  41.         perror("listen");
  42.         exit(1);
  43.     }
  45.     struct sockaddr_in peer_addr;
  46.     muduo::memZero(&peer_addr, sizeof(peer_addr));
  47.     socklen_t addrlen = 0;
  48.     int sockfd = ::accept(listenfd, reinterpret_cast<struct sockaddr*>(&peer_addr), &addrlen);
  49.     if (sockfd < 0)
  50.     {
  51.         perror("accept");
  52.         exit(1);
  53.     }
  54.     ::close(listenfd);
  55.     return sockfd;
  56. }
  58. static int write_n(int sockfd, const void* buf, int length)
  59. {
  60.     int written = 0;
  61.     while (written < length)
  62.     {
  63.         ssize_t nw = ::write(sockfd, static_cast<const char*>(buf) + written, length - written);
  64.         if (nw > 0)
  65.         {
  66.             written += static_cast<int>(nw);
  67.         }
  68.         else if (nw == 0)
  69.         {
  70.             break;  // EOF
  71.         }
  72.         else if (errno != EINTR)
  73.         {
  74.             perror("write");
  75.             break;
  76.         }
  77.     }
  78.     return written;
  79. }
  81. static int read_n(int sockfd, void* buf, int length)
  82. {
  83.     int nread = 0;
  84.     while (nread < length)
  85.     {
  86.         ssize_t nr = ::read(sockfd, static_cast<char*>(buf) + nread, length - nread);
  87.         if (nr > 0)
  88.         {
  89.             nread += static_cast<int>(nr);
  90.         }
  91.         else if (nr == 0)
  92.         {
  93.             break;  // EOF
  94.         }
  95.         else if (errno != EINTR)
  96.         {
  97.             perror("read");
  98.             break;
  99.         }
  100.     }
  101.     return nread;
  102. }
  104. void transmit(const Options& opt)
  105. {
  106.     struct sockaddr_in addr = resolveOrDie(opt.host.c_str(), opt.port);
  107.     printf("connecting to %s:%d\n", inet_ntoa(addr.sin_addr), opt.port);
  109.     int sockfd = ::socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
  110.     assert(sockfd >= 0);
  111.     int ret = ::connect(sockfd, reinterpret_cast<struct sockaddr*>(&addr), sizeof(addr));
  112.     if (ret)
  113.     {
  114.         perror("connect");
  115.         printf("Unable to connect %s\n", opt.host.c_str());
  116.         ::close(sockfd);
  117.         return;
  118.     }
  120.     printf("connected\n");
  121.     muduo::Timestamp start(muduo::Timestamp::now());
  122.     struct SessionMessage sessionMessage = { 0, 0 };
  123.     sessionMessage.number = htonl(opt.number);
  124.     sessionMessage.length = htonl(opt.length);
  125.     if (write_n(sockfd, &sessionMessage, sizeof(sessionMessage)) != sizeof(sessionMessage))
  126.     {
  127.         perror("write SessionMessage");
  128.         exit(1);
  129.     }
  131.     const int total_len = static_cast<int>(sizeof(int32_t) + opt.length);
  132.     PayloadMessage* payload = static_cast<PayloadMessage*>(::malloc(total_len));
  133.     assert(payload);
  134.     payload->length = htonl(opt.length);
  135.     for (int i = 0; i < opt.length; ++i)
  136.     {
  137.         payload->data[i] = "0123456789ABCDEF"[i % 16];
  138.     }
  140.     double total_mb = 1.0 * opt.length * opt.number / 1024 / 1024;
  141.     printf("%.3f MiB in total\n", total_mb);
  143.     for (int i = 0; i < opt.number; ++i)
  144.     {
  145.         int nw = write_n(sockfd, payload, total_len);
  146.         assert(nw == total_len);
  148.         int ack = 0;
  149.         int nr = read_n(sockfd, &ack, sizeof(ack));
  150.         assert(nr == sizeof(ack));
  151.         ack = ntohl(ack);
  152.         assert(ack == opt.length);
  153.     }
  155.     ::free(payload);
  156.     ::close(sockfd);
  157.     double elapsed = timeDifference(muduo::Timestamp::now(), start);
  158.     printf("%.3f seconds\n%.3f MiB/s\n", elapsed, total_mb / elapsed);
  159. }
  161. void receive(const Options& opt)
  162. {
  163.     // 接受连接
  164.     int sockfd = acceptOrDie(opt.port);
  166.     struct SessionMessage sessionMessage = { 0, 0 };
  167.     // 读取8个字节长度
  168.     if (read_n(sockfd, &sessionMessage, sizeof(sessionMessage)) != sizeof(sessionMessage))
  169.     {
  170.         perror("read SessionMessage");
  171.         exit(1);
  172.     }
  174.     // 网络字节序转为本机字节序
  175.     sessionMessage.number = ntohl(sessionMessage.number);
  176.     sessionMessage.length = ntohl(sessionMessage.length);
  177.     printf("receive number = %d\nreceive length = %d\n",
  178.            sessionMessage.number, sessionMessage.length);
  179.     // 缓冲区: 4个字节 + 消息的长度
  180.     const int total_len = static_cast<int>(sizeof(int32_t) + sessionMessage.length);
  181.     // 如果是外网使用: 接受到超长的消息, malloc会分配很大的内存, 可能造成DDos攻击; 可以通过限制length的长度来防止DDos攻击
  182.     PayloadMessage* payload = static_cast<PayloadMessage*>(::malloc(total_len)); // 不定长消息
  183.     assert(payload);
  185.     for (int i = 0; i < sessionMessage.number; ++i)
  186.     {
  187.         payload->length = 0;
  188.         if (read_n(sockfd, &payload->length, sizeof(payload->length)) != sizeof(payload->length))
  189.         {
  190.             perror("read length");
  191.             exit(1);
  192.         }
  193.         payload->length = ntohl(payload->length);
  194.         assert(payload->length == sessionMessage.length); // 验证消息长度
  196.         // 读取数据
  197.         if (read_n(sockfd, payload->data, payload->length) != payload->length) 
  198.         {
  199.             perror("read payload data");
  200.             exit(1);
  201.         }
  203.         // 构造响应
  204.         int32_t ack = htonl(payload->length);
  205.         if (write_n(sockfd, &ack, sizeof(ack)) != sizeof(ack))
  206.         {
  207.             perror("write ack");
  208.             exit(1);
  209.         }
  210.     }
  211.     ::free(payload);
  212.     ::close(sockfd);
  213. }

[

](https://github.com/chenshuo/muduo/blob/master/examples/ace/ttcp/common.h)

NON-BLOCKING

C++

https://github.com/chenshuo/recipes/blob/master/tpc/bin/ttcp.cc

  1. #include "Acceptor.h"
  2. #include "InetAddress.h"
  3. #include "TcpStream.h"
  5. #include <iostream>
  7. #include <boost/program_options.hpp>
  9. #include <sys/time.h>
  11. struct Options
  12. {
  13.   uint16_t port;
  14.   int length;
  15.   int number;
  16.   bool transmit, receive, nodelay;
  17.   std::string host;
  18.   Options()
  19.     : port(0), length(0), number(0),
  20.       transmit(false), receive(false), nodelay(false)
  21.   {
  22.   }
  23. };
  25. struct SessionMessage
  26. {
  27.   int32_t number;
  28.   int32_t length;
  29. } __attribute__ ((__packed__));
  31. struct PayloadMessage
  32. {
  33.   int32_t length;
  34.   char data[0];
  35. };
  37. double now()
  38. {
  39.   struct timeval tv = { 0, 0 };
  40.   gettimeofday(&tv, NULL);
  41.   return tv.tv_sec + tv.tv_usec / 1000000.0;
  42. }
  44. // FIXME: rewrite with getopt(3).
  45. bool parseCommandLine(int argc, char* argv[], Options* opt)
  46. {
  47.   namespace po = boost::program_options;
  49.   po::options_description desc("Allowed options");
  50.   desc.add_options()
  51.       ("help,h", "Help")
  52.       ("port,p", po::value<uint16_t>(&opt->port)->default_value(5001), "TCP port")
  53.       ("length,l", po::value<int>(&opt->length)->default_value(65536), "Buffer length")
  54.       ("number,n", po::value<int>(&opt->number)->default_value(8192), "Number of buffers")
  55.       ("trans,t",  po::value<std::string>(&opt->host), "Transmit")
  56.       ("recv,r", "Receive")
  57.       ("nodelay,D", "set TCP_NODELAY")
  58.       ;
  60.   po::variables_map vm;
  61.   po::store(po::parse_command_line(argc, argv, desc), vm);
  62.   po::notify(vm);
  64.   opt->transmit = vm.count("trans");
  65.   opt->receive = vm.count("recv");
  66.   opt->nodelay = vm.count("nodelay");
  67.   if (vm.count("help"))
  68.   {
  69.     std::cout << desc << std::endl;
  70.     return false;
  71.   }
  73.   if (opt->transmit == opt->receive)
  74.   {
  75.     printf("either -t or -r must be specified.\n");
  76.     return false;
  77.   }
  79.   printf("port = %d\n", opt->port);
  80.   if (opt->transmit)
  81.   {
  82.     printf("buffer length = %d\n", opt->length);
  83.     printf("number of buffers = %d\n", opt->number);
  84.   }
  85.   else
  86.   {
  87.     printf("accepting...\n");
  88.   }
  89.   return true;
  90. }
  92. void transmit(const Options& opt)
  93. {
  94.   InetAddress addr();
  95.   if (!InetAddress::resolve(opt.host.c_str(), opt.port, &addr))
  96.   {
  97.     printf("Unable to resolve %s\n", opt.host.c_str());
  98.     return;
  99.   }
  101.   printf("connecting to %s\n", addr.toIpPort().c_str());
  102.   TcpStreamPtr stream(TcpStream::connect(addr));
  103.   if (!stream)
  104.   {
  105.     printf("Unable to connect %s\n", addr.toIpPort().c_str());
  106.     perror("");
  107.     return;
  108.   }
  110.   if (opt.nodelay)
  111.   {
  112.     stream->setTcpNoDelay(true);
  113.   }
  114.   printf("connected\n");
  115.   double start = now();
  116.   struct SessionMessage sessionMessage = { 0, 0 };
  117.   sessionMessage.number = htonl(opt.number);
  118.   sessionMessage.length = htonl(opt.length);
  119.   if (stream->sendAll(&sessionMessage, sizeof(sessionMessage)) != sizeof(sessionMessage))
  120.   {
  121.     perror("write SessionMessage");
  122.     return;
  123.   }
  125.   const int total_len = sizeof(int32_t) + opt.length;
  126.   PayloadMessage* payload = static_cast<PayloadMessage*>(::malloc(total_len));
  127.   std::unique_ptr<PayloadMessage, void (*)(void*)> freeIt(payload, ::free);
  128.   assert(payload);
  129.   payload->length = htonl(opt.length);
  130.   for (int i = 0; i < opt.length; ++i)
  131.   {
  132.     payload->data[i] = "0123456789ABCDEF"[i % 16];
  133.   }
  135.   double total_mb = 1.0 * opt.length * opt.number / 1024 / 1024;
  136.   printf("%.3f MiB in total\n", total_mb);
  138.   for (int i = 0; i < opt.number; ++i)
  139.   {
  140.     int nw = stream->sendAll(payload, total_len);
  141.     assert(nw == total_len);
  143.     int ack = 0;
  144.     int nr = stream->receiveAll(&ack, sizeof(ack));
  145.     assert(nr == sizeof(ack));
  146.     ack = ntohl(ack);
  147.     assert(ack == opt.length);
  148.   }
  150.   double elapsed = now() - start;
  151.   printf("%.3f seconds\n%.3f MiB/s\n", elapsed, total_mb / elapsed);
  152. }
  154. void receive(const Options& opt)
  155. {
  156.   Acceptor acceptor(InetAddress(opt.port));
  157.   TcpStreamPtr stream(acceptor.accept());
  158.   if (!stream)
  159.   {
  160.     return;
  161.   }
  162.   struct SessionMessage sessionMessage = { 0, 0 };
  163.   if (stream->receiveAll(&sessionMessage, sizeof(sessionMessage)) != sizeof(sessionMessage))
  164.   {
  165.     perror("read SessionMessage");
  166.     return;
  167.   }
  169.   sessionMessage.number = ntohl(sessionMessage.number);
  170.   sessionMessage.length = ntohl(sessionMessage.length);
  171.   printf("receive buffer length = %d\nreceive number of buffers = %d\n",
  172.          sessionMessage.length, sessionMessage.number);
  173.   double total_mb = 1.0 * sessionMessage.number * sessionMessage.length / 1024 / 1024;
  174.   printf("%.3f MiB in total\n", total_mb);
  176.   const int total_len = sizeof(int32_t) + sessionMessage.length;
  177.   PayloadMessage* payload = static_cast<PayloadMessage*>(::malloc(total_len));
  178.   std::unique_ptr<PayloadMessage, void (*)(void*)> freeIt(payload, ::free);
  179.   assert(payload);
  181.   double start = now();
  182.   for (int i = 0; i < sessionMessage.number; ++i)
  183.   {
  184.     payload->length = 0;
  185.     if (stream->receiveAll(&payload->length, sizeof(payload->length)) != sizeof(payload->length))
  186.     {
  187.       perror("read length");
  188.       return;
  189.     }
  190.     payload->length = ntohl(payload->length);
  191.     assert(payload->length == sessionMessage.length);
  192.     if (stream->receiveAll(payload->data, payload->length) != payload->length)
  193.     {
  194.       perror("read payload data");
  195.       return;
  196.     }
  197.     int32_t ack = htonl(payload->length);
  198.     if (stream->sendAll(&ack, sizeof(ack)) != sizeof(ack))
  199.     {
  200.       perror("write ack");
  201.       return;
  202.     }
  203.   }
  204.   double elapsed = now() - start;
  205.   printf("%.3f seconds\n%.3f MiB/s\n", elapsed, total_mb / elapsed);
  206. }
  208. int main(int argc, char* argv[])
  209. {
  210.   Options options;
  211.   if (parseCommandLine(argc, argv, &options))
  212.   {
  213.     if (options.transmit)
  214.     {
  215.       transmit(options);
  216.     }
  217.     else if (options.receive)
  218.     {
  219.       receive(options);
  220.     }
  221.     else
  222.     {
  223.       assert(0);
  224.     }
  225.   }
  226. }

[

](https://github.com/chenshuo/recipes/blob/master/tpc/bin/ttcp.cc)