Merge branch 'unstable' into luajit

.travis.yml

@@ -0,0 +1,31 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+language: cpp
+
+arch:
+  - amd64
+  - arm64
+
+os: linux
+dist: focal
+
+services:
+  - docker
+
+script:
+  - docker build .

Dockerfile

@@ -15,23 +15,27 @@
 # specific language governing permissions and limitations
 # under the License.
 
-FROM ubuntu:22.04 as build
+FROM ubuntu:focal as build
 
-RUN apt update && apt install -y cmake make git autoconf libtool g++
+# workaround tzdata install hanging
+ENV TZ=Asia/Shanghai
+RUN ln -snf /usr/share/zoneinfo/$TZ /etc/localtime && echo $TZ > /etc/timezone
+
+RUN apt update
+RUN apt install -y cmake make git autoconf libtool g++
 WORKDIR /kvrocks
 
 COPY . .
-RUN mkdir docker-build && ./build.sh docker-build
-
+RUN ./build.sh build
 
-FROM ubuntu:22.04
+FROM ubuntu:focal
 
 WORKDIR /kvrocks
 
-COPY --from=build /kvrocks/docker-build/kvrocks ./bin/
+COPY --from=build /kvrocks/build/kvrocks ./bin/
 
 COPY ./kvrocks.conf  ./conf/
-RUN sed  -i -e 's|dir /tmp/kvrocks|dir /var/lib/kvrocks|g' ./conf/kvrocks.conf
+RUN sed -i -e 's%dir /tmp/kvrocks%dir /var/lib/kvrocks%g' ./conf/kvrocks.conf
 VOLUME /var/lib/kvrocks
 
 EXPOSE 6666:6666 

src/log_collector.cc

@@ -53,36 +53,33 @@ LogCollector<T>::~LogCollector() {
 template <class T>
 ssize_t LogCollector<T>::Size() {
   size_t n;
-  mu_.lock();
+  std::lock_guard<std::mutex> guard(mu_);
   n = entries_.size();
-  mu_.unlock();
   return n;
 }
 
 template <class T>
 void LogCollector<T>::Reset() {
-  mu_.lock();
+  std::lock_guard<std::mutex> guard(mu_);
   while (!entries_.empty()) {
     delete entries_.front();
     entries_.pop_front();
   }
-  mu_.unlock();
 }
 
 template <class T>
 void LogCollector<T>::SetMaxEntries(int64_t max_entries) {
-  mu_.lock();
+  std::lock_guard<std::mutex> guard(mu_);
   while (max_entries > 0 && static_cast<int64_t>(entries_.size()) > max_entries) {
     delete entries_.back();
     entries_.pop_back();
   }
   max_entries_ = max_entries;
-  mu_.unlock();
 }
 
 template <class T>
 void LogCollector<T>::PushEntry(T *entry) {
-  mu_.lock();
+  std::lock_guard<std::mutex> guard(mu_);
   entry->id = ++id_;
   entry->time = time(nullptr);
   if (max_entries_ > 0 && !entries_.empty()
@@ -91,15 +88,14 @@ void LogCollector<T>::PushEntry(T *entry) {
     entries_.pop_back();
   }
   entries_.push_front(entry);
-  mu_.unlock();
 }
 
 template <class T>
 std::string LogCollector<T>::GetLatestEntries(int64_t cnt) {
   size_t n;
   std::string output;
 
-  mu_.lock();
+  std::lock_guard<std::mutex> guard(mu_);
   if (cnt > 0) {
     n = std::min(entries_.size(), static_cast<size_t>(cnt));
   } else {
@@ -110,7 +106,6 @@ std::string LogCollector<T>::GetLatestEntries(int64_t cnt) {
     output.append(entry->ToRedisString());
     if (--n == 0) break;
   }
-  mu_.unlock();
   return output;
 }
 

src/replication.cc

@@ -952,8 +952,7 @@ rocksdb::Status ReplicationThread::ParseWriteBatch(const std::string &batch_stri
       break;
     case kBatchTypePropagate:
       if (write_batch_handler.Key() == Engine::kPropagateScriptCommand) {
-        std::vector<std::string> tokens;
-        Util::TokenizeRedisProtocol(write_batch_handler.Value(), &tokens);
+        std::vector<std::string> tokens = Util::TokenizeRedisProtocol(write_batch_handler.Value());
         if (!tokens.empty()) {
           srv_->ExecPropagatedCommand(tokens);
         }

src/server.cc

@@ -191,13 +191,12 @@ void Server::Join() {
 }
 
 Status Server::AddMaster(std::string host, uint32_t port, bool force_reconnect) {
-  slaveof_mu_.lock();
+  std::lock_guard<std::mutex> guard(slaveof_mu_);
 
   // Don't check host and port if 'force_reconnect' argument is set to true
   if (!force_reconnect &&
       !master_host_.empty() && master_host_ == host &&
       master_port_ == port) {
-    slaveof_mu_.unlock();
     return Status::OK();
   }
 
@@ -228,12 +227,11 @@ Status Server::AddMaster(std::string host, uint32_t port, bool force_reconnect)
   } else {
     replication_thread_ = nullptr;
   }
-  slaveof_mu_.unlock();
   return s;
 }
 
 Status Server::RemoveMaster() {
-  slaveof_mu_.lock();
+  std::lock_guard<std::mutex> guard(slaveof_mu_);
   if (!master_host_.empty()) {
     master_host_.clear();
     master_port_ = 0;
@@ -242,7 +240,6 @@ Status Server::RemoveMaster() {
     replication_thread_ = nullptr;
     storage_->ShiftReplId();
   }
-  slaveof_mu_.unlock();
   return Status::OK();
 }
 
@@ -260,7 +257,7 @@ Status Server::AddSlave(Redis::Connection *conn, rocksdb::SequenceNumber next_re
 }
 
 void Server::DisconnectSlaves() {
-  slave_threads_mu_.lock();
+  std::lock_guard<std::mutex> guard(slaveof_mu_);
   for (const auto &slave_thread : slave_threads_) {
     if (!slave_thread->IsStopped()) slave_thread->Stop();
   }
@@ -270,12 +267,11 @@ void Server::DisconnectSlaves() {
     slave_thread->Join();
     delete slave_thread;
   }
-  slave_threads_mu_.unlock();
 }
 
 void Server::cleanupExitedSlaves() {
   std::list<FeedSlaveThread *> exited_slave_threads;
-  slave_threads_mu_.lock();
+  std::lock_guard<std::mutex> guard(slaveof_mu_);
   for (const auto &slave_thread : slave_threads_) {
     if (slave_thread->IsStopped())
       exited_slave_threads.emplace_back(slave_thread);
@@ -287,7 +283,6 @@ void Server::cleanupExitedSlaves() {
     t->Join();
     delete t;
   }
-  slave_threads_mu_.unlock();
 }
 
 void Server::FeedMonitorConns(Redis::Connection *conn, const std::vector<std::string> &tokens) {
@@ -832,15 +827,13 @@ void Server::GetRoleInfo(std::string *info) {
 
 std::string Server::GetLastRandomKeyCursor() {
   std::string cursor;
-  last_random_key_cursor_mu_.lock();
+  std::lock_guard<std::mutex> guard(last_random_key_cursor_mu_);
   cursor = last_random_key_cursor_;
-  last_random_key_cursor_mu_.unlock();
   return cursor;
 }
 void Server::SetLastRandomKeyCursor(const std::string &cursor) {
-  last_random_key_cursor_mu_.lock();
+  std::lock_guard<std::mutex> guard(last_random_key_cursor_mu_);
   last_random_key_cursor_ = cursor;
-  last_random_key_cursor_mu_.unlock();
 }
 
 int Server::GetUnixTime() {
@@ -1266,11 +1259,10 @@ std::string Server::GetClientsStr() {
   for (const auto &t : worker_threads_) {
     clients.append(t->GetWorker()->GetClientsStr());
   }
-  slave_threads_mu_.lock();
+  std::lock_guard<std::mutex> guard(slave_threads_mu_);
   for (const auto &st : slave_threads_) {
     clients.append(st->GetConn()->ToString());
   }
-  slave_threads_mu_.unlock();
   return clients;
 }