/ - Diff - OpenNebula - OpenNebula Development pages

         // -------------------------------------------------------------------------
         RaftReplicaManager replica_manager;
         HeartBeatManager   heartbeat_manager;
         unsigned int commit;
         std::map<int, unsigned int> next;
-...
         // Internal Raft functions
         // -------------------------------------------------------------------------
     	/**
     	 *  Send the heartbeat to the followers.
     	 */
     	void send_heartbeat();
     	/**
     	 *  Request votes of followers
     	 */
         void request_vote();

         ReplicaThread * thread_factory(int follower_id);
     };
     class HeartBeatManager : public ReplicaManager
+    {
     public:
         HeartBeatManager():ReplicaManager(){};
         virtual ~HeartBeatManager(){};
     private:
         ReplicaThread * thread_factory(int follower_id);
     };
     #endif /*REPLICA_MANAGER_H_*/

         FedReplicaManager * frm;
     };
     // -----------------------------------------------------------------------------
     // Thread to send hearbeats to each follower
     // -----------------------------------------------------------------------------
     class HeartBeatThread : public ReplicaThread
+    {
     public:
         HeartBeatThread(int follower_id);
         virtual ~HeartBeatThread(){};
     private:
         /**
          *  Error statistics for follower
          */
         time_t last_error;
         int num_errors;
         /**
          * Specific logic for the replicate process
          */
         int replicate();
         /**
          * Pointers to other components
          */
         RaftManager * raftm;
     };
     #endif

     	replica_manager.add_replica_thread(follower_id);
     	heartbeat_manager.add_replica_thread(follower_id);
     	pthread_mutex_unlock(&mutex);
     };
-...
     	replica_manager.delete_replica_thread(follower_id);
     	heartbeat_manager.delete_replica_thread(follower_id);
     	pthread_mutex_unlock(&mutex);
     };
-...
+        }
         replica_manager.start_replica_threads(_follower_ids);
         heartbeat_manager.start_replica_threads(_follower_ids);
         pthread_mutex_unlock(&mutex);
-...
+        }
         replica_manager.stop_replica_threads();
         heartbeat_manager.stop_replica_threads();
         state = FOLLOWER;
-...
     		if ((sec < the_time.tv_sec) || (sec == the_time.tv_sec &&
     				nsec <= the_time.tv_nsec))
+    		{
                 pthread_mutex_unlock(&mutex);
     			send_heartbeat();
                 pthread_mutex_lock(&mutex);
     			heartbeat_manager.replicate();
                 clock_gettime(CLOCK_REALTIME, &last_heartbeat);
-...
     /* -------------------------------------------------------------------------- */
     /* -------------------------------------------------------------------------- */
     void RaftManager::send_heartbeat()
+    {
         std::map<int, std::string> _servers;
         std::map<int, std::string>::iterator it;
     	LogDBRecord lr;
     	bool success;
     	unsigned int fterm;
     	std::string error;
     	lr.index = 0;
     	lr.prev_index = 0;
     	lr.term = 0;
     	lr.prev_term = 0;
     	lr.sql = "";
     	lr.timestamp = 0;
         pthread_mutex_lock(&mutex);
         if ( state != LEADER )
+        {
             pthread_mutex_unlock(&mutex);
             return;
+        }
         _servers = servers;
         pthread_mutex_unlock(&mutex);
         for (it = _servers.begin(); it != _servers.end() ; ++it )
+        {
             if ( it->first == server_id )
+            {
                 continue;
+            }
     		int rc = xmlrpc_replicate_log(it->first, &lr, success, fterm, error);
     		if ( rc == -1 )
+    		{
                 static time_t last_error = 0;
                 static int    num_errors = 0;
                 num_errors++;
                 if ( last_error == 0 )
+                {
                     last_error = time(0);
                     num_errors = 1;
+                }
                 else if ( last_error + 60 < time(0) )
+                {
                     if ( num_errors > 10 )
+                    {
                         std::ostringstream oss;
                         oss << "Detetected error condition on follower "
                             << it->first <<". Last error was: " << error;
                         NebulaLog::log("RCM", Log::INFO, oss);
+                    }
                     last_error = 0;
+                }
+    		}
     		else if ( success == false && fterm > term )
+    		{
     			std::ostringstream oss;
                 oss << "Follower " << it->first << " term (" << fterm
                     << ") is higher than current (" << term << ")";
             	NebulaLog::log("RCM", Log::INFO, oss);
     			follower(fterm);
     			break;
+    		}
+        }
+    }
     /* -------------------------------------------------------------------------- */
     /* -------------------------------------------------------------------------- */
     void RaftManager::request_vote()
+    {
         unsigned int lindex, lterm, fterm, _term;
-...
     	pthread_mutex_lock(&mutex);
     	if ( state == SOLO )
+    	{
     		lindex = 0;
     		lterm  = 0;
+    	}
         oss << "<RAFT>"
             << "<SERVER_ID>" << server_id << "</SERVER_ID>"
             << "<STATE>"     << state << "</STATE>"

src/raft/ReplicaManager.cc
164	164	{
165	165	return new RaftReplicaThread(follower_id);
166	166	}
	167
	168	// -----------------------------------------------------------------------------
	169
	170	ReplicaThread * HeartBeatManager::thread_factory(int follower_id)
	171	{
	172	return new HeartBeatThread(follower_id);
	173	}

         frm = nd.get_frm();
     };
     // -----------------------------------------------------------------------------
     // -----------------------------------------------------------------------------
     int FedReplicaThread::replicate()
+    {
         std::string error;
-...
         return 0;
+    }
     // -----------------------------------------------------------------------------
     // -----------------------------------------------------------------------------
     HeartBeatThread::HeartBeatThread(int fid):ReplicaThread(fid), last_error(0),
         num_errors(0)
+    {
         Nebula& nd = Nebula::instance();
         raftm = nd.get_raftm();
     };
     // -----------------------------------------------------------------------------
     // -----------------------------------------------------------------------------
     int HeartBeatThread::replicate()
+    {
         int rc;
     	bool success;
     	std::string error;
     	unsigned int fterm;
         unsigned int term  = raftm->get_term();
     	LogDBRecord lr;
     	lr.index = 0;
     	lr.prev_index = 0;
     	lr.term = 0;
     	lr.prev_term = 0;
     	lr.sql = "";
     	lr.timestamp = 0;
         rc = raftm->xmlrpc_replicate_log(follower_id, &lr, success, fterm, error);
         if ( rc == -1 )
+        {
             num_errors++;
             if ( last_error == 0 )
+            {
                 last_error = time(0);
                 num_errors = 1;
+            }
             else if ( last_error + 60 < time(0) )
+            {
                 if ( num_errors > 10 )
+                {
                     std::ostringstream oss;
                     oss << "Detetected error condition on follower "
                         << follower_id <<". Last error was: " << error;
                     NebulaLog::log("RCM", Log::INFO, oss);
+                }
                 last_error = 0;
+            }
+        }
         else if ( success == false && fterm > term )
+        {
             std::ostringstream oss;
             oss << "Follower " << follower_id << " term (" << fterm
                 << ") is higher than current (" << term << ")";
             NebulaLog::log("RCM", Log::INFO, oss);
             raftm->follower(fterm);
+        }
         return 0;
+    }

OpenNebula

Revision 0a2f18a7