Merge pull request #34 from vikramrajkumar/master

Log rotation improvements

.gitignore

@@ -33,9 +33,11 @@ ZERO_CHECK
 Debug/
 Release/
 
+CMakeCache.txt
 CMakeFiles
 Makefile
-cmake_install.cmake
+*.cmake
+
 libfc.a
 libfc_debug.a
 
@@ -45,6 +47,7 @@ fc_automoc.cpp
 git_revision.cpp
 GitSHA3.cpp
 
+lzma_test
 ntp_test
 task_cancel_test
 udt_client

CMakeLists.txt

@@ -33,7 +33,7 @@ endif()
 SET (ORIGINAL_LIB_SUFFIXES ${CMAKE_FIND_LIBRARY_SUFFIXES})
 
 SET(BOOST_COMPONENTS)
-LIST(APPEND BOOST_COMPONENTS thread date_time system filesystem program_options signals serialization chrono unit_test_framework context locale)
+LIST(APPEND BOOST_COMPONENTS thread date_time system filesystem program_options signals serialization chrono unit_test_framework context locale iostreams)
 
 IF( WIN32 )
   MESSAGE(STATUS "Configuring fc to build on Win32")
@@ -237,6 +237,9 @@ target_link_libraries( udt_server fc udt )
 add_executable( udt_client tests/udtc.cpp )
 target_link_libraries( udt_client fc udt )
 
+add_executable( lzma_test tests/lzma_test.cpp )
+target_link_libraries( lzma_test fc )
+
 #add_executable( test_compress tests/compress.cpp )
 #target_link_libraries( test_compress fc )
 #add_executable( test_aes tests/aes_test.cpp )

include/fc/compress/lzma.hpp

@@ -1,9 +1,19 @@
 #pragma once
+
+#include <fc/filesystem.hpp>
 #include <vector>
 
 namespace fc {
 
-  std::vector<char> lzma_compress( const std::vector<char>& in );
+std::vector<char> lzma_compress( const std::vector<char>& in );
-  std::vector<char> lzma_decompress( const std::vector<char>& compressed );
+std::vector<char> lzma_decompress( const std::vector<char>& compressed );
+
+void lzma_compress_file( const path& src_path,
+                         const path& dst_path,
+                         unsigned char level = 5,
+                         unsigned int dict_size = (1 << 20) );
+
+void lzma_decompress_file( const path& src_path,
+                           const path& dst_path );
 
 } // namespace fc

include/fc/log/file_appender.hpp

@@ -1,13 +1,12 @@
 #pragma once
+
+#include <fc/filesystem.hpp>
 #include <fc/log/appender.hpp>
 #include <fc/log/logger.hpp>
-#include <fc/filesystem.hpp>
 #include <fc/time.hpp>
 
 namespace fc {
 
-class varaint;
-
 class file_appender : public appender {
     public:
          struct config {
@@ -20,6 +19,7 @@ class file_appender : public appender {
             bool                               rotate;
             microseconds                       rotation_interval;
             microseconds                       rotation_limit;
+            bool                               rotation_compression;
          };
          file_appender( const variant& args );
          ~file_appender();
@@ -32,4 +32,5 @@ class file_appender : public appender {
 } // namespace fc
 
 #include <fc/reflect/reflect.hpp>
-FC_REFLECT( fc::file_appender::config, (format)(filename)(flush)(truncate)(rotate)(rotation_interval)(rotation_limit) )
+FC_REFLECT( fc::file_appender::config,
+            (format)(filename)(flush)(truncate)(rotate)(rotation_interval)(rotation_limit)(rotation_compression) )

include/fc/thread/thread.hpp

@@ -9,7 +9,7 @@ namespace fc {
 
   class thread {
     public:
-      thread( const char* name = "" );
+      thread( const std::string& name = "" );
       thread( thread&& m );
       thread& operator=(thread&& t );
 

src/compress/lzma.cpp

@@ -1,8 +1,11 @@
+#include <boost/iostreams/device/mapped_file.hpp>
 #include <fc/compress/lzma.hpp>
 #include <fc/exception/exception.hpp>
-
+#include <fc/io/fstream.hpp>
 #include <lzma_c.h>
 
+#include <iostream>
+
 namespace fc {
 
 std::vector<char> lzma_compress(const std::vector<char>& in)
@@ -50,4 +53,148 @@ std::vector<char> lzma_decompress( const std::vector<char>& compressed )
   return out;
 }
 
+struct lzma_file_ctx
+{
+    const unsigned char* src_buf;
+    size_t src_len;
+
+    path dst_path;
+};
+
+static int lzma_file_input_callback( void* input_ctx, void* input_buf, size_t* input_len )
+{
+    FC_ASSERT( input_ctx != NULL );
+    FC_ASSERT( input_buf != NULL );
+
+    const auto ctx = ( struct lzma_file_ctx* )input_ctx;
+    const auto size = ( ctx->src_len < *input_len ) ? ctx->src_len : *input_len;
+
+    if( size > 0 )
+    {
+        memcpy( input_buf, ( void * )ctx->src_buf, size );
+        ctx->src_buf += size;
+        ctx->src_len -= size;
+    }
+
+    *input_len = size;
+
+    return 0;
+}
+
+static size_t lzma_file_output_callback( void* output_ctx, const void* output_buf, size_t output_len )
+{
+    FC_ASSERT( output_ctx != NULL );
+    FC_ASSERT( output_buf != NULL );
+
+    const auto ctx = ( struct lzma_file_ctx* )output_ctx;
+
+    if( output_len > 0 )
+    {
+        size_t dst_len = 0;
+        if( !exists( ctx->dst_path ) )
+        {
+            ofstream fs( ctx->dst_path );
+            fs.close();
+        }
+        else
+        {
+            dst_len = file_size( ctx->dst_path );
+        }
+
+        resize_file( ctx->dst_path, dst_len + output_len );
+
+        boost::iostreams::mapped_file_sink dst_file;
+        dst_file.open( ctx->dst_path.string() );
+        FC_ASSERT( dst_file.is_open() );
+
+        memcpy( ( void* )(dst_file.data() + dst_len), output_buf, output_len);
+
+        dst_file.close();
+    }
+
+    return output_len;
+}
+
+void lzma_compress_file( const path& src_path,
+                         const path& dst_path,
+                         unsigned char level,
+                         unsigned int dict_size )
+{
+    FC_ASSERT( exists( src_path ) );
+    FC_ASSERT( !exists( dst_path ) );
+
+    boost::iostreams::mapped_file_source src_file;
+    src_file.open( src_path.string() );
+    FC_ASSERT( src_file.is_open() );
+
+    elzma_compress_handle handle = NULL;
+    handle = elzma_compress_alloc();
+    FC_ASSERT( handle != NULL );
+
+    struct lzma_file_ctx ctx;
+    ctx.src_buf = ( const unsigned char* )src_file.data();
+    ctx.src_len = src_file.size();
+    ctx.dst_path = dst_path;
+
+    auto rc = elzma_compress_config( handle,
+                                     ELZMA_LC_DEFAULT,
+                                     ELZMA_LP_DEFAULT,
+                                     ELZMA_PB_DEFAULT,
+                                     level,
+                                     dict_size,
+                                     elzma_file_format::ELZMA_lzma,
+                                     ctx.src_len );
+
+    try
+    {
+        FC_ASSERT( rc == ELZMA_E_OK );
+    }
+    catch( ... )
+    {
+        elzma_compress_free( &handle );
+        throw;
+    }
+
+    rc = elzma_compress_run( handle,
+                             lzma_file_input_callback,
+                             ( void * )&ctx,
+                             lzma_file_output_callback,
+                             ( void * )&ctx,
+                             NULL,
+                             NULL );
+
+    elzma_compress_free( &handle );
+    FC_ASSERT( rc == ELZMA_E_OK );
+}
+
+void lzma_decompress_file( const path& src_path,
+                           const path& dst_path )
+{
+    FC_ASSERT( exists( src_path ) );
+    FC_ASSERT( !exists( dst_path ) );
+
+    boost::iostreams::mapped_file_source src_file;
+    src_file.open( src_path.string() );
+    FC_ASSERT( src_file.is_open() );
+
+    elzma_decompress_handle handle = NULL;
+    handle = elzma_decompress_alloc();
+    FC_ASSERT( handle != NULL );
+
+    struct lzma_file_ctx ctx;
+    ctx.src_buf = ( const unsigned char* )src_file.data();
+    ctx.src_len = src_file.size();
+    ctx.dst_path = dst_path;
+
+    auto rc = elzma_decompress_run( handle,
+                                    lzma_file_input_callback,
+                                    ( void * )&ctx,
+                                    lzma_file_output_callback,
+                                    ( void * )&ctx,
+                                    elzma_file_format::ELZMA_lzma );
+
+    elzma_decompress_free( &handle );
+    FC_ASSERT( rc == ELZMA_E_OK );
+}
+
 } // namespace fc

src/log/file_appender.cpp

@@ -1,22 +1,31 @@
-#include <fc/log/file_appender.hpp>
+#include <fc/compress/lzma.hpp>
-#include <boost/thread/mutex.hpp>
+#include <fc/exception/exception.hpp>
-#include <fc/thread/scoped_lock.hpp>
-#include <boost/thread/mutex.hpp>
 #include <fc/io/fstream.hpp>
-#include <fc/variant.hpp>
+#include <fc/log/file_appender.hpp>
 #include <fc/reflect/variant.hpp>
+#include <fc/thread/scoped_lock.hpp>
+#include <fc/thread/thread.hpp>
+#include <fc/variant.hpp>
+#include <boost/date_time/posix_time/posix_time.hpp>
+#include <boost/thread/mutex.hpp>
 #include <iomanip>
+#include <queue>
 #include <sstream>
-#include <fc/filesystem.hpp>
-
 
 namespace fc {
+
+   static const string  compression_extension( ".lzma" );
+
    class file_appender::impl : public fc::retainable {
       public:
-         config              cfg;
+         config                     cfg;
-         ofstream            out;
+         ofstream                   out;
-         boost::mutex        slock;
+         boost::mutex               slock;
-         time_point_sec      current_file_start_time;
+
+      private:
+         future<void>               _rotation_task;
+         time_point_sec             _current_file_start_time;
+         std::unique_ptr<thread>    _compression_thread;
 
          time_point_sec get_file_start_time( const time_point_sec& timestamp, const microseconds& interval )
          {
@@ -24,36 +33,138 @@ namespace fc {
              const auto file_number = timestamp.sec_since_epoch() / interval_seconds;
              return time_point_sec( file_number * interval_seconds );
          }
+
+         string timestamp_to_string( const time_point_sec& timestamp )
+         {
+             auto ptime = boost::posix_time::from_time_t( time_t ( timestamp.sec_since_epoch() ) );
+             return boost::posix_time::to_iso_string( ptime );
+         }
+
+         time_point_sec string_to_timestamp( const string& str )
+         {
+             return time_point::from_iso_string( str );
+         }
+
+         void compress_file( const string& filename )
+         {
+             FC_ASSERT( cfg.rotate && cfg.rotation_compression );
+             FC_ASSERT( _compression_thread );
+             if( !_compression_thread->is_current() )
+             {
+                 _compression_thread->async( [this, filename]() { compress_file( filename ); } ).wait();
+                 return;
+             }
+
+             try
+             {
+                 lzma_compress_file( filename, filename + compression_extension );
+                 remove_all( filename );
+             }
+             catch( ... )
+             {
+             }
+         }
+
+      public:
+         impl( const config& c) : cfg( c )
+         {
+             if( cfg.rotate )
+             {
+                 if( cfg.rotation_compression )
+                     _compression_thread.reset( new thread( "compression") );
+
+                 _rotation_task = async( [this]() { rotate_files( true ); } );
+             }
+         }
+
+         ~impl()
+         {
+             try
+             {
+                 _rotation_task.cancel_and_wait();
+                 if( _compression_thread ) _compression_thread->quit();
+             }
+             catch( ... )
+             {
+             }
+         }
+
+         void rotate_files( bool initializing = false )
+         {
+             FC_ASSERT( cfg.rotate );
+             const auto now = time_point::now();
+             const auto start_time = get_file_start_time( now, cfg.rotation_interval );
+             const auto timestamp_string = timestamp_to_string( start_time );
+             const auto link_filename = cfg.filename.string();
+             const auto log_filename = link_filename + "." + timestamp_string;
+
+             if( !initializing )
+             {
+                 if( start_time <= _current_file_start_time ) return;
+                 fc::scoped_lock<boost::mutex> lock( slock );
+                 out.flush();
+                 out.close();
+             }
+
+             out.open( log_filename.c_str() );
+             remove_all( link_filename );
+             create_hard_link( log_filename, link_filename );
+
+             /* Delete old log files */
+             const auto limit_time = now - cfg.rotation_limit;
+             auto itr = directory_iterator( fc::path( link_filename ).parent_path() );
+             for( ; itr != directory_iterator(); itr++ )
+             {
+                 try
+                 {
+                     const auto current_filename = itr->string();
+                     auto current_pos = current_filename.find( link_filename );
+                     if( current_pos != 0 ) continue;
+                     current_pos = link_filename.size() + 1;
+                     const auto current_timestamp_str = string( current_filename.begin() + current_pos, /* substr not working */
+                                                                current_filename.begin() + current_pos + timestamp_string.size() );
+                     const auto current_timestamp = string_to_timestamp( current_timestamp_str );
+                     if( current_timestamp < start_time )
+                     {
+                         if( current_timestamp < limit_time || file_size( current_filename ) <= 0 )
+                         {
+                             remove_all( current_filename );
+                             continue;
+                         }
+
+                         if( !cfg.rotation_compression ) continue;
+                         if( current_filename.find( compression_extension ) != string::npos ) continue;
+                         compress_file( current_filename );
+                     }
+                 }
+                 catch( ... )
+                 {
+                 }
+             }
+
+             _current_file_start_time = start_time;
+             _rotation_task = schedule( [this]() { rotate_files(); }, now + cfg.rotation_interval );
+         }
    };
    file_appender::config::config( const fc::path& p  )
    :format( "${timestamp} ${thread_name} ${context} ${file}:${line} ${method} ${level}]  ${message}" ),
-   filename(p),flush(true),truncate(true),rotate(false){}
+   filename(p),flush(true),truncate(true),rotate(false),rotation_compression(true){}
 
    file_appender::file_appender( const variant& args )
-   :my( new impl() )
+   :my( new impl( args.as<config>() ) )
    {
       std::string log_filename;
       try
       {
-         my->cfg = args.as<config>(); 
          log_filename = my->cfg.filename.string();
 
          fc::create_directories( fc::path( log_filename ).parent_path() );
 
-         if( my->cfg.rotate )
+         if( !my->cfg.rotate ) my->out.open( log_filename.c_str() );
-         {
-             const auto start_time = my->get_file_start_time( time_point::now(), my->cfg.rotation_interval );
-             // TODO: Convert to proper timestamp string
-             log_filename += "." + std::to_string( start_time.sec_since_epoch() );
-             my->current_file_start_time = start_time;
-         }
-
-         my->out.open( log_filename.c_str() );
       }
       catch( ... )
       {
-         std::cerr << "error opening log file: " << my->cfg.filename.string() << "\n";
+         std::cerr << "error opening log file: " << log_filename << "\n";
-         //elog( "%s", fc::except_str().c_str() );
       }
    }
    file_appender::~file_appender(){}
@@ -61,7 +172,6 @@ namespace fc {
    // MS THREAD METHOD  MESSAGE \t\t\t File:Line
    void file_appender::log( const log_message& m )
    {
-      const auto now = time_point::now();
       std::stringstream line;
       //line << (m.get_context().get_timestamp().time_since_epoch().count() % (1000ll*1000ll*60ll*60))/1000 <<"ms ";
       line << std::string(m.get_context().get_timestamp()) << " ";
@@ -88,49 +198,11 @@ namespace fc {
 
       // fc::string fmt_str = fc::format_string( my->cfg.format, mutable_variant_object(m.get_context())( "message", message)  );
 
-      /* Write to log file (rotating file beforehand if necessary) */
       {
         fc::scoped_lock<boost::mutex> lock( my->slock );
-
-        if( my->cfg.rotate )
-        {
-            const auto start_time = my->get_file_start_time( now, my->cfg.rotation_interval );
-            if( start_time > my->current_file_start_time )
-            {
-               my->out.close();
-               auto log_filename = my->cfg.filename.string();
-
-               /* Delete old log files */
-               // TODO: Delete on startup as well
-               const auto limit_time = now - my->cfg.rotation_limit;
-               auto itr = directory_iterator( fc::path( log_filename ).parent_path() );
-               for( ; itr != directory_iterator(); itr++ )
-               {
-                   const auto current_filename = itr->string();
-                   const auto current_pos = current_filename.find( log_filename );
-                   if( current_pos != 0 ) continue;
-                   const auto current_timestamp = current_filename.substr( log_filename.size() + 1 );
-                   try
-                   {
-                       if( std::stoi( current_timestamp ) < limit_time.sec_since_epoch() )
-                           remove_all( current_filename );
-                       //else compress if not already compressed
-                   }
-                   catch( ... )
-                   {
-                   }
-               }
-
-               // TODO: Convert to proper timestamp string
-               log_filename += "." + std::to_string( start_time.sec_since_epoch() );
-               my->out.open( log_filename.c_str() );
-
-               my->current_file_start_time = start_time;
-            }
-        }
-
         my->out << line.str() << "\t\t\t" << m.get_context().get_file() <<":"<<m.get_context().get_line_number()<<"\n";
         if( my->cfg.flush ) my->out.flush();
       }
    }
-}
+
+} // fc

src/thread/thread.cpp

@@ -70,11 +70,11 @@ namespace fc {
       return t;
    }
 
-   thread::thread( const char* name  ) {
+   thread::thread( const std::string& name  ) {
       promise<void>::ptr p(new promise<void>());
       boost::thread* t = new boost::thread( [this,p,name]() {
           try {
-		    set_thread_name(name); // set thread's name for the debugger to display
+		    set_thread_name(name.c_str()); // set thread's name for the debugger to display
             this->my = new thread_d(*this);
             current_thread() = this;
             p->set_value();
@@ -126,17 +126,15 @@ namespace fc {
    void          thread::debug( const fc::string& d ) { /*my->debug(d);*/ }
 
    void thread::quit() {
-     //if quiting from a different thread, start quit task on thread.
+     //if quitting from a different thread, start quit task on thread.
      //If we have and know our attached boost thread, wait for it to finish, then return.
       if( &current() != this ) {
           async( [=](){quit();} );//.wait();
           if( my->boost_thread ) {
             auto n = name();
-            ilog( "joining... ${n}", ("n",n) );//n.c_str() );
             my->boost_thread->join();
             delete my;
             my = nullptr;
-            ilog( "done joining...${n}", ("n",n) ); //n.c_str() );
           }
           return;
       }

tests/lzma_test.cpp

@@ -0,0 +1,24 @@
+#include <fc/compress/lzma.hpp>
+#include <fc/filesystem.hpp>
+
+#include <iostream>
+#include <string>
+
+using namespace fc;
+
+int main( int argc, char** argv )
+{
+    if( argc != 2 )
+    {
+        std::cout << "usage: " << argv[0] << " <filename>\n";
+        exit( -1 );
+    }
+
+    auto src = std::string( argv[1] );
+    auto dst = src + ".compressed";
+    lzma_compress_file( src, dst );
+
+    lzma_decompress_file( dst, src + ".decompressed" );
+
+    return 0;
+}