Changeset 283

Timestamp:

02/18/08 09:06:30 (9 months ago)

Author:

rm

Message:

bring spread replication in to things. rename to just ReplicationBackend?.

Location:

trunk/thrudoc

Files:

• examples/thrudoc.conf (modified) (1 diff)
• src/Makefile.am (modified) (2 diffs)
• src/ReplicationBackend.cpp (moved) (moved from trunk/thrudoc/src/SpreadReplicationBackend.cpp) (30 diffs)
• src/ReplicationBackend.h (moved) (moved from trunk/thrudoc/src/SpreadReplicationBackend.h) (6 diffs)
• src/app_helpers.cpp (modified) (4 diffs)

trunk/thrudoc/examples/thrudoc.conf

@@ -38 +38 @@
 #SPREAD_GROUP = thrudoc
 
+# spread replication backend
+#REPLICATION_NAME = 4803@localhost
+#REPLICATION_PRIVATE_NAME = ds_s1
+#REPLICATION_GROUP = thrudoc
+#REPLICATION_STATUS_FILE=replication_status
+#REPLICATION_STATUS_FLUSH_FREQUENCY=30
+
 #ENABLE_BLOOM_FILTER=1
 

trunk/thrudoc/src/Makefile.am

@@ -11 +11 @@
 include_thrudocdir = $(includedir)/thrudoc
 include_thrudoc_HEADERS = \
-                             app_helpers.h      \
-                             BDBBackend.h       \
-                             BloomBackend.h     \
-                             DiskBackend.h      \
-                             LogBackend.h       \
-                             MemcachedBackend.h \
-                             MySQLBackend.h     \
-                             NBackend.h         \
-                             NullBackend.h      \
-                             S3Backend.h        \
-                             SpreadBackend.h    \
-                             StatsBackend.h     \
+                             app_helpers.h              \
+                             BDBBackend.h               \
+                             BloomBackend.h             \
+                             DiskBackend.h              \
+                             LogBackend.h               \
+                             MemcachedBackend.h         \
+                             MySQLBackend.h             \
+                             NBackend.h                 \
+                             NullBackend.h              \
+                             ReplicationBackend.h       \
+                             S3Backend.h                \
+                             SpreadBackend.h            \
+                             StatsBackend.h             \
                              ThrudocBackend.h
 
@@ -40 +41 @@
                   NBackend.cpp                          \
                   NullBackend.cpp                       \
+                  ReplicationBackend.cpp                \
                   S3Backend.cpp                         \
                   StatsBackend.cpp                      \

trunk/thrudoc/src/ReplicationBackend.cpp

@@ -7 +7 @@
 #if HAVE_LIBSPREAD && HAVE_LIBUUID
 
-#include "SpreadReplicationBackend.h"
+#include "ReplicationBackend.h"
 
 #include <sys/types.h>
@@ -21 +21 @@
 #define UUID_LEN 37
 #define MESSAGE_OVERHEAD 4
-#define SPREAD_BACKEND_MAX_MESSAGE_SIZE MAX_BUCKET_SIZE + MAX_KEY_SIZE + MAX_VALUE_SIZE + UUID_LEN + MESSAGE_OVERHEAD
+#define REPLICATION_BACKEND_MAX_MESSAGE_SIZE MAX_BUCKET_SIZE + MAX_KEY_SIZE + MAX_VALUE_SIZE + UUID_LEN + MESSAGE_OVERHEAD
 
 #define ORIG_MESSAGE_TYPE 1
@@ -35 +35 @@
 string SP_error_to_string (int error);
 
-class SpreadReplicationWait 
+class ReplicationWait 
 {
     public:
         // uuid is just for logging/debugging purposes
-        SpreadReplicationWait (string uuid, uint32_t max_wait)
+        ReplicationWait (string uuid, uint32_t max_wait)
         {
             char buf[128];
-            sprintf (buf, "SpreadReplicationWait: uuid=%s, max_wait=%d", 
+            sprintf (buf, "ReplicationWait: uuid=%s, max_wait=%d", 
                      uuid.c_str (), max_wait);
             LOG4CXX_DEBUG (logger, buf);
@@ -56 +56 @@
         }
 
-        ~SpreadReplicationWait ()
-        {
-            LOG4CXX_DEBUG (logger, "~SpreadReplicationWait: uuid=" + 
+        ~ReplicationWait ()
+        {
+            LOG4CXX_DEBUG (logger, "~ReplicationWait: uuid=" + 
                            this->uuid);
             pthread_cond_destroy (&this->condition);
@@ -132 +132 @@
 };
 
-class SpreadReplicationMessage
+class ReplicationMessage
 {
     public:
@@ -138 +138 @@
         {
             max_groups = 5;
-            buf_size = SPREAD_BACKEND_MAX_MESSAGE_SIZE;
+            buf_size = REPLICATION_BACKEND_MAX_MESSAGE_SIZE;
 
             buf_len = SP_receive (spread_mailbox, &service_type, sender,
@@ -180 +180 @@
         }
 
-        int parse (const string & spread_private_group)
+        int parse (const string & replication_private_group)
         {
             // TODO: need better/safer parsing than scanf... maybe thrift ser.
@@ -196 +196 @@
             }
             else if (type == REPLAY_MESSAGE_TYPE &&
-                     spread_private_group == groups[0])
+                     replication_private_group == groups[0])
             {
                 LOG4CXX_DEBUG (logger, "parse: replay");
@@ -276 +276 @@
         int buf_size;
         int buf_len;
-        char buf[SPREAD_BACKEND_MAX_MESSAGE_SIZE];
+        char buf[REPLICATION_BACKEND_MAX_MESSAGE_SIZE];
 
         // repli command stuff
@@ -287 +287 @@
 
 // private
-LoggerPtr SpreadReplicationWait::logger (Logger::getLogger ("SpreadReplicationWait"));
-LoggerPtr SpreadReplicationMessage::logger (Logger::getLogger ("SpreadReplicationMessage"));
-LoggerPtr SpreadReplicationBackend::logger (Logger::getLogger ("SpreadReplicationBackend"));
-
-SpreadReplicationBackend::SpreadReplicationBackend (shared_ptr<ThrudocBackend> backend,
-                                                    const string & spread_name,
-                                                    const string & spread_private_name,
-                                                    const string & spread_group)
-{
-    LOG4CXX_INFO (logger, string ("SpreadReplicationBackend: spread_name=") +
-                  spread_name + ", spread_private_name=" +
-                  spread_private_name + ", spread_group=" + spread_group);
+LoggerPtr ReplicationWait::logger (Logger::getLogger ("ReplicationWait"));
+LoggerPtr ReplicationMessage::logger (Logger::getLogger ("ReplicationMessage"));
+LoggerPtr ReplicationBackend::logger (Logger::getLogger ("ReplicationBackend"));
+
+ReplicationBackend::ReplicationBackend (shared_ptr<ThrudocBackend> backend,
+                                        const string & replication_name,
+                                        const string & replication_private_name,
+                                        const string & replication_group,
+                                        const string & replication_status_file,
+                                        const int replication_status_flush_frequency)
+{
+    char buf[1024];
+    sprintf (buf, "ReplicationBackend: replication_name=%s, replication_private_name=%s, replication_group=%s, replication_status_file=%s, replication_status_flush_frequency=%d",
+             replication_name.c_str (), replication_private_name.c_str (),
+             replication_group.c_str (), replication_status_file.c_str (),
+             replication_status_flush_frequency);
+    LOG4CXX_INFO (logger, buf);
 
     this->set_backend (backend);
-    this->spread_name = spread_name;
-    this->spread_private_name = spread_private_name;
-    this->spread_group = spread_group;
+    this->replication_name = replication_name;
+    this->replication_private_name = replication_private_name;
+    this->replication_group = replication_group;
+    this->replication_status_file = replication_status_file;
+    this->replication_status_flush_frequency = replication_status_flush_frequency;
 
     char private_group[MAX_GROUP_NAME];
-    int ret = SP_connect (this->spread_name.c_str (),
-                          this->spread_private_name.c_str (), 0, 1,
+    int ret = SP_connect (this->replication_name.c_str (),
+                          this->replication_private_name.c_str (), 0, 1,
                           &this->spread_mailbox, private_group);
     if (ret < 0)
@@ -318 +325 @@
     }
 
-    this->spread_private_group = string (private_group);
-    LOG4CXX_INFO (logger, "SpreadBackend: private_group=" +
-                  this->spread_private_group);
-
-    ret = SP_join (this->spread_mailbox, this->spread_group.c_str ());
+    this->replication_private_group = string (private_group);
+    LOG4CXX_INFO (logger, "ReplicationBackend: private_group=" +
+                  this->replication_private_group);
+
+    ret = SP_join (this->spread_mailbox, this->replication_group.c_str ());
     if (ret < 0)
     {
@@ -333 +340 @@
 
     int fd;
-    fd = ::open ("last_uuid", 0x0, S_IRUSR | S_IWUSR| S_IRGRP | S_IROTH);
+    fd = ::open (this->replication_status_file.c_str (), 0x0, 
+                 S_IRUSR | S_IWUSR| S_IRGRP | S_IROTH);
     listener_live = true; // we're live unless we load a last_uuid in a sec
     if (fd)
     {
+        LOG4CXX_DEBUG (logger, "ReplicationBackend: opened status file=" +
+                       this->replication_status_file);
         char buf[64] = "";
         ::read (fd, buf, 64);
-        last_uuid = buf;
+        this->last_uuid = buf;
         ::close (fd);
-        if (!last_uuid.empty ())
+        if (!this->last_uuid.empty ())
         {
             listener_live = false;
-            request_next (last_uuid);
-            LOG4CXX_INFO (logger, "SpreadReplicationBackend: loaded last_uuid=" +
-                          last_uuid);
+            request_next (this->last_uuid);
+            LOG4CXX_INFO (logger, "ReplicationBackend: found last_uuid=" +
+                          this->last_uuid);
         }
     }
@@ -355 +365 @@
                        (void *)this) != 0)
     {
-        char error[] = "SpreadReplicationBackend: start_listener_thread failed\n";
+        char error[] = "ReplicationBackend: start_listener_thread failed\n";
         LOG4CXX_ERROR (logger, error);
         ThrudocException e;
@@ -363 +373 @@
 }
 
-SpreadReplicationBackend::~SpreadReplicationBackend ()
-{
-    LOG4CXX_INFO (logger, "~SpreadReplicationBackend");
+ReplicationBackend::~ReplicationBackend ()
+{
+    LOG4CXX_INFO (logger, "~ReplicationBackend");
     // we're no longer live, don't accept connections
     this->listener_live = false;
@@ -400 +410 @@
 // TODO: implement circuit breaker pattern around spread...
 
-void SpreadReplicationBackend::put (const string & bucket, const string & key,
+void ReplicationBackend::put (const string & bucket, const string & key,
                                     const string & value)
 {
-    char msg[SPREAD_BACKEND_MAX_MESSAGE_SIZE];
+    char msg[REPLICATION_BACKEND_MAX_MESSAGE_SIZE];
     string uuid = generate_uuid ();
-    snprintf (msg, SPREAD_BACKEND_MAX_MESSAGE_SIZE, "%s p %s %s %s",
+    snprintf (msg, REPLICATION_BACKEND_MAX_MESSAGE_SIZE, "%s p %s %s %s",
               uuid.c_str (), bucket.c_str (), key.c_str (), value.c_str ());
 
@@ -411 +421 @@
 }
 
-void SpreadReplicationBackend::remove (const string & bucket,
+void ReplicationBackend::remove (const string & bucket,
                                        const string & key )
 {
-    char msg[SPREAD_BACKEND_MAX_MESSAGE_SIZE];
+    char msg[REPLICATION_BACKEND_MAX_MESSAGE_SIZE];
     string uuid = generate_uuid ();
-    snprintf (msg, SPREAD_BACKEND_MAX_MESSAGE_SIZE, "%s r %s %s",
+    snprintf (msg, REPLICATION_BACKEND_MAX_MESSAGE_SIZE, "%s r %s %s",
               uuid.c_str (), bucket.c_str (), key.c_str ());
 
@@ -422 +432 @@
 }
 
-string SpreadReplicationBackend::admin (const string & op, const string & data)
+string ReplicationBackend::admin (const string & op, const string & data)
 {
     if (op == "replay_from")
@@ -430 +440 @@
         return "done";
     }
-    char msg[SPREAD_BACKEND_MAX_MESSAGE_SIZE];
+    char msg[REPLICATION_BACKEND_MAX_MESSAGE_SIZE];
     string uuid = generate_uuid ();
-    snprintf (msg, SPREAD_BACKEND_MAX_MESSAGE_SIZE, "%s a %s %s",
+    snprintf (msg, REPLICATION_BACKEND_MAX_MESSAGE_SIZE, "%s a %s %s",
               uuid.c_str (), op.c_str (), data.c_str ());
 
@@ -438 +448 @@
 }
 
-string SpreadReplicationBackend::send_and_wait_for_resp (const char * msg,
+string ReplicationBackend::send_and_wait_for_resp (const char * msg,
                                                          string uuid)
 {
     LOG4CXX_DEBUG (logger, "wait_for_resp: begin uuid=" + uuid);
     string ret;
-    shared_ptr<SpreadReplicationWait> wait (new SpreadReplicationWait (uuid, 
+    shared_ptr<ReplicationWait> wait (new ReplicationWait (uuid, 
                                                                        2));
     // install wait
@@ -451 +461 @@
     }
     // send out multi-cast message
-    SP_multicast (this->spread_mailbox, SAFE_MESS, this->spread_group.c_str (),
+    SP_multicast (this->spread_mailbox, SAFE_MESS, 
+                  this->replication_group.c_str (),
                   ORIG_MESSAGE_TYPE, strlen (msg), msg);
     // wait here until we have the result
@@ -471 +482 @@
 }
 
-void SpreadReplicationBackend::validate (const std::string & bucket,
+void ReplicationBackend::validate (const std::string & bucket,
                                          const std::string * key,
                                          const std::string * value)
@@ -504 +515 @@
 }
 
-string SpreadReplicationBackend::generate_uuid ()
+string ReplicationBackend::generate_uuid ()
 {
     uuid_t uuid;
@@ -513 +524 @@
 }
 
-void * SpreadReplicationBackend::start_listener_thread (void * ptr)
+void * ReplicationBackend::start_listener_thread (void * ptr)
 {
     LOG4CXX_INFO (logger, "start_listener_thread: ");
-    (((SpreadReplicationBackend*)ptr)->listener_thread_run ());
+    (((ReplicationBackend*)ptr)->listener_thread_run ());
     return NULL;
 }
 
-void SpreadReplicationBackend::listener_thread_run ()
+void ReplicationBackend::listener_thread_run ()
 {
     struct pollfd fds[] = { {this->spread_mailbox, POLLIN, 0} };
@@ -541 +552 @@
                 this->listener_thread_go = false;
             }
-            if (last_flush + 30 < time (0))
-            {
-                LOG4CXX_DEBUG (logger, "flushing last_uuid=" + last_uuid);
+            if ((last_flush + this->replication_status_flush_frequency) < 
+                time (0))
+            {
+                LOG4CXX_DEBUG (logger, "listener_thread_run: flushing last_uuid=" +
+                               this->last_uuid);
                 int fd;
-                fd = ::open ("last_uuid", O_RDWR | O_TRUNC | O_CREAT, 
+                fd = ::open (this->replication_status_file.c_str (),
+                             O_RDWR | O_TRUNC | O_CREAT, 
                              S_IRUSR | S_IWUSR| S_IRGRP | S_IROTH);
                 ::write (fd, this->last_uuid.c_str (), 
@@ -564 +578 @@
 }
 
-void SpreadReplicationBackend::handle_message ()
-{
-    SpreadReplicationMessage * message = new SpreadReplicationMessage ();
+void ReplicationBackend::handle_message ()
+{
+    ReplicationMessage * message = new ReplicationMessage ();
     message->receive (this->spread_mailbox);
-    int type = message->parse (this->spread_private_group);
+    int type = message->parse (this->replication_private_group);
     if (type == ORIG_MESSAGE_TYPE)
     {
@@ -583 +597 @@
             while (!pending_messages.empty ())
             {
-                SpreadReplicationMessage * drain = pending_messages.front ();
+                ReplicationMessage * drain = pending_messages.front ();
                 pending_messages.pop ();
                 LOG4CXX_DEBUG (logger, string ("handle_message: drain.uuid=") +
@@ -608 +622 @@
         {
             // it's a message
-            SpreadReplicationMessage * first_queued = pending_messages.front ();
+            ReplicationMessage * first_queued = pending_messages.front ();
             if (first_queued == NULL ||
                 first_queued->get_uuid () != message->get_uuid ())
@@ -645 +659 @@
 }
 
-void SpreadReplicationBackend::do_message (SpreadReplicationMessage * message)
+void ReplicationBackend::do_message (ReplicationMessage * message)
 {
     string ret;
@@ -696 +710 @@
 
     // if we sent this message signal to the waiting thread that it's complete
-    if (this->spread_private_group == message->get_sender ())
+    if (this->replication_private_group == message->get_sender ())
     {
         RWGuard g (this->pending_waits_mutex, false);
-        std::map<std::string, boost::shared_ptr<SpreadReplicationWait> >::iterator
+        std::map<std::string, boost::shared_ptr<ReplicationWait> >::iterator
             i = pending_waits.find (message->get_uuid ());
         if (i != pending_waits.end ())
@@ -708 +722 @@
 }
 
-void SpreadReplicationBackend::request_next (string uuid)
+void ReplicationBackend::request_next (string uuid)
 {
     // we don't want our own message back here...
     SP_multicast (this->spread_mailbox, RELIABLE_MESS | SELF_DISCARD,
-                  this->spread_group.c_str (),
+                  this->replication_group.c_str (),
                   REPLAY_MESSAGE_TYPE, uuid.length (), uuid.c_str ());
 }

trunk/thrudoc/src/ReplicationBackend.h

@@ -3 +3 @@
  **/
 
-#ifndef __SPREAD_REPLICIATION_BACKEND_H_
-#define __SPREAD_REPLICIATION_BACKEND_H_
+#ifndef _REPLICIATION_BACKEND_H_
+#define _REPLICIATION_BACKEND_H_
 
 #if HAVE_LIBMEMCACHED && HAVE_LIBUUID
@@ -18 +18 @@
 #include "ThrudocPassthruBackend.h"
 
-class SpreadReplicationWait;
-class SpreadReplicationMessage;
+class ReplicationWait;
+class ReplicationMessage;
 
 /* NOTE: this is not exactly a normal passthrough backend even tho it kinda
  * looks like one. we use the passthrough, but there's spread in between
  * passing things down */
-class SpreadReplicationBackend : public ThrudocPassthruBackend
+class ReplicationBackend : public ThrudocPassthruBackend
 {
     public:
-        SpreadReplicationBackend (boost::shared_ptr<ThrudocBackend> backend,
-                                  const std::string & spread_name, 
-                                  const std::string & spread_private_name,
-                                  const std::string & spread_group);
-        ~SpreadReplicationBackend ();
+        ReplicationBackend (boost::shared_ptr<ThrudocBackend> backend,
+                            const std::string & replication_name, 
+                            const std::string & replication_private_name,
+                            const std::string & replication_group,
+                            const std::string & replication_status_file,
+                            const int replication_status_flush_frequency);
+        ~ReplicationBackend ();
 
         void put (const std::string & bucket, const std::string & key,
@@ -48 +50 @@
         static void * start_listener_thread (void * ptr);
 
-        std::string spread_name;
-        std::string spread_private_name;
-        std::string spread_group;
-        std::string spread_private_group;
+        std::string replication_name;
+        std::string replication_private_name;
+        std::string replication_group;
+        std::string replication_private_group;
+        std::string replication_status_file;
+        int replication_status_flush_frequency;
         mailbox spread_mailbox;
         pthread_t listener_thread;
@@ -57 +61 @@
         bool listener_live;
         std::string last_uuid;
-        std::queue<SpreadReplicationMessage *> pending_messages;
-        std::map<std::string, boost::shared_ptr<SpreadReplicationWait> > 
+        std::queue<ReplicationMessage *> pending_messages;
+        std::map<std::string, boost::shared_ptr<ReplicationWait> > 
             pending_waits;
         facebook::thrift::concurrency::ReadWriteMutex pending_waits_mutex;
@@ -65 +69 @@
         void listener_thread_run ();
         void handle_message ();
-        void do_message (SpreadReplicationMessage * message);
+        void do_message (ReplicationMessage * message);
         void request_next (std::string message);
 };
@@ -71 +75 @@
 #endif /* HAVE_LIBMEMCACHED */
 
-#endif
+#endif /* _REPLICIATION_BACKEND_H_ */

trunk/thrudoc/src/app_helpers.cpp

@@ -18 +18 @@
 #include "S3Backend.h"
 #include "SpreadBackend.h"
+#include "ReplicationBackend.h"
 #include "StatsBackend.h"
 #include "ThrudocBackend.h"
@@ -174 +175 @@
     string spread_private_name =
         ConfigManager->read<string>("SPREAD_PRIVATE_NAME", "");
+    // Spread replication (sorta) passthrough
+    string replication_private_name =
+        ConfigManager->read<string>("REPLICATION_PRIVATE_NAME", "");
 #if HAVE_LIBSPREAD
     string spread_name =
@@ -184 +188 @@
             (new SpreadBackend (backend, spread_name, spread_private_name,
                                 spread_group));
+
+    string replication_name =
+        ConfigManager->read<string>("REPLICATION_NAME", "4803");
+    string replication_group =
+        ConfigManager->read<string>("REPLICATION_GROUP", "thrudoc");
+    string replication_status_file =
+        ConfigManager->read<string>("REPLICATION_STATUS_FILE", 
+                                    "replication_status");
+    int replication_status_flush_frequency =
+        ConfigManager->read<int>("REPLICATION_STATUS_FLUSH_FREQUENCY", 30);
+
+    if (!replication_private_name.empty ())
+        backend = shared_ptr<ThrudocBackend>
+            (new ReplicationBackend (backend, replication_name, 
+                                     replication_private_name,
+                                     replication_group, 
+                                     replication_status_file,
+                                     replication_status_flush_frequency));
 #else
     if (!spread_private_name.empty ())
@@ -189 +211 @@
         LOG4CXX_ERROR (logger, "SPREAD_PRIVATE_NAME supplied, but spread support not complied in");
         fprintf (stderr, "SPREAD_PRIVATE_NAME supplied, but spread support not complied in\n");
+        exit (1);
+    }
+    if (!replication_private_name.empty ())
+    {
+        LOG4CXX_ERROR (logger, "REPLICATION_PRIVATE_NAME supplied, but spread support not complied in");
+        fprintf (stderr, "REPLCATION_PRIVATE_NAME supplied, but spread support not complied in\n");
         exit (1);
     }