Merge pull request #21 from neuront/master

cluster status info

common.hpp

@@ -5,7 +5,7 @@
 
 #include "utils/typetraits.hpp"
 
-#define VERSION "0.7.4-2015-10-22"
+#define VERSION "0.7.5-2015-11-12"
 
 namespace cerb {
 

core/globals.cpp

@@ -8,6 +8,7 @@ thread_local cerb::msize_t cerb_global::allocated_buffer(0);
 
 static std::mutex remote_addrs_mutex;
 static std::set<util::Address> remote_addrs;
+static bool cluster_ok = false;
 
 void cerb_global::set_remotes(std::set<util::Address> remotes)
 {
@@ -32,3 +33,13 @@ bool cerb_global::cluster_req_full_cov()
 {
     return ::req_full_cov;
 }
+
+void cerb_global::set_cluster_ok(bool ok)
+{
+    ::cluster_ok = ok;
+}
+
+bool cerb_global::cluster_ok()
+{
+    return ::cluster_ok;
+}

core/globals.hpp

@@ -20,6 +20,9 @@ namespace cerb_global {
     void set_cluster_req_full_cov(bool c);
     bool cluster_req_full_cov();
 
+    void set_cluster_ok(bool ok);
+    bool cluster_ok();
+
 }
 
 #endif /* __CERBERUS_GLOBALS_HPP__ */

core/proxy.cpp

@@ -128,6 +128,7 @@ void Proxy::_set_slot_map(std::vector<RedisNode> const& map,
     _server_map.replace_map(map, this);
     _slot_map_expired = false;
     cerb_global::set_remotes(std::move(remotes));
+    cerb_global::set_cluster_ok(true);
     LOG(INFO) << "Slot map updated";
     LOG(DEBUG) << "Retry MOVED or ASK: " << this->_retrying_commands.size();
     if (this->_retrying_commands.empty()) {
@@ -178,6 +179,7 @@ void Proxy::_update_slot_map_failed()
     }
 
     this->_move_closed_slot_updaters();
+    cerb_global::set_cluster_ok(false);
     LOG(DEBUG) << "Failed to retrieve slot map, discard all commands.";
     _server_map.clear();
     std::vector<util::sref<DataCommand>> cmds(std::move(this->_retrying_commands));

core/server.cpp

@@ -26,11 +26,11 @@ void Server::on_events(int events)
         try {
             this->_recv_from();
         } catch (BadRedisMessage& e) {
-            LOG(FATAL) << "Receive bad message from server " << this->str()
+            LOG(ERROR) << "Receive bad message from server " << this->str()
                        << " because: " << e.what()
                        << " dump buffer (before close): "
                        << this->_buffer.to_string();
-            exit(1);
+            return this->close_conn();
         }
     }
     this->_push_to_buffer_set();
@@ -70,8 +70,7 @@ void Server::_recv_from()
             LOG(ERROR) << "::: " << rsp->get_buffer().to_string();
         }
         LOG(ERROR) << "Rest buffer: " << this->_buffer.to_string();
-        LOG(FATAL) << "Exit";
-        exit(1);
+        return this->close_conn();
     }
     LOG(DEBUG) << "+responses size: " << responses.size();
     LOG(DEBUG) << "+rest buffer: " << this->_buffer.size();

core/stats.cpp

@@ -39,6 +39,7 @@ std::string cerb::stats_all()
     return util::join("", {
         "version:" VERSION
         "\nthreads:", util::str(msize_t(cerb_global::all_threads.size())),
+        "\ncluster_ok:", cerb_global::cluster_ok() ? "1" : "0",
         "\nread_slave:", ::read_slave ? "1" : "0",
         "\nclients_count:", util::join(",", clients_counts),
         "\nlong_connections_count:", util::join(",", long_conns_counts),