mario split log along with producer.put, it may cause the log with d…

…iffent

 sizes on master and slave;
 add daemonize config item;

conf/pika.conf

@@ -16,3 +16,5 @@ timeout : 60
 requirepass : 
 # Dump Prefix
 dump_prefix : 
+# daemonize  [yes | no]
+#daemonize : yes

include/pika_conf.h

@@ -21,6 +21,7 @@ class PikaConf : public BaseConf
     char* db_path()         { RWLock l(&rwlock_, false); return db_path_; }
     int write_buffer_size() { RWLock l(&rwlock_, false); return write_buffer_size_; }
     int timeout()           { RWLock l(&rwlock_, false); return timeout_; }
+    bool daemonize()        { RWLock l(&rwlock_, false); return daemonize_; }
     char* requirepass()     { RWLock l(&rwlock_, false); return requirepass_; }
     char* conf_path()       { RWLock l(&rwlock_, false); return conf_path_; }
     char* dump_prefix()     { RWLock l(&rwlock_, false); return dump_prefix_; }
@@ -46,6 +47,7 @@ class PikaConf : public BaseConf
     char db_path_[PIKA_WORD_SIZE];
     int write_buffer_size_;
     int log_level_;
+    bool daemonize_;
     int timeout_;
     char requirepass_[PIKA_WORD_SIZE];
     char dump_prefix_[PIKA_WORD_SIZE];

src/pika.cc

@@ -79,8 +79,7 @@ static void usage()
             "usage: pika [-hd] [-c conf/file]\n"
             "\t-h               -- show this help\n"
             "\t-c conf/file     -- config file \n"
-            "\t-d               -- daemonize\n"
-            "  example: ./output/bin/pika -c ./conf/pika.conf -d\n"
+            "  example: ./output/bin/pika -c ./conf/pika.conf\n"
            );
 }
 
@@ -112,7 +111,6 @@ void create_pid_file(void) {
 int main(int argc, char **argv)
 {
     bool path_opt = false;
-    bool daemon_flag = false;
     char c;
     char path[PIKA_LINE_SIZE];
 
@@ -121,7 +119,7 @@ int main(int argc, char **argv)
         exit(-1);
     }
 
-    while (-1 != (c = getopt(argc, argv, "c:hd"))) {
+    while (-1 != (c = getopt(argc, argv, "c:h"))) {
         switch (c) {
         case 'c':
             snprintf(path, PIKA_LINE_SIZE, "%s", optarg);
@@ -130,9 +128,6 @@ int main(int argc, char **argv)
         case 'h':
             usage();
             return 0;
-        case 'd':
-            daemon_flag = true;
-            break;
         default:
             usage();
             return 0;
@@ -156,7 +151,7 @@ int main(int argc, char **argv)
     /*
      * daemonize if needed
      */
-    if (daemon_flag) daemonize();
+    if (g_pikaConf->daemonize()) daemonize();
 
     /*
      * init the glog config
@@ -168,7 +163,7 @@ int main(int argc, char **argv)
      */
     pika_signal_setup();
 
-    if (daemon_flag) create_pid_file();
+    if (g_pikaConf->daemonize()) create_pid_file();
 
 
     /*
@@ -416,7 +411,7 @@ int main(int argc, char **argv)
     /*
      * shutdown server
      */
-    if (daemon_flag) {
+    if (g_pikaConf->daemonize()) {
         unlink(PIKA_DEFAULT_PID_FILE);
     }
 

src/pika_admin.cc

@@ -408,6 +408,10 @@ void ConfigCmd::Do(std::list<std::string> &argv, std::string &ret) {
             ret = "*2\r\n";
             EncodeString(&ret, "requirepass");
             EncodeString(&ret, g_pikaConf->requirepass());
+        } else if (conf_item == "daemonize") {
+            ret = "*2\r\n";
+            EncodeString(&ret, "daemonize");
+            EncodeString(&ret, g_pikaConf->daemonize() ? "yes" : "no");
         } else {
             ret = "-ERR No such configure item\r\n";
         }

src/pika_conf.cc

@@ -18,5 +18,15 @@ PikaConf::PikaConf(const char* path) :
     getConfStr("requirepass", requirepass_);
     getConfStr("dump_prefix", dump_prefix_);
 
+    char str[PIKA_WORD_SIZE];
+    getConfStr("daemonize", str);
+
+    if (strcmp(str, "yes") == 0) {
+        daemonize_ = true;
+    } else {
+        daemonize_ = false;
+    }
+
+
     pthread_rwlock_init(&rwlock_, NULL);
 }

third/mario/src/mario.cc

@@ -73,7 +73,7 @@ Mario::Mario(int32_t retry)
     }
 
     producer_ = new Producer(writefile_, version_);
-    env_->StartThread(&Mario::SplitLogWork, this);
+    //env_->StartThread(&Mario::SplitLogWork, this);
 
 }
 
@@ -219,7 +219,7 @@ Status Mario::AppendBlank(WritableFile *file, uint64_t len) {
     }
 
     // Append a msg which occupy the remain part of the last block
-    uint32_t n = (len % kBlockSize) - kHeaderSize;
+    uint32_t n = (uint32_t) ((len % kBlockSize) - kHeaderSize);
 
     char buf[kBlockSize];
     buf[0] = static_cast<char>(n & 0xff);
@@ -386,6 +386,24 @@ Status Mario::Put(const std::string &item)
 
     {
     MutexLock l(&mutex_);
+
+    /* Check to roll log file */
+    uint64_t filesize = writefile_->Filesize();
+    //log_info("filesize %llu kMmapSize %llu\n", filesize, kMmapSize);
+    if (filesize > kMmapSize) {
+        //log_info("roll file filesize %llu kMmapSize %llu\n", filesize, kMmapSize);
+        delete producer_;
+        delete writefile_;
+        pronum_++;
+        std::string profile = NewFileName(filename_, pronum_);
+        env_->NewWritableFile(profile, &writefile_);
+        version_->set_pro_offset(0);
+        version_->set_pronum(pronum_);
+        version_->StableSave();
+        version_->debug();
+        producer_ = new Producer(writefile_, version_);
+    }
+
     s = producer_->Produce(Slice(item.data(), item.size()));
     if (s.ok()) {
         version_->plus_item_num();
@@ -403,6 +421,24 @@ Status Mario::Put(const char* item, int len)
 
     {
     MutexLock l(&mutex_);
+
+    /* Check to roll log file */
+    uint64_t filesize = writefile_->Filesize();
+    if (filesize > kMmapSize) {
+        //log_info("roll file filesize %llu kMmapSize %llu\n", filesize, kMmapSize);
+        delete producer_;
+        delete writefile_;
+        pronum_++;
+        std::string profile = NewFileName(filename_, pronum_);
+        env_->NewWritableFile(profile, &writefile_);
+        version_->set_pro_offset(0);
+        version_->set_pronum(pronum_);
+        version_->StableSave();
+        version_->debug();
+        producer_ = new Producer(writefile_, version_);
+    }
+
+
     s = producer_->Produce(Slice(item, len));
     if (s.ok()) {
         version_->plus_item_num();