#include "rpc.h"
#include <arpa/inet.h>
-#include <stdio.h>
+#include <iostream>
+#include <vector>
+#include <thread>
#include <stdlib.h>
-#include <string.h>
#include <getopt.h>
+#include <sys/types.h>
+#include <unistd.h>
#include "jsl_log.h"
-#include "gettime.h"
#include "lang/verify.h"
#define NUM_CL 2
+char log_thread_prefix = 'r';
+
+using std::string;
+using std::cout;
+using std::endl;
+using std::vector;
+using std::thread;
+
rpcs *server; // server rpc object
rpcc *clients[NUM_CL]; // client rpc object
-struct sockaddr_in dst; //server's ip address
+string dst; //server's ip address
int port;
-pthread_attr_t attr;
// server-side handlers. they must be methods of some class
// to simplify rpcs::reg(). a server process can have handlers
// from multiple classes.
class srv {
public:
- int handle_22(const std::string a, const std::string b, std::string & r);
- int handle_fast(const int a, int &r);
- int handle_slow(const int a, int &r);
- int handle_bigrep(const int a, std::string &r);
+ int handle_22(string & r, const string a, const string b);
+ int handle_fast(int &r, const int a);
+ int handle_slow(int &r, const int a);
+ int handle_bigrep(string &r, const size_t a);
};
// a handler. a and b are arguments, r is the result.
// at these argument types, so this function definition
// does what a .x file does in SunRPC.
int
-srv::handle_22(const std::string a, std::string b, std::string &r)
+srv::handle_22(string &r, const string a, string b)
{
r = a + b;
return 0;
}
int
-srv::handle_fast(const int a, int &r)
+srv::handle_fast(int &r, const int a)
{
r = a + 1;
return 0;
}
int
-srv::handle_slow(const int a, int &r)
+srv::handle_slow(int &r, const int a)
{
usleep(random() % 5000);
r = a + 2;
}
int
-srv::handle_bigrep(const int len, std::string &r)
+srv::handle_bigrep(string &r, const size_t len)
{
- r = std::string(len, 'x');
+ r = string((size_t)len, 'x');
return 0;
}
void startserver()
{
- server = new rpcs(port);
- server->reg(22, &service, &srv::handle_22);
- server->reg(23, &service, &srv::handle_fast);
- server->reg(24, &service, &srv::handle_slow);
- server->reg(25, &service, &srv::handle_bigrep);
+ server = new rpcs((unsigned int)port);
+ server->reg(22, &srv::handle_22, &service);
+ server->reg(23, &srv::handle_fast, &service);
+ server->reg(24, &srv::handle_slow, &service);
+ server->reg(25, &srv::handle_bigrep, &service);
}
void
testmarshall()
{
marshall m;
- req_header rh(1,2,3,4,5);
+ request_header rh{1,2,3,4,5};
m.pack_req_header(rh);
VERIFY(m.size()==RPC_HEADER_SZ);
int i = 12345;
unsigned long long l = 1223344455L;
- std::string s = std::string("hallo....");
+ string s = "hallo....";
m << i;
m << l;
m << s;
char *b;
- int sz;
+ size_t sz;
m.take_buf(&b,&sz);
- VERIFY(sz == (int)(RPC_HEADER_SZ+sizeof(i)+sizeof(l)+s.size()+sizeof(int)));
+ VERIFY(sz == RPC_HEADER_SZ+sizeof(i)+sizeof(l)+s.size()+sizeof(int));
unmarshall un(b,sz);
- req_header rh1;
+ request_header rh1;
un.unpack_req_header(&rh1);
VERIFY(memcmp(&rh,&rh1,sizeof(rh))==0);
int i1;
unsigned long long l1;
- std::string s1;
+ string s1;
un >> i1;
un >> l1;
un >> s1;
VERIFY(i1==i && l1==l && s1==s);
}
-void *
-client1(void *xx)
+void
+client1(size_t cl)
{
-
// test concurrency.
- int which_cl = ((unsigned long) xx ) % NUM_CL;
+ size_t which_cl = cl % NUM_CL;
for(int i = 0; i < 100; i++){
int arg = (random() % 2000);
- std::string rep;
- int ret = clients[which_cl]->call(25, arg, rep);
+ string rep;
+ int ret = clients[which_cl]->call(25, rep, arg);
VERIFY(ret == 0);
- if ((int)rep.size()!=arg) {
- printf("repsize wrong %d!=%d\n", (int)rep.size(), arg);
- }
+ if ((int)rep.size()!=arg)
+ cout << "repsize wrong " << rep.size() << "!=" << arg << endl;
VERIFY((int)rep.size() == arg);
}
int arg = (random() % 1000);
int rep;
- struct timespec start,end;
- clock_gettime(CLOCK_REALTIME, &start);
+ auto start = std::chrono::steady_clock::now();
- int ret = clients[which_cl]->call(which ? 23 : 24, arg, rep);
- clock_gettime(CLOCK_REALTIME, &end);
- int diff = diff_timespec(end, start);
+ int ret = clients[which_cl]->call(which ? 23 : 24, rep, arg);
+ auto end = std::chrono::steady_clock::now();
+ auto diff = std::chrono::duration_cast<std::chrono::milliseconds>(end - start).count();
if (ret != 0)
- printf("%d ms have elapsed!!!\n", diff);
+ cout << diff << " ms have elapsed!!!" << endl;
VERIFY(ret == 0);
VERIFY(rep == (which ? arg+1 : arg+2));
}
-
- return 0;
}
-void *
-client2(void *xx)
+void
+client2(size_t cl)
{
- int which_cl = ((unsigned long) xx ) % NUM_CL;
+ size_t which_cl = cl % NUM_CL;
time_t t1;
time(&t1);
while(time(0) - t1 < 10){
int arg = (random() % 2000);
- std::string rep;
- int ret = clients[which_cl]->call(25, arg, rep);
- if ((int)rep.size()!=arg) {
- printf("ask for %d reply got %d ret %d\n",
- arg, (int)rep.size(), ret);
- }
+ string rep;
+ int ret = clients[which_cl]->call(25, rep, arg);
+ if ((int)rep.size()!=arg)
+ cout << "ask for " << arg << " reply got " << rep.size() << " ret " << ret << endl;
VERIFY((int)rep.size() == arg);
}
- return 0;
}
-void *
+void
client3(void *xx)
{
rpcc *c = (rpcc *) xx;
for(int i = 0; i < 4; i++){
int rep;
- int ret = c->call(24, i, rep, rpcc::to(3000));
+ int ret = c->call_timeout(24, rpcc::to(3000), rep, i);
VERIFY(ret == rpc_const::timeout_failure || rep == i+2);
}
- return 0;
}
void
simple_tests(rpcc *c)
{
- printf("simple_tests\n");
+ cout << "simple_tests" << endl;
// an RPC call to procedure #22.
// rpcc::call() looks at the argument types to decide how
// to marshall the RPC call packet, and how to unmarshall
// the reply packet.
- std::string rep;
- int intret = c->call(22, (std::string)"hello", (std::string)" goodbye", rep);
+ string rep;
+ int intret = c->call(22, rep, (string)"hello", (string)" goodbye");
VERIFY(intret == 0); // this is what handle_22 returns
VERIFY(rep == "hello goodbye");
- printf(" -- string concat RPC .. ok\n");
+ cout << " -- string concat RPC .. ok" << endl;
// small request, big reply (perhaps req via UDP, reply via TCP)
- intret = c->call(25, 70000, rep, rpcc::to(200000));
+ intret = c->call_timeout(25, rpcc::to(200000), rep, 70000);
VERIFY(intret == 0);
VERIFY(rep.size() == 70000);
- printf(" -- small request, big reply .. ok\n");
-
-#if 0
- // too few arguments
- intret = c->call(22, (std::string)"just one", rep);
- VERIFY(intret < 0);
- printf(" -- too few arguments .. failed ok\n");
-
- // too many arguments; proc #23 expects just one.
- intret = c->call(23, 1001, 1002, rep);
- VERIFY(intret < 0);
- printf(" -- too many arguments .. failed ok\n");
-
- // wrong return value size
- int wrongrep;
- intret = c->call(23, (std::string)"hello", (std::string)" goodbye", wrongrep);
- VERIFY(intret < 0);
- printf(" -- wrong ret value size .. failed ok\n");
-#endif
+ cout << " -- small request, big reply .. ok" << endl;
// specify a timeout value to an RPC that should succeed (udp)
int xx = 0;
- intret = c->call(23, 77, xx, rpcc::to(3000));
+ intret = c->call_timeout(23, rpcc::to(3000), xx, 77);
VERIFY(intret == 0 && xx == 78);
- printf(" -- no suprious timeout .. ok\n");
+ cout << " -- no spurious timeout .. ok" << endl;
// specify a timeout value to an RPC that should succeed (tcp)
{
- std::string arg(1000, 'x');
- std::string rep;
- c->call(22, arg, (std::string)"x", rep, rpcc::to(3000));
- VERIFY(rep.size() == 1001);
- printf(" -- no suprious timeout .. ok\n");
+ string arg(1000, 'x');
+ string rep2;
+ c->call_timeout(22, rpcc::to(3000), rep2, arg, (string)"x");
+ VERIFY(rep2.size() == 1001);
+ cout << " -- no spurious timeout .. ok" << endl;
}
// huge RPC
- std::string big(1000000, 'x');
- intret = c->call(22, big, (std::string)"z", rep);
+ string big(1000000, 'x');
+ intret = c->call(22, rep, big, (string)"z");
VERIFY(rep.size() == 1000001);
- printf(" -- huge 1M rpc request .. ok\n");
+ cout << " -- huge 1M rpc request .. ok" << endl;
// specify a timeout value to an RPC that should timeout (udp)
- struct sockaddr_in non_existent;
- memset(&non_existent, 0, sizeof(non_existent));
- non_existent.sin_family = AF_INET;
- non_existent.sin_addr.s_addr = inet_addr("127.0.0.1");
- non_existent.sin_port = htons(7661);
+ string non_existent = "127.0.0.1:7661";
rpcc *c1 = new rpcc(non_existent);
time_t t0 = time(0);
intret = c1->bind(rpcc::to(3000));
time_t t1 = time(0);
VERIFY(intret < 0 && (t1 - t0) <= 4);
- printf(" -- rpc timeout .. ok\n");
- printf("simple_tests OK\n");
+ cout << " -- rpc timeout .. ok" << endl;
+ cout << "simple_tests OK" << endl;
}
void
-concurrent_test(int nt)
+concurrent_test(size_t nt)
{
// create threads that make lots of calls in parallel,
// to test thread synchronization for concurrent calls
// and dispatches.
- int ret;
+ cout << "start concurrent_test (" << nt << " threads) ...";
- printf("start concurrent_test (%d threads) ...", nt);
+ vector<thread> th(nt);
- pthread_t th[nt];
- for(int i = 0; i < nt; i++){
- ret = pthread_create(&th[i], &attr, client1, (void *) (uintptr_t)i);
- VERIFY(ret == 0);
- }
+ for(size_t i = 0; i < nt; i++)
+ th[i] = thread(client1, i);
- for(int i = 0; i < nt; i++){
- VERIFY(pthread_join(th[i], NULL) == 0);
- }
- printf(" OK\n");
+ for(size_t i = 0; i < nt; i++)
+ th[i].join();
+
+ cout << " OK" << endl;
}
void
lossy_test()
{
- int ret;
-
- printf("start lossy_test ...");
+ cout << "start lossy_test ...";
VERIFY(setenv("RPC_LOSSY", "5", 1) == 0);
if (server) {
VERIFY(clients[i]->bind()==0);
}
- int nt = 1;
- pthread_t th[nt];
- for(int i = 0; i < nt; i++){
- ret = pthread_create(&th[i], &attr, client2, (void *) (uintptr_t)i);
- VERIFY(ret == 0);
- }
- for(int i = 0; i < nt; i++){
- VERIFY(pthread_join(th[i], NULL) == 0);
- }
- printf(".. OK\n");
+ size_t nt = 1;
+
+ vector<thread> th(nt);
+
+ for(size_t i = 0; i < nt; i++)
+ th[i] = thread(client2, i);
+
+ for(size_t i = 0; i < nt; i++)
+ th[i].join();
+
+ cout << ".. OK" << endl;
VERIFY(setenv("RPC_LOSSY", "0", 1) == 0);
}
rpcc *client1;
rpcc *client = clients[0];
- printf("failure_test\n");
+ cout << "failure_test" << endl;
delete server;
client1 = new rpcc(dst);
VERIFY (client1->bind(rpcc::to(3000)) < 0);
- printf(" -- create new client and try to bind to failed server .. failed ok\n");
+ cout << " -- create new client and try to bind to failed server .. failed ok" << endl;
delete client1;
startserver();
- std::string rep;
- int intret = client->call(22, (std::string)"hello", (std::string)" goodbye", rep);
+ string rep;
+ int intret = client->call(22, rep, (string)"hello", (string)" goodbye");
VERIFY(intret == rpc_const::oldsrv_failure);
- printf(" -- call recovered server with old client .. failed ok\n");
+ cout << " -- call recovered server with old client .. failed ok" << endl;
delete client;
VERIFY (client->bind() >= 0);
VERIFY (client->bind() < 0);
- intret = client->call(22, (std::string)"hello", (std::string)" goodbye", rep);
+ intret = client->call(22, rep, (string)"hello", (string)" goodbye");
VERIFY(intret == 0);
VERIFY(rep == "hello goodbye");
- printf(" -- delete existing rpc client, create replacement rpc client .. ok\n");
+ cout << " -- delete existing rpc client, create replacement rpc client .. ok" << endl;
- int nt = 10;
- int ret;
- printf(" -- concurrent test on new rpc client w/ %d threads ..", nt);
+ size_t nt = 10;
+ cout << " -- concurrent test on new rpc client w/ " << nt << " threads ..";
- pthread_t th[nt];
- for(int i = 0; i < nt; i++){
- ret = pthread_create(&th[i], &attr, client3, (void *) client);
- VERIFY(ret == 0);
- }
+ vector<thread> th(nt);
- for(int i = 0; i < nt; i++){
- VERIFY(pthread_join(th[i], NULL) == 0);
- }
- printf("ok\n");
+ for(size_t i = 0; i < nt; i++)
+ th[i] = thread(client3, client);
+
+ for(size_t i = 0; i < nt; i++)
+ th[i].join();
+
+ cout << "ok" << endl;
delete server;
delete client;
startserver();
clients[0] = client = new rpcc(dst);
VERIFY (client->bind() >= 0);
- printf(" -- delete existing rpc client and server, create replacements.. ok\n");
+ cout << " -- delete existing rpc client and server, create replacements.. ok" << endl;
- printf(" -- concurrent test on new client and server w/ %d threads ..", nt);
- for(int i = 0; i < nt; i++){
- ret = pthread_create(&th[i], &attr, client3, (void *)client);
- VERIFY(ret == 0);
- }
+ cout << " -- concurrent test on new client and server w/ " << nt << " threads ..";
- for(int i = 0; i < nt; i++){
- VERIFY(pthread_join(th[i], NULL) == 0);
- }
- printf("ok\n");
+ for(size_t i = 0; i < nt; i++)
+ th[i] = thread(client3, client);
+
+ for(size_t i = 0; i < nt; i++)
+ th[i].join();
- printf("failure_test OK\n");
+ cout << "ok" << endl;
+
+ cout << "failure_test OK" << endl;
}
int
bool isclient = false;
bool isserver = false;
- srandom(getpid());
+ srandom((uint32_t)getpid());
port = 20000 + (getpid() % 10000);
- char ch = 0;
+ int ch = 0;
while ((ch = getopt(argc, argv, "csd:p:l"))!=-1) {
switch (ch) {
case 'c':
break;
case 'l':
VERIFY(setenv("RPC_LOSSY", "5", 1) == 0);
+ break;
default:
break;
}
}
if (debug_level > 0) {
- //__loginit.initNow();
- jsl_set_debug(debug_level);
+ JSL_DEBUG_LEVEL = debug_level;
jsl_log(JSL_DBG_1, "DEBUG LEVEL: %d\n", debug_level);
}
testmarshall();
- pthread_attr_init(&attr);
- // set stack size to 32K, so we don't run out of memory
- pthread_attr_setstacksize(&attr, 32*1024);
-
if (isserver) {
- printf("starting server on port %d RPC_HEADER_SZ %d\n", port, RPC_HEADER_SZ);
+ cout << "starting server on port " << port << " RPC_HEADER_SZ " << (int)RPC_HEADER_SZ << endl;
startserver();
}
if (isclient) {
// server's address.
- memset(&dst, 0, sizeof(dst));
- dst.sin_family = AF_INET;
- dst.sin_addr.s_addr = inet_addr("127.0.0.1");
- dst.sin_port = htons(port);
+ dst = "127.0.0.1:" + std::to_string(port);
// start the client. bind it to the server.
failure_test();
}
- printf("rpctest OK\n");
+ cout << "rpctest OK" << endl;
exit(0);
}
projects / invirt/third/libt4.git / blobdiff