So many changes. Broken.

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

diff --git a/rpc/rpc.cc b/rpc/rpc.cc
index de33675..e33b25e 100644 (file)
--- a/rpc/rpc.cc
+++ b/rpc/rpc.cc
@@ -52,7 +52,7 @@
 // x exited worker threads).
 //
 
 // x exited worker threads).
 //
 

-#include "rpc.h"
+#include "include/rpc/rpc.h"

 
 #include <arpa/inet.h>
 #include <netinet/tcp.h>
 
 #include <arpa/inet.h>
 #include <netinet/tcp.h>

@@ -63,17 +63,11 @@
 using std::list;
 using namespace std::chrono;
 
 using std::list;
 using namespace std::chrono;
 

-inline void set_rand_seed() {
-    auto now = time_point_cast<nanoseconds>(steady_clock::now());
-    srandom((uint32_t)now.time_since_epoch().count()^(uint32_t)getpid());
-}
-

 static sockaddr_in make_sockaddr(const string & hostandport);
 
 rpcc::rpcc(const string & d) : dst_(make_sockaddr(d))
 {
 static sockaddr_in make_sockaddr(const string & hostandport);
 
 rpcc::rpcc(const string & d) : dst_(make_sockaddr(d))
 {

-    set_rand_seed();
-    clt_nonce_ = (nonce_t)random();
+    clt_nonce_ = (nonce_t)global->random_generator();

 
     char *loss_env = getenv("RPC_LOSSY");
     if (loss_env)
 
     char *loss_env = getenv("RPC_LOSSY");
     if (loss_env)

@@ -85,7 +79,10 @@ rpcc::rpcc(const string & d) : dst_(make_sockaddr(d))
 // IMPORTANT: destruction should happen only when no external threads
 // are blocked inside rpcc or will use rpcc in the future
 rpcc::~rpcc() {
 // IMPORTANT: destruction should happen only when no external threads
 // are blocked inside rpcc or will use rpcc in the future
 rpcc::~rpcc() {

-    cancel();
+    lock ml(m_);
+    cancel(ml);
+
+    lock cl(chan_m_);

     IF_LEVEL(2) LOG << "delete nonce " << clt_nonce_ << " chan " << (chan_?(int)chan_->fd:-1);
     chan_.reset();
     VERIFY(calls_.size() == 0);
     IF_LEVEL(2) LOG << "delete nonce " << clt_nonce_ << " chan " << (chan_?(int)chan_->fd:-1);
     chan_.reset();
     VERIFY(calls_.size() == 0);

@@ -104,9 +101,29 @@ int rpcc::bind(milliseconds to) {
     return ret;
 }
 
     return ret;
 }
 

+shared_ptr<rpcc> rpcc::bind_cached(const string & destination) {
+    auto client = global->get_handle(destination);
+    lock cl = lock(client->bind_m_);
+    if (!client->bind_done_) {
+        LOG_NONMEMBER << "bind(\"" << destination << "\")";
+        int ret = client->bind(1000ms);
+        if (ret < 0) {
+            LOG_NONMEMBER << "bind failure! " << destination << " " << ret;
+            client.reset();
+        } else {
+            LOG_NONMEMBER << "bind succeeded " << destination;
+        }
+    }
+    return client;
+}
+
+void rpcc::unbind_cached(const string & destination) {
+    global->erase_handle(destination);
+}
+

 // Cancel all outstanding calls
 // Cancel all outstanding calls

-void rpcc::cancel(void) {
-    lock ml(m_);
+void rpcc::cancel(lock & m_lock) {
+    VERIFY(m_lock);

     if (calls_.size()) {
         LOG << "force callers to fail";
         for (auto & p : calls_) {
     if (calls_.size()) {
         LOG << "force callers to fail";
         for (auto & p : calls_) {

@@ -122,20 +139,21 @@ void rpcc::cancel(void) {
 
         destroy_wait_ = true;
         while (calls_.size () > 0)
 
         destroy_wait_ = true;
         while (calls_.size () > 0)

-            destroy_wait_c_.wait(ml);
+            destroy_wait_c_.wait(m_lock);

 
         LOG << "done";
     }
 }
 
 
         LOG << "done";
     }
 }
 

-int rpcc::call1(proc_id_t proc, milliseconds to, string & rep, marshall & req) {
+int rpcc::call_marshalled(const rpc_protocol::proc_t & proc, milliseconds to, string & rep, const marshall & req) {

 
     caller ca(0, &rep);
     xid_t xid_rep;
 
     caller ca(0, &rep);
     xid_t xid_rep;

+    string datagram;

     {
         lock ml(m_);
 
     {
         lock ml(m_);
 

-        if ((proc != rpc_protocol::bind.id && !bind_done_) || (proc == rpc_protocol::bind.id && bind_done_)) {
+        if ((proc.id != rpc_protocol::bind.id && !bind_done_) || (proc.id == rpc_protocol::bind.id && bind_done_)) {

             IF_LEVEL(1) LOG << "rpcc has not been bound to dst or binding twice";
             return rpc_protocol::bind_failure;
         }
             IF_LEVEL(1) LOG << "rpcc has not been bound to dst or binding twice";
             return rpc_protocol::bind_failure;
         }

@@ -146,9 +164,9 @@ int rpcc::call1(proc_id_t proc, milliseconds to, string & rep, marshall & req) {
         ca.xid = xid_++;
         calls_[ca.xid] = &ca;
 
         ca.xid = xid_++;
         calls_[ca.xid] = &ca;
 

-        req.write_header(rpc_protocol::request_header{
-                ca.xid, proc, clt_nonce_, srv_nonce_, xid_rep_window_.front()
-                });
+        datagram = marshall::datagram(rpc_protocol::request_header{
+                ca.xid, proc.id, clt_nonce_, srv_nonce_, xid_rep_window_.front()
+            }, req);

         xid_rep = xid_rep_window_.front();
     }
 
         xid_rep = xid_rep_window_.front();
     }
 

@@ -173,10 +191,10 @@ int rpcc::call1(proc_id_t proc, milliseconds to, string & rep, marshall & req) {
                     }
                     if (forgot.isvalid())
                         ch->send(forgot.buf);
                     }
                     if (forgot.isvalid())
                         ch->send(forgot.buf);

-                    ch->send(req);
+                    ch->send(datagram);

                 }
                 else IF_LEVEL(1) LOG << "not reachable";
                 }
                 else IF_LEVEL(1) LOG << "not reachable";

-                IF_LEVEL(2) LOG << clt_nonce_ << " just sent req proc " << std::hex << proc
+                IF_LEVEL(2) LOG << clt_nonce_ << " just sent req proc " << std::hex << proc.id

                                 << " xid " << std::dec << ca.xid << " clt_nonce " << clt_nonce_;
             }
             transmit = false; // only send once on a given channel
                                 << " xid " << std::dec << ca.xid << " clt_nonce " << clt_nonce_;
             }
             transmit = false; // only send once on a given channel

@@ -225,7 +243,7 @@ int rpcc::call1(proc_id_t proc, milliseconds to, string & rep, marshall & req) {
     {
         lock ml(m_);
         if (!dup_req_.isvalid()) {
     {
         lock ml(m_);
         if (!dup_req_.isvalid()) {

-            dup_req_.buf = req;
+            dup_req_.buf = datagram;

             dup_req_.xid = ca.xid;
         }
         if (xid_rep > xid_rep_done_)
             dup_req_.xid = ca.xid;
         }
         if (xid_rep > xid_rep_done_)

@@ -234,12 +252,12 @@ int rpcc::call1(proc_id_t proc, milliseconds to, string & rep, marshall & req) {
 
     lock cal(ca.m);
 
 
     lock cal(ca.m);
 

-    IF_LEVEL(2) LOG << clt_nonce_ << " call done for req proc " << std::hex << proc
+    IF_LEVEL(2) LOG << clt_nonce_ << " call done for req proc " << std::hex << proc.id

                     << " xid " << std::dec << ca.xid << " " << inet_ntoa(dst_.sin_addr) << ":"
                     << ntoh(dst_.sin_port) << " done? " << ca.done << " ret " << ca.intret;
 
     // destruction of req automatically frees its buffer
                     << " xid " << std::dec << ca.xid << " " << inet_ntoa(dst_.sin_addr) << ":"
                     << ntoh(dst_.sin_port) << " done? " << ca.done << " ret " << ca.intret;
 
     // destruction of req automatically frees its buffer

-    return (ca.done? ca.intret : rpc_protocol::timeout_failure);
+    return ca.done ? ca.intret : rpc_protocol::timeout_failure;

 }
 
 void rpcc::get_latest_connection(shared_ptr<connection> & ch) {
 }
 
 void rpcc::get_latest_connection(shared_ptr<connection> & ch) {

@@ -251,19 +269,12 @@ void rpcc::get_latest_connection(shared_ptr<connection> & ch) {
         ch = chan_;
 }
 
         ch = chan_;
 }
 

-// PollMgr's thread is being used to
-// make this upcall from connection object to rpcc.
-// this funtion must not block.
-//
-// this function keeps no reference for connection *c
-bool
-rpcc::got_pdu(const shared_ptr<connection> &, const string & b)
-{
-    unmarshall rep(b, true);
+// Runs in poll_mgr's thread as an upcall from the connection object to the
+// rpcc.  Does not call blocking RPC handlers.
+bool rpcc::got_pdu(const shared_ptr<connection> &, const string & b) {

     rpc_protocol::reply_header h;
     rpc_protocol::reply_header h;

-    rep.read_header(h);

 
 

-    if (!rep.ok()) {
+    if (!unmarshall::datagram(b, h)) {

         IF_LEVEL(1) LOG << "unmarshall header failed!!!";
         return true;
     }
         IF_LEVEL(1) LOG << "unmarshall header failed!!!";
         return true;
     }

@@ -314,8 +325,7 @@ compress:
 
 rpcs::rpcs(in_port_t p1) : port_(p1)
 {
 
 rpcs::rpcs(in_port_t p1) : port_(p1)
 {

-    set_rand_seed();
-    nonce_ = (nonce_t)random();
+    nonce_ = (nonce_t)global->random_generator();

     IF_LEVEL(2) LOG << "created with nonce " << nonce_;
 
     reg(rpc_protocol::bind, &rpcs::rpcbind, this);
     IF_LEVEL(2) LOG << "created with nonce " << nonce_;
 
     reg(rpc_protocol::bind, &rpcs::rpcbind, this);

@@ -342,30 +352,28 @@ bool rpcs::got_pdu(const shared_ptr<connection> & c, const string & b) {
 }
 
 void rpcs::dispatch(shared_ptr<connection> c, const string & buf) {
 }
 
 void rpcs::dispatch(shared_ptr<connection> c, const string & buf) {

-    unmarshall req(buf, true);
-

     rpc_protocol::request_header h;
     rpc_protocol::request_header h;

-    req.read_header(h);
-    proc_id_t proc = h.proc;

 
 

-    if (!req.ok()) {
+    auto req = unmarshall::datagram(buf, h);
+
+    if (!req) {

         IF_LEVEL(1) LOG << "unmarshall header failed";
         return;
     }
 
         IF_LEVEL(1) LOG << "unmarshall header failed";
         return;
     }
 

+    proc_id_t proc = h.proc;
+

     IF_LEVEL(2) LOG << "rpc " << h.xid << " (proc " << std::hex << proc << ", last_rep "
                     << std::dec << h.xid_rep << ") from clt " << h.clt_nonce << " for srv instance " << h.srv_nonce;
 
     IF_LEVEL(2) LOG << "rpc " << h.xid << " (proc " << std::hex << proc << ", last_rep "
                     << std::dec << h.xid_rep << ") from clt " << h.clt_nonce << " for srv instance " << h.srv_nonce;
 

-    marshall rep;

     rpc_protocol::reply_header rh{h.xid,0};
 
     // is client sending to an old instance of server?
     if (h.srv_nonce != 0 && h.srv_nonce != nonce_) {
         IF_LEVEL(2) LOG << "rpc for an old server instance " << h.srv_nonce
     rpc_protocol::reply_header rh{h.xid,0};
 
     // is client sending to an old instance of server?
     if (h.srv_nonce != 0 && h.srv_nonce != nonce_) {
         IF_LEVEL(2) LOG << "rpc for an old server instance " << h.srv_nonce

-                        << " (current " << nonce_ << ") proc " << std::hex << h.proc;
+                        << " (current " << nonce_ << ") proc " << std::hex << proc;

         rh.ret = rpc_protocol::oldsrv_failure;
         rh.ret = rpc_protocol::oldsrv_failure;

-        rep.write_header(rh);
-        c->send(rep);
+        c->send(marshall::datagram(rh));

         return;
     }
 
         return;
     }
 

@@ -376,7 +384,6 @@ void rpcs::dispatch(shared_ptr<connection> c, const string & buf) {
         if (procs_.count(proc) < 1) {
             LOG << "unknown proc 0x" << std::hex << proc << " with h.srv_nonce=" << h.srv_nonce << ", my srv_nonce=" << nonce_;
             VERIFY(0);
         if (procs_.count(proc) < 1) {
             LOG << "unknown proc 0x" << std::hex << proc << " with h.srv_nonce=" << h.srv_nonce << ", my srv_nonce=" << nonce_;
             VERIFY(0);

-            return;

         }
 
         f = procs_[proc];
         }
 
         f = procs_[proc];

@@ -403,27 +410,29 @@ void rpcs::dispatch(shared_ptr<connection> c, const string & buf) {
             conns_[h.clt_nonce] = c;
     }
 
             conns_[h.clt_nonce] = c;
     }
 

-    string b1;
+    string stored_reply;

 
 

-    switch (check_duplicate_and_update(h.clt_nonce, h.xid, h.xid_rep, b1)) {
+    switch (check_duplicate_and_update(h.clt_nonce, h.xid, h.xid_rep, stored_reply)) {

         case NEW: // new request
         case NEW: // new request

-            rh.ret = (*f)(std::forward<unmarshall>(req), rep);
-            if (rh.ret == rpc_protocol::unmarshall_args_failure) {
-                LOG << "failed to unmarshall the arguments. You are "
-                    << "probably calling RPC 0x" << std::hex << proc << " with the wrong "
-                    << "types of arguments.";
-                VERIFY(0);
-            }
-            VERIFY(rh.ret >= 0);
+            {
+                marshall rep;
+                rh.ret = (*f)(std::forward<unmarshall>(req), rep);
+                if (rh.ret == rpc_protocol::unmarshall_args_failure) {
+                    LOG << "failed to unmarshall the arguments. You are "
+                        << "probably calling RPC 0x" << std::hex << proc << " with the wrong "
+                        << "types of arguments.";
+                    VERIFY(0);
+                }
+                VERIFY(rh.ret >= 0);

 
 

-            rep.write_header(rh);
-            b1 = rep;
+                stored_reply = marshall::datagram(rh, rep);
+            }

 
 

-            IF_LEVEL(2) LOG << "sending and saving reply of size " << b1.size() << " for rpc "
+            IF_LEVEL(2) LOG << "sending and saving reply of size " << stored_reply.size() << " for rpc "

                             << h.xid << ", proc " << std::hex << proc << " ret " << std::dec
                             << rh.ret << ", clt " << h.clt_nonce;
 
                             << h.xid << ", proc " << std::hex << proc << " ret " << std::dec
                             << rh.ret << ", clt " << h.clt_nonce;
 

-            add_reply(h.clt_nonce, h.xid, b1);
+            add_reply(h.clt_nonce, h.xid, stored_reply);

 
             // get the latest connection to the client
             {
 
             // get the latest connection to the client
             {

@@ -432,18 +441,17 @@ void rpcs::dispatch(shared_ptr<connection> c, const string & buf) {
                     c = conns_[h.clt_nonce];
             }
 
                     c = conns_[h.clt_nonce];
             }
 

-            c->send(rep);
+            c->send(stored_reply);

             break;
         case INPROGRESS: // server is working on this request
             break;
         case DONE: // duplicate and we still have the response
             break;
         case INPROGRESS: // server is working on this request
             break;
         case DONE: // duplicate and we still have the response

-            c->send(b1);
+            c->send(stored_reply);

             break;
         case FORGOTTEN: // very old request and we don't have the response anymore
             IF_LEVEL(2) LOG << "very old request " << h.xid << " from " << h.clt_nonce;
             rh.ret = rpc_protocol::atmostonce_failure;
             break;
         case FORGOTTEN: // very old request and we don't have the response anymore
             IF_LEVEL(2) LOG << "very old request " << h.xid << " from " << h.clt_nonce;
             rh.ret = rpc_protocol::atmostonce_failure;

-            rep.write_header(rh);
-            c->send(rep);
+            c->send(marshall::datagram(rh));

             break;
     }
 }
             break;
     }
 }