Changeset 313

Timestamp:

02/27/08 00:43:56 (9 months ago)

Author:

rm

Message:

Replication:

• listen for replay next requests on both the main group and direct to the recorder.
  
• improve Spread's drain behavior in run
  
• provide access to group membership in Spread
  
• ReplicationBackend? now keeps track of who it's syncing from, asks everyone the first time and uses the first to answer.
  

Location:

trunk

Files:

• thrucommon/src/ReplicationRecorder.cpp (modified) (2 diffs)
• thrucommon/src/Spread.cpp (modified) (3 diffs)
• thrucommon/src/Spread.h (modified) (1 diff)
• thrucommon/src/thrurecorder.conf (added)
• thrudoc/src/ReplicationBackend.cpp (modified) (3 diffs)
• thrudoc/src/ReplicationBackend.h (modified) (1 diff)

trunk/thrucommon/src/ReplicationRecorder.cpp

@@ -47 +47 @@
                             live_callback_info);
 
-    // subscribe to "replay" messages
+    // subscribe to "replay" messages, both broadcast and direct
     SubscriberCallbackInfo * replay_callback_info = 
         new SubscriberCallbackInfo ();
@@ -54 +54 @@
     this->spread.subscribe ("", this->replication_group, REPLAY_MESSAGE_TYPE,
                             replay_callback_info);
+    this->spread.subscribe ("", this->spread.get_private_group (),
+                            REPLAY_MESSAGE_TYPE, replay_callback_info);
 }
 

trunk/thrucommon/src/Spread.cpp

@@ -167 +167 @@
     }
 
-    // send out any pending message
-    this->drain_pending ();
-
     service service_type = 0;
     char sender[MAX_GROUP_NAME];
@@ -182 +179 @@
     while (!count || i < count)
     {
+        // drain any pending messages
+        this->drain_pending ();
+
         // created here so that if we have to grow them they'll be recreated
         // larger, see errorhandling below
@@ -272 +272 @@
             }
         }
-        // drain any pending messages
-        this->drain_pending ();
     }
     LOG4CXX_DEBUG (logger, "run:    done");

trunk/thrucommon/src/Spread.h

@@ -80 +80 @@
         }
 
+        const std::set<std::string> get_group_members (std::string group)
+        {
+            return this->groups[group];
+        }
+
     private:
         static log4cxx::LoggerPtr logger;

trunk/thrudoc/src/ReplicationBackend.cpp

@@ -409 +409 @@
 }
 
+// TODO: handle when we don't hear from our replay host for a while...
 bool ReplicationBackend::handle_replay_message 
-(const std::string & /* sender */,
+(const std::string & sender,
  const std::vector<std::string> & /* groups */,
  const int /* message_type */, const char * message, const int message_len)
 {
     LOG4CXX_DEBUG (logger, "handle_replay_message:");
+
+    // first one to answer becomes the person we'll ask in the future
+    if (this->current_replay_name.empty ())
+    {
+        this->current_replay_name = sender;
+    }
+    else if (this->current_replay_name != sender)
+    {
+        // this isn't from our offical replay host, skip it
+        return true;
+    }
 
     // TODO: don't recreate these every time...
@@ -454 +466 @@
         // actually happened since the last message we recorded
         this->listener_live = true;
+        // unset our current_replay_name, so we'll get one again next time
+        this->current_replay_name = "";
     }
 
@@ -579 +593 @@
 void ReplicationBackend::request_next (string uuid)
 {
+    string who_to_ask = this->current_replay_name;
+    if (who_to_ask.empty ())
+    {
+        // we don't have a current replay from, so ask them all, first one to
+        // respond will be the the new replay host
+        who_to_ask = this->replication_group;
+    }
+    LOG4CXX_DEBUG (logger, "request_next: who_to_ask=" + who_to_ask);
+
     // we don't want our own message back here...
-    this->spread.send (RELIABLE_MESS | SELF_DISCARD, this->replication_group,
+    this->spread.send (RELIABLE_MESS | SELF_DISCARD, who_to_ask,
                        REPLAY_MESSAGE_TYPE, uuid.c_str (), uuid.length ());
 }

trunk/thrudoc/src/ReplicationBackend.h

@@ -73 +73 @@
         Spread spread;
         std::string replication_group;
+        std::string current_replay_name;
         std::string replication_status_file;
         int replication_status_flush_frequency;