// x exited worker threads).
//
-#include "rpc.h"
+#include "include/rpc/rpc.h"
#include <arpa/inet.h>
#include <netinet/tcp.h>
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();
caller ca(0, &rep);
xid_t xid_rep;
- marshall datagram = req;
+ string datagram;
{
lock ml(m_);
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()
- });
+ }, req);
xid_rep = xid_rep_window_.front();
}
// 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;
- rep.read_header(h);
- if (!rep.ok()) {
+ if (!unmarshall::datagram(b, h)) {
IF_LEVEL(1) LOG << "unmarshall header failed!!!";
return true;
}
}
void rpcs::dispatch(shared_ptr<connection> c, const string & buf) {
- unmarshall req(buf, true);
-
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;
}
+ 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;
- marshall rep;
rpc_protocol::reply_header rh{h.xid,0};
// is client sending to an old instance of server?
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;
}
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
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;
}
}