Use defended_call for online/suspect

rescrv · May 25, 2015 · 8431b25 · 8431b25
1 parent 5866411
commit 8431b25
Show file tree

Hide file tree

Showing 5 changed files with 54 additions and 11 deletions.
diff --git a/common/coordinator_link.cc b/common/coordinator_link.cc
@@ -95,6 +95,17 @@ coordinator_link :: rpc(const char* func,
                                  REPLICANT_CALL_ROBUST, status, output, output_sz);
 }
 
+int64_t
+coordinator_link :: rpc_defended(const char* enter_func,
+                                 const char* enter_data, size_t enter_data_sz,
+                                 const char* exit_func,
+                                 const char* exit_data, size_t exit_data_sz,
+                                 replicant_returncode* status)
+{
+    return replicant_client_defended_call(m_repl, "hyperdex", enter_func, enter_data, enter_data_sz,
+                                          exit_func, exit_data, exit_data_sz, status);
+}
+
 int64_t
 coordinator_link :: backup(replicant_returncode* status,
                            char** output, size_t* output_sz)

diff --git a/common/coordinator_link.h b/common/coordinator_link.h
@@ -65,6 +65,11 @@ class coordinator_link
                     const char* data, size_t data_sz,
                     replicant_returncode* status,
                     char** output, size_t* output_sz);
+        int64_t rpc_defended(const char* enter_func,
+                             const char* enter_data, size_t enter_data_sz,
+                             const char* exit_func,
+                             const char* exit_data, size_t exit_data_sz,
+                             replicant_returncode* status);
         int64_t backup(replicant_returncode* status,
                        char** output, size_t* output_sz);
         int64_t wait(const char* cond, uint64_t state,

diff --git a/coordinator/coordinator.cc b/coordinator/coordinator.cc
@@ -359,12 +359,6 @@ coordinator :: server_online(rsm_context* ctx,
         generate_next_configuration(ctx);
     }
 
-#if 0
-    char buf[sizeof(uint64_t)];
-    e::pack64be(sid.get(), buf);
-    uint64_t client = replicant_state_machine_get_client(ctx);
-    replicant_state_machine_suspect(ctx, client, "server_suspect",  buf, sizeof(uint64_t));
-#endif
     return generate_response(ctx, COORD_SUCCESS);
 }
 

diff --git a/daemon/coordinator_link_wrapper.cc b/daemon/coordinator_link_wrapper.cc
@@ -771,13 +771,16 @@ coordinator_link_wrapper :: ensure_available()
     }
 
     size_t sz = sizeof(uint64_t) + pack_size(m_daemon->m_bind_to);
-    std::auto_ptr<e::buffer> buf(e::buffer::create(sz));
-    buf->pack() << m_daemon->m_us << m_daemon->m_bind_to;
+    std::auto_ptr<e::buffer> enter_buf(e::buffer::create(sz));
+    enter_buf->pack() << m_daemon->m_us << m_daemon->m_bind_to;
+    char exit_buf[sizeof(uint64_t)];
+    e::pack64be(m_daemon->m_us.get(), exit_buf);
     e::intrusive_ptr<coord_rpc> rpc = new coord_rpc_available();
     rpc->msg << "server online";
-    m_online_id = make_rpc_nosync("server_online",
-                                  reinterpret_cast<const char*>(buf->data()), buf->size(),
-                                  rpc);
+    m_online_id = make_rpc_defended("server_online",
+                                    enter_buf->cdata(), enter_buf->size(),
+                                    "server_suspect", exit_buf, sizeof(uint64_t),
+                                    rpc);
 }
 
 class coordinator_link_wrapper::coord_rpc_config_ack : public coord_rpc
@@ -1056,6 +1059,31 @@ coordinator_link_wrapper :: make_rpc_nosync(const char* func,
     return id;
 }
 
+int64_t
+coordinator_link_wrapper :: make_rpc_defended(const char* enter_func,
+                                              const char* enter_data, size_t enter_data_sz,
+                                              const char* exit_func,
+                                              const char* exit_data, size_t exit_data_sz,
+                                              e::intrusive_ptr<coord_rpc> rpc)
+{
+    int64_t id = m_coord->rpc_defended(enter_func, enter_data, enter_data_sz,
+                                       exit_func, exit_data, exit_data_sz,
+                                       &rpc->status);
+
+    if (id < 0)
+    {
+        LOG(ERROR) << "coordinator error: " << rpc->msg.str()
+                   << ": " << m_coord->error_message()
+                   << " @ " << m_coord->error_location();
+    }
+    else
+    {
+        m_rpcs.insert(std::make_pair(id, rpc));
+    }
+
+    return id;
+}
+
 int64_t
 coordinator_link_wrapper :: wait_nosync(const char* cond, uint64_t state,
                                         e::intrusive_ptr<coord_rpc> rpc)

diff --git a/daemon/coordinator_link_wrapper.h b/daemon/coordinator_link_wrapper.h
@@ -121,6 +121,11 @@ class coordinator_link_wrapper
         int64_t make_rpc_nosync(const char* func,
                                 const char* data, size_t data_sz,
                                 e::intrusive_ptr<coord_rpc> rpc);
+        int64_t make_rpc_defended(const char* enter_func,
+                                  const char* enter_data, size_t enter_data_sz,
+                                  const char* exit_func,
+                                  const char* exit_data, size_t exit_data_sz,
+                                  e::intrusive_ptr<coord_rpc> rpc);
         int64_t wait_nosync(const char* cond, uint64_t state,
                             e::intrusive_ptr<coord_rpc> rpc);