Sandbox/grpc (#103)

* Add gRPC server/client sample and automated build in sandbox/grpc_server

- Add EchoService sample using echo.proto
- CMake now auto-generates C++/gRPC code from .proto into build dir
- Add sample server (grpc_server) and client (grpc_client, grpc_client_safe) implementations
- Support for AddressSanitizer-disabled builds
- Generated files are kept out of the source tree

Author: umegane

.clang-tidy

@@ -54,5 +54,5 @@ CheckOptions:
     value: 1
   - key: readability-function-cognitive-complexity.IgnoreMacros
     value: true
-HeaderFilterRegex: '.*/(include|src|examples)/.*\.h$'
+HeaderFilterRegex: '^(?!.*/build/).*/(include|src|examples)/.*\.h$'
 ...

.github/workflows/ci-build.yml

@@ -35,6 +35,11 @@ jobs:
         with:
           submodules: true
 
+      - name: Install gRPC
+        run: |
+          apt-get update
+          apt-get install -y libabsl-dev
+
       - name: CMake_Build
         run: |
           mkdir build

CMakeLists.txt

@@ -41,6 +41,7 @@ option(ENABLE_ALTIMETER "enable altimeter logging" OFF)
 option(INSTALL_EXPERIMENTAL_TOOLS "Install experimental tools like tgreplica" OFF)
 option(BUILD_REPLICATION_TESTS "Build replication tests" OFF)
 option(BUILD_SANDBOX_TOOLS "build sandbox (temporary) tools" OFF)
+option(USE_GRPC_CONFIG "Use CMake Config mode for gRPC instead of pkg-config" OFF)
 
 if (FORCE_INSTALL_RPATH)
     message(DEPRECATION "FORCE_INSTALL_RPATH is obsoleted")
@@ -77,6 +78,23 @@ if (ENABLE_ALTIMETER)
     find_package(fmt REQUIRED)
 endif()
 
+
+# gRPC/protobuf
+find_package(Protobuf REQUIRED)
+
+
+
+if(USE_GRPC_CONFIG)
+    message(STATUS "Using gRPC CMake Config mode")
+    find_package(gRPC CONFIG REQUIRED)
+else()
+    message(STATUS "Using gRPC pkg-config mode")
+    find_package(PkgConfig REQUIRED)
+    pkg_check_modules(GRPC REQUIRED grpc++)
+endif()
+
+find_program(GRPC_CPP_PLUGIN grpc_cpp_plugin REQUIRED)
+
 add_subdirectory(third_party) # should be before enable_testing()
 
 include(GNUInstallDirs)

README.md

@@ -1,5 +1,6 @@
 # Limestone - a datastore engine
 
+
 ## Requirements
 
 * CMake `>= 3.16`
@@ -16,7 +17,7 @@ git submodule update --init --recursive
 ```dockerfile
 FROM ubuntu:22.04
 
-RUN apt update -y && apt install -y git build-essential cmake ninja-build libboost-filesystem-dev libboost-system-dev libboost-container-dev libboost-thread-dev libgoogle-glog-dev libgflags-dev doxygen libleveldb-dev librocksdb-dev pkg-config nlohmann-json3-dev
+RUN apt update -y && apt install -y git build-essential cmake ninja-build libboost-filesystem-dev libboost-system-dev libboost-container-dev libboost-thread-dev libgoogle-glog-dev libgflags-dev doxygen libleveldb-dev librocksdb-dev pkg-config nlohmann-json3-dev libgrpc-dev libgrpc++-dev protobuf-compiler-grpc libabsl-dev
 # libleveldb-dev is not required if -DRECOVERY_SORTER_KVSLIB=ROCKSDB
 # librocksdb-dev is not required if -DRECOVERY_SORTER_KVSLIB=LEVELDB
 ```
@@ -47,6 +48,7 @@ available options:
 * `-DRECOVERY_SORTER_KVSLIB=<library>` - select the eKVS library using at recovery process. (`LEVELDB` or `ROCKSDB` (default), case-insensitive)
 * `-DRECOVERY_SORTER_PUT_ONLY=OFF` - don't use (faster) put-only method at recovery process
 * `-DBUILD_REPLICATION_TESTS=ON` - (temporary) enable experimental replication tests (excluded by default)
+* `-DUSE_GRPC_CONFIG=ON` - use CMake Config mode for gRPC instead of pkg-config (recommended on Ubuntu 24.04 or later for faster configuration)
 
 * for debugging only
   * `-DENABLE_SANITIZER=OFF` - disable sanitizers (requires `-DCMAKE_BUILD_TYPE=Debug`)

sandbox/CMakeLists.txt

@@ -1,18 +1,20 @@
 set(TOOLS
-    startup_speed_benchmark
-    wal_dump
+    startup_speed_benchmark/startup_speed_benchmark.cpp
+    wal_dump/wal_dump.cpp
+    grpc/grpc_server.cpp
+    grpc/grpc_client.cpp
+    grpc/server_streaming_sample_client.cpp
+    grpc/server_streaming_sample_server.cpp
 )
 
-foreach(tool IN LISTS TOOLS)
-    add_executable(${tool} ${tool}/${tool}.cpp)
-
-    target_include_directories(${tool}
-        PRIVATE ${CMAKE_SOURCE_DIR}/src/limestone
-        PRIVATE ${CMAKE_SOURCE_DIR}/include
-    )
-
-    target_link_libraries(${tool}
-        PRIVATE limestone-impl
-        PRIVATE glog::glog
+foreach(tool_src IN LISTS TOOLS)
+    get_filename_component(tool_name ${tool_src} NAME_WE)
+    add_executable(${tool_name} ${tool_src})
+    target_include_directories(${tool_name} PRIVATE
+        ${PROJECT_SOURCE_DIR}/src
+        ${PROJECT_SOURCE_DIR}/src/limestone
+        ${PROJECT_SOURCE_DIR}/include
+        ${CMAKE_BINARY_DIR}/src   # for generated protobuf headers (safe to include always)
     )
+    target_link_libraries(${tool_name} PRIVATE limestone-impl)
 endforeach()

sandbox/grpc/grpc_client.cpp

@@ -0,0 +1,54 @@
+#include <iostream>
+#include <memory>
+#include <string>
+
+#include <grpcpp/grpcpp.h>
+#include <glog/logging.h>
+#include "limestone/grpc/client/echo_client.h"
+
+int main(int argc, char** argv) {
+    // Initialize glog
+    google::InitGoogleLogging(argv[0]);
+    FLAGS_logtostderr = 1;  // Log to stderr instead of file
+    
+    std::string target_str = "localhost:50051";
+    std::string message = "Hello, gRPC!";
+    
+    if (argc > 1) {
+        message = argv[1];
+    }
+
+    // Message size check
+    if (message.size() > 100) {
+        std::cerr << "Error: Message too long (max 100 chars)" << std::endl;
+        return 1;
+    }
+
+    LOG(INFO) << "Starting limestone gRPC echo client";
+    LOG(INFO) << "Connecting to: " << target_str;
+    
+    try {
+        // Create echo client
+        limestone::grpc::client::echo_client client(target_str);
+        
+        // Send echo request
+        std::string response;
+        ::grpc::Status status = client.echo(message, response);
+        
+        if (status.ok()) {
+            std::cout << "Server replied: " << response << std::endl;
+            LOG(INFO) << "Echo successful: " << response;
+        } else {
+            std::cout << "RPC failed: " << status.error_message() << std::endl;
+            LOG(ERROR) << "RPC failed: " << status.error_code() 
+                       << ": " << status.error_message();
+            return 1;
+        }
+    } catch (const std::exception& e) {
+        std::cerr << "Exception: " << e.what() << std::endl;
+        LOG(ERROR) << "Exception occurred: " << e.what();
+        return 1;
+    }
+
+    return 0;
+}

sandbox/grpc/grpc_server.cpp

@@ -0,0 +1,34 @@
+#include <iostream>
+#include <memory>
+#include <string>
+
+#include <grpcpp/grpcpp.h>
+#include <glog/logging.h>
+#include "limestone/grpc/service/echo_service_impl.h"
+
+using grpc::Server;
+using grpc::ServerBuilder;
+
+void RunServer() {
+    std::string server_address("0.0.0.0:50051");
+    limestone::grpc::service::echo_service_impl service;
+
+    ServerBuilder builder;
+    builder.AddListeningPort(server_address, grpc::InsecureServerCredentials());
+    builder.RegisterService(&service);
+    std::unique_ptr<Server> server(builder.BuildAndStart());
+    
+    LOG(INFO) << "Echo server listening on " << server_address;
+    std::cout << "Server listening on " << server_address << std::endl;
+    server->Wait();
+}
+
+int main(int argc, char** argv) {
+    // Initialize glog
+    google::InitGoogleLogging(argv[0]);
+    FLAGS_logtostderr = 1;  // Log to stderr instead of file
+    
+    LOG(INFO) << "Starting limestone gRPC echo server";
+    RunServer();
+    return 0;
+}

sandbox/grpc/server_streaming_sample_client.cpp

@@ -0,0 +1,106 @@
+// Command-line client for server-streaming sample services
+#include <grpcpp/grpcpp.h>
+#include <iostream>
+#include <glog/logging.h>
+#include <fstream>
+#include <vector>
+#include <string>
+#include "server_streaming_sample.pb.h"
+#include "server_streaming_sample.grpc.pb.h"
+
+void print_usage(const char* prog_name) {
+    std::cout << "Usage: " << prog_name << " <service> <server_address> [options]\n"
+              << "  service: file_size | random_bytes\n"
+              << "  server_address: host:port\n"
+              << "  file_size options: <file_path>\n"
+              << "  random_bytes options: <size>\n";
+}
+
+int main(int argc, char* argv[]) {
+    if (argc < 3) {
+        print_usage(argv[0]);
+        return 1;
+    }
+    std::string service = argv[1];
+    std::string server_address = argv[2];
+    grpc::ChannelArguments args;
+    args.SetMaxSendMessageSize(64 * 1024 * 1024);    // 64MB
+    args.SetMaxReceiveMessageSize(64 * 1024 * 1024); // 64MB
+    auto channel = grpc::CreateCustomChannel(server_address, grpc::InsecureChannelCredentials(), args);
+
+    if (service == "file_size") {
+        using namespace std::chrono;
+        LOG(INFO) << "[file_size] start";
+        auto t0 = steady_clock::now();
+        if (argc < 4) {
+            std::cerr << "Missing file_path argument for file_size service\n";
+            return 1;
+        }
+        std::string file_path = argv[3];
+        std::ifstream file(file_path, std::ios::binary);
+        if (!file) {
+            std::cerr << "Failed to open file: " << file_path << "\n";
+            return 1;
+        }
+        limestone::grpc::FileSizeService::Stub stub(channel);
+        grpc::ClientContext context;
+        limestone::grpc::FileSizeResponse response;
+        std::unique_ptr<grpc::ClientWriter<limestone::grpc::FileChunk>> writer(
+            stub.GetFileSize(&context, &response));
+        constexpr size_t buffer_size = 32 * 1024 * 1024; // 32MB
+        std::vector<char> buffer(buffer_size);
+        while (file.read(buffer.data(), buffer.size()) || file.gcount() > 0) {
+            limestone::grpc::FileChunk chunk;
+            chunk.set_data(std::string(buffer.data(), file.gcount()));
+            if (!writer->Write(chunk)) {
+                std::cerr << "Failed to write chunk to server\n";
+                break;
+            }
+        }
+        writer->WritesDone();
+        grpc::Status status = writer->Finish();
+        auto t1 = steady_clock::now();
+        auto ms = duration_cast<milliseconds>(t1 - t0).count();
+        if (status.ok()) {
+            std::cout << "File size: " << response.size() << " bytes\n";
+        } else {
+            std::cerr << "RPC failed: " << status.error_message() << "\n";
+        }
+        LOG(INFO) << "[file_size] end: elapsed " << ms << " ms";
+    } else if (service == "random_bytes") {
+        using namespace std::chrono;
+        LOG(INFO) << "[random_bytes] start";
+        auto t0 = steady_clock::now();
+        if (argc < 4) {
+            std::cerr << "Missing size argument for random_bytes service\n";
+            return 1;
+        }
+        int64_t size = std::stoll(argv[3]);
+        limestone::grpc::RandomBytesService::Stub stub(channel);
+        grpc::ClientContext context;
+        limestone::grpc::RandomBytesRequest request;
+        request.set_size(size);
+        std::unique_ptr<grpc::ClientReader<limestone::grpc::RandomBytesChunk>> reader(
+            stub.GenerateRandomBytes(&context, request));
+        int64_t received = 0;
+        limestone::grpc::RandomBytesChunk chunk;
+        while (reader->Read(&chunk)) {
+            received += chunk.data().size();
+            // For demonstration, do not print the data
+        }
+        grpc::Status status = reader->Finish();
+        auto t1 = steady_clock::now();
+        auto ms = duration_cast<milliseconds>(t1 - t0).count();
+        if (status.ok()) {
+            std::cout << "Received " << received << " bytes of random data\n";
+        } else {
+            std::cerr << "RPC failed: " << status.error_message() << "\n";
+        }
+        LOG(INFO) << "[random_bytes] end: elapsed " << ms << " ms";
+    } else {
+        std::cerr << "Unknown service: " << service << "\n";
+        print_usage(argv[0]);
+        return 1;
+    }
+    return 0;
+}

sandbox/grpc/server_streaming_sample_server.cpp

@@ -0,0 +1,24 @@
+// Command-line server for server-streaming sample services
+#include <grpcpp/grpcpp.h>
+#include "grpc/service/server_streaming_sample_service.h"
+#include <iostream>
+
+int main(int argc, char* argv[]) {
+    std::string server_address = "0.0.0.0:50051";
+    if (argc > 1) {
+        server_address = argv[1];
+    }
+    grpc::ServerBuilder builder;
+    builder.SetMaxReceiveMessageSize(64 * 1024 * 1024); // 64MB
+    builder.SetMaxSendMessageSize(64 * 1024 * 1024);    // 64MB
+    builder.SetMaxMessageSize(64 * 1024 * 1024);        // 64MB
+    builder.AddListeningPort(server_address, grpc::InsecureServerCredentials());
+    limestone::grpc::service::FileSizeServiceImpl file_size_service;
+    limestone::grpc::service::RandomBytesServiceImpl random_bytes_service;
+    builder.RegisterService(&file_size_service);
+    builder.RegisterService(&random_bytes_service);
+    std::unique_ptr<grpc::Server> server(builder.BuildAndStart());
+    std::cout << "Server listening on " << server_address << std::endl;
+    server->Wait();
+    return 0;
+}