Skip to content

Commit cc4a8dc

Browse files
chenshuochenshuo
committed
add examples/ace/ttcp
1 parent 125a440 commit cc4a8dc

6 files changed

Lines changed: 551 additions & 0 deletions

File tree

‎examples/ace/ttcp/CMakeLists.txt‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,10 @@
1+
2+
if(BOOSTPO_LIBRARY)
3+
add_executable(ttcp_blocking ttcp_blocking.cc common.cc main.cc)
4+
target_link_libraries(ttcp_blocking muduo_base boost_program_options)
5+
set_target_properties(ttcp_blocking PROPERTIES COMPILE_FLAGS "-Wno-error=old-style-cast")
6+
7+
add_executable(ttcp_muduo ttcp.cc common.cc main.cc)
8+
target_link_libraries(ttcp_muduo muduo_net boost_program_options)
9+
endif()
10+

‎examples/ace/ttcp/common.cc‎

Lines changed: 75 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,75 @@
1+
#include <examples/ace/ttcp/common.h>
2+
3+
#include <boost/program_options.hpp>
4+
5+
#include <iostream>
6+
7+
#include <netdb.h>
8+
#include <stdio.h>
9+
10+
namespace po = boost::program_options;
11+
12+
bool parseCommandLine(int argc, char* argv[], Options* opt)
13+
{
14+
po::options_description desc("Allowed options");
15+
desc.add_options()
16+
("help,h", "Help")
17+
("port,p", po::value<uint16_t>(&opt->port)->default_value(5001), "TCP port")
18+
("length,l", po::value<int>(&opt->length)->default_value(65536), "Buffer length")
19+
("number,n", po::value<int>(&opt->number)->default_value(8192), "Number of buffers")
20+
("trans,t", po::value<std::string>(&opt->host), "Transmit")
21+
("recv,r", "Receive")
22+
("nodelay,D", "set TCP_NODELAY")
23+
;
24+
25+
po::variables_map vm;
26+
po::store(po::parse_command_line(argc, argv, desc), vm);
27+
po::notify(vm);
28+
29+
opt->transmit = vm.count("trans");
30+
opt->receive = vm.count("recv");
31+
opt->nodelay = vm.count("nodelay");
32+
if (vm.count("help"))
33+
{
34+
std::cout << desc << std::endl;
35+
return false;
36+
}
37+
38+
if (opt->transmit == opt->receive)
39+
{
40+
printf("either -t or -r must be specified.\n");
41+
return false;
42+
}
43+
44+
printf("port = %d\n", opt->port);
45+
if (opt->transmit)
46+
{
47+
printf("buffer length = %d\n", opt->length);
48+
printf("number of buffers = %d\n", opt->number);
49+
}
50+
else
51+
{
52+
printf("accepting...\n");
53+
}
54+
return true;
55+
}
56+
57+
#pragma GCC diagnostic ignored "-Wold-style-cast"
58+
59+
struct sockaddr_in resolveOrDie(const char* host, uint16_t port)
60+
{
61+
struct hostent* he = ::gethostbyname(host);
62+
if (!he)
63+
{
64+
perror("gethostbyname");
65+
exit(1);
66+
}
67+
assert(he->h_addrtype == AF_INET && he->h_length == sizeof(uint32_t));
68+
struct sockaddr_in addr;
69+
bzero(&addr, sizeof(addr));
70+
addr.sin_family = AF_INET;
71+
addr.sin_port = htons(port);
72+
addr.sin_addr = *reinterpret_cast<struct in_addr*>(he->h_addr);
73+
return addr;
74+
}
75+

‎examples/ace/ttcp/common.h‎

Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,37 @@
1+
#pragma once
2+
3+
#include <string>
4+
#include <stdint.h>
5+
6+
struct Options
7+
{
8+
uint16_t port;
9+
int length;
10+
int number;
11+
bool transmit, receive, nodelay;
12+
std::string host;
13+
Options()
14+
: port(0), length(0), number(0),
15+
transmit(false), receive(false), nodelay(false)
16+
{
17+
}
18+
};
19+
20+
bool parseCommandLine(int argc, char* argv[], Options* opt);
21+
struct sockaddr_in resolveOrDie(const char* host, uint16_t port);
22+
23+
struct SessionMessage
24+
{
25+
int32_t number;
26+
int32_t length;
27+
} __attribute__ ((__packed__));
28+
29+
struct PayloadMessage
30+
{
31+
int32_t length;
32+
char data[0];
33+
};
34+
35+
void transmit(const Options& opt);
36+
37+
void receive(const Options& opt);

‎examples/ace/ttcp/main.cc‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,24 @@
1+
#include <examples/ace/ttcp/common.h>
2+
3+
#include <assert.h>
4+
5+
int main(int argc, char* argv[])
6+
{
7+
Options options;
8+
if (parseCommandLine(argc, argv, &options))
9+
{
10+
if (options.transmit)
11+
{
12+
transmit(options);
13+
}
14+
else if (options.receive)
15+
{
16+
receive(options);
17+
}
18+
else
19+
{
20+
assert(0);
21+
}
22+
}
23+
}
24+

‎examples/ace/ttcp/ttcp.cc‎

Lines changed: 212 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,212 @@
1+
#include <examples/ace/ttcp/common.h>
2+
3+
#include <muduo/base/Logging.h>
4+
#include <muduo/net/EventLoop.h>
5+
#include <muduo/net/TcpClient.h>
6+
#include <muduo/net/TcpServer.h>
7+
8+
#include <boost/bind.hpp>
9+
10+
#include <stdio.h>
11+
12+
using namespace muduo;
13+
using namespace muduo::net;
14+
15+
EventLoop* g_loop;
16+
17+
struct Context
18+
{
19+
int count;
20+
int64_t bytes;
21+
SessionMessage session;
22+
Buffer output;
23+
24+
Context()
25+
: count(0),
26+
bytes(0)
27+
{
28+
session.number = 0;
29+
session.length = 0;
30+
}
31+
};
32+
33+
/////////////////////////////////////////////////////////////////////
34+
// T R A N S M I T
35+
/////////////////////////////////////////////////////////////////////
36+
37+
namespace trans
38+
{
39+
40+
void onConnection(const Options& opt, const TcpConnectionPtr& conn)
41+
{
42+
if (conn->connected())
43+
{
44+
printf("connected\n");
45+
Context context;
46+
context.count = 1;
47+
context.bytes = opt.length;
48+
context.session.number = opt.number;
49+
context.session.length = opt.length;
50+
context.output.appendInt32(opt.length);
51+
context.output.ensureWritableBytes(opt.length);
52+
for (int i = 0; i < opt.length; ++i)
53+
{
54+
context.output.beginWrite()[i] = "0123456789ABCDEF"[i % 16];
55+
}
56+
context.output.hasWritten(opt.length);
57+
conn->setContext(context);
58+
59+
SessionMessage sessionMessage = { 0, 0 };
60+
sessionMessage.number = htonl(opt.number);
61+
sessionMessage.length = htonl(opt.length);
62+
conn->send(&sessionMessage, sizeof(sessionMessage));
63+
64+
conn->send(context.output.toStringPiece());
65+
}
66+
else
67+
{
68+
const Context& context = boost::any_cast<Context>(conn->getContext());
69+
LOG_INFO << "payload bytes " << context.bytes;
70+
conn->getLoop()->quit();
71+
}
72+
}
73+
74+
void onMessage(const TcpConnectionPtr& conn, Buffer* buf, Timestamp time)
75+
{
76+
Context* context = boost::any_cast<Context>(conn->getMutableContext());
77+
while (buf->readableBytes() >= sizeof(int32_t))
78+
{
79+
int32_t length = buf->readInt32();
80+
if (length == context->session.length)
81+
{
82+
if (context->count < context->session.number)
83+
{
84+
conn->send(context->output.toStringPiece());
85+
++context->count;
86+
context->bytes += length;
87+
}
88+
else
89+
{
90+
conn->shutdown();
91+
break;
92+
}
93+
}
94+
else
95+
{
96+
conn->shutdown();
97+
break;
98+
}
99+
}
100+
}
101+
102+
}
103+
104+
void transmit(const Options& opt)
105+
{
106+
InetAddress addr(resolveOrDie(opt.host.c_str(), opt.port));
107+
muduo::Timestamp start(muduo::Timestamp::now());
108+
EventLoop loop;
109+
g_loop = &loop;
110+
TcpClient client(&loop, addr, "TtcpClient");
111+
client.setConnectionCallback(
112+
boost::bind(&trans::onConnection, opt, _1));
113+
client.setMessageCallback(
114+
boost::bind(&trans::onMessage, _1, _2, _3));
115+
client.connect();
116+
loop.loop();
117+
double elapsed = timeDifference(muduo::Timestamp::now(), start);
118+
double total_mb = 1.0 * opt.length * opt.number / 1024 / 1024;
119+
printf("%.3f MiB\n%.3f MiB/s\n", total_mb, total_mb / elapsed);
120+
}
121+
122+
/////////////////////////////////////////////////////////////////////
123+
// R E C E I V E
124+
/////////////////////////////////////////////////////////////////////
125+
126+
namespace receiving
127+
{
128+
129+
void onConnection(const TcpConnectionPtr& conn)
130+
{
131+
if (conn->connected())
132+
{
133+
Context context;
134+
conn->setContext(context);
135+
}
136+
else
137+
{
138+
const Context& context = boost::any_cast<Context>(conn->getContext());
139+
LOG_INFO << "payload bytes " << context.bytes;
140+
conn->getLoop()->quit();
141+
}
142+
}
143+
144+
void onMessage(const TcpConnectionPtr& conn, Buffer* buf, Timestamp time)
145+
{
146+
while (buf->readableBytes() >= sizeof(int32_t))
147+
{
148+
Context* context = boost::any_cast<Context>(conn->getMutableContext());
149+
SessionMessage& session = context->session;
150+
if (session.number == 0 && session.length == 0)
151+
{
152+
if (buf->readableBytes() >= sizeof(SessionMessage))
153+
{
154+
session.number = buf->readInt32();
155+
session.length = buf->readInt32();
156+
context->output.appendInt32(session.length);
157+
printf("receive number = %d\nreceive length = %d\n",
158+
session.number, session.length);
159+
}
160+
else
161+
{
162+
break;
163+
}
164+
}
165+
else
166+
{
167+
const unsigned total_len = session.length + static_cast<int>(sizeof(int32_t));
168+
const int32_t length = buf->peekInt32();
169+
if (length == session.length)
170+
{
171+
if (buf->readableBytes() >= total_len)
172+
{
173+
buf->retrieve(total_len);
174+
conn->send(context->output.toStringPiece());
175+
++context->count;
176+
context->bytes += length;
177+
if (context->count >= session.number)
178+
{
179+
conn->shutdown();
180+
break;
181+
}
182+
}
183+
else
184+
{
185+
break;
186+
}
187+
}
188+
else
189+
{
190+
printf("wrong length %d\n", length);
191+
conn->shutdown();
192+
break;
193+
}
194+
}
195+
}
196+
}
197+
198+
}
199+
200+
void receive(const Options& opt)
201+
{
202+
EventLoop loop;
203+
g_loop = &loop;
204+
InetAddress listenAddr(opt.port);
205+
TcpServer server(&loop, listenAddr, "TtcpReceive");
206+
server.setConnectionCallback(
207+
boost::bind(&receiving::onConnection, _1));
208+
server.setMessageCallback(
209+
boost::bind(&receiving::onMessage, _1, _2, _3));
210+
server.start();
211+
loop.loop();
212+
}

0 commit comments

Comments
 (0)