Simplify the daemon coordinator link

rescrv · Jul 3, 2015 · 188ca2c · 188ca2c
1 parent 809a644
commit 188ca2c
Show file tree

Hide file tree

Showing 14 changed files with 740 additions and 1,408 deletions.
diff --git a/Makefile.am b/Makefile.am
@@ -186,7 +186,6 @@ EXTRA_DIST += initscripts/sysv/init.d/hyperdex-daemon
 noinst_HEADERS += daemon/auth.h
 noinst_HEADERS += daemon/background_thread.h
 noinst_HEADERS += daemon/communication.h
-noinst_HEADERS += daemon/coordinator_link_wrapper.h
 noinst_HEADERS += daemon/daemon.h
 noinst_HEADERS += daemon/datalayer_checkpointer_thread.h
 noinst_HEADERS += daemon/datalayer_encodings.h
@@ -231,7 +230,6 @@ hyperdex_daemon_SOURCES += common/attribute.cc
 hyperdex_daemon_SOURCES += common/attribute_check.cc
 hyperdex_daemon_SOURCES += common/auth_wallet.cc
 hyperdex_daemon_SOURCES += common/configuration.cc
-hyperdex_daemon_SOURCES += common/coordinator_link.cc
 hyperdex_daemon_SOURCES += common/coordinator_returncode.cc
 hyperdex_daemon_SOURCES += common/datatype_document.cc
 hyperdex_daemon_SOURCES += common/datatype_float.cc
@@ -265,7 +263,7 @@ hyperdex_daemon_SOURCES += cityhash/city.cc
 hyperdex_daemon_SOURCES += daemon/auth.cc
 hyperdex_daemon_SOURCES += daemon/background_thread.cc
 hyperdex_daemon_SOURCES += daemon/communication.cc
-hyperdex_daemon_SOURCES += daemon/coordinator_link_wrapper.cc
+hyperdex_daemon_SOURCES += daemon/coordinator_link.cc
 hyperdex_daemon_SOURCES += daemon/daemon.cc
 hyperdex_daemon_SOURCES += daemon/datalayer.cc
 hyperdex_daemon_SOURCES += daemon/datalayer_checkpointer_thread.cc
@@ -304,6 +302,7 @@ hyperdex_daemon_LDADD += $(REPLICANT_LIBS)
 hyperdex_daemon_LDADD += $(HYPERLEVELDB_LIBS)
 hyperdex_daemon_LDADD += $(BUSYBEE_LIBS)
 hyperdex_daemon_LDADD += $(E_LIBS)
+hyperdex_daemon_LDADD += $(PO6_LIBS)
 hyperdex_daemon_LDADD += $(POPT_LIBS) ${GLOG_LIBS} $(LRT_LDFLAGS) -lpthread
 man/hyperdex-daemon.1: man/hyperdex-daemon.1.h2m daemon/main.cc | hyperdex-daemon$(EXEEXT)
 	$(help2man_verbose)help2man $(HELP2MAN_FLAGS) --section 1 --output $@ --include $< ${abs_top_builddir}/hyperdex-daemon$(EXEEXT)

diff --git a/coordinator/coordinator.cc b/coordinator/coordinator.cc
@@ -939,8 +939,8 @@ coordinator :: config_stable(rsm_context* ctx,
 void
 coordinator :: checkpoint(rsm_context* ctx)
 {
-    rsm_cond_broadcast(ctx, "checkp");
     ++m_checkpoint;
+    broadcast_checkpoint_information(ctx);
     rsm_log(ctx, "establishing checkpoint %" PRIu64 "\n", m_checkpoint);
     assert(m_checkpoint_stable_through <= m_checkpoint);
     std::vector<server_id> sids;
@@ -1883,6 +1883,7 @@ coordinator :: generate_next_configuration(rsm_context* ctx)
     check_stable_condition(ctx);
     generate_cached_configuration(ctx);
     rsm_cond_broadcast_data(ctx, "config", m_latest_config->cdata(), m_latest_config->size());
+    broadcast_checkpoint_information(ctx);
 }
 
 void
@@ -2059,8 +2060,8 @@ coordinator :: check_checkpoint_stable_condition(rsm_context* ctx, bool reissue)
     while (m_checkpoint_stable_through < m_checkpoint_stable_barrier.min_version())
     {
         stabilized = true;
-        rsm_cond_broadcast(ctx, "checkps");
         ++m_checkpoint_stable_through;
+        broadcast_checkpoint_information(ctx);
     }
 
     bool gc = false;
@@ -2088,8 +2089,8 @@ coordinator :: check_checkpoint_stable_condition(rsm_context* ctx, bool reissue)
     while (m_checkpoint_gc_through + outstanding_checkpoints < m_checkpoint_stable_barrier.min_version())
     {
         gc = true;
-        rsm_cond_broadcast(ctx, "checkpgc");
         ++m_checkpoint_gc_through;
+        broadcast_checkpoint_information(ctx);
     }
 
     if (gc && m_checkpoint_gc_through > 0)
@@ -2132,3 +2133,14 @@ coordinator :: check_checkpoint_stable_condition(rsm_context* ctx, bool reissue)
         generate_next_configuration(ctx);
     }
 }
+
+void
+coordinator :: broadcast_checkpoint_information(rsm_context* ctx)
+{
+    std::string input;
+    e::packer(&input) << m_version
+                      << m_checkpoint
+                      << m_checkpoint_stable_through
+                      << m_checkpoint_gc_through;
+    rsm_cond_broadcast_data(ctx, "checkpoint", input.data(), input.size());
+}
diff --git a/coordinator/coordinator.h b/coordinator/coordinator.h
@@ -193,6 +193,7 @@ class coordinator
         void regions_in_space(space_ptr s, std::vector<region_id>* rids);
         // checkpoints
         void check_checkpoint_stable_condition(rsm_context* ctx, bool reissue);
+        void broadcast_checkpoint_information(rsm_context* ctx);
 
     private:
         // meta state

diff --git a/coordinator/transitions.cc b/coordinator/transitions.cc
@@ -72,9 +72,7 @@ hyperdex_coordinator_create(struct rsm_context* ctx)
     rsm_cond_create(ctx, "config");
     rsm_cond_create(ctx, "ack");
     rsm_cond_create(ctx, "stable");
-    rsm_cond_create(ctx, "checkp");
-    rsm_cond_create(ctx, "checkps");
-    rsm_cond_create(ctx, "checkpgc");
+    rsm_cond_create(ctx, "checkpoint");
     return new (std::nothrow) coordinator();
 }
 

diff --git a/daemon/communication.cc b/daemon/communication.cc
@@ -517,10 +517,6 @@ communication :: handle_disruption(uint64_t id)
 {
     if (m_daemon->m_config.get_address(server_id(id)) != po6::net::location())
     {
-        m_daemon->m_coord.report_tcp_disconnect(server_id(id));
-        // XXX If the above line changes, then we need to sometimes tell
-        // the transfer manager to resend all that is unacked  Right now, it
-        // will cause a deadlock.
-        // m_daemon->m_stm.retransmit(server_id(id));
+        m_daemon->m_coord->report_tcp_disconnect(m_daemon->m_config.version(), server_id(id));
     }
 }