From eabc4027c89b37f5d5ae96e29c1f0c92e29c0f2b Mon Sep 17 00:00:00 2001 From: Philipp Moritz Date: Tue, 20 Feb 2018 13:37:09 -0800 Subject: [PATCH] Hiredis asio integration (#1547) --- cmake/Modules/ThirdpartyToolchain.cmake | 4 + src/ray/CMakeLists.txt | 1 + src/ray/gcs/CMakeLists.txt | 1 + src/ray/gcs/asio.cc | 110 ++++++++++++++++++++++++ src/ray/gcs/asio.h | 73 ++++++++++++++++ src/ray/gcs/asio_test.cc | 54 ++++++++++++ 6 files changed, 243 insertions(+) create mode 100644 src/ray/gcs/asio.cc create mode 100644 src/ray/gcs/asio.h create mode 100644 src/ray/gcs/asio_test.cc diff --git a/cmake/Modules/ThirdpartyToolchain.cmake b/cmake/Modules/ThirdpartyToolchain.cmake index 5b4c24275..95893a762 100644 --- a/cmake/Modules/ThirdpartyToolchain.cmake +++ b/cmake/Modules/ThirdpartyToolchain.cmake @@ -81,3 +81,7 @@ if(RAY_BUILD_TESTS OR RAY_BUILD_BENCHMARKS) add_dependencies(gflags gflags_ep) endif() + +set(Boost_USE_STATIC_LIBS ON) +find_package(Boost COMPONENTS system filesystem REQUIRED) +include_directories(${Boost_INCLUDE_DIR}) diff --git a/src/ray/CMakeLists.txt b/src/ray/CMakeLists.txt index e40d8cc10..5d043b680 100644 --- a/src/ray/CMakeLists.txt +++ b/src/ray/CMakeLists.txt @@ -27,6 +27,7 @@ set(RAY_SRCS gcs/client.cc gcs/tables.cc gcs/redis_context.cc + gcs/asio.cc ) install(FILES diff --git a/src/ray/gcs/CMakeLists.txt b/src/ray/gcs/CMakeLists.txt index ec0b6adfc..5c75984d9 100644 --- a/src/ray/gcs/CMakeLists.txt +++ b/src/ray/gcs/CMakeLists.txt @@ -20,6 +20,7 @@ add_custom_command( add_custom_target(gen_gcs_fbs DEPENDS ${GCS_FBS_OUTPUT_FILES}) ADD_RAY_TEST(client_test STATIC_LINK_LIBS ray_static ${PLASMA_STATIC_LIB} ${ARROW_STATIC_LIB} gtest gtest_main pthread) +ADD_RAY_TEST(asio_test STATIC_LINK_LIBS ray_static ${PLASMA_STATIC_LIB} ${ARROW_STATIC_LIB} gtest gtest_main pthread ${Boost_SYSTEM_LIBRARY}) install(FILES client.h diff --git a/src/ray/gcs/asio.cc b/src/ray/gcs/asio.cc new file mode 100644 index 000000000..7b6ccc259 --- /dev/null +++ b/src/ray/gcs/asio.cc @@ -0,0 +1,110 @@ +#include "asio.h" + +#include "ray/util/logging.h" + +RedisAsioClient::RedisAsioClient(boost::asio::io_service &io_service, + redisAsyncContext *async_context) + : async_context_(async_context), + socket_(io_service), + read_requested_(false), + write_requested_(false), + read_in_progress_(false), + write_in_progress_(false) { + // gives access to c->fd + redisContext *c = &(async_context->c); + + // hiredis is already connected + // use the existing native socket + socket_.assign(boost::asio::ip::tcp::v4(), c->fd); + + // register hooks with the hiredis async context + async_context->ev.addRead = call_C_addRead; + async_context->ev.delRead = call_C_delRead; + async_context->ev.addWrite = call_C_addWrite; + async_context->ev.delWrite = call_C_delWrite; + async_context->ev.cleanup = call_C_cleanup; + + // C wrapper functions will use this pointer to call class members. + async_context->ev.data = this; +} + +void RedisAsioClient::operate() { + if (read_requested_ && !read_in_progress_) { + read_in_progress_ = true; + socket_.async_read_some(boost::asio::null_buffers(), + boost::bind(&RedisAsioClient::handle_read, this, + boost::asio::placeholders::error)); + } + + if (write_requested_ && !write_in_progress_) { + write_in_progress_ = true; + socket_.async_write_some(boost::asio::null_buffers(), + boost::bind(&RedisAsioClient::handle_write, this, + boost::asio::placeholders::error)); + } +} + +void RedisAsioClient::handle_read(boost::system::error_code error_code) { + RAY_CHECK(!error_code || error_code == boost::asio::error::would_block); + read_in_progress_ = false; + redisAsyncHandleRead(async_context_); + + if (error_code == boost::asio::error::would_block) { + operate(); + } +} + +void RedisAsioClient::handle_write(boost::system::error_code error_code) { + RAY_CHECK(!error_code || error_code == boost::asio::error::would_block); + write_in_progress_ = false; + redisAsyncHandleWrite(async_context_); + + if (error_code == boost::asio::error::would_block) { + operate(); + } +} + +void RedisAsioClient::add_read() { + read_requested_ = true; + operate(); +} + +void RedisAsioClient::del_read() { + read_requested_ = false; +} + +void RedisAsioClient::add_write() { + write_requested_ = true; + operate(); +} + +void RedisAsioClient::del_write() { + write_requested_ = false; +} + +void RedisAsioClient::cleanup() {} + +static inline RedisAsioClient *cast_to_client(void *private_data) { + RAY_CHECK(private_data != nullptr); + return static_cast(private_data); +} + +extern "C" void call_C_addRead(void *private_data) { + cast_to_client(private_data)->add_read(); +} + +extern "C" void call_C_delRead(void *private_data) { + cast_to_client(private_data)->del_read(); +} + +extern "C" void call_C_addWrite(void *private_data) { + cast_to_client(private_data)->add_write(); +} + +extern "C" void call_C_delWrite(void *private_data) { + cast_to_client(private_data)->del_write(); +} + +extern "C" void call_C_cleanup(void *private_data) { + cast_to_client(private_data)->cleanup(); +} diff --git a/src/ray/gcs/asio.h b/src/ray/gcs/asio.h new file mode 100644 index 000000000..187d3d7ac --- /dev/null +++ b/src/ray/gcs/asio.h @@ -0,0 +1,73 @@ +// 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. +// +// Adapted from https://github.com/ryangraham/hiredis-boostasio-adapter +// (Copyright 2018 Ryan Graham) + +#ifndef RAY_GCS_ASIO_H +#define RAY_GCS_ASIO_H + +#include "hiredis/async.h" +#include "hiredis/hiredis.h" + +#include +#include +#include + +#include +#include +#include + +using boost::asio::ip::tcp; + +class RedisAsioClient { + public: + RedisAsioClient(boost::asio::io_service &io_service, redisAsyncContext *ac); + + void operate(); + + void handle_read(boost::system::error_code ec); + void handle_write(boost::system::error_code ec); + void add_read(); + void del_read(); + void add_write(); + void del_write(); + void cleanup(); + + private: + redisAsyncContext *async_context_; + boost::asio::ip::tcp::socket socket_; + // Hiredis wanted to add a read operation to the event loop + // but the read might not have happened yet + bool read_requested_; + // Hiredis wanted to add a write operation to the event loop + // but the read might not have happened yet + bool write_requested_; + // A read is currently in progress + bool read_in_progress_; + // A write is currently in progress + bool write_in_progress_; +}; + +// C wrappers for class member functions +extern "C" void call_C_addRead(void *private_data); +extern "C" void call_C_delRead(void *private_data); +extern "C" void call_C_addWrite(void *private_data); +extern "C" void call_C_delWrite(void *private_data); +extern "C" void call_C_cleanup(void *private_data); + +#endif // RAY_GCS_ASIO_H diff --git a/src/ray/gcs/asio_test.cc b/src/ray/gcs/asio_test.cc new file mode 100644 index 000000000..53c4ef167 --- /dev/null +++ b/src/ray/gcs/asio_test.cc @@ -0,0 +1,54 @@ +#include + +#include "gtest/gtest.h" + +#include "asio.h" + +boost::asio::io_service io_service; + +// For this test to work, you need to have a redis-server in your PATH + +class TestRedisAsioClient : public ::testing::Test { + public: + TestRedisAsioClient() { + int r = system("redis-server > /dev/null & sleep 1"); + std::cout << "TestRedisAsioClient: redis-server status code was " << r + << std::endl; + } + ~TestRedisAsioClient() { + int r = system("redis-cli -c shutdown"); + std::cout << "TestRedisAsioClient: redis-cli status code was " << r + << std::endl; + } +}; + +void ConnectCallback(const redisAsyncContext *c, int status) { + ASSERT_EQ(status, REDIS_OK); +} + +void DisconnectCallback(const redisAsyncContext *c, int status) { + ASSERT_EQ(status, REDIS_OK); +} + +void GetCallback(redisAsyncContext *c, void *r, void *privdata) { + redisReply *reply = reinterpret_cast(r); + ASSERT_TRUE(reply != nullptr); + ASSERT_TRUE(std::string(reinterpret_cast(reply->str)) == "test"); + redisAsyncDisconnect(c); + io_service.stop(); +} + +TEST_F(TestRedisAsioClient, TestRedisCommands) { + redisAsyncContext *ac = redisAsyncConnect("127.0.0.1", 6379); + ASSERT_TRUE(ac->err == 0); + + RedisAsioClient client(io_service, ac); + + redisAsyncSetConnectCallback(ac, ConnectCallback); + redisAsyncSetDisconnectCallback(ac, DisconnectCallback); + + redisAsyncCommand(ac, NULL, NULL, "SET key test"); + redisAsyncCommand(ac, GetCallback, nullptr, "GET key"); + + io_service.run(); +}