All sleep calls via std::this_thread

[invirt/third/libt4.git] / rpc / rpctest.cc

diff --git a/rpc/rpctest.cc b/rpc/rpctest.cc
index 1963ada..0435ab1 100644 (file)
--- a/rpc/rpctest.cc
+++ b/rpc/rpctest.cc
@@ -6,15 +6,20 @@
 #include <arpa/inet.h>
 #include <getopt.h>
 #include <unistd.h>
 #include <arpa/inet.h>
 #include <getopt.h>
 #include <unistd.h>

+#include <string.h>
+#include "threaded_log.h"

 
 #define NUM_CL 2
 
 
 #define NUM_CL 2
 

-char log_thread_prefix = 'r';
+static rpcs *server;  // server rpc object
+static rpcc *clients[NUM_CL];  // client rpc object
+static string * dst; //server's ip address
+static in_port_t port;

 
 

-rpcs *server;  // server rpc object
-rpcc *clients[NUM_CL];  // client rpc object
-string dst; //server's ip address
-in_port_t port;
+using std::cout;
+using std::endl;
+using namespace std::chrono;
+using std::vector;

 
 // server-side handlers. they must be methods of some class
 // to simplify rpcs::reg(). a server process can have handlers
 
 // server-side handlers. they must be methods of some class
 // to simplify rpcs::reg(). a server process can have handlers

@@ -22,9 +27,9 @@ in_port_t port;
 class srv {
     public:
         int handle_22(string & r, const string a, const string b);
 class srv {
     public:
         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);
+        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);

 };
 
 namespace srv_protocol {
 };
 
 namespace srv_protocol {

@@ -34,7 +39,7 @@ namespace srv_protocol {
     REMOTE_PROCEDURE(23, fast, (int &, int));
     REMOTE_PROCEDURE(24, slow, (int &, int));
     REMOTE_PROCEDURE(25, bigrep, (string &, size_t));
     REMOTE_PROCEDURE(23, fast, (int &, int));
     REMOTE_PROCEDURE(24, slow, (int &, int));
     REMOTE_PROCEDURE(25, bigrep, (string &, size_t));

-};
+}

 
 // a handler. a and b are arguments, r is the result.
 // there can be multiple arguments but only one result.
 
 // a handler. a and b are arguments, r is the result.
 // there can be multiple arguments but only one result.

@@ -43,30 +48,31 @@ namespace srv_protocol {
 // rpcs::reg() decides how to unmarshall by looking
 // at these argument types, so this function definition
 // does what a .x file does in SunRPC.
 // rpcs::reg() decides how to unmarshall by looking
 // at these argument types, so this function definition
 // does what a .x file does in SunRPC.

-int srv::handle_22(string &r, const string a, string b) {
+int srv::handle_22(string & r, const string a, string b) {

     r = a + b;
     return 0;
 }
 
     r = a + b;
     return 0;
 }
 

-int srv::handle_fast(int &r, const int a) {
+int srv::handle_fast(int & r, const int a) {

     r = a + 1;
     return 0;
 }
 
     r = a + 1;
     return 0;
 }
 

-int srv::handle_slow(int &r, const int a) {
-    usleep(random() % 500);
+int srv::handle_slow(int & r, const int a) {
+    int us = std::uniform_int_distribution<>(0,500)(global->random_generator);
+    std::this_thread::sleep_for(microseconds(us));

     r = a + 2;
     return 0;
 }
 
     r = a + 2;
     return 0;
 }
 

-int srv::handle_bigrep(string &r, const size_t len) {
-    r = string((size_t)len, 'x');
+int srv::handle_bigrep(string & r, const size_t len) {
+    r = string(len, 'x');

     return 0;
 }
 
     return 0;
 }
 

-srv service;
+static srv service;

 
 

-void startserver() {
+static void startserver() {

     server = new rpcs(port);
     server->reg(srv_protocol::_22, &srv::handle_22, &service);
     server->reg(srv_protocol::fast, &srv::handle_fast, &service);
     server = new rpcs(port);
     server->reg(srv_protocol::_22, &srv::handle_22, &service);
     server->reg(srv_protocol::fast, &srv::handle_fast, &service);

@@ -75,41 +81,49 @@ void startserver() {
     server->start();
 }
 
     server->start();
 }
 

-void testmarshall() {
+static void testmarshall() {

     marshall m;
     rpc_protocol::request_header rh{1,2,3,4,5};
     marshall m;
     rpc_protocol::request_header rh{1,2,3,4,5};

-    m.pack_header(rh);
+    m.write_header(rh);

     VERIFY(((string)m).size()==rpc_protocol::RPC_HEADER_SZ);
     int i = 12345;
     unsigned long long l = 1223344455L;
     VERIFY(((string)m).size()==rpc_protocol::RPC_HEADER_SZ);
     int i = 12345;
     unsigned long long l = 1223344455L;

+    size_t sz = 101010101;

     string s = "hallo....";
     string s = "hallo....";

+    string bin("\x00\x00\x00\x00\x00\x00\x00\x40\x00\x00\x7f\xe5", 12);

     m << i;
     m << l;
     m << s;
     m << i;
     m << l;
     m << s;

+    m << sz;
+    m << bin;

 
     string b = m;
 
     string b = m;

-    VERIFY(b.size() == rpc_protocol::RPC_HEADER_SZ+sizeof(i)+sizeof(l)+s.size()+sizeof(int));
+    VERIFY(b.size() == rpc_protocol::RPC_HEADER_SZ+sizeof(i)+sizeof(l)+sizeof(uint32_t)+s.size()+sizeof(uint32_t)+sizeof(uint32_t)+bin.size());

 
     unmarshall un(b, true);
     rpc_protocol::request_header rh1;
 
     unmarshall un(b, true);
     rpc_protocol::request_header rh1;

-    un.unpack_header(rh1);
+    un.read_header(rh1);

     VERIFY(memcmp(&rh,&rh1,sizeof(rh))==0);
     int i1;
     unsigned long long l1;
     string s1;
     VERIFY(memcmp(&rh,&rh1,sizeof(rh))==0);
     int i1;
     unsigned long long l1;
     string s1;

+    string bin1;
+    size_t sz1;

     un >> i1;
     un >> l1;
     un >> s1;
     un >> i1;
     un >> l1;
     un >> s1;

+    un >> sz1;
+    un >> bin1;

     VERIFY(un.okdone());
     VERIFY(un.okdone());

-    VERIFY(i1==i && l1==l && s1==s);
+    VERIFY(i1==i && l1==l && s1==s && sz1==sz && bin1==bin);

 }
 
 }
 

-void client1(size_t cl) {
+static void client1(size_t cl) {

     // test concurrency.
     size_t which_cl = cl % NUM_CL;
 
     for(int i = 0; i < 100; i++){
     // test concurrency.
     size_t which_cl = cl % NUM_CL;
 
     for(int i = 0; i < 100; i++){

-        unsigned long arg = (random() % 2000);
+        auto arg = std::uniform_int_distribution<unsigned long>(0,2000)(global->random_generator);

         string rep;
         int ret = clients[which_cl]->call(srv_protocol::bigrep, rep, arg);
         VERIFY(ret == 0);
         string rep;
         int ret = clients[which_cl]->call(srv_protocol::bigrep, rep, arg);
         VERIFY(ret == 0);

@@ -121,9 +135,9 @@ void client1(size_t cl) {
     // test rpc replies coming back not in the order of
     // the original calls -- i.e. does xid reply dispatch work.
     for(int i = 0; i < 100; i++){
     // test rpc replies coming back not in the order of
     // the original calls -- i.e. does xid reply dispatch work.
     for(int i = 0; i < 100; i++){

-        int which = (random() % 2);
-        int arg = (random() % 1000);
-        int rep;
+        bool which = std::bernoulli_distribution()(global->random_generator);
+        int arg = std::uniform_int_distribution<>(0,1000)(global->random_generator);
+        int rep = -1;

 
         auto start = steady_clock::now();
 
 
         auto start = steady_clock::now();
 

@@ -137,14 +151,14 @@ void client1(size_t cl) {
     }
 }
 
     }
 }
 

-void client2(size_t cl) {
+static void client2(size_t cl) {

     size_t which_cl = cl % NUM_CL;
 
     time_t t1;
     time(&t1);
 
     while(time(0) - t1 < 10){
     size_t which_cl = cl % NUM_CL;
 
     time_t t1;
     time(&t1);
 
     while(time(0) - t1 < 10){

-        unsigned long arg = (random() % 2000);
+        auto arg = std::uniform_int_distribution<unsigned long>(0,2000)(global->random_generator);

         string rep;
         int ret = clients[which_cl]->call(srv_protocol::bigrep, rep, arg);
         if ((unsigned long)rep.size()!=arg)
         string rep;
         int ret = clients[which_cl]->call(srv_protocol::bigrep, rep, arg);
         if ((unsigned long)rep.size()!=arg)

@@ -153,7 +167,7 @@ void client2(size_t cl) {
     }
 }
 
     }
 }
 

-void client3(void *xx) {
+static void client3(void *xx) {

     rpcc *c = (rpcc *) xx;
 
     for(int i = 0; i < 4; i++){
     rpcc *c = (rpcc *) xx;
 
     for(int i = 0; i < 4; i++){

@@ -163,7 +177,7 @@ void client3(void *xx) {
     }
 }
 
     }
 }
 

-void simple_tests(rpcc *c) {
+static void simple_tests(rpcc *c) {

     cout << "simple_tests" << endl;
     // an RPC call to procedure #22.
     // rpcc::call() looks at the argument types to decide how
     cout << "simple_tests" << endl;
     // an RPC call to procedure #22.
     // rpcc::call() looks at the argument types to decide how

@@ -199,6 +213,7 @@ void simple_tests(rpcc *c) {
     // huge RPC
     string big(1000000, 'x');
     intret = c->call(srv_protocol::_22, rep, big, (string)"z");
     // huge RPC
     string big(1000000, 'x');
     intret = c->call(srv_protocol::_22, rep, big, (string)"z");

+    VERIFY(intret == 0);

     VERIFY(rep.size() == 1000001);
     cout << "   -- huge 1M rpc request .. ok" << endl;
 
     VERIFY(rep.size() == 1000001);
     cout << "   -- huge 1M rpc request .. ok" << endl;
 

@@ -213,7 +228,7 @@ void simple_tests(rpcc *c) {
     cout << "simple_tests OK" << endl;
 }
 
     cout << "simple_tests OK" << endl;
 }
 

-void concurrent_test(size_t nt) {
+static void concurrent_test(size_t nt) {

     // create threads that make lots of calls in parallel,
     // to test thread synchronization for concurrent calls
     // and dispatches.
     // create threads that make lots of calls in parallel,
     // to test thread synchronization for concurrent calls
     // and dispatches.

@@ -230,7 +245,7 @@ void concurrent_test(size_t nt) {
     cout << " OK" << endl;
 }
 
     cout << " OK" << endl;
 }
 

-void lossy_test() {
+static void lossy_test() {

     cout << "start lossy_test ...";
     VERIFY(setenv("RPC_LOSSY", "5", 1) == 0);
 
     cout << "start lossy_test ...";
     VERIFY(setenv("RPC_LOSSY", "5", 1) == 0);
 

@@ -241,7 +256,7 @@ void lossy_test() {
 
     for (int i = 0; i < NUM_CL; i++) {
         delete clients[i];
 
     for (int i = 0; i < NUM_CL; i++) {
         delete clients[i];

-        clients[i] = new rpcc(dst);
+        clients[i] = new rpcc(*dst);

         VERIFY(clients[i]->bind()==0);
     }
 
         VERIFY(clients[i]->bind()==0);
     }
 

@@ -259,7 +274,7 @@ void lossy_test() {
     VERIFY(setenv("RPC_LOSSY", "0", 1) == 0);
 }
 
     VERIFY(setenv("RPC_LOSSY", "0", 1) == 0);
 }
 

-void failure_test() {
+static void failure_test() {

     rpcc *client1;
     rpcc *client = clients[0];
 
     rpcc *client1;
     rpcc *client = clients[0];
 

@@ -267,7 +282,7 @@ void failure_test() {
 
     delete server;
 
 
     delete server;
 

-    client1 = new rpcc(dst);
+    client1 = new rpcc(*dst);

     VERIFY (client1->bind(milliseconds(3000)) < 0);
     cout << "   -- create new client and try to bind to failed server .. failed ok" << endl;
 
     VERIFY (client1->bind(milliseconds(3000)) < 0);
     cout << "   -- create new client and try to bind to failed server .. failed ok" << endl;
 

@@ -282,7 +297,7 @@ void failure_test() {
 
     delete client;
 
 
     delete client;
 

-    clients[0] = client = new rpcc(dst);
+    clients[0] = client = new rpcc(*dst);

     VERIFY (client->bind() >= 0);
     VERIFY (client->bind() < 0);
 
     VERIFY (client->bind() >= 0);
     VERIFY (client->bind() < 0);
 

@@ -310,7 +325,7 @@ void failure_test() {
     delete client;
 
     startserver();
     delete client;
 
     startserver();

-    clients[0] = client = new rpcc(dst);
+    clients[0] = client = new rpcc(*dst);

     VERIFY (client->bind() >= 0);
     cout << "   -- delete existing rpc client and server, create replacements.. ok" << endl;
 
     VERIFY (client->bind() >= 0);
     cout << "   -- delete existing rpc client and server, create replacements.. ok" << endl;
 

@@ -328,6 +343,7 @@ void failure_test() {
 }
 
 int main(int argc, char *argv[]) {
 }
 
 int main(int argc, char *argv[]) {

+    global = new t4_state('r');

 
     setvbuf(stdout, NULL, _IONBF, 0);
     setvbuf(stderr, NULL, _IONBF, 0);
 
     setvbuf(stdout, NULL, _IONBF, 0);
     setvbuf(stderr, NULL, _IONBF, 0);

@@ -336,7 +352,6 @@ int main(int argc, char *argv[]) {
     bool isclient = false;
     bool isserver = false;
 
     bool isclient = false;
     bool isserver = false;
 

-    srandom((uint32_t)getpid());

     port = 20000 + (getpid() % 10000);
 
     int ch = 0;
     port = 20000 + (getpid() % 10000);
 
     int ch = 0;

@@ -367,8 +382,8 @@ int main(int argc, char *argv[]) {
     }
 
     if (debug_level > 0) {
     }
 
     if (debug_level > 0) {

-        DEBUG_LEVEL = debug_level;
-        IF_LEVEL(1) LOG_NONMEMBER("DEBUG LEVEL: " << debug_level);
+        global->DEBUG_LEVEL = debug_level;
+        IF_LEVEL(1) LOG_NONMEMBER << "DEBUG LEVEL: " << debug_level;

     }
 
     testmarshall();
     }
 
     testmarshall();

@@ -380,7 +395,7 @@ int main(int argc, char *argv[]) {
 
     if (isclient) {
         // server's address.
 
     if (isclient) {
         // server's address.

-        dst = "127.0.0.1:" + to_string(port);
+        dst = new string("127.0.0.1:" + std::to_string(port));

 
 
         // start the client.  bind it to the server.
 
 
         // start the client.  bind it to the server.

@@ -389,7 +404,7 @@ int main(int argc, char *argv[]) {
         // be only one rpcc per process. you probably need one
         // rpcc per server.
         for (int i = 0; i < NUM_CL; i++) {
         // be only one rpcc per process. you probably need one
         // rpcc per server.
         for (int i = 0; i < NUM_CL; i++) {

-            clients[i] = new rpcc(dst);
+            clients[i] = new rpcc(*dst);

             VERIFY (clients[i]->bind() == 0);
         }
 
             VERIFY (clients[i]->bind() == 0);
         }
 

@@ -406,5 +421,5 @@ int main(int argc, char *argv[]) {
     }
 
     while (1)
     }
 
     while (1)

-        usleep(100000);
+        std::this_thread::sleep_for(milliseconds(100));

 }
 }