Local resume task queue per task group.

CMakeLists.txt

@@ -41,14 +41,14 @@ SET(CPACK_DEBIAN_PACKAGE_MAINTAINER "brpc authors")
 INCLUDE(CPack)
 
 if(CMAKE_CXX_COMPILER_ID STREQUAL "GNU")
-    # require at least gcc 4.8
-    if(CMAKE_CXX_COMPILER_VERSION VERSION_LESS 4.8)
-        message(FATAL_ERROR "GCC is too old, please install a newer version supporting C++11")
+    # require at least gcc 8
+    if(CMAKE_CXX_COMPILER_VERSION VERSION_LESS 8)  # ref: https://gcc.gnu.org/projects/cxx-status.html
+        message(FATAL_ERROR "GCC is too old, please install a newer version supporting C++17")
     endif()
 elseif(CMAKE_CXX_COMPILER_ID STREQUAL "Clang")
-    # require at least clang 3.3
-    if(CMAKE_CXX_COMPILER_VERSION VERSION_LESS 3.3)
-        message(FATAL_ERROR "Clang is too old, please install a newer version supporting C++11")
+    # require at least clang 5
+    if(CMAKE_CXX_COMPILER_VERSION VERSION_LESS 5)  # ref: https://clang.llvm.org/cxx_status.html
+        message(FATAL_ERROR "Clang is too old, please install a newer version supporting C++17")
     endif()
 else()
     message(WARNING "You are using an unsupported compiler! Compilation has only been tested with Clang and GCC.")
@@ -121,21 +121,21 @@ set(CMAKE_CPP_FLAGS "${CMAKE_CPP_FLAGS} ${DEBUG_SYMBOL} ${THRIFT_CPP_FLAG}")
 set(CMAKE_CXX_FLAGS "${CMAKE_CPP_FLAGS} -O2 -pipe -Wall -W -fPIC -fstrict-aliasing -Wno-invalid-offsetof -Wno-unused-parameter -fno-omit-frame-pointer")
 set(CMAKE_C_FLAGS "${CMAKE_CPP_FLAGS} -O2 -pipe -Wall -W -fPIC -fstrict-aliasing -Wno-unused-parameter -fno-omit-frame-pointer")
 
-macro(use_cxx11)
+macro(use_cxx17)
 if(CMAKE_VERSION VERSION_LESS "3.1.3")
     if(CMAKE_CXX_COMPILER_ID STREQUAL "GNU")
-        set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -std=c++11")
+        set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -std=c++17")
     endif()
     if(CMAKE_CXX_COMPILER_ID STREQUAL "Clang")
-        set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -std=c++11")
+        set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -std=c++17")
     endif()
 else()
-    set(CMAKE_CXX_STANDARD 11)
+    set(CMAKE_CXX_STANDARD 17)
     set(CMAKE_CXX_STANDARD_REQUIRED ON)
 endif()
-endmacro(use_cxx11)
+endmacro(use_cxx17)
 
-use_cxx11()
+use_cxx17()
 
 if(CMAKE_CXX_COMPILER_ID STREQUAL "GNU")
     #required by butil/crc32.cc to boost performance for 10x
@@ -169,8 +169,9 @@ if(WITH_SNAPPY)
 endif()
 
 if(WITH_GLOG)
+    message(NOTICE "BRPC WITH_GLOG=ON")
     find_path(GLOG_INCLUDE_PATH NAMES glog/logging.h)
-    find_library(GLOG_LIB NAMES glog)
+    find_library(GLOG_LIB NAMES glog VERSION ">=0.6.0" REQUIRE)
     if((NOT GLOG_INCLUDE_PATH) OR (NOT GLOG_LIB))
         message(FATAL_ERROR "Fail to find glog")
     endif()

src/brpc/policy/redis_protocol.cpp

@@ -33,6 +33,7 @@
 #include "brpc/redis.h"
 #include "brpc/redis_command.h"
 #include "brpc/policy/redis_protocol.h"
+#include "bvar/latency_recorder.h"
 
 namespace brpc {
 
@@ -59,6 +60,7 @@ class RedisConnContext : public Destroyable  {
 public:
     explicit RedisConnContext(const RedisService* rs)
         : redis_service(rs)
+        , in_transaction(false)
         , batched_size(0) {}
 
     ~RedisConnContext();
@@ -68,7 +70,10 @@ class RedisConnContext : public Destroyable  {
     const RedisService* redis_service;
     // If user starts a transaction, transaction_handler indicates the
     // handler pointer that runs the transaction command.
-    std::unique_ptr<RedisCommandHandler> transaction_handler;
+    std::unique_ptr<TransactionHandler> transaction_handler;
+    // Whether this connection has begun a transaction. If true, the commands
+    // received will be handled by transaction_handler.
+    bool in_transaction;
     // >0 if command handler is run in batched mode.
     int batched_size;
 
@@ -82,15 +87,33 @@ int ConsumeCommand(RedisConnContext* ctx,
                    butil::IOBufAppender* appender) {
     RedisReply output(&ctx->arena);
     RedisCommandHandlerResult result = REDIS_CMD_HANDLED;
-    if (ctx->transaction_handler) {
+    if (ctx->in_transaction) {
+        assert(ctx->transaction_handler != nullptr);
         result = ctx->transaction_handler->Run(args, &output, flush_batched);
         if (result == REDIS_CMD_HANDLED) {
             ctx->transaction_handler.reset(NULL);
+            ctx->in_transaction = false;
         } else if (result == REDIS_CMD_BATCHED) {
             LOG(ERROR) << "BATCHED should not be returned by a transaction handler.";
             return -1;
         }
-    } else {
+    }
+    else if (args[0] == "watch" || args[0] == "unwatch") {
+        if (!ctx->transaction_handler) {
+            ctx->transaction_handler.reset(ctx->redis_service->NewTransactionHandler());
+            ctx->in_transaction = false;
+        }
+        if (!ctx->transaction_handler) {
+            output.SetError("ERR Transaction not supported.");
+        } else {
+            result = ctx->transaction_handler->Run(args, &output, flush_batched);
+            if (result == REDIS_CMD_BATCHED) {
+                LOG(ERROR) << "BATCHED should not be returned by a transaction handler.";
+                return -1;
+            }
+        }
+    }
+    else {
         RedisCommandHandler* ch = ctx->redis_service->FindCommandHandler(args[0]);
         if (!ch) {
             char buf[64];
@@ -103,7 +126,16 @@ int ConsumeCommand(RedisConnContext* ctx,
                     LOG(ERROR) << "CONTINUE should not be returned in a batched process.";
                     return -1;
                 }
-                ctx->transaction_handler.reset(ch->NewTransactionHandler());
+                if (ctx->transaction_handler == nullptr) {
+                    ctx->transaction_handler.reset(ctx->redis_service->NewTransactionHandler());
+                }
+                if (ctx->transaction_handler != nullptr) {
+                    ctx->transaction_handler->Begin();
+                    ctx->in_transaction = true;
+                }
+                else {
+                    output.SetError("ERR Transaction not supported.");
+                }
             } else if (result == REDIS_CMD_BATCHED) {
                 ctx->batched_size++;
             }
@@ -144,6 +176,9 @@ void RedisConnContext::Destroy() {
 
 // ========== impl of RedisConnContext ==========
 
+inline bvar::LatencyRecorder socket_write_latency("socket", "write");
+inline bvar::LatencyRecorder consume_cmd_latency("socket", "consume_cmd");
+
 ParseResult ParseRedisMessage(butil::IOBuf* source, Socket* socket,
                               bool read_eof, const void* arg) {
     if (read_eof || source->empty()) {
@@ -174,22 +209,28 @@ ParseResult ParseRedisMessage(butil::IOBuf* source, Socket* socket,
             if (err != PARSE_OK) {
                 break;
             }
+            int64_t start_time_us = butil::cpuwide_time_us();
             if (ConsumeCommand(ctx, current_args, false, &appender) != 0) {
                 return MakeParseError(PARSE_ERROR_ABSOLUTELY_WRONG);
             }
+            consume_cmd_latency << (butil::cpuwide_time_us() - start_time_us);
             current_args.swap(next_args);
         }
+        int64_t start_time_us = butil::cpuwide_time_us();
         if (ConsumeCommand(ctx, current_args,
                     true /*must be the last message*/, &appender) != 0) {
             return MakeParseError(PARSE_ERROR_ABSOLUTELY_WRONG);
         }
+        consume_cmd_latency << (butil::cpuwide_time_us() - start_time_us);
         butil::IOBuf sendbuf;
         appender.move_to(sendbuf);
         CHECK(!sendbuf.empty());
         Socket::WriteOptions wopt;
         wopt.ignore_eovercrowded = true;
+        start_time_us = butil::cpuwide_time_us();
         LOG_IF(WARNING, socket->Write(&sendbuf, &wopt) != 0)
             << "Fail to send redis reply";
+        socket_write_latency << (butil::cpuwide_time_us() - start_time_us);
         if(ctx->parser.ParsedArgsSize() == 0) {
             ctx->arena.clear();
         }

src/brpc/redis.cpp

@@ -467,6 +467,11 @@ RedisCommandHandler* RedisService::FindCommandHandler(const butil::StringPiece&
     return NULL;
 }
 
+TransactionHandler* RedisService::NewTransactionHandler() const {
+    LOG(ERROR) << "NewTransactionHandler is not implemented";
+    return NULL;
+}
+
 RedisCommandHandler* RedisCommandHandler::NewTransactionHandler() {
     LOG(ERROR) << "NewTransactionHandler is not implemented";
     return NULL;

src/brpc/redis.h

@@ -221,6 +221,7 @@ std::ostream& operator<<(std::ostream& os, const RedisRequest&);
 std::ostream& operator<<(std::ostream& os, const RedisResponse&);
 
 class RedisCommandHandler;
+class TransactionHandler;
 
 // Container of CommandHandlers.
 // Assign an instance to ServerOption.redis_service to enable redis support. 
@@ -231,6 +232,9 @@ class RedisService {
     // Call this function to register `handler` that can handle command `name`.
     bool AddCommandHandler(const std::string& name, RedisCommandHandler* handler);
 
+    // Create a transaction handler to handle commands inside a transaction.
+    virtual TransactionHandler* NewTransactionHandler() const;
+
     // This function should not be touched by user and used by brpc deverloper only.
     RedisCommandHandler* FindCommandHandler(const butil::StringPiece& name) const;
 
@@ -243,6 +247,8 @@ enum RedisCommandHandlerResult {
     REDIS_CMD_HANDLED = 0,
     REDIS_CMD_CONTINUE = 1,
     REDIS_CMD_BATCHED = 2,
+    REDIS_CMD_TXN_START = 3,
+    REDIS_CMD_TXN_FINISH = 4,
 };
 
 // The Command handler for a redis request. User should impletement Run().
@@ -289,6 +295,11 @@ class RedisCommandHandler {
     virtual RedisCommandHandler* NewTransactionHandler();
 };
 
+class TransactionHandler : public RedisCommandHandler {
+public:
+    virtual bool Begin() = 0;
+};
+
 } // namespace brpc
 
 #endif  // BRPC_REDIS_H

src/bthread/condition_variable.cpp

@@ -43,6 +43,7 @@ extern "C" {
 
 extern int bthread_mutex_unlock(bthread_mutex_t*);
 extern int bthread_mutex_lock_contended(bthread_mutex_t*);
+extern void bthread_flush();
 
 int bthread_cond_init(bthread_cond_t* __restrict c,
                       const bthread_condattr_t*) {
@@ -65,7 +66,10 @@ int bthread_cond_signal(bthread_cond_t* c) {
     butil::atomic<int>* const saved_seq = ic->seq;
     saved_seq->fetch_add(1, butil::memory_order_release);
     // don't touch ic any more
-    bthread::butex_wake(saved_seq);
+    bool no_signal = true;
+    bthread::butex_wake(saved_seq, no_signal);
+    // flush unsignaled tasks manually
+    bthread_flush();
     return 0;
 }
 

src/bthread/condition_variable.h

@@ -45,7 +45,7 @@ class ConditionVariable {
     DISALLOW_COPY_AND_ASSIGN(ConditionVariable);
 public:
     typedef bthread_cond_t*         native_handler_type;
-    
+
     ConditionVariable() {
         CHECK_EQ(0, bthread_cond_init(&_cond, NULL));
     }