So many changes. Broken.

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

diff --git a/rpc/rpc.cc b/rpc/rpc.cc
index 8f868f5..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>

@@ -106,7 +106,7 @@ shared_ptr<rpcc> rpcc::bind_cached(const string & destination) {
     lock cl = lock(client->bind_m_);
     if (!client->bind_done_) {
         LOG_NONMEMBER << "bind(\"" << destination << "\")";
     lock cl = lock(client->bind_m_);
     if (!client->bind_done_) {
         LOG_NONMEMBER << "bind(\"" << destination << "\")";

-        int ret = client->bind(milliseconds(1000));
+        int ret = client->bind(1000ms);

         if (ret < 0) {
             LOG_NONMEMBER << "bind failure! " << destination << " " << ret;
             client.reset();
         if (ret < 0) {
             LOG_NONMEMBER << "bind failure! " << destination << " " << ret;
             client.reset();

@@ -149,7 +149,7 @@ int rpcc::call_marshalled(const rpc_protocol::proc_t & proc, milliseconds to, st
 
     caller ca(0, &rep);
     xid_t xid_rep;
 
     caller ca(0, &rep);
     xid_t xid_rep;

-    marshall datagram = req;
+    string datagram;

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

@@ -164,9 +164,9 @@ int rpcc::call_marshalled(const rpc_protocol::proc_t & proc, milliseconds to, st
         ca.xid = xid_++;
         calls_[ca.xid] = &ca;
 
         ca.xid = xid_++;
         calls_[ca.xid] = &ca;
 

-        datagram.write_header(rpc_protocol::request_header{
+        datagram = marshall::datagram(rpc_protocol::request_header{

                 ca.xid, proc.id, clt_nonce_, srv_nonce_, xid_rep_window_.front()
                 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();
     }
 

@@ -272,11 +272,9 @@ void rpcc::get_latest_connection(shared_ptr<connection> & ch) {
 // 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) {
 // 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) {

-    unmarshall rep(b, true);

     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;
     }

@@ -354,21 +352,20 @@ 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?
     rpc_protocol::reply_header rh{h.xid,0};
 
     // is client sending to an old instance of server?

@@ -376,8 +373,7 @@ void rpcs::dispatch(shared_ptr<connection> c, const string & buf) {
         IF_LEVEL(2) LOG << "rpc for an old server instance " << h.srv_nonce
                         << " (current " << nonce_ << ") proc " << std::hex << proc;
         rh.ret = rpc_protocol::oldsrv_failure;
         IF_LEVEL(2) LOG << "rpc for an old server instance " << h.srv_nonce
                         << " (current " << nonce_ << ") proc " << std::hex << proc;
         rh.ret = rpc_protocol::oldsrv_failure;

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

         return;
     }
 
         return;
     }
 

@@ -418,17 +414,19 @@ void rpcs::dispatch(shared_ptr<connection> c, const string & buf) {
 
     switch (check_duplicate_and_update(h.clt_nonce, h.xid, h.xid_rep, stored_reply)) {
         case NEW: // new request
 
     switch (check_duplicate_and_update(h.clt_nonce, h.xid, h.xid_rep, stored_reply)) {
         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);
-            stored_reply = rep;
+                stored_reply = marshall::datagram(rh, rep);
+            }

 
             IF_LEVEL(2) LOG << "sending and saving reply of size " << stored_reply.size() << " for rpc "
                             << h.xid << ", proc " << std::hex << proc << " ret " << std::dec
 
             IF_LEVEL(2) LOG << "sending and saving reply of size " << stored_reply.size() << " for rpc "
                             << h.xid << ", proc " << std::hex << proc << " ret " << std::dec

@@ -453,8 +451,7 @@ void rpcs::dispatch(shared_ptr<connection> c, const string & buf) {
         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;
         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;
     }
 }