Merge branch 'OpenAtomFoundation:unstable' into ttl_logic_time

.github/pr-title-checker-config.json

@@ -4,7 +4,7 @@
     "color": "B60205"
   },
   "CHECKS": {
-    "regexp": "^(feat|fix|test|refactor|chore|upgrade|style|docs|perf|build|ci|revert)(\\(.*\\))?:.*",
+    "regexp": "^(feat|fix|test|refactor|chore|upgrade|style|docs|perf|build|ci|revert)(\\(.*\\))?:[^\u4e00-\u9fa5]+$",
     "ignoreLabels": [
       "ignore-title"
     ]

.github/workflows/pr-title-checker.yaml

@@ -12,7 +12,7 @@ jobs:
   check:
     runs-on: ubuntu-latest
     steps:
-      - uses: thehanimo/[email protected].1
+      - uses: thehanimo/[email protected].2
         with:
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
           pass_on_octokit_error: false

.github/workflows/release.yml

@@ -16,7 +16,7 @@ jobs:
         include:
           - os: ubuntu-latest
             name: ${{ github.event.repository.name }}-${{ github.ref_name }}-ubuntu-amd64.tar.gz
-          - os: macos-latest
+          - os: macos-12
             name: ${{ github.event.repository.name }}-${{ github.ref_name }}-macos-amd64.tar.gz
 
     runs-on: ${{ matrix.os }}

CMakeLists.txt

@@ -56,6 +56,9 @@ endif()
 if(CMAKE_SYSTEM_NAME MATCHES "Darwin")
    set(CMAKE_CXX_FLAGS "-pthread")
    add_definitions(-DOS_MACOSX)
+elseif (CMAKE_SYSTEM_NAME MATCHES "FreeBSD")
+    set(CMAKE_CXX_FLAGS "-pthread")
+    add_definitions(-DOS_FREEBSD)
 elseif(CMAKE_SYSTEM_NAME MATCHES "Linux")
    if(CMAKE_CXX_COMPILER_ID STREQUAL "Clang")
      set(CMAKE_EXE_LINKER_FLAGS "-stdlib=libc++ -fuse-ld=lld -lc++ -lc++abi ${CMAKE_EXE_LINKER_FLAGS}")
@@ -66,7 +69,7 @@ elseif(CMAKE_SYSTEM_NAME MATCHES "Linux")
    endif()
    add_definitions(-DOS_LINUX)
 else()
-   message(FATAL_ERROR "only support linux or macOs")
+   message(FATAL_ERROR "only support linux or macOs or FreeBSD")
 endif()
 
 if(HOST_ARCH MATCHES "x86_64" OR HOST_ARCH MATCHES "i386")

conf/pika.conf

@@ -317,6 +317,11 @@ max-write-buffer-num : 2
 # whether the key exists. Setting this value too high may hurt performance.
 min-write-buffer-number-to-merge : 1
 
+# The total size of wal files, when reaches this limit, rocksdb will force the flush of column-families
+# whose memtables are backed by the oldest live WAL file. Also used to control the rocksdb open time when
+# process restart.
+max-total-wal-size : 1073741824
+
 # rocksdb level0_stop_writes_trigger
 level0-stop-writes-trigger : 36
 

include/pika_db.h

@@ -24,11 +24,11 @@ struct KeyScanInfo {
   time_t start_time = 0;
   std::string s_start_time;
   int32_t duration = -3;
-  std::vector<storage::KeyInfo> key_infos;  // the order is strings, hashes, lists, zsets, sets
+  std::vector<storage::KeyInfo> key_infos;  // the order is strings, hashes, lists, zsets, sets, streams
   bool key_scaning_ = false;
   KeyScanInfo() :
                   s_start_time("0"),
-                  key_infos({{0, 0, 0, 0}, {0, 0, 0, 0}, {0, 0, 0, 0}, {0, 0, 0, 0}, {0, 0, 0, 0}})
+                  key_infos({{0, 0, 0, 0}, {0, 0, 0, 0}, {0, 0, 0, 0}, {0, 0, 0, 0}, {0, 0, 0, 0}, {0, 0, 0, 0}})
   {}
 };
 

include/pika_server.h

@@ -7,12 +7,14 @@
 #define PIKA_SERVER_H_
 
 #include <shared_mutex>
-#if defined(__APPLE__)
+
+#if defined(__APPLE__) || defined(__FreeBSD__)
 #  include <sys/mount.h>
 #  include <sys/param.h>
 #else
 #  include <sys/statfs.h>
 #endif
+
 #include <memory>
 #include <set>
 

src/acl.cc

@@ -489,15 +489,14 @@ void Acl::InitLimitUser(const std::string& bl, bool limit_exist) {
   auto u = GetUser(DefaultLimitUser);
   if (limit_exist) {
     if (!bl.empty()) {
-      u->SetUser("+@all");
       for(auto& cmd : blacklist) {
         cmd = pstd::StringTrim(cmd, " ");
         u->SetUser("-" + cmd);
       }
       u->SetUser("on");
-      if (!pass.empty()) {
-        u->SetUser(">"+pass);
-      }
+    }
+    if (!pass.empty()) {
+      u->SetUser(">"+pass);
     }
   } else {
     if (pass.empty()) {

src/net/CMakeLists.txt

@@ -15,7 +15,7 @@ add_subdirectory(examples)
 
 if(${CMAKE_SYSTEM_NAME} MATCHES "Linux")
   list(FILTER DIR_SRCS EXCLUDE REGEX ".net_kqueue.*")
-elseif(${CMAKE_SYSTEM_NAME} MATCHES "Darwin")
+elseif(${CMAKE_SYSTEM_NAME} MATCHES "Darwin" OR ${CMAKE_SYSTEM_NAME} MATCHES "FreeBSD")
   list(FILTER DIR_SRCS EXCLUDE REGEX ".net_epoll.*")
 endif()
 

src/net/src/net_interfaces.cc

@@ -12,10 +12,14 @@
 #include <arpa/inet.h>
 #include <ifaddrs.h>
 
-#if defined(__APPLE__)
+#if defined(__APPLE__) || defined(__FreeBSD__)
+#  include <arpa/inet.h>
+#  include <ifaddrs.h>
 #  include <net/if.h>
+#  include <netinet/in.h>
 #  include <sys/ioctl.h>
 #  include <sys/socket.h>
+#  include <sys/types.h>
 #  include <unistd.h>
 
 #  include "pstd/include/pstd_defer.h"
@@ -31,7 +35,7 @@
 #include "pstd/include/xdebug.h"
 
 std::string GetDefaultInterface() {
-#if defined(__APPLE__)
+#if defined(__APPLE__) || defined(__FreeBSD__)
   std::string name("lo0");
 
   int fd = socket(AF_INET, SOCK_DGRAM, 0);

src/pika_admin.cc

@@ -1201,7 +1201,7 @@ void InfoCmd::InfoKeyspace(std::string& info) {
       key_scan_info = db_item.second->GetKeyScanInfo();
       key_infos = key_scan_info.key_infos;
       duration = key_scan_info.duration;
-      if (key_infos.size() != 5) {
+      if (key_infos.size() != (size_t)(storage::DataType::kNones)) {
         info.append("info keyspace error\r\n");
         return;
       }
@@ -1227,6 +1227,8 @@ void InfoCmd::InfoKeyspace(std::string& info) {
                  << ", invalid_keys=" << key_infos[3].invaild_keys << "\r\n";
       tmp_stream << db_name << " Sets_keys=" << key_infos[4].keys << ", expires=" << key_infos[4].expires
                  << ", invalid_keys=" << key_infos[4].invaild_keys << "\r\n\r\n";
+      tmp_stream << db_name << " Streams_keys=" << key_infos[5].keys << ", expires=" << key_infos[5].expires
+                 << ", invalid_keys=" << key_infos[5].invaild_keys << "\r\n\r\n";
     }
   }
   info.append(tmp_stream.str());
@@ -2840,11 +2842,14 @@ void DbsizeCmd::Do() {
     }
     KeyScanInfo key_scan_info = dbs->GetKeyScanInfo();
     std::vector<storage::KeyInfo> key_infos = key_scan_info.key_infos;
-    if (key_infos.size() != 5) {
+    if (key_infos.size() != (size_t)(storage::DataType::kNones)) {
       res_.SetRes(CmdRes::kErrOther, "keyspace error");
       return;
     }
-    uint64_t dbsize = key_infos[0].keys + key_infos[1].keys + key_infos[2].keys + key_infos[3].keys + key_infos[4].keys;
+    uint64_t dbsize = 0;
+    for (auto info : key_infos) {
+      dbsize += info.keys;
+    }
     res_.AppendInteger(static_cast<int64_t>(dbsize));
   }
 }
@@ -3043,6 +3048,7 @@ void PKPatternMatchDelCmd::DoInitial() {
   pattern_ = argv_[1];
 }
 
+//TODO: may lead to inconsistent between rediscache and db, because currently it only cleans db
 void PKPatternMatchDelCmd::Do() {
   int ret = 0;
   rocksdb::Status s = db_->storage()->PKPatternMatchDel(type_, pattern_, &ret);

src/pika_binlog_transverter.cc

@@ -169,6 +169,7 @@ bool PikaBinlogTransverter::BinlogItemWithoutContentDecode(BinlogType type, cons
     LOG(ERROR) << "Binlog Item type error, expect type:" << type << " actualy type: " << binlog_type;
     return false;
   }
+  storage::g_storage_logictime->UpdateLogicTime(binlog_item->exec_time());
   pstd::GetFixed32(&binlog_str, &binlog_item->exec_time_);
   pstd::GetFixed32(&binlog_str, &binlog_item->term_id_);
   pstd::GetFixed64(&binlog_str, &binlog_item->logic_id_);

src/pika_client_conn.cc

@@ -153,11 +153,8 @@ std::shared_ptr<Cmd> PikaClientConn::DoCmd(const PikaCmdArgsType& argv, const st
     return c_ptr;
   }
 
-  if (g_pika_server->readonly(current_db_) && opt != kCmdNameExec) {
-    storage::g_storage_logictime->SetProtectionMode(true);
-  } else{
-    storage::g_storage_logictime->SetProtectionMode(false);
-  }
+  bool is_readonly = g_pika_server->readonly(current_db_);
+  storage::g_storage_logictime->SetProtectionMode(is_readonly);
 
   if (c_ptr->is_write()) {
     if (g_pika_server->IsDBBinlogIoError(current_db_)) {

src/pika_command.cc

@@ -130,7 +130,7 @@ void InitCmdTable(CmdTable* cmd_table) {
   cmd_table->insert(std::pair<std::string, std::unique_ptr<Cmd>>(kCmdNamePadding, std::move(paddingptr)));
 
   std::unique_ptr<Cmd> pkpatternmatchdelptr =
-      std::make_unique<PKPatternMatchDelCmd>(kCmdNamePKPatternMatchDel, 3, kCmdFlagsWrite | kCmdFlagsAdmin);
+      std::make_unique<PKPatternMatchDelCmd>(kCmdNamePKPatternMatchDel, 2, kCmdFlagsWrite | kCmdFlagsAdmin);
   cmd_table->insert(
       std::pair<std::string, std::unique_ptr<Cmd>>(kCmdNamePKPatternMatchDel, std::move(pkpatternmatchdelptr)));
   std::unique_ptr<Cmd> dummyptr = std::make_unique<DummyCmd>(kCmdDummy, 0, kCmdFlagsWrite);

src/pika_conf.cc

@@ -339,6 +339,12 @@ int PikaConf::Load() {
     max_write_buffer_size_ = PIKA_CACHE_SIZE_DEFAULT;  // 10Gb
   }
 
+  // max-total-wal-size
+  GetConfInt64("max-total-wal-size", &max_total_wal_size_);
+  if (max_total_wal_size_ < 0) {
+    max_total_wal_size_ = 0;
+  }
+
   // rate-limiter-mode
   rate_limiter_mode_ = 1;
   GetConfInt("rate-limiter-mode", &rate_limiter_mode_);

src/pika_db.cc

@@ -95,20 +95,11 @@ bool DB::IsKeyScaning() {
 
 void DB::RunKeyScan() {
   Status s;
-  std::vector<storage::KeyInfo> new_key_infos(5);
+  std::vector<storage::KeyInfo> new_key_infos;
 
   InitKeyScan();
   std::shared_lock l(dbs_rw_);
-  std::vector<storage::KeyInfo> tmp_key_infos;
-  s = GetKeyNum(&tmp_key_infos);
-  if (s.ok()) {
-    for (size_t idx = 0; idx < tmp_key_infos.size(); ++idx) {
-      new_key_infos[idx].keys += tmp_key_infos[idx].keys;
-      new_key_infos[idx].expires += tmp_key_infos[idx].expires;
-      new_key_infos[idx].avg_ttl += tmp_key_infos[idx].avg_ttl;
-      new_key_infos[idx].invaild_keys += tmp_key_infos[idx].invaild_keys;
-    }
-  }
+  s = GetKeyNum(&new_key_infos);
   key_scan_info_.duration = static_cast<int32_t>(time(nullptr) - key_scan_info_.start_time);
 
   std::lock_guard lm(key_scan_protector_);

src/pika_kv.cc

@@ -533,7 +533,7 @@ void MgetCmd::Do() {
     cache_miss_keys_ = keys_;
   }
   db_value_status_array_.clear();
-  s_ = db_->storage()->MGet(cache_miss_keys_, &db_value_status_array_);
+  s_ = db_->storage()->MGetWithTTL(cache_miss_keys_, &db_value_status_array_);
   if (!s_.ok()) {
     if (s_.IsInvalidArgument()) {
       res_.SetRes(CmdRes::kMultiKey);

src/pika_monotonic_time.cc

@@ -3,7 +3,7 @@
 // LICENSE file in the root directory of this source tree. An additional grant
 // of patent rights can be found in the PATENTS file in the same directory.
 
-#ifdef __APPLE__  // Mac
+#if defined(__APPLE__)  // Mac
 #include <mach/mach_time.h>
 
 #include "include/pika_monotonic_time.h"
@@ -17,7 +17,18 @@ monotime getMonotonicUs() {
   return nanos / 1000;
 }
 
-#elif __linux__  // Linux
+#elif defined(__FreeBSD__)  // FreeBSD
+#include <time.h>
+
+#include "include/pika_monotonic_time.h"
+
+monotime getMonotonicUs() {
+  struct timespec ts;
+  clock_gettime(CLOCK_MONOTONIC, &ts);
+  return (ts.tv_sec * 1000000) + (ts.tv_nsec / 1000);
+}
+
+#elif defined(__linux__)  // Linux
 
 #ifdef __x86_64__  // x86_64
 

src/pika_rsync_service.cc

@@ -15,6 +15,10 @@
 #include "include/pika_conf.h"
 #include "include/pika_define.h"
 
+#ifdef __FreeBSD__
+#  include <sys/wait.h>
+#endif
+
 extern std::unique_ptr<PikaConf> g_pika_conf;
 
 PikaRsyncService::PikaRsyncService(const std::string& raw_path, const int port) : raw_path_(raw_path), port_(port) {

src/pstd/src/rsync.cc

@@ -8,6 +8,11 @@
 #include "pstd/include/rsync.h"
 #include "pstd/include/xdebug.h"
 
+#ifdef __FreeBSD__
+#  include <sys/types.h>
+#  include <sys/wait.h>
+#endif
+
 namespace pstd {
 // Clean files for rsync info, such as the lock, log, pid, conf file
 static bool CleanRsyncInfo(const std::string& path) { return pstd::DeleteDirIfExist(path + kRsyncSubDir); }

src/storage/include/storage/storage.h

@@ -1114,6 +1114,7 @@ class Storage {
 
   // For scan keys in data base
   std::atomic<bool> scan_keynum_exit_ = {false};
+  Status MGetWithTTL(const Slice& key, std::string* value, int64_t* ttl);
 };
 
 }  //  namespace storage

src/storage/src/base_filter.h

@@ -27,8 +27,12 @@ class BaseMetaFilter : public rocksdb::CompactionFilter {
   BaseMetaFilter() = default;
   bool Filter(int level, const rocksdb::Slice& key, const rocksdb::Slice& value, std::string* new_value,
               bool* value_changed) const override {
-    int64_t unix_time = g_storage_logictime->Now();
-    auto cur_time = static_cast<uint32_t>(unix_time);
+    int64_t unix_time;
+    if (g_storage_logictime->ProtectionMode()){
+    auto cur_time = static_cast<uint32_t>(g_storage_logictime->Now());
+    }
+    rocksdb::Env::Default()->GetCurrentTime(&unix_time);
+    auto cur_time = static_cast<uint64_t>(unix_time);
     /*
      * For the filtering of meta information, because the field designs of string
      * and list are different, their filtering policies are written separately.

src/storage/src/base_value_format.h

@@ -135,8 +135,11 @@ class ParsedInternalValue {
     if (etime_ == 0) {
       return false;
     }
-
-    int64_t unix_time = g_storage_logictime->Now();
+    int64_t unix_time;
+    if (g_storage_logictime->ProtectionMode()){
+      unix_time = g_storage_logictime->Now();
+    }
+    rocksdb::Env::Default()->GetCurrentTime(&unix_time);
     return etime_ < unix_time;
   }
 

src/storage/src/coding.h

@@ -6,17 +6,20 @@
 #ifndef SRC_CODING_H_
 #define SRC_CODING_H_
 
+#undef STORAGE_PLATFORM_IS_LITTLE_ENDIAN
+
 #if defined(__APPLE__)
 #  include <machine/endian.h>  // __BYTE_ORDER
 #  define __BYTE_ORDER __DARWIN_BYTE_ORDER
 #  define __LITTLE_ENDIAN __DARWIN_LITTLE_ENDIAN
 #elif defined(__FreeBSD__)
-#  include <sys/endian.h>  // __BYTE_ORDER
+#  include <sys/endian.h>
+#  include <sys/types.h>
+#  define STORAGE_PLATFORM_IS_LITTLE_ENDIAN (_BYTE_ORDER == _LITTLE_ENDIAN)
 #else
 #  include <endian.h>  // __BYTE_ORDER
 #endif
 
-#undef STORAGE_PLATFORM_IS_LITTLE_ENDIAN
 #ifndef STORAGE_PLATFORM_IS_LITTLE_ENDIAN
 #  define STORAGE_PLATFORM_IS_LITTLE_ENDIAN (__BYTE_ORDER == __LITTLE_ENDIAN)
 #endif

src/storage/src/redis.cc

@@ -387,7 +387,7 @@ Status Redis::ScanKeyNum(std::vector<KeyInfo>* key_infos) {
   if (!s.ok()) {
     return s;
   }
-  s = ScanSetsKeyNum(&((*key_infos)[5]));
+  s = ScanStreamsKeyNum(&((*key_infos)[5]));
   if (!s.ok()) {
     return s;
   }

src/storage/src/redis.h

@@ -156,6 +156,7 @@ class Redis {
   Status Get(const Slice& key, std::string* value);
   Status MGet(const Slice& key, std::string* value);
   Status GetWithTTL(const Slice& key, std::string* value, int64_t* ttl);
+  Status MGetWithTTL(const Slice& key, std::string* value, int64_t* ttl);
   Status GetBit(const Slice& key, int64_t offset, int32_t* ret);
   Status Getrange(const Slice& key, int64_t start_offset, int64_t end_offset, std::string* ret);
   Status GetrangeWithValue(const Slice& key, int64_t start_offset, int64_t end_offset,