Changeset 339

Timestamp:

04/04/08 01:31:30 (8 months ago)

Author:

rm

Message:

move common file logging code to thrucommon (replay needs a pass) and give thrudex replay support

Location:

trunk

Files:

• thrucommon/configure.in (modified) (2 diffs)
• thrucommon/src (modified) (1 prop)
• thrucommon/src/FileLogger.cpp (added)
• thrucommon/src/FileLogger.h (added)
• thrucommon/src/Makefile.am (modified) (2 diffs)
• thrudex (modified) (1 prop)
• thrudex/configure.in (modified) (1 diff)
• thrudex/src (modified) (1 prop)
• thrudex/src/LogBackend.cpp (added)
• thrudex/src/LogBackend.h (added)
• thrudex/src/Makefile.am (modified) (4 diffs)
• thrudex/src/app_helpers.cpp (added)
• thrudex/src/app_helpers.h (added)
• thrudex/src/thrudex.cpp (moved) (moved from trunk/thrudex/src/main.cpp) (2 diffs)
• thrudex/src/thrudex_replay.cpp (copied) (copied from trunk/thrudoc/src/thrudoc_replay.cpp) (11 diffs)
• thrudoc/src/LogBackend.cpp (modified) (10 diffs)
• thrudoc/src/LogBackend.h (modified) (3 diffs)
• thrudoc/src/thrudoc.cpp (modified) (2 diffs)
• thrudoc/src/thrudoc_replay.cpp (modified) (2 diffs)

trunk/thrucommon/configure.in

@@ -83 +83 @@
 fi
 
+# boost filesystem
+if test "$with_disk" != "no" -o "$with_bdb" != "no"; then
+    AC_CHECK_LIB(boost_filesystem, [open], [],
+                 AC_MSG_ERROR([You must install boost_filesystem; please install libraries and development files]))
+    AC_CHECK_HEADERS([boost/filesystem/path.hpp],,
+                     AC_MSG_ERROR([boost/filesystem/path.hpp missing; please add db_cxx development files]))
+fi
+
 # log4cxx
 AC_CHECK_LIB(log4cxx, [open], [],
@@ -92 +100 @@
 AC_C_CONST
 AC_CHECK_FUNCS([clock_gettime fdatasync fsync gettimeofday memset socket strcasecmp strchr strdup])
-AC_CHECK_HEADERS([netinet/in.h stdlib.h string.h sys/param.h])
+AC_CHECK_HEADERS([netinet/in.h stdlib.h string.h sys/param.h cstdlib])
 AC_C_INLINE
 AC_FUNC_ALLOCA

trunk/thrucommon/src

@@ -7 +7 @@
 thrucommon.pc
 thrurecorder
+*.loT

@@ -7 +7 @@
 thrucommon.pc
 thrurecorder
+*.loT

trunk/thrucommon/src/Makefile.am

@@ -15 +15 @@
                              ConfigFile.h               \
                              Hashing.h                  \
+                             FileLogger.h               \
                              memcache++.h               \
                              MemcacheHandle.h           \
@@ -33 +34 @@
                            ConfigFile.cpp                       \
                            Hashing.cpp                          \
+                           FileLogger.cpp                       \
                            ReplicationRecorder.cpp              \
                            Spread.cpp                           \

trunk/thrudex

@@ -8 +8 @@
 COPYING
 INSTALL
+libtool
 Makefile
 Makefile.in

@@ -8 +8 @@
 COPYING
 INSTALL
+libtool
 Makefile
 Makefile.in

trunk/thrudex/configure.in

@@ -58 +58 @@
 fi
 
+# boost filesystem
+if test "$with_disk" != "no" -o "$with_bdb" != "no"; then
+    AC_CHECK_LIB(boost_filesystem, [open], [],
+                 AC_MSG_ERROR([You must install boost_filesystem; please install libraries and development files]))
+    AC_CHECK_HEADERS([boost/filesystem/path.hpp],,
+                     AC_MSG_ERROR([boost/filesystem/path.hpp missing; please add db_cxx development files]))
+fi
+
 # log4cxx
 AC_CHECK_LIB(log4cxx, [open], [],

trunk/thrudex/src

@@ -1 +1 @@
 .deps
 gen-cpp
+.libs
 Makefile
 Makefile.in
 *.swp
 thrudex
+thrudex_bench
+thrudex_replay

@@ -1 +1 @@
 .deps
 gen-cpp
+.libs
 Makefile
 Makefile.in
 *.swp
 thrudex
+thrudex_bench
+thrudex_replay

trunk/thrudex/src/Makefile.am

@@ -1 +1 @@
 pkgconfigdir = $(libdir)/pkgconfig
 
-bin_PROGRAMS = thrudex thrudex_bench
+bin_PROGRAMS = thrudex thrudex_bench thrudex_replay
 noinst_PROGRAMS = thrudex_bench
 lib_LTLIBRARIES = libthrudex.la
@@ -13 +13 @@
                   gen-cpp/Thrudex_types.cpp             \
                   gen-cpp/Thrudex_constants.cpp         \
+                  app_helpers.cpp                       \
                   ThrudexHandler.cpp                    \
+                  LogBackend.cpp                        \
                   ThrudexBackend.cpp                    \
                   CLuceneBackend.cpp                    \
@@ -22 +24 @@
 
 thrudex_SOURCES = \
-                  main.cpp
+                  thrudex.cpp
 
 
@@ -29 +31 @@
 
 LDADD = $(top_builddir)/src/libthrudex.la
+
+thrudex_replay_CPPFLAGS = -Wall -Igen-cpp $(CURL_CFLAGS) $(MEMCACHED_CFLAGS) $(THRIFTNB_CFLAGS) $(THRUCOMMON_CFLAGS) $(SSL_CFLAGS) $(UUID_CFLAGS)
+thrudex_replay_LDADD = $(LDADD)
+thrudex_replay_LDFLAGS = -Wall -rpath $(pkglibdir) $(CURL_LIBS) $(MEMCACHED_LIBS) $(THRIFTNB_LIBS) $(THRUCOMMON_LIBS) $(SSL_LIBS) $(UUID_LIBS) 
+thrudex_replay_SOURCES = \
+                  thrudex_replay.cpp
 
 thrudex_bench_SOURCES = \

trunk/thrudex/src/thrudex.cpp

@@ -35 +35 @@
 #include "log4cxx/helpers/exception.h"
 
+#include "app_helpers.h"
 #include "ThrudexHandler.h"
 #include "ThrudexBackend.h"
@@ -83 +84 @@
         ConfigManager->readFile( conf_file );
 
-        string index_root   = ConfigManager->read<string>("INDEX_ROOT");
+        PropertyConfigurator::configure(conf_file);
 
-        int    thread_count = ConfigManager->read<int>("THREAD_COUNT",3);
-        int    server_port  = ConfigManager->read<int>("SERVER_PORT",9099);
+        int thread_count = ConfigManager->read<int>("THREAD_COUNT",3);
+        int server_port  = ConfigManager->read<int>("SERVER_PORT",9099);
+        string which = ConfigManager->read<string>("BACKEND","CLucene");
 
-        PropertyConfigurator::configure(conf_file);
-        setlocale(LC_CTYPE, "en_US.utf8");  //unicode support
-
-        LOG4CXX_INFO(logger, "Starting up");
-
-        string which      = ConfigManager->read<string>("BACKEND","CLucene");
-        shared_ptr<ThrudexBackend>    backend(new CLuceneBackend(index_root));
-
+        shared_ptr<ThrudexBackend> backend = create_backend (which, 
+                                                             thread_count);
 
         if (ConfigManager->read<int>("KEEP_STATS", 0))

trunk/thrudex/src/thrudex_replay.cpp

@@ -9 +9 @@
 
 #ifdef HAVE_CONFIG_H
-#include "thrudoc_config.h"
+#include "thrudex_config.h"
 #endif
 /* hack to work around thrift and log4cxx installing config.h's */
@@ -41 +41 @@
 #include "ConfigFile.h"
 #include "LogBackend.h"
-#include "ThrudocBackend.h"
-#include "ThrudocHandler.h"
-#include "ThrudocHandler.h"
+#include "ThrudexBackend.h"
+#include "ThrudexHandler.h"
+#include "ThrudexHandler.h"
 #include "ThruFileTransport.h"
 
 using namespace boost;
-using namespace thrudoc;
+using namespace thrudex;
 using namespace facebook::thrift;
 using namespace facebook::thrift::concurrency;
@@ -57 +57 @@
 using namespace std;
 
-LoggerPtr logger (Logger::getLogger ("thrudoc_replay"));
+LoggerPtr logger (Logger::getLogger ("thrudex_replay"));
 
 //print usage and die
 inline void usage ()
 {
-    cerr<<"thrudoc -f /path/to/thrudoc.conf"<<endl;
-    cerr<<"\tor create ~/.thrudoc"<<endl;
+    cerr<<"thrudex -f /path/to/thrudex.conf"<<endl;
+    cerr<<"\tor create ~/.thrudex"<<endl;
     cerr<<"\t-nb creates non-blocking server"<<endl;
     exit (-1);
@@ -71 +71 @@
 {
     public:
-        Replayer (shared_ptr<ThrudocBackend> backend, string current_filename,
+        Replayer (shared_ptr<ThrudexBackend> backend, string current_filename,
                   uint32_t delay_seconds) 
         {
@@ -83 +83 @@
             this->delay_seconds = delay_seconds;
 
-            this->handler = shared_ptr<ThrudocHandler>
-                (new ThrudocHandler (this->backend));
-            this->processor = shared_ptr<ThrudocProcessor>
-                (new ThrudocProcessor (this->handler));
+            shared_ptr<ThrudexHandler> handler = shared_ptr<ThrudexHandler>
+                (new ThrudexHandler (this->backend));
+            this->processor = shared_ptr<ThrudexProcessor>
+                (new ThrudexProcessor (handler));
 
             this->last_position_flush = 0;
@@ -105 +105 @@
                 }
             }
-            catch (ThrudocException e)
+            catch (ThrudexException e)
             {
                 LOG4CXX_WARN (logger, "last_position unknown, assuming epoch");
@@ -212 +212 @@
 
         TBinaryProtocolFactory protocol_factory;
-        shared_ptr<ThrudocBackend> backend;
-        shared_ptr<ThrudocHandler> handler;
-        shared_ptr<ThrudocProcessor> processor;
+        shared_ptr<ThrudexBackend> backend;
+        shared_ptr<ThrudexProcessor> processor;
         string current_filename;
         int64_t current_position;
@@ -226 +225 @@
 {
 
-    string conf_file = string (getenv ("HOME"))+"/.thrudoc";
+    string conf_file = string (getenv ("HOME"))+"/.thrudex";
     bool nonblocking = true;
 
@@ -263 +262 @@
         else 
         {
-            ThrudocException e;
+            ThrudexException e;
             e.what = "error opening log index file";
             LOG4CXX_ERROR (logger, e.what);
@@ -271 +270 @@
         if (log_filename.empty ())
         {
-            ThrudocException e;
+            ThrudexException e;
             e.what = "error log index file empty";
             LOG4CXX_ERROR (logger, e.what);
@@ -282 +281 @@
         // create our backend
         string which = ConfigManager->read<string> ("BACKEND", "mysql");
-        shared_ptr<ThrudocBackend> backend = create_backend (which, 1);
+        shared_ptr<ThrudexBackend> backend = create_backend (which, 1);
 
         int32_t delay_seconds = 

trunk/thrudoc/src/LogBackend.cpp

@@ -45 +45 @@
 
     this->set_backend (backend);
-    this->log_directory = log_directory;
-    this->num_ops = 0;
-    this->max_ops = max_ops;
-    this->sync_wait = sync_wait;
 
     // create our serializer
@@ -55 +51 @@
     msg_client = shared_ptr<ThrudocClient>(new ThrudocClient (msg_protocol));
 
-    // if our log directory doesn't exist, create it
-    if (!fs::is_directory (log_directory))
-    {
-        try
-        {
-            fs::create_directories (log_directory);
-        }
-        catch (exception e)
-        {
-            LOG4CXX_ERROR (logger, string ("log error: ") + e.what ());
-            throw e;
-        }
-    }
-
-    // try opening up the index
-    index_file.open (log_directory + "/" + LOG_FILE_PREFIX + "index", ios::in);
-
-    // get the last log file name
-    LOG4CXX_DEBUG (logger, "reading index");
-    string old_log_filename;
-    while (index_file.good ())
-    {
-        char row[256];
-        index_file.getline (row, 256);
-        if (strlen (row))
-        {
-            LOG4CXX_DEBUG (logger, string ("    log file=") + row);
-            old_log_filename = string (row);
-        }
-    }
-    // we're done reading
-    index_file.close ();
-    LOG4CXX_INFO (logger, string ("old_log_filename=") + old_log_filename);
-
-    // our new logfile
-    string new_log_filename = get_log_filename ();
-    LOG4CXX_INFO (logger, string ("new_log_filename=") + new_log_filename);
-
-    // if there's an old log
-    if (!old_log_filename.empty ())
-    {
-        // open the old log
-        open_log_client (old_log_filename, true);
-        // write a nextLog of the new logfile so that replayers can chain
-        send_nextLog (new_log_filename);
-    }
-    // then open the new log
-    open_log_client (new_log_filename, false);
-
-    // write the new logfile to the index
-    index_file.open (log_directory + "/" + LOG_FILE_PREFIX + "index", 
-                         ios::out | ios::app);
-    if (!index_file.is_open ())
-    {
-        ThrudocException e;
-        e.what = "error opening log index file";
-        LOG4CXX_ERROR (logger, e.what);
-        throw e;
-    }
-
-    index_file.write ((new_log_filename + "\n").c_str (), 
-                      new_log_filename.length () + 1);
-
-    // make sure our addition makes it to disk
-    index_file.flush ();
-}
-
-LogBackend::~LogBackend ()
-{
-    log_transport->close ();
-    index_file.close ();
+    file_logger = new FileLogger(log_directory, LOG_FILE_PREFIX, max_ops, 
+                                 sync_wait);
 }
 
 void LogBackend::put (const string & bucket, const string & key,
-          const string & value)
+                      const string & value)
 {
     this->get_backend ()->put (bucket, key, value);
@@ -140 +67 @@
         msg_transport->resetBuffer ();
 
-        send_log (raw_msg);
+        file_logger->send_log (raw_msg);
     }
     catch (TException e)
@@ -163 +90 @@
         msg_transport->resetBuffer ();
 
-        send_log (raw_msg);
+        file_logger->send_log (raw_msg);
     }
     catch (TException e)
@@ -179 +106 @@
     if (op == "roll_log")
     {
-        this->roll_log ();
+        file_logger->roll_log ();
         return "done";
     }
@@ -193 +120 @@
             msg_transport->resetBuffer ();
 
-            send_log (raw_msg);
+            file_logger->send_log (raw_msg);
         }
         catch (TException e)
@@ -219 +146 @@
         msg_transport->resetBuffer ();
 
-        send_log (raw_msg);
+        file_logger->send_log (raw_msg);
     }
     catch (TException e)
@@ -244 +171 @@
         msg_transport->resetBuffer ();
 
-        send_log (raw_msg);
+        file_logger->send_log (raw_msg);
     }
     catch (TException e)
@@ -257 +184 @@
     return ret;
 }
-
-string LogBackend::get_log_filename ()
-{
-    char buf[64];
-    sprintf (buf, "%s%d", LOG_FILE_PREFIX, (int)time (NULL));
-    return buf; 
-}
-
-void LogBackend::open_log_client (string log_filename, bool imediate_sync)
-{
-    LOG4CXX_INFO (logger, "open_log_client: log_filename=" + log_filename + 
-                  ", imediate_sync=" + (imediate_sync ? "true" : "false"));
-
-    // flush old log file
-    if (log_transport)
-        log_transport->flush ();
-
-    // and open up a new one
-    log_transport = shared_ptr<ThruFileWriterTransport>
-        (new ThruFileWriterTransport (log_directory + "/" + log_filename, 
-                                      imediate_sync ? 0 : this->sync_wait));
-    shared_ptr<TProtocol> log_protocol (new TBinaryProtocol (log_transport));
-    log_client = shared_ptr<EventLogClient> (new EventLogClient (log_protocol));
-}
-
 
 Event LogBackend::create_event (const string & message)
@@ -309 +211 @@
 }
 
-void LogBackend::send_nextLog (string new_log_filename)
-{
-    log_client->send_nextLog (new_log_filename);
-}
-
-void LogBackend::send_log (string raw_message)
-{
-    log_client->send_log (this->create_event (raw_message));
-
-    // this is going to be fuzzy b/c of multi-thread, but that's ok
-    this->num_ops++;
-
-    if (this->num_ops >= this->max_ops)
-    {
-        Guard g(log_mutex); 
-        if (this->num_ops >= this->max_ops)
-        {   
-            this->roll_log ();
-            this->num_ops = 0;
-        }
-    }
-}
-
-void LogBackend::roll_log ()
-{
-    string new_log_filename = get_log_filename ();
-    LOG4CXX_INFO (logger, "roll_log: new logfile=" + new_log_filename);
-
-    // time for a new file
-    // point to it in the old one
-    send_nextLog (new_log_filename);
-    // and open the new one
-    open_log_client (new_log_filename, false);
-    // add it to the index
-    index_file.write ((new_log_filename + "\n").c_str (), 
-                      new_log_filename.length () + 1);
-    index_file.flush ();
-}
-
 #endif /* HAVE_LIBBOOST_FILESYSTEM */

trunk/thrudoc/src/LogBackend.h

@@ -22 +22 @@
 
 #include "ThrudocPassthruBackend.h"
-#include "EventLog.h"
-#include "ThruFileTransport.h"
+
+#include <EventLog.h>
+#include <FileLogger.h>
 
 #define LOG_FILE_PREFIX "thrudoc-log."
@@ -33 +34 @@
                     const std::string &log_directory, unsigned int max_ops,
                     unsigned int sync_wait);
-        ~LogBackend ();
 
         void put (const std::string & bucket, const std::string & key,
@@ -53 +53 @@
         boost::shared_ptr<facebook::thrift::transport::TMemoryBuffer> msg_transport;
         boost::shared_ptr<thrudoc::ThrudocClient> msg_client;
+        Event create_event (const std::string &msg);
 
-        // this will be used to write to the log file
-        boost::shared_ptr<ThruFileWriterTransport> log_transport;
-        boost::shared_ptr<EventLogClient> log_client;
-
-        facebook::thrift::concurrency::Mutex log_mutex;
-        std::string log_directory;
-        boost::filesystem::fstream index_file;
-        unsigned int num_ops;
-        unsigned int max_ops;
-        unsigned int sync_wait;
-
-        std::string get_log_filename ();
-        void open_log_client (std::string log_filename, bool imediate_sync);
-        void roll_log ();
-        Event create_event (const std::string &msg);
-        void send_log (std::string raw_message);
-        void send_nextLog (std::string new_log_filename);
+        FileLogger * file_logger;
 };
 

trunk/thrudoc/src/thrudoc.cpp

@@ -17 +17 @@
 #include <concurrency/PosixThreadFactory.h>
 #include <protocol/TBinaryProtocol.h>
+#include <protocol/TCountingProtocol.h>
 #include <server/TSimpleServer.h>
 #include <server/TThreadPoolServer.h>
@@ -93 +94 @@
         //Startup Services
         shared_ptr<TProtocolFactory>
-            protocolFactory (new TBinaryProtocolFactory ());
+            protF (new TBinaryProtocolFactory ());
+        shared_ptr<TProtocolFactory>
+            protocolFactory (new TCountingProtocolFactory (protF));
 
         string which = ConfigManager->read<string> ("BACKEND", "mysql");

trunk/thrudoc/src/thrudoc_replay.cpp

@@ -83 +83 @@
             this->delay_seconds = delay_seconds;
 
-            this->handler = shared_ptr<ThrudocHandler>
+            shared_ptr<ThrudocHandler> handler = shared_ptr<ThrudocHandler>
                 (new ThrudocHandler (this->backend));
             this->processor = shared_ptr<ThrudocProcessor>
-                (new ThrudocProcessor (this->handler));
+                (new ThrudocProcessor (handler));
 
             this->last_position_flush = 0;
@@ -213 +213 @@
         TBinaryProtocolFactory protocol_factory;
         shared_ptr<ThrudocBackend> backend;
-        shared_ptr<ThrudocHandler> handler;
         shared_ptr<ThrudocProcessor> processor;
         string current_filename;

@@ -7 +7 @@
 thrucommon.pc
 thrurecorder
+*.loT

@@ -8 +8 @@
 COPYING
 INSTALL
+libtool
 Makefile
 Makefile.in

@@ -1 +1 @@
 .deps
 gen-cpp
+.libs
 Makefile
 Makefile.in
 *.swp
 thrudex
+thrudex_bench
+thrudex_replay