#include <getopt.h>
#include <unistd.h>
#include <string.h>
+#include "threaded_log.h"
#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 string * dst; //server's ip address
static 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
// from multiple classes.
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 {
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.
// 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;
}
-int srv::handle_fast(int &r, const int a) {
+int srv::handle_fast(int & r, const int a) {
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) {
+ usleep(std::uniform_int_distribution<useconds_t>(0,500)(global->random_generator));
r = a + 2;
return 0;
}
-int srv::handle_bigrep(string &r, const size_t len) {
+int srv::handle_bigrep(string & r, const size_t len) {
r = string(len, 'x');
return 0;
}
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->start();
}
-void testmarshall() {
+static void testmarshall() {
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;
size_t sz = 101010101;
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 << sz;
+ m << bin;
string b = m;
- VERIFY(b.size() == rpc_protocol::RPC_HEADER_SZ+sizeof(i)+sizeof(l)+s.size()+sizeof(int)+sizeof(uint32_t));
+ 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;
- un.unpack_header(rh1);
+ un.read_header(rh1);
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 >> sz1;
+ un >> bin1;
VERIFY(un.okdone());
- VERIFY(i1==i && l1==l && s1==s && sz1==sz);
+ 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++){
- 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);
// 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();
}
}
-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){
- 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)
}
}
-void client3(void *xx) {
+static void client3(void *xx) {
rpcc *c = (rpcc *) xx;
for(int i = 0; i < 4; i++){
}
}
-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 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.
cout << " OK" << endl;
}
-void lossy_test() {
+static void lossy_test() {
cout << "start lossy_test ...";
VERIFY(setenv("RPC_LOSSY", "5", 1) == 0);
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(setenv("RPC_LOSSY", "0", 1) == 0);
}
-void failure_test() {
+static void failure_test() {
rpcc *client1;
rpcc *client = clients[0];
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;
delete client;
- clients[0] = client = new rpcc(dst);
+ clients[0] = client = new rpcc(*dst);
VERIFY (client->bind() >= 0);
VERIFY (client->bind() < 0);
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;
}
int main(int argc, char *argv[]) {
+ global = new t4_state('r');
setvbuf(stdout, NULL, _IONBF, 0);
setvbuf(stderr, NULL, _IONBF, 0);
bool isclient = false;
bool isserver = false;
- srandom((uint32_t)getpid());
port = 20000 + (getpid() % 10000);
int ch = 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();
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.
// 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);
}
projects / invirt/third/libt4.git / blobdiff