/ - Diff - OpenNebula - OpenNebula Development pages

     #include "ActionManager.h"
     #include "ReplicaManager.h"
     #include "ReplicaRequest.h"
     #include "Template.h"
     extern "C" void * raft_manager_loop(void *arg);
-...
         //   - match, highest log replicated in this server <follower, match>
     	//   - servers, list of servers in zone and xml-rpc edp <follower, edp>
         // -------------------------------------------------------------------------
         ReplicaManager replica_manager;
         RaftReplicaManager replica_manager;
         unsigned int commit;

     #ifndef REPLICA_MANAGER_H_
     #define REPLICA_MANAGER_H_
     #include <xmlrpc-c/client.hpp>
     #include <string>
     #include <map>
     #include <vector>
     class LogDBRecord;
     class ReplicaThread;
     extern "C" void * replication_thread(void *arg);
     // -----------------------------------------------------------------------------
     // -----------------------------------------------------------------------------
     // Replication thread class
     // -----------------------------------------------------------------------------
     // -----------------------------------------------------------------------------
     class ReplicaThread
+    {
     public:
         ReplicaThread(int follower_id);
         virtual ~ReplicaThread(){};
         /**
          *  Main replication logic for the thread, it sends log records to followers
          *  and handle errors
          */
         void do_replication();
         /**
          *  Notify this replica thread that are new records in the log to replicate
          */
         void add_request();
         /**
          *  Exists the replication thread
          */
         void finalize();
         pthread_t thread_id() const
+        {
             return _thread_id;
+        }
     private:
         /**
          * C linkage function to start the thread
          *   @param arg pointer to "this"
          */
         friend void * replication_thread(void *arg);
         // -------------------------------------------------------------------------
         // pthread synchronization variables
         // -------------------------------------------------------------------------
         pthread_t _thread_id;
         pthread_mutex_t mutex;
         pthread_cond_t cond;
         bool _finalize;
         bool _pending_requests;
         time_t retry_timeout;
         static const time_t max_retry_timeout;
         // -------------------------------------------------------------------------
         // Information of the replication target server and leader
         // -------------------------------------------------------------------------
         int follower_id;
     };
     // -----------------------------------------------------------------------------
     // -----------------------------------------------------------------------------
     // Replication Manager
     // Replication Manager. This is a generic replication manager it starts, stops
     // and send control events to replica threads.
     // -----------------------------------------------------------------------------
     // -----------------------------------------------------------------------------
     class ReplicaManager
+    {
     public:
         ReplicaManager(){};
         virtual ~ReplicaManager()
+        {
             stop_replica_threads();
         };
         /**
          *  Start the replication threads, one for each server in the zone
          */
-...
          */
         void add_replica_thread(int follower_id);
     protected:
         ReplicaManager(){};
         virtual ~ReplicaManager()
+        {
             stop_replica_threads();
         };
         virtual ReplicaThread * thread_factory(int follower_id) = 0;
     private:
         /**
          *  The replication thread pool
-...
         ReplicaThread * get_thread(int server_id);
     };
     // -----------------------------------------------------------------------------
     // -----------------------------------------------------------------------------
     // RaftReplicaManager to manage the raft replication thread pool
     // -----------------------------------------------------------------------------
     // -----------------------------------------------------------------------------
     class RaftReplicaManager : public ReplicaManager
+    {
     public:
         RaftReplicaManager():ReplicaManager(){};
         virtual ~RaftReplicaManager(){};
     private:
         ReplicaThread * thread_factory(int follower_id);
     };
     #endif /*REPLICA_MANAGER_H_*/

     /* -------------------------------------------------------------------------- */
     /* Copyright 2002-2016, OpenNebula Project, OpenNebula Systems                */
     /*                                                                            */
     /* Licensed under the Apache License, Version 2.0 (the "License"); you may    */
     /* not use this file except in compliance with the License. You may obtain    */
     /* a copy of the License at                                                   */
     /*                                                                            */
     /* http://www.apache.org/licenses/LICENSE-2.0                                 */
     /*                                                                            */
     /* Unless required by applicable law or agreed to in writing, software        */
     /* distributed under the License is distributed on an "AS IS" BASIS,          */
     /* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.   */
     /* See the License for the specific language governing permissions and        */
     /* limitations under the License.                                             */
     /* -------------------------------------------------------------------------- */
     #ifndef REPLICA_THREAD_H_
     #define REPLICA_THREAD_H_
     #include <pthread.h>
     extern "C" void * replication_thread(void *arg);
     // -----------------------------------------------------------------------------
     // -----------------------------------------------------------------------------
     // Replication thread class. This is a generic replicaton thread, it is used
     // to send information to a given server (follower). This class needs to be
     // specialized to implement the specific replication logic.
     // -----------------------------------------------------------------------------
     // -----------------------------------------------------------------------------
     class ReplicaThread
+    {
     public:
         ReplicaThread(int _follower_id):follower_id(_follower_id), _finalize(false),
             _pending_requests(false), retry_timeout(2)
+        {
             pthread_mutex_init(&mutex, 0);
             pthread_cond_init(&cond, 0);
         };
         virtual ~ReplicaThread(){};
         /**
          *  Notify this replica thread that are new records in the log to replicate
          */
         void add_request();
         /**
          *  Exists the replication thread
          */
         void finalize();
         /**
          *  @return the ID of the thread
          */
         pthread_t thread_id() const
+        {
             return _thread_id;
+        }
     protected:
         /**
          * Specific logic for the replicate process
          */
         virtual int replicate() = 0;
         /**
          * ID of follower to replicate state to
          */
         int follower_id;
     private:
         /**
          *  Wrapper function to handle the replication loop and timeouts. It makes
          *  use of the virtual function to replicate to actually start the replica-
          *  tion process.
          */
         void do_replication();
         /**
          * C linkage function to start the thread
          *   @param arg pointer to "this"
          */
         friend void * replication_thread(void *arg);
         // -------------------------------------------------------------------------
         // pthread synchronization variables
         // -------------------------------------------------------------------------
         pthread_t _thread_id;
         pthread_mutex_t mutex;
         pthread_cond_t cond;
         bool _finalize;
         bool _pending_requests;
         time_t retry_timeout;
         static const time_t max_retry_timeout;
     };
     // -----------------------------------------------------------------------------
     // Raft replication thread, it implements the Ratf replication algorithm on
     // followers
     // -----------------------------------------------------------------------------
     class LogDB;
     class RaftManager;
     class RaftReplicaThread : public ReplicaThread
+    {
     public:
         RaftReplicaThread(int follower_id);
         virtual ~RaftReplicaThread(){};
     private:
         /**
          * Specific logic for the replicate process
          */
         int replicate();
         /**
          * Pointers to other components
          */
         LogDB * logdb;
         RaftManager * raftm;
     };
     #endif

         std::ostringstream oss;
         unsigned int granted_votes = 0;
         unsigned int granted_votes;
         unsigned int votes2go;
         Nebula& nd    = Nebula::instance();
-...
             raft_state.to_xml(raft_state_xml);
             votes2go = num_servers / 2;
             votes2go      = num_servers / 2;
             granted_votes = 0;
             _term      = term;
             _server_id = server_id;

     /* -------------------------------------------------------------------------- */
     #include "ReplicaManager.h"
     #include "ReplicaThread.h"
     #include "Nebula.h"
     #include "NebulaLog.h"
     #include "ZoneServer.h"
     // -----------------------------------------------------------------------------
     // -----------------------------------------------------------------------------
     // Replication thread class & pool
     // -----------------------------------------------------------------------------
     // -----------------------------------------------------------------------------
     const time_t ReplicaThread::max_retry_timeout = 300;
     // -----------------------------------------------------------------------------
     // -----------------------------------------------------------------------------
     extern "C" void * replication_thread(void *arg)
+    {
         ReplicaThread * rt;
         if ( arg == 0 )
+        {
             return 0;
+        }
         rt = static_cast<ReplicaThread *>(arg);
         rt->_thread_id = pthread_self();
         rt->do_replication();
         return 0;
+    }
     // -----------------------------------------------------------------------------
     // -----------------------------------------------------------------------------
     ReplicaThread::ReplicaThread(int f):_finalize(false), _pending_requests(false),
         retry_timeout(2), follower_id(f)
+    {
         pthread_mutex_init(&mutex, 0);
         pthread_cond_init(&cond, 0);
     };
     // -----------------------------------------------------------------------------
     // -----------------------------------------------------------------------------
     void ReplicaThread::do_replication()
+    {
         Nebula& nd    = Nebula::instance();
         LogDB * logdb = nd.get_logdb();
         RaftManager * raftm = nd.get_raftm();
         unsigned int term = raftm->get_term();
         bool retry_request = false;
         std::string error;
         while ( _finalize == false )
+        {
             pthread_mutex_lock(&mutex);
             while ( _pending_requests == false )
+            {
                 struct timespec timeout;
                 timeout.tv_sec  = time(NULL) + retry_timeout;
                 timeout.tv_nsec = 0;
                 if ( pthread_cond_timedwait(&cond, &mutex, &timeout) == ETIMEDOUT )
+                {
                     _pending_requests = retry_request;
+                }
                 if ( _finalize )
+                {
                     return;
+                }
+            }
             _pending_requests = false;
             pthread_mutex_unlock(&mutex);
             // ---------------------------------------------------------------------
             // Get parameters to call append entries on follower
             // ---------------------------------------------------------------------
             LogDBRecord lr;
             int next_index = raftm->get_next_index(follower_id);
             bool success = false;
             unsigned int follower_term = -1;
             if ( logdb->get_log_record(next_index, lr) != 0 )
+            {
                 ostringstream ess;
                 ess << "Failed to load log record at index: " << next_index;
                 NebulaLog::log("RCM", Log::ERROR, ess);
                 continue;
+            }
             int xml_rc = raftm->xmlrpc_replicate_log(follower_id, &lr, success,
                 follower_term, error);
             if ( xml_rc == -1 )
+            {
                 if ( retry_timeout < max_retry_timeout )
+                {
                     retry_timeout = 2 * retry_timeout;
+                }
                 retry_request = true;
                 continue;
+            }
             else
+            {
                 retry_timeout = 2;
                 retry_request = false;
+            }
             if ( success )
+            {
                 raftm->replicate_success(follower_id);
+            }
             else
+            {
                 if ( follower_term > term )
+                {
                     ostringstream ess;
                     ess << "Follower " << follower_id << " term (" << follower_term
                         << ") is higher than current (" << term << ")";
                     NebulaLog::log("RCM", Log::INFO, ess);
                     raftm->follower(follower_term);
+                }
                 else
+                {
                     raftm->replicate_failure(follower_id);
+                }
+            }
+        }
+    }
     // -----------------------------------------------------------------------------
     // -----------------------------------------------------------------------------
     void ReplicaThread::finalize()
+    {
         pthread_mutex_lock(&mutex);
         _finalize = true;
         _pending_requests = false;
         pthread_cond_signal(&cond);
         pthread_mutex_unlock(&mutex);
+    }
     // -----------------------------------------------------------------------------
     // -----------------------------------------------------------------------------
     void ReplicaThread::add_request()
+    {
         pthread_mutex_lock(&mutex);
         _pending_requests = true;
         pthread_cond_signal(&cond);
         pthread_mutex_unlock(&mutex);
+    }
     // -----------------------------------------------------------------------------
     // -----------------------------------------------------------------------------
     // -----------------------------------------------------------------------------
     // -----------------------------------------------------------------------------
     ReplicaThread * ReplicaManager::get_thread(int server_id)
+    {
-...
             return;
+        }
         ReplicaThread * rthread = new ReplicaThread(follower_id);
         ReplicaThread * rthread = thread_factory(follower_id);
         thread_pool.insert(std::make_pair(follower_id, rthread));
-...
         pthread_attr_destroy(&pattr);
     };
     // -----------------------------------------------------------------------------
     ReplicaThread * RaftReplicaManager::thread_factory(int follower_id)
+    {
         return new RaftReplicaThread(follower_id);
+    }

     /* -------------------------------------------------------------------------- */
     /* Copyright 2002-2016, OpenNebula Project, OpenNebula Systems                */
     /*                                                                            */
     /* Licensed under the Apache License, Version 2.0 (the "License"); you may    */
     /* not use this file except in compliance with the License. You may obtain    */
     /* a copy of the License at                                                   */
     /*                                                                            */
     /* http://www.apache.org/licenses/LICENSE-2.0                                 */
     /*                                                                            */
     /* Unless required by applicable law or agreed to in writing, software        */
     /* distributed under the License is distributed on an "AS IS" BASIS,          */
     /* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.   */
     /* See the License for the specific language governing permissions and        */
     /* limitations under the License.                                             */
     /* -------------------------------------------------------------------------- */
     #include <errno.h>
     #include <string>
     #include "LogDB.h"
     #include "RaftManager.h"
     #include "ReplicaThread.h"
     #include "Nebula.h"
     #include "NebulaLog.h"
     // -----------------------------------------------------------------------------
     // -----------------------------------------------------------------------------
     // Replication thread class & pool
     // -----------------------------------------------------------------------------
     // -----------------------------------------------------------------------------
     const time_t ReplicaThread::max_retry_timeout = 300;
     // -----------------------------------------------------------------------------
     // -----------------------------------------------------------------------------
     extern "C" void * replication_thread(void *arg)
+    {
         ReplicaThread * rt;
         if ( arg == 0 )
+        {
             return 0;
+        }
         rt = static_cast<ReplicaThread *>(arg);
         rt->_thread_id = pthread_self();
         rt->do_replication();
         return 0;
+    }
     // -----------------------------------------------------------------------------
     // -----------------------------------------------------------------------------
     void ReplicaThread::do_replication()
+    {
         int rc;
         bool retry_request = false;
         while ( _finalize == false )
+        {
             pthread_mutex_lock(&mutex);
             while ( _pending_requests == false )
+            {
                 struct timespec timeout;
                 timeout.tv_sec  = time(NULL) + retry_timeout;
                 timeout.tv_nsec = 0;
                 if ( pthread_cond_timedwait(&cond, &mutex, &timeout) == ETIMEDOUT )
+                {
                     _pending_requests = retry_request;
+                }
                 if ( _finalize )
+                {
                     return;
+                }
+            }
             _pending_requests = false;
             pthread_mutex_unlock(&mutex);
             rc = replicate();
             if ( rc == -1 )
+            {
                 if ( retry_timeout < max_retry_timeout )
+                {
                     retry_timeout = 2 * retry_timeout;
+                }
                 retry_request = true;
+            }
             else
+            {
                 retry_timeout = 2;
                 retry_request = false;
+            }
+        }
+    }
     // -----------------------------------------------------------------------------
     // -----------------------------------------------------------------------------
     void ReplicaThread::finalize()
+    {
         pthread_mutex_lock(&mutex);
         _finalize = true;
         _pending_requests = false;
         pthread_cond_signal(&cond);
         pthread_mutex_unlock(&mutex);
+    }
     // -----------------------------------------------------------------------------
     // -----------------------------------------------------------------------------
     void ReplicaThread::add_request()
+    {
         pthread_mutex_lock(&mutex);
         _pending_requests = true;
         pthread_cond_signal(&cond);
         pthread_mutex_unlock(&mutex);
+    }
     // -----------------------------------------------------------------------------
     // -----------------------------------------------------------------------------
     RaftReplicaThread::RaftReplicaThread(int fid):ReplicaThread(fid)
+    {
         Nebula& nd = Nebula::instance();
         logdb = nd.get_logdb();
         raftm = nd.get_raftm();
     };
     // -----------------------------------------------------------------------------
     // -----------------------------------------------------------------------------
     int RaftReplicaThread::replicate()
+    {
         std::string error;
         LogDBRecord lr;
         bool success = false;
         unsigned int follower_term = -1;
         unsigned int term  = raftm->get_term();
         int next_index = raftm->get_next_index(follower_id);
         if ( logdb->get_log_record(next_index, lr) != 0 )
+        {
             ostringstream ess;
             ess << "Failed to load log record at index: " << next_index;
             NebulaLog::log("RCM", Log::ERROR, ess);
             return -1;
+        }
         if ( raftm->xmlrpc_replicate_log(follower_id, &lr, success, follower_term,
                     error) != 0 )
+        {
             return -1;
+        }
         if ( success )
+        {
             raftm->replicate_success(follower_id);
+        }
         else
+        {
             if ( follower_term > term )
+            {
                 ostringstream ess;
                 ess << "Follower " << follower_id << " term (" << follower_term
                     << ") is higher than current (" << term << ")";
                 NebulaLog::log("RCM", Log::INFO, ess);
                 raftm->follower(follower_term);
+            }
             else
+            {
                 raftm->replicate_failure(follower_id);
+            }
+        }
         return 0;
+    }
     // -----------------------------------------------------------------------------
     // -----------------------------------------------------------------------------
     /*
     void FederationReplicaThread::replicate()
+    {
         std::string error;
         LogDBRecord lr;
         bool success = false;
         unsigned int follower_term = -1;
         int next_index = raftm->get_next_index(follower_id);
         if ( fedlogdb->get_log_record(next_index, lr) != 0 )
+        {
             ostringstream ess;
             ess << "Failed to load log record at index: " << next_index;
             NebulaLog::log("RCM", Log::ERROR, ess);
             return;
+        }
         if ( fedm->xmlrpc_replicate_log(follower_id, &lr, success, error) != 0 )
+        {
             return -1;
+        }
         if ( success )
+        {
             fedm->replicate_success(follower_id);
+        }
         else
+        {
             fedm->replicate_failure(follower_id);
+        }
         return 0;
+    }
     */

     # Sources to generate the library
     source_files=[
         'RaftManager.cc',
         'ReplicaManager.cc'
         'ReplicaManager.cc',
         'ReplicaThread.cc'
+    ]
     # Build library

             failure_response(ACTION, att);
             return;
+        }
         else if ( candidate_term > current_term  )
+        {
             std::ostringstream oss;
             oss << "New term (" << candidate_term << ") discovered from candidate "
                 << candidate_id;
             NebulaLog::log("ReM", Log::INFO, oss);
             raftm->follower(leader_term);
+        }
         if ((log_term > candidate_log_term) || ((log_term == candidate_log_term) &&
             (log_index > candidate_log_index)))

OpenNebula

Revision 741a55d1