Skip to content

Commit

Permalink
Format code by clang-format.
Browse files Browse the repository at this point in the history
  • Loading branch information
@jovany-wang
jovany-wang committed Sep 2, 2018
1 parent c331d97 commit 93608ff
Show file tree
Hide file tree
Showing 66 changed files with 1,946 additions and 1,754 deletions.
107 changes: 56 additions & 51 deletions 10m/10m-cli.cc
View file
Original file line number Diff line number Diff line change
Expand Up @@ -12,9 +12,10 @@ struct Report {
Report() { memset(this, 0, sizeof(*this)); }
};

int main(int argc, const char* argv[]) {
int main(int argc, const char *argv[]) {
if (argc < 9) {
printf("usage %s <host> <begin port> <end port> <conn count> <create seconds> <subprocesses> <hearbeat interval> <send size> <management port>\n", argv[0]);
printf("usage %s <host> <begin port> <end port> <conn count> <create seconds> <subprocesses> <hearbeat interval> <send size> <management port>\n",
argv[0]);
return 1;
}
int c = 1;
Expand All @@ -30,18 +31,18 @@ int main(int argc, const char* argv[]) {
int man_port = atoi(argv[c++]);

int pid = 1;
for (int i = 0; i < processes; i ++) {
for (int i = 0; i < processes; i++) {
pid = fork();
if (pid == 0) { // a child process, break
if (pid == 0) { // a child process, break
sleep(1);
break;
}
}
Signal::signal(SIGPIPE, []{});
Signal::signal(SIGPIPE, [] {});
EventBase base;
if (pid == 0) { //child process
if (pid == 0) { // child process
char *buf = new char[bsz];
ExitCaller ec1([=] {delete[] buf; });
ExitCaller ec1([=] { delete[] buf; });
Slice msg(buf, bsz);
char heartbeat[] = "heartbeat";
int send = 0;
Expand All @@ -51,94 +52,98 @@ int main(int argc, const char* argv[]) {

vector<TcpConnPtr> allConns;
info("creating %d connections", conn_count);
for (int k = 0; k < create_seconds * 10; k ++) {
base.runAfter(100*k, [&]{
for (int k = 0; k < create_seconds * 10; k++) {
base.runAfter(100 * k, [&] {
int c = conn_count / create_seconds / 10;
for (int i = 0; i < c; i++) {
short port = begin_port + (i % (end_port - begin_port));
auto con = TcpConn::createConnection(&base, host, port, 20*1000);
auto con = TcpConn::createConnection(&base, host, port, 20 * 1000);
allConns.push_back(con);
con->setReconnectInterval(20*1000);
con->onMsg(new LengthCodec, [&](const TcpConnPtr& con, const Slice& msg) {
if (heartbeat_interval == 0) { // echo the msg if no interval
con->setReconnectInterval(20 * 1000);
con->onMsg(new LengthCodec, [&](const TcpConnPtr &con, const Slice &msg) {
if (heartbeat_interval == 0) { // echo the msg if no interval
con->sendMsg(msg);
send++;
}
recved ++;
recved++;
});
con->onState([&, i](const TcpConnPtr &con) {
TcpConn::State st = con->getState();
if (st == TcpConn::Connected) {
connected++;
// send ++;
// con->sendMsg(msg);
} else if (st == TcpConn::Failed || st == TcpConn::Closed) { //连接出错
if (st == TcpConn::Closed) { connected--; }
// send ++;
// con->sendMsg(msg);
} else if (st == TcpConn::Failed || st == TcpConn::Closed) { //连接出错
if (st == TcpConn::Closed) {
connected--;
}
retry++;
}
});
}

});
}
if (heartbeat_interval) {
base.runAfter(heartbeat_interval * 1000, [&] {
for (int i = 0; i < heartbeat_interval*10; i ++) {
base.runAfter(i*100, [&,i]{
size_t block = allConns.size() / heartbeat_interval / 10;
for (size_t j=i*block; j<(i+1)*block && j<allConns.size(); j++) {
if (allConns[j]->getState() == TcpConn::Connected) {
allConns[j]->sendMsg(msg);
send++;
}
}
});
}
}, heartbeat_interval * 1000);
base.runAfter(heartbeat_interval * 1000,
[&] {
for (int i = 0; i < heartbeat_interval * 10; i++) {
base.runAfter(i * 100, [&, i] {
size_t block = allConns.size() / heartbeat_interval / 10;
for (size_t j = i * block; j < (i + 1) * block && j < allConns.size(); j++) {
if (allConns[j]->getState() == TcpConn::Connected) {
allConns[j]->sendMsg(msg);
send++;
}
}
});
}
},
heartbeat_interval * 1000);
}
TcpConnPtr report = TcpConn::createConnection(&base, "127.0.0.1", man_port, 3000);
report->onMsg(new LineCodec, [&](const TcpConnPtr& con, Slice msg) {
report->onMsg(new LineCodec, [&](const TcpConnPtr &con, Slice msg) {
if (msg == "exit") {
info("recv exit msg from master, so exit");
base.exit();
}
});
report->onState([&](const TcpConnPtr& con) {
report->onState([&](const TcpConnPtr &con) {
if (con->getState() == TcpConn::Closed) {
base.exit();
}
});
base.runAfter(2000, [&]() {
report->sendMsg(util::format("%d connected: %ld retry: %ld send: %ld recved: %ld",
getpid(), connected, retry, send, recved));
}, 100);
base.runAfter(2000,
[&]() { report->sendMsg(util::format("%d connected: %ld retry: %ld send: %ld recved: %ld", getpid(), connected, retry, send, recved)); },
100);
base.loop();
} else { // master process
} else { // master process
map<int, Report> subs;
TcpServerPtr master = TcpServer::startServer(&base, "127.0.0.1", man_port);
master->onConnMsg(new LineCodec, [&](const TcpConnPtr& con, Slice msg) {
master->onConnMsg(new LineCodec, [&](const TcpConnPtr &con, Slice msg) {
auto fs = msg.split(' ');
if (fs.size() != 9) {
error("number of fields is %lu expected 7", fs.size());
return;
}
Report& c = subs[atoi(fs[0].data())];
Report &c = subs[atoi(fs[0].data())];
c.connected = atoi(fs[2].data());
c.retry= atoi(fs[4].data());
c.retry = atoi(fs[4].data());
c.sended = atoi(fs[6].data());
c.recved = atoi(fs[8].data());
});
base.runAfter(3000, [&](){
for(auto& s: subs) {
Report& r = s.second;
printf("pid: %6ld connected %6ld retry %6ld sended %6ld recved %6ld\n", (long)s.first, r.connected, r.retry, r.sended, r.recved);
}
printf("\n");
}, 3000);
Signal::signal(SIGCHLD, []{
base.runAfter(3000,
[&]() {
for (auto &s : subs) {
Report &r = s.second;
printf("pid: %6ld connected %6ld retry %6ld sended %6ld recved %6ld\n", (long) s.first, r.connected, r.retry, r.sended, r.recved);
}
printf("\n");
},
3000);
Signal::signal(SIGCHLD, [] {
int status = 0;
wait(&status);
error("wait result: status: %d is signaled: %d signal: %d", status, WIFSIGNALED(status), WTERMSIG(status));
error("wait result: status: %d is signaled: %d signal: %d", status, WIFSIGNALED(status), WTERMSIG(status));
});
base.loop();
}
Expand Down
54 changes: 27 additions & 27 deletions 10m/10m-svr.cc
View file
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@ struct Report {
Report() { memset(this, 0, sizeof(*this)); }
};

int main(int argc, const char* argv[]) {
int main(int argc, const char *argv[]) {
if (argc < 5) {
printf("usage: %s <begin port> <end port> <subprocesses> <management port>\n", argv[0]);
return 1;
Expand All @@ -20,75 +20,75 @@ int main(int argc, const char* argv[]) {
int processes = atoi(argv[3]);
int man_port = atoi(argv[4]);
int pid = 1;
for (int i = 0; i < processes; i ++) {
for (int i = 0; i < processes; i++) {
pid = fork();
if (pid == 0) { // a child process, break
if (pid == 0) { // a child process, break
break;
}
}
EventBase base;
if (pid == 0) { //child process
usleep(100*1000); // wait master to listen management port
if (pid == 0) { // child process
usleep(100 * 1000); // wait master to listen management port
vector<TcpServerPtr> svrs;
long connected = 0, closed = 0, recved = 0;
for (int i = 0; i < end_port-begin_port; i ++) {
for (int i = 0; i < end_port - begin_port; i++) {
TcpServerPtr p = TcpServer::startServer(&base, "", begin_port + i, true);
p->onConnCreate([&]{
p->onConnCreate([&] {
TcpConnPtr con(new TcpConn);
con->onState([&](const TcpConnPtr& con) {
con->onState([&](const TcpConnPtr &con) {
auto st = con->getState();
if (st == TcpConn::Connected) {
connected ++;
connected++;
} else if (st == TcpConn::Closed || st == TcpConn::Failed) {
closed ++;
connected --;
closed++;
connected--;
}
});
con->onMsg(new LengthCodec, [&](const TcpConnPtr& con, Slice msg) {
recved ++;
con->onMsg(new LengthCodec, [&](const TcpConnPtr &con, Slice msg) {
recved++;
con->sendMsg(msg);
});
return con;
});
svrs.push_back(p);
}
TcpConnPtr report = TcpConn::createConnection(&base, "127.0.0.1", man_port, 3000);
report->onMsg(new LineCodec, [&](const TcpConnPtr& con, Slice msg) {
report->onMsg(new LineCodec, [&](const TcpConnPtr &con, Slice msg) {
if (msg == "exit") {
info("recv exit msg from master, so exit");
base.exit();
}
});
report->onState([&](const TcpConnPtr& con) {
report->onState([&](const TcpConnPtr &con) {
if (con->getState() == TcpConn::Closed) {
base.exit();
}
});
base.runAfter(100, [&]() {
report->sendMsg(util::format("%d connected: %ld closed: %ld recved: %ld", getpid(), connected, closed, recved));
}, 100);
base.runAfter(100, [&]() { report->sendMsg(util::format("%d connected: %ld closed: %ld recved: %ld", getpid(), connected, closed, recved)); }, 100);
base.loop();
} else {
map<int, Report> subs;
TcpServerPtr master = TcpServer::startServer(&base, "127.0.0.1", man_port);
master->onConnMsg(new LineCodec, [&](const TcpConnPtr& con, Slice msg) {
master->onConnMsg(new LineCodec, [&](const TcpConnPtr &con, Slice msg) {
auto fs = msg.split(' ');
if (fs.size() != 7) {
error("number of fields is %lu expected 7", fs.size());
return;
}
Report& c = subs[atoi(fs[0].data())];
Report &c = subs[atoi(fs[0].data())];
c.connected = atoi(fs[2].data());
c.closed = atoi(fs[4].data());
c.recved = atoi(fs[6].data());
});
base.runAfter(3000, [&](){
for(auto& s: subs) {
Report r = s.second;
printf("pid: %6d connected %6ld closed: %6ld recved %6ld\n", s.first, r.connected, r.closed, r.recved);
}
printf("\n");
}, 3000);
base.runAfter(3000,
[&]() {
for (auto &s : subs) {
Report r = s.second;
printf("pid: %6d connected %6ld closed: %6ld recved %6ld\n", s.first, r.connected, r.closed, r.recved);
}
printf("\n");
},
3000);
base.loop();
}
info("program exited");
Expand Down
35 changes: 17 additions & 18 deletions examples/chat.cc
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4,50 +4,50 @@
using namespace std;
using namespace handy;

int main(int argc, const char* argv[]) {
int main(int argc, const char *argv[]) {
setloglevel("TRACE");
map<intptr_t, TcpConnPtr> users; //生命周期比连接更长，必须放在Base前
map<intptr_t, TcpConnPtr> users; //生命周期比连接更长，必须放在Base前
EventBase base;
Signal::signal(SIGINT, [&]{ base.exit(); });
Signal::signal(SIGINT, [&] { base.exit(); });

int userid = 1;
TcpServerPtr chat = TcpServer::startServer(&base, "", 2099);
exitif(chat == NULL, "start tcpserver failed");
chat->onConnCreate([&]{
chat->onConnCreate([&] {
TcpConnPtr con(new TcpConn);
con->onState([&](const TcpConnPtr& con) {
con->onState([&](const TcpConnPtr &con) {
if (con->getState() == TcpConn::Connected) {
con->context<int>() = userid;
const char* welcome = "<id> <msg>: send msg to <id>\n<msg>: send msg to all\n\nhello %d";
const char *welcome = "<id> <msg>: send msg to <id>\n<msg>: send msg to all\n\nhello %d";
con->sendMsg(util::format(welcome, userid));
users[userid] = con;
userid ++;
userid++;
} else if (con->getState() == TcpConn::Closed) {
users.erase(con->context<int>());
}
});
con->onMsg(new LineCodec, [&](const TcpConnPtr& con, Slice msg){
if (msg.size() == 0) { //忽略空消息
con->onMsg(new LineCodec, [&](const TcpConnPtr &con, Slice msg) {
if (msg.size() == 0) { //忽略空消息
return;
}
int cid = con->context<int>();
char* p = (char*)msg.data();
char *p = (char *) msg.data();
intptr_t id = strtol(p, &p, 10);
p += *p == ' '; //忽略一个空格
string resp = util::format("%ld# %.*s", cid, msg.end()-p, p);
p += *p == ' '; //忽略一个空格
string resp = util::format("%ld# %.*s", cid, msg.end() - p, p);

int sended = 0;
if (id == 0) { //发给其他所有用户
for(auto& pc: users) {
if (id == 0) { //发给其他所有用户
for (auto &pc : users) {
if (pc.first != cid) {
sended ++;
sended++;
pc.second->sendMsg(resp);
}
}
} else { //发给特定用户
} else { //发给特定用户
auto p1 = users.find(id);
if (p1 != users.end()) {
sended ++;
sended++;
p1->second->sendMsg(resp);
}
}
Expand All @@ -59,4 +59,3 @@ int main(int argc, const char* argv[]) {
info("program exited");
return 0;
}

10 changes: 4 additions & 6 deletions examples/codec-cli.cc
100755 → 100644
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3,16 +3,14 @@
using namespace std;
using namespace handy;

int main(int argc, const char* argv[]) {
int main(int argc, const char *argv[]) {
setloglevel("TRACE");
EventBase base;
Signal::signal(SIGINT, [&]{ base.exit(); });
Signal::signal(SIGINT, [&] { base.exit(); });
TcpConnPtr con = TcpConn::createConnection(&base, "127.0.0.1", 2099, 3000);
con->setReconnectInterval(3000);
con->onMsg(new LengthCodec, [](const TcpConnPtr& con, Slice msg) {
info("recv msg: %.*s", (int)msg.size(), msg.data());
});
con->onState([=](const TcpConnPtr& con) {
con->onMsg(new LengthCodec, [](const TcpConnPtr &con, Slice msg) { info("recv msg: %.*s", (int) msg.size(), msg.data()); });
con->onState([=](const TcpConnPtr &con) {
info("onState called state: %d", con->getState());
if (con->getState() == TcpConn::Connected) {
con->sendMsg("hello");
Expand Down
Loading

0 comments on commit 93608ff

Please sign in to comment.