From cac5f476004025a8f709a9f938c8fdbff333b804 Mon Sep 17 00:00:00 2001 From: Philipp Moritz Date: Thu, 14 Dec 2017 14:54:09 -0800 Subject: [PATCH] First Part of Internal Ray API Refactor (#1173) * add Ray status class * add C++ util files * add ID types * more APIs * build system integration * add test infrastructure and implement some APIs * add more tests * fix bugs * add task table tests * update * add toolchain file * fix * test * link with pthread * update * fix * more fixes * fixes * always vendor gtest and gflags * linting * fixes * add constants file * comments * more fixes * fix linting --- .travis.yml | 1 + CMakeLists.txt | 29 +++ cmake/Modules/BuildUtils.cmake | 228 ++++++++++++++++++++ cmake/Modules/ThirdpartyToolchain.cmake | 83 +++++++ src/common/cmake/Common.cmake | 4 +- src/common/redis_module/ray_redis_module.cc | 54 +++++ src/ray/CMakeLists.txt | 52 +++++ src/ray/constants.h | 12 ++ src/ray/gcs/CMakeLists.txt | 26 +++ src/ray/gcs/client.cc | 45 ++++ src/ray/gcs/client.h | 81 +++++++ src/ray/gcs/client_test.cc | 83 +++++++ src/ray/gcs/format/gcs.fbs | 41 ++++ src/ray/gcs/redis_context.cc | 139 ++++++++++++ src/ray/gcs/redis_context.h | 63 ++++++ src/ray/gcs/tables.cc | 7 + src/ray/gcs/tables.h | 195 +++++++++++++++++ src/ray/id.cc | 68 ++++++ src/ray/id.h | 55 +++++ src/ray/ray.pc.in | 11 + src/ray/status.cc | 88 ++++++++ src/ray/status.h | 184 ++++++++++++++++ src/ray/test/run_gcs_tests.sh | 14 ++ src/ray/util/CMakeLists.txt | 6 + src/ray/util/logging.h | 145 +++++++++++++ src/ray/util/macros.h | 44 ++++ src/ray/util/visibility.h | 44 ++++ 27 files changed, 1801 insertions(+), 1 deletion(-) create mode 100644 cmake/Modules/BuildUtils.cmake create mode 100644 cmake/Modules/ThirdpartyToolchain.cmake create mode 100644 src/ray/CMakeLists.txt create mode 100644 src/ray/constants.h create mode 100644 src/ray/gcs/CMakeLists.txt create mode 100644 src/ray/gcs/client.cc create mode 100644 src/ray/gcs/client.h create mode 100644 src/ray/gcs/client_test.cc create mode 100644 src/ray/gcs/format/gcs.fbs create mode 100644 src/ray/gcs/redis_context.cc create mode 100644 src/ray/gcs/redis_context.h create mode 100644 src/ray/gcs/tables.cc create mode 100644 src/ray/gcs/tables.h create mode 100644 src/ray/id.cc create mode 100644 src/ray/id.h create mode 100644 src/ray/ray.pc.in create mode 100644 src/ray/status.cc create mode 100644 src/ray/status.h create mode 100644 src/ray/test/run_gcs_tests.sh create mode 100644 src/ray/util/CMakeLists.txt create mode 100644 src/ray/util/logging.h create mode 100644 src/ray/util/macros.h create mode 100644 src/ray/util/visibility.h diff --git a/.travis.yml b/.travis.yml index 9cee5ecba..64c663854 100644 --- a/.travis.yml +++ b/.travis.yml @@ -94,6 +94,7 @@ install: - ./.travis/install-cython-examples.sh - cd python/ray/core + - bash ../../../src/ray/test/run_gcs_tests.sh - bash ../../../src/common/test/run_tests.sh - bash ../../../src/plasma/test/run_tests.sh - bash ../../../src/local_scheduler/test/run_tests.sh diff --git a/CMakeLists.txt b/CMakeLists.txt index 546e58974..c2f6aebf7 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -2,11 +2,40 @@ cmake_minimum_required(VERSION 3.4) project(ray) +set(CMAKE_MODULE_PATH ${CMAKE_MODULE_PATH} "${CMAKE_SOURCE_DIR}/cmake/Modules") + +# This ensures that things like gnu++11 get passed correctly +set(CMAKE_CXX_STANDARD 11) + +# We require a C++11 compliant compiler +set(CMAKE_CXX_STANDARD_REQUIRED ON) + +option(RAY_BUILD_STATIC + "Build the libarrow static libraries" + ON) + +option(RAY_BUILD_SHARED + "Build the libarrow shared libraries" + ON) + +option(RAY_BUILD_TESTS + "Build the Ray googletest unit tests" + ON) + +include(ExternalProject) +include(GNUInstallDirs) +include(BuildUtils) +enable_testing() + +include(ThirdpartyToolchain) + set(ARROW_DIR "${CMAKE_CURRENT_LIST_DIR}/src/thirdparty/arrow/" CACHE STRING "Path of the arrow source directory") include_directories("${ARROW_DIR}/cpp/src/") +include_directories("${CMAKE_CURRENT_LIST_DIR}/src/") +add_subdirectory(${CMAKE_CURRENT_LIST_DIR}/src/ray/) add_subdirectory(${CMAKE_CURRENT_LIST_DIR}/src/common/) add_subdirectory(${CMAKE_CURRENT_LIST_DIR}/src/plasma/) add_subdirectory(${CMAKE_CURRENT_LIST_DIR}/src/local_scheduler/) diff --git a/cmake/Modules/BuildUtils.cmake b/cmake/Modules/BuildUtils.cmake new file mode 100644 index 000000000..d3e148b10 --- /dev/null +++ b/cmake/Modules/BuildUtils.cmake @@ -0,0 +1,228 @@ +function(ADD_THIRDPARTY_LIB LIB_NAME) + set(options) + set(one_value_args SHARED_LIB STATIC_LIB) + set(multi_value_args DEPS) + cmake_parse_arguments(ARG "${options}" "${one_value_args}" "${multi_value_args}" ${ARGN}) + if(ARG_UNPARSED_ARGUMENTS) + message(SEND_ERROR "Error: unrecognized arguments: ${ARG_UNPARSED_ARGUMENTS}") + endif() + + if(ARG_STATIC_LIB AND ARG_SHARED_LIB) + if(NOT ARG_STATIC_LIB) + message(FATAL_ERROR "No static or shared library provided for ${LIB_NAME}") + endif() + + SET(AUG_LIB_NAME "${LIB_NAME}_static") + add_library(${AUG_LIB_NAME} STATIC IMPORTED) + set_target_properties(${AUG_LIB_NAME} + PROPERTIES IMPORTED_LOCATION "${ARG_STATIC_LIB}") + message("Added static library dependency ${LIB_NAME}: ${ARG_STATIC_LIB}") + + SET(AUG_LIB_NAME "${LIB_NAME}_shared") + add_library(${AUG_LIB_NAME} SHARED IMPORTED) + + if(MSVC) + # Mark the ”.lib” location as part of a Windows DLL + set_target_properties(${AUG_LIB_NAME} + PROPERTIES IMPORTED_IMPLIB "${ARG_SHARED_LIB}") + else() + set_target_properties(${AUG_LIB_NAME} + PROPERTIES IMPORTED_LOCATION "${ARG_SHARED_LIB}") + endif() + if(ARG_DEPS) + set_target_properties(${AUG_LIB_NAME} + PROPERTIES IMPORTED_LINK_INTERFACE_LIBRARIES "${ARG_DEPS}") + endif() + message("Added shared library dependency ${LIB_NAME}: ${ARG_SHARED_LIB}") + elseif(ARG_STATIC_LIB) + add_library(${LIB_NAME} STATIC IMPORTED) + set_target_properties(${LIB_NAME} + PROPERTIES IMPORTED_LOCATION "${ARG_STATIC_LIB}") + SET(AUG_LIB_NAME "${LIB_NAME}_static") + add_library(${AUG_LIB_NAME} STATIC IMPORTED) + set_target_properties(${AUG_LIB_NAME} + PROPERTIES IMPORTED_LOCATION "${ARG_STATIC_LIB}") + if(ARG_DEPS) + set_target_properties(${AUG_LIB_NAME} + PROPERTIES IMPORTED_LINK_INTERFACE_LIBRARIES "${ARG_DEPS}") + endif() + message("Added static library dependency ${LIB_NAME}: ${ARG_STATIC_LIB}") + elseif(ARG_SHARED_LIB) + add_library(${LIB_NAME} SHARED IMPORTED) + set_target_properties(${LIB_NAME} + PROPERTIES IMPORTED_LOCATION "${ARG_SHARED_LIB}") + SET(AUG_LIB_NAME "${LIB_NAME}_shared") + add_library(${AUG_LIB_NAME} SHARED IMPORTED) + + if(MSVC) + # Mark the ”.lib” location as part of a Windows DLL + set_target_properties(${AUG_LIB_NAME} + PROPERTIES IMPORTED_IMPLIB "${ARG_SHARED_LIB}") + else() + set_target_properties(${AUG_LIB_NAME} + PROPERTIES IMPORTED_LOCATION "${ARG_SHARED_LIB}") + endif() + message("Added shared library dependency ${LIB_NAME}: ${ARG_SHARED_LIB}") + if(ARG_DEPS) + set_target_properties(${AUG_LIB_NAME} + PROPERTIES IMPORTED_LINK_INTERFACE_LIBRARIES "${ARG_DEPS}") + endif() + else() + message(FATAL_ERROR "No static or shared library provided for ${LIB_NAME}") + endif() +endfunction() + +function(ADD_RAY_LIB LIB_NAME) + set(options) + set(one_value_args SHARED_LINK_FLAGS) + set(multi_value_args SOURCES STATIC_LINK_LIBS STATIC_PRIVATE_LINK_LIBS SHARED_LINK_LIBS SHARED_PRIVATE_LINK_LIBS DEPENDENCIES) + cmake_parse_arguments(ARG "${options}" "${one_value_args}" "${multi_value_args}" ${ARGN}) + if(ARG_UNPARSED_ARGUMENTS) + message(SEND_ERROR "Error: unrecognized arguments: ${ARG_UNPARSED_ARGUMENTS}") + endif() + + add_library(${LIB_NAME}_objlib OBJECT + ${ARG_SOURCES} + ) + + if (ARG_DEPENDENCIES) + add_dependencies(${LIB_NAME}_objlib ${ARG_DEPENDENCIES}) + endif() + + # Necessary to make static linking into other shared libraries work properly + set_property(TARGET ${LIB_NAME}_objlib PROPERTY POSITION_INDEPENDENT_CODE 1) + + set(RUNTIME_INSTALL_DIR bin) + + if (RAY_BUILD_SHARED) + add_library(${LIB_NAME}_shared SHARED $) + + if(APPLE) + # On OS X, you can avoid linking at library load time and instead + # expecting that the symbols have been loaded separately. This happens + # with libpython* where there can be conflicts between system Python and + # the Python from a thirdparty distribution + set(ARG_SHARED_LINK_FLAGS + "-undefined dynamic_lookup ${ARG_SHARED_LINK_FLAGS}") + endif() + + set_target_properties(${LIB_NAME}_shared + PROPERTIES + LIBRARY_OUTPUT_DIRECTORY "${BUILD_OUTPUT_ROOT_DIRECTORY}" + RUNTIME_OUTPUT_DIRECTORY "${BUILD_OUTPUT_ROOT_DIRECTORY}" + PDB_OUTPUT_DIRECTORY "${BUILD_OUTPUT_ROOT_DIRECTORY}" + LINK_FLAGS "${ARG_SHARED_LINK_FLAGS}" + OUTPUT_NAME ${LIB_NAME} + VERSION "${RAY_ABI_VERSION}" + SOVERSION "${RAY_SO_VERSION}") + + target_link_libraries(${LIB_NAME}_shared + LINK_PUBLIC ${ARG_SHARED_LINK_LIBS} + LINK_PRIVATE ${ARG_SHARED_PRIVATE_LINK_LIBS}) + + if (RAY_RPATH_ORIGIN) + if (APPLE) + set(_lib_install_rpath "@loader_path") + else() + set(_lib_install_rpath "\$ORIGIN") + endif() + set_target_properties(${LIB_NAME}_shared PROPERTIES + INSTALL_RPATH ${_lib_install_rpath}) + endif() + + install(TARGETS ${LIB_NAME}_shared + RUNTIME DESTINATION ${RUNTIME_INSTALL_DIR} + LIBRARY DESTINATION ${CMAKE_INSTALL_LIBDIR} + ARCHIVE DESTINATION ${CMAKE_INSTALL_LIBDIR}) + endif() + + if (RAY_BUILD_STATIC) + if (MSVC) + set(LIB_NAME_STATIC ${LIB_NAME}_static) + else() + set(LIB_NAME_STATIC ${LIB_NAME}) + endif() + add_library(${LIB_NAME}_static STATIC $) + set_target_properties(${LIB_NAME}_static + PROPERTIES + LIBRARY_OUTPUT_DIRECTORY "${BUILD_OUTPUT_ROOT_DIRECTORY}" + OUTPUT_NAME ${LIB_NAME_STATIC}) + + target_link_libraries(${LIB_NAME}_static + LINK_PUBLIC ${ARG_STATIC_LINK_LIBS} + LINK_PRIVATE ${ARG_STATIC_PRIVATE_LINK_LIBS}) + + install(TARGETS ${LIB_NAME}_static + RUNTIME DESTINATION ${RUNTIME_INSTALL_DIR} + LIBRARY DESTINATION ${CMAKE_INSTALL_LIBDIR} + ARCHIVE DESTINATION ${CMAKE_INSTALL_LIBDIR}) + endif() + + if (APPLE) + set_target_properties(${LIB_NAME}_shared + PROPERTIES + BUILD_WITH_INSTALL_RPATH ON + INSTALL_NAME_DIR "@rpath") + endif() + +endfunction() + +############################################################ +# Testing +############################################################ +# Add a new test case, with or without an executable that should be built. +# +# REL_TEST_NAME is the name of the test. It may be a single component +# (e.g. monotime-test) or contain additional components (e.g. +# net/net_util-test). Either way, the last component must be a globally +# unique name. +# +# The unit test is added with a label of "unittest" to support filtering with +# ctest. +# +# Arguments after the test name will be passed to set_tests_properties(). +function(ADD_RAY_TEST REL_TEST_NAME) + set(options NO_VALGRIND) + set(single_value_args) + set(multi_value_args STATIC_LINK_LIBS) + cmake_parse_arguments(ARG "${options}" "${one_value_args}" "${multi_value_args}" ${ARGN}) + if(ARG_UNPARSED_ARGUMENTS) + message(SEND_ERROR "Error: unrecognized arguments: ${ARG_UNPARSED_ARGUMENTS}") + endif() + + if(NO_TESTS OR NOT RAY_BUILD_STATIC) + return() + endif() + get_filename_component(TEST_NAME ${REL_TEST_NAME} NAME_WE) + + if(EXISTS ${CMAKE_CURRENT_SOURCE_DIR}/${REL_TEST_NAME}.cc) + # This test has a corresponding .cc file, set it up as an executable. + set(TEST_PATH "${EXECUTABLE_OUTPUT_PATH}/${TEST_NAME}") + add_executable(${TEST_NAME} "${REL_TEST_NAME}.cc") + + if (ARG_STATIC_LINK_LIBS) + # Customize link libraries + target_link_libraries(${TEST_NAME} ${ARG_STATIC_LINK_LIBS}) + else() + target_link_libraries(${TEST_NAME} ${RAY_TEST_LINK_LIBS}) + endif() + add_dependencies(unittest ${TEST_NAME}) + else() + # No executable, just invoke the test (probably a script) directly. + set(TEST_PATH ${CMAKE_CURRENT_SOURCE_DIR}/${REL_TEST_NAME}) + endif() + + if (RAY_TEST_MEMCHECK AND NOT ARG_NO_VALGRIND) + SET_PROPERTY(TARGET ${TEST_NAME} + APPEND_STRING PROPERTY + COMPILE_FLAGS " -RAY_VALGRIND") + add_test(${TEST_NAME} + bash -c "cd ${EXECUTABLE_OUTPUT_PATH}; valgrind --tool=memcheck --leak-check=full --leak-check-heuristics=stdstring --error-exitcode=1 ${TEST_PATH}") + elseif(MSVC) + add_test(${TEST_NAME} ${TEST_PATH}) + else() + add_test(${TEST_NAME} + ${BUILD_SUPPORT_DIR}/run-test.sh ${CMAKE_BINARY_DIR} test ${TEST_PATH}) + endif() + set_tests_properties(${TEST_NAME} PROPERTIES LABELS "unittest") +endfunction() diff --git a/cmake/Modules/ThirdpartyToolchain.cmake b/cmake/Modules/ThirdpartyToolchain.cmake new file mode 100644 index 000000000..5b4c24275 --- /dev/null +++ b/cmake/Modules/ThirdpartyToolchain.cmake @@ -0,0 +1,83 @@ +set(GFLAGS_VERSION "2.2.0") +set(GTEST_VERSION "1.8.0") +set(GBENCHMARK_VERSION "1.1.0") + +if(RAY_BUILD_TESTS OR RAY_BUILD_BENCHMARKS) + add_custom_target(unittest ctest -L unittest) + + if(APPLE) + set(GTEST_CMAKE_CXX_FLAGS "-fPIC -DGTEST_USE_OWN_TR1_TUPLE=1 -Wno-unused-value -Wno-ignored-attributes") + elseif(NOT MSVC) + set(GTEST_CMAKE_CXX_FLAGS "-fPIC") + endif() + string(TOUPPER ${CMAKE_BUILD_TYPE} UPPERCASE_BUILD_TYPE) + set(GTEST_CMAKE_CXX_FLAGS "${EP_CXX_FLAGS} ${CMAKE_CXX_FLAGS_${UPPERCASE_BUILD_TYPE}} ${GTEST_CMAKE_CXX_FLAGS}") + + set(GTEST_PREFIX "${CMAKE_CURRENT_BINARY_DIR}/googletest_ep-prefix/src/googletest_ep") + set(GTEST_INCLUDE_DIR "${GTEST_PREFIX}/include") + set(GTEST_STATIC_LIB + "${GTEST_PREFIX}/lib/${CMAKE_STATIC_LIBRARY_PREFIX}gtest${CMAKE_STATIC_LIBRARY_SUFFIX}") + set(GTEST_MAIN_STATIC_LIB + "${GTEST_PREFIX}/lib/${CMAKE_STATIC_LIBRARY_PREFIX}gtest_main${CMAKE_STATIC_LIBRARY_SUFFIX}") + set(GTEST_CMAKE_ARGS -DCMAKE_BUILD_TYPE=${CMAKE_BUILD_TYPE} + -DCMAKE_INSTALL_PREFIX=${GTEST_PREFIX} + -DCMAKE_CXX_FLAGS=${GTEST_CMAKE_CXX_FLAGS}) + if (MSVC AND NOT ARROW_USE_STATIC_CRT) + set(GTEST_CMAKE_ARGS ${GTEST_CMAKE_ARGS} -Dgtest_force_shared_crt=ON) + endif() + + ExternalProject_Add(googletest_ep + URL "https://github.com/google/googletest/archive/release-${GTEST_VERSION}.tar.gz" + BUILD_BYPRODUCTS ${GTEST_STATIC_LIB} ${GTEST_MAIN_STATIC_LIB} + CMAKE_ARGS ${GTEST_CMAKE_ARGS} + ${EP_LOG_OPTIONS}) + + message(STATUS "GTest include dir: ${GTEST_INCLUDE_DIR}") + message(STATUS "GTest static library: ${GTEST_STATIC_LIB}") + include_directories(SYSTEM ${GTEST_INCLUDE_DIR}) + ADD_THIRDPARTY_LIB(gtest + STATIC_LIB ${GTEST_STATIC_LIB}) + ADD_THIRDPARTY_LIB(gtest_main + STATIC_LIB ${GTEST_MAIN_STATIC_LIB}) + + add_dependencies(gtest googletest_ep) + add_dependencies(gtest_main googletest_ep) + + set(GFLAGS_CMAKE_CXX_FLAGS ${EP_CXX_FLAGS}) + + set(GFLAGS_URL "https://github.com/gflags/gflags/archive/v${GFLAGS_VERSION}.tar.gz") + set(GFLAGS_PREFIX "${CMAKE_CURRENT_BINARY_DIR}/gflags_ep-prefix/src/gflags_ep") + set(GFLAGS_HOME "${GFLAGS_PREFIX}") + set(GFLAGS_INCLUDE_DIR "${GFLAGS_PREFIX}/include") + if(MSVC) + set(GFLAGS_STATIC_LIB "${GFLAGS_PREFIX}/lib/gflags_static.lib") + else() + set(GFLAGS_STATIC_LIB "${GFLAGS_PREFIX}/lib/libgflags.a") + endif() + set(GFLAGS_CMAKE_ARGS -DCMAKE_BUILD_TYPE=${CMAKE_BUILD_TYPE} + -DCMAKE_INSTALL_PREFIX=${GFLAGS_PREFIX} + -DBUILD_SHARED_LIBS=OFF + -DBUILD_STATIC_LIBS=ON + -DBUILD_PACKAGING=OFF + -DBUILD_TESTING=OFF + -BUILD_CONFIG_TESTS=OFF + -DINSTALL_HEADERS=ON + -DCMAKE_CXX_FLAGS_${UPPERCASE_BUILD_TYPE}=${EP_CXX_FLAGS} + -DCMAKE_C_FLAGS_${UPPERCASE_BUILD_TYPE}=${EP_C_FLAGS} + -DCMAKE_CXX_FLAGS=${GFLAGS_CMAKE_CXX_FLAGS}) + + ExternalProject_Add(gflags_ep + URL ${GFLAGS_URL} + ${EP_LOG_OPTIONS} + BUILD_IN_SOURCE 1 + BUILD_BYPRODUCTS "${GFLAGS_STATIC_LIB}" + CMAKE_ARGS ${GFLAGS_CMAKE_ARGS}) + + message(STATUS "GFlags include dir: ${GFLAGS_INCLUDE_DIR}") + message(STATUS "GFlags static library: ${GFLAGS_STATIC_LIB}") + include_directories(SYSTEM ${GFLAGS_INCLUDE_DIR}) + ADD_THIRDPARTY_LIB(gflags + STATIC_LIB ${GFLAGS_STATIC_LIB}) + + add_dependencies(gflags gflags_ep) +endif() diff --git a/src/common/cmake/Common.cmake b/src/common/cmake/Common.cmake index 983b3db2f..134f35680 100644 --- a/src/common/cmake/Common.cmake +++ b/src/common/cmake/Common.cmake @@ -22,6 +22,8 @@ if (NOT TARGET flatbuffers_ep) "-DFLATBUFFERS_BUILD_TESTS=OFF") endif() +set(FBS_DEPENDS flatbuffers_ep) + set(FLATBUFFERS_INCLUDE_DIR "${FLATBUFFERS_PREFIX}/include") set(FLATBUFFERS_STATIC_LIB "${FLATBUFFERS_PREFIX}/lib/libflatbuffers.a") set(FLATBUFFERS_COMPILER "${FLATBUFFERS_PREFIX}/bin/flatc") @@ -33,7 +35,7 @@ include_directories(SYSTEM ${FLATBUFFERS_INCLUDE_DIR}) # Custom CFLAGS -set(CMAKE_C_FLAGS "-g -Wall -Wextra -Werror=implicit-function-declaration -Wno-sign-compare -Wno-unused-parameter -Wno-type-limits -Wno-missing-field-initializers --std=c99 -D_XOPEN_SOURCE=500 -D_POSIX_C_SOURCE=200809L -fPIC -std=c99") +set(CMAKE_C_FLAGS "-g -Wall -Wextra -Werror=implicit-function-declaration -Wno-sign-compare -Wno-unused-parameter -Wno-type-limits -Wno-missing-field-initializers --std=c99 -fPIC -std=c99") # Code for finding Python find_package(PythonInterp REQUIRED) diff --git a/src/common/redis_module/ray_redis_module.cc b/src/common/redis_module/ray_redis_module.cc index ab14fd06f..a6827e0e0 100644 --- a/src/common/redis_module/ray_redis_module.cc +++ b/src/common/redis_module/ray_redis_module.cc @@ -388,6 +388,49 @@ bool PublishObjectNotification(RedisModuleCtx *ctx, return true; } +// This is a temporary redis command that will be removed once +// the GCS uses https://github.com/pcmoritz/credis. +int TableAdd_RedisCommand(RedisModuleCtx *ctx, + RedisModuleString **argv, + int argc) { + if (argc != 3) { + return RedisModule_WrongArity(ctx); + } + + RedisModuleString *id = argv[1]; + RedisModuleString *data = argv[2]; + + // Set the keys in the table + RedisModuleKey *key = + OpenPrefixedKey(ctx, "T:", id, REDISMODULE_READ | REDISMODULE_WRITE); + RedisModule_StringSet(key, data); + RedisModule_CloseKey(key); + + return RedisModule_ReplyWithSimpleString(ctx, "OK"); +} + +// This is a temporary redis command that will be removed once +// the GCS uses https://github.com/pcmoritz/credis. +int TableLookup_RedisCommand(RedisModuleCtx *ctx, + RedisModuleString **argv, + int argc) { + if (argc != 2) { + return RedisModule_WrongArity(ctx); + } + + RedisModuleString *id = argv[1]; + + RedisModuleKey *key = OpenPrefixedKey(ctx, "T:", id, REDISMODULE_READ); + size_t len = 0; + const char *buf = RedisModule_StringDMA(key, &len, REDISMODULE_READ); + + RedisModule_ReplyWithStringBuffer(ctx, buf, len); + + RedisModule_CloseKey(key); + + return REDISMODULE_OK; +} + /** * Add a new entry to the object table or update an existing one. * @@ -1149,6 +1192,17 @@ int RedisModule_OnLoad(RedisModuleCtx *ctx, return REDISMODULE_ERR; } + if (RedisModule_CreateCommand(ctx, "ray.table_add", TableAdd_RedisCommand, + "write", 0, 0, 0) == REDISMODULE_ERR) { + return REDISMODULE_ERR; + } + + if (RedisModule_CreateCommand(ctx, "ray.table_lookup", + TableLookup_RedisCommand, "readonly", 0, 0, + 0) == REDISMODULE_ERR) { + return REDISMODULE_ERR; + } + if (RedisModule_CreateCommand(ctx, "ray.object_table_lookup", ObjectTableLookup_RedisCommand, "readonly", 0, 0, 0) == REDISMODULE_ERR) { diff --git a/src/ray/CMakeLists.txt b/src/ray/CMakeLists.txt new file mode 100644 index 000000000..c1be6700b --- /dev/null +++ b/src/ray/CMakeLists.txt @@ -0,0 +1,52 @@ +cmake_minimum_required(VERSION 3.2) +set(CMAKE_CXX_FLAGS "-g -Wall -Werror -std=c++11") + +include_directories(${CMAKE_CURRENT_LIST_DIR}/../common/thirdparty/ae) + +add_subdirectory(util) +add_subdirectory(gcs) + +include(${CMAKE_CURRENT_LIST_DIR}/../common/cmake/Common.cmake) + +set(AE_SRCS + ${CMAKE_CURRENT_LIST_DIR}/../common/thirdparty/ae/ae.c +) + +set(HIREDIS_SRCS + ${CMAKE_CURRENT_LIST_DIR}/../common/thirdparty/hiredis/async.c + ${CMAKE_CURRENT_LIST_DIR}/../common/thirdparty/hiredis/dict.c + ${CMAKE_CURRENT_LIST_DIR}/../common/thirdparty/hiredis/hiredis.c + ${CMAKE_CURRENT_LIST_DIR}/../common/thirdparty/hiredis/net.c + ${CMAKE_CURRENT_LIST_DIR}/../common/thirdparty/hiredis/read.c + ${CMAKE_CURRENT_LIST_DIR}/../common/thirdparty/hiredis/sds.c +) + +set(RAY_SRCS + id.cc + status.cc + gcs/client.cc + gcs/tables.cc + gcs/redis_context.cc +) + +install(FILES + api.h + id.h + status.h + DESTINATION "${CMAKE_INSTALL_INCLUDEDIR}/ray") + +# pkg-config support +configure_file(ray.pc.in + "${CMAKE_CURRENT_BINARY_DIR}/ray.pc" + @ONLY) +install( + FILES "${CMAKE_CURRENT_BINARY_DIR}/ray.pc" + DESTINATION "${CMAKE_INSTALL_LIBDIR}/pkgconfig/") + +ADD_RAY_LIB(ray + SOURCES ${RAY_SRCS} ${AE_SRCS} ${HIREDIS_SRCS} + DEPENDENCIES gen_gcs_fbs + SHARED_LINK_LIBS + STATIC_LINK_LIBS) + +set(RAY_TEST_LINK_LIBS ray_static gtest gtest_main) diff --git a/src/ray/constants.h b/src/ray/constants.h new file mode 100644 index 000000000..a084543c4 --- /dev/null +++ b/src/ray/constants.h @@ -0,0 +1,12 @@ +#ifndef RAY_CONSTANTS_H_ +#define RAY_CONSTANTS_H_ + +/// Length of Ray IDs in bytes. +constexpr int64_t kUniqueIDSize = 20; + +/// Prefix for the object table keys in redis. +constexpr char kObjectTablePrefix[] = "ObjectTable"; +/// Prefix for the task table keys in redis. +constexpr char kTaskTablePrefix[] = "TaskTable"; + +#endif // RAY_CONSTANTS_H_ diff --git a/src/ray/gcs/CMakeLists.txt b/src/ray/gcs/CMakeLists.txt new file mode 100644 index 000000000..624f78264 --- /dev/null +++ b/src/ray/gcs/CMakeLists.txt @@ -0,0 +1,26 @@ +include(${CMAKE_CURRENT_LIST_DIR}/../../common/cmake/Common.cmake) + +set(GCS_FBS_SRC "${CMAKE_CURRENT_LIST_DIR}/format/gcs.fbs") +set(OUTPUT_DIR ${CMAKE_CURRENT_LIST_DIR}/format/) + +set(GCS_FBS_OUTPUT_FILES + "${OUTPUT_DIR}/gcs_generated.h") + +add_custom_command( + OUTPUT ${GCS_FBS_OUTPUT_FILES} + # The --gen-object-api flag generates a C++ class MessageT for each + # flatbuffers message Message, which can be used to store deserialized + # messages in data structures. This is currently used for ObjectInfo for + # example. + COMMAND ${FLATBUFFERS_COMPILER} -c -o ${OUTPUT_DIR} ${GCS_FBS_SRC} --gen-object-api + DEPENDS ${FBS_DEPENDS} + COMMENT "Running flatc compiler on ${GCS_FBS_SRC}" + VERBATIM) + +add_custom_target(gen_gcs_fbs DEPENDS ${GCS_FBS_OUTPUT_FILES}) + +ADD_RAY_TEST(client_test STATIC_LINK_LIBS ray_static gtest gtest_main pthread) + +install(FILES + client.h + DESTINATION "${CMAKE_INSTALL_INCLUDEDIR}/ray/gcs") diff --git a/src/ray/gcs/client.cc b/src/ray/gcs/client.cc new file mode 100644 index 000000000..601468a67 --- /dev/null +++ b/src/ray/gcs/client.cc @@ -0,0 +1,45 @@ +#include "ray/gcs/client.h" + +#include "ray/gcs/redis_context.h" + +namespace ray { + +namespace gcs { + +AsyncGcsClient::AsyncGcsClient() {} + +AsyncGcsClient::~AsyncGcsClient() {} + +Status AsyncGcsClient::Connect(const std::string &address, int port) { + context_.reset(new RedisContext()); + RETURN_NOT_OK(context_->Connect(address, port)); + object_table_.reset(new ObjectTable(context_)); + task_table_.reset(new TaskTable(context_)); + return Status::OK(); +} + +Status Attach(plasma::EventLoop &event_loop) { + // TODO(pcm): Implement this via + // context()->AttachToEventLoop(event loop) + return Status::OK(); +} + +ObjectTable &AsyncGcsClient::object_table() { + return *object_table_; +} + +TaskTable &AsyncGcsClient::task_table() { + return *task_table_; +} + +FunctionTable &AsyncGcsClient::function_table() { + return *function_table_; +} + +ClassTable &AsyncGcsClient::class_table() { + return *class_table_; +} + +} // namespace gcs + +} // namespace ray diff --git a/src/ray/gcs/client.h b/src/ray/gcs/client.h new file mode 100644 index 000000000..a1db98b30 --- /dev/null +++ b/src/ray/gcs/client.h @@ -0,0 +1,81 @@ +#ifndef RAY_GCS_CLIENT_H +#define RAY_GCS_CLIENT_H + +#include +#include + +#include "plasma/events.h" +#include "ray/id.h" +#include "ray/status.h" +#include "ray/gcs/tables.h" +#include "ray/util/logging.h" + +namespace ray { + +namespace gcs { + +class RedisContext; + +class AsyncGcsClient { + public: + AsyncGcsClient(); + ~AsyncGcsClient(); + + Status Connect(const std::string &address, int port); + Status Attach(plasma::EventLoop &event_loop); + + inline FunctionTable &function_table(); + // TODO: Some API for getting the error on the driver + inline ClassTable &class_table(); + inline ActorTable &actor_table(); + inline CustomSerializerTable &custom_serializer_table(); + inline ConfigTable &config_table(); + ObjectTable &object_table(); + TaskTable &task_table(); + inline ErrorTable &error_table(); + + // We also need something to export generic code to run on workers from the + // driver (to set the PYTHONPATH) + + using GetExportCallback = std::function; + Status AddExport(const std::string &driver_id, std::string &export_data); + Status GetExport(const std::string &driver_id, + int64_t export_index, + const GetExportCallback &done_callback); + + std::shared_ptr context() { return context_; } + + private: + std::unique_ptr function_table_; + std::unique_ptr class_table_; + std::unique_ptr object_table_; + std::unique_ptr task_table_; + std::shared_ptr context_; +}; + +class SyncGcsClient { + Status LogEvent(const std::string &key, + const std::string &value, + double timestamp); + Status NotifyError(const std::map &error_info); + Status RegisterFunction(const JobID &job_id, + const FunctionID &function_id, + const std::string &language, + const std::string &name, + const std::string &data); + Status RetrieveFunction(const JobID &job_id, + const FunctionID &function_id, + std::string *name, + std::string *data); + + Status AddExport(const std::string &driver_id, std::string &export_data); + Status GetExport(const std::string &driver_id, + int64_t export_index, + std::string *data); +}; + +} // namespace gcs + +} // namespace ray + +#endif // RAY_GCS_CLIENT_H diff --git a/src/ray/gcs/client_test.cc b/src/ray/gcs/client_test.cc new file mode 100644 index 000000000..14f593e42 --- /dev/null +++ b/src/ray/gcs/client_test.cc @@ -0,0 +1,83 @@ +#include "gtest/gtest.h" + +// TODO(pcm): get rid of this and replace with the type safe plasma event loop +extern "C" { +#include "hiredis/async.h" +#include "hiredis/hiredis.h" +#include "hiredis/adapters/ae.h" +} + +#include "ray/gcs/client.h" +#include "ray/gcs/tables.h" + +namespace ray { + +aeEventLoop *loop; + +class TestGcs : public ::testing::Test { + public: + TestGcs() { + RAY_CHECK_OK(client_.Connect("127.0.0.1", 6379)); + job_id_ = UniqueID::from_random(); + } + + protected: + gcs::AsyncGcsClient client_; + UniqueID job_id_; +}; + +void ObjectAdded(gcs::AsyncGcsClient *client, + const UniqueID &id, + std::shared_ptr data) { + ASSERT_EQ(data->managers, std::vector({"A", "B"})); +} + +void Lookup(gcs::AsyncGcsClient *client, + const UniqueID &id, + std::shared_ptr data) { + ASSERT_EQ(data->managers, std::vector({"A", "B"})); + aeStop(loop); +} + +TEST_F(TestGcs, TestObjectTable) { + loop = aeCreateEventLoop(1024); + RAY_CHECK_OK(client_.context()->AttachToEventLoop(loop)); + auto data = std::make_shared(); + data->managers.push_back("A"); + data->managers.push_back("B"); + ObjectID object_id = ObjectID::from_random(); + RAY_CHECK_OK( + client_.object_table().Add(job_id_, object_id, data, &ObjectAdded)); + RAY_CHECK_OK( + client_.object_table().Lookup(job_id_, object_id, &Lookup, &Lookup)); + aeMain(loop); + aeDeleteEventLoop(loop); +} + +void TaskAdded(gcs::AsyncGcsClient *client, + const TaskID &id, + std::shared_ptr data) { + ASSERT_EQ(data->scheduling_state, 3); +} + +void TaskLookup(gcs::AsyncGcsClient *client, + const TaskID &id, + std::shared_ptr data) { + ASSERT_EQ(data->scheduling_state, 3); + aeStop(loop); +} + +TEST_F(TestGcs, TestTaskTable) { + loop = aeCreateEventLoop(1024); + RAY_CHECK_OK(client_.context()->AttachToEventLoop(loop)); + auto data = std::make_shared(); + data->scheduling_state = 3; + TaskID task_id = TaskID::from_random(); + RAY_CHECK_OK(client_.task_table().Add(job_id_, task_id, data, &TaskAdded)); + RAY_CHECK_OK( + client_.task_table().Lookup(job_id_, task_id, &TaskLookup, &TaskLookup)); + aeMain(loop); + aeDeleteEventLoop(loop); +} + +} // namespace diff --git a/src/ray/gcs/format/gcs.fbs b/src/ray/gcs/format/gcs.fbs new file mode 100644 index 000000000..333f2f233 --- /dev/null +++ b/src/ray/gcs/format/gcs.fbs @@ -0,0 +1,41 @@ +enum Language:int { + PYTHON = 0, + CPP = 1, + JAVA = 2 +} + +table FunctionTableData { + language: Language; + name: string; + data: string; +} + +table ObjectTableData { + task_id: string; + object_size: long; + is_put: bool; + never_created: bool; + managers: [string]; +} + +table TaskTableData { + scheduling_state: int; + scheduler_id: string; + execution_arg_ids: [string]; + task_info: string; +} + +table ClassTableData { +} + +table ActorTableData { +} + +table ErrorTableData { +} + +table CustomSerializerData { +} + +table ConfigTableData { +} diff --git a/src/ray/gcs/redis_context.cc b/src/ray/gcs/redis_context.cc new file mode 100644 index 000000000..2dc7d2d2f --- /dev/null +++ b/src/ray/gcs/redis_context.cc @@ -0,0 +1,139 @@ +#include "ray/gcs/redis_context.h" + +#include + +extern "C" { +#include "hiredis/async.h" +#include "hiredis/hiredis.h" +#include "hiredis/adapters/ae.h" +} + +// TODO(pcm): Integrate into the C++ tree. +#include "state/ray_config.h" + +namespace ray { + +namespace gcs { + +// This is a global redis callback which will be registered for every +// asynchronous redis call. It dispatches the appropriate callback +// that was registered with the RedisCallbackManager. +void GlobalRedisCallback(void *c, void *r, void *privdata) { + if (r == NULL) { + return; + } + int64_t callback_index = reinterpret_cast(privdata); + redisReply *reply = reinterpret_cast(r); + std::string data = ""; + if (reply->type == REDIS_REPLY_NIL) { + } else if (reply->type == REDIS_REPLY_STRING) { + data = std::string(reply->str, reply->len); + } else if (reply->type == REDIS_REPLY_STATUS) { + } else if (reply->type == REDIS_REPLY_ERROR) { + RAY_LOG(ERROR) << "Redis error " << reply->str; + } else { + RAY_LOG(FATAL) << "Fatal redis error of type " << reply->type + << " and with string " << reply->str; + } + RedisCallbackManager::instance().get(callback_index)(data); +} + +int64_t RedisCallbackManager::add(const RedisCallback &function) { + callbacks_.emplace(num_callbacks, std::unique_ptr( + new RedisCallback(function))); + return num_callbacks++; +} + +RedisCallbackManager::RedisCallback &RedisCallbackManager::get( + int64_t callback_index) { + return *callbacks_[callback_index]; +} + +#define REDIS_CHECK_ERROR(CONTEXT, REPLY) \ + if (REPLY == nullptr || REPLY->type == REDIS_REPLY_ERROR) { \ + return Status::RedisError(CONTEXT->errstr); \ + } + +RedisContext::~RedisContext() { + if (context_) { + redisFree(context_); + } + if (async_context_) { + redisAsyncFree(async_context_); + } +} + +Status RedisContext::Connect(const std::string &address, int port) { + int connection_attempts = 0; + context_ = redisConnect(address.c_str(), port); + while (context_ == nullptr || context_->err) { + if (connection_attempts >= + RayConfig::instance().redis_db_connect_retries()) { + if (context_ == nullptr) { + RAY_LOG(FATAL) << "Could not allocate redis context."; + } + if (context_->err) { + RAY_LOG(FATAL) << "Could not establish connection to redis " << address + << ":" << port; + } + break; + } + RAY_LOG(WARNING) << "Failed to connect to Redis, retrying."; + // Sleep for a little. + usleep(RayConfig::instance().redis_db_connect_wait_milliseconds() * 1000); + context_ = redisConnect(address.c_str(), port); + connection_attempts += 1; + } + redisReply *reply = reinterpret_cast( + redisCommand(context_, "CONFIG SET notify-keyspace-events Kl")); + REDIS_CHECK_ERROR(context_, reply); + + // Connect to async context + async_context_ = redisAsyncConnect(address.c_str(), port); + if (async_context_ == nullptr || async_context_->err) { + RAY_LOG(FATAL) << "Could not establish connection to redis " << address + << ":" << port; + } + return Status::OK(); +} + +Status RedisContext::AttachToEventLoop(aeEventLoop *loop) { + if (redisAeAttach(loop, async_context_) != REDIS_OK) { + return Status::RedisError("could not attach redis event loop"); + } else { + return Status::OK(); + } +} + +Status RedisContext::RunAsync(const std::string &command, + const UniqueID &id, + uint8_t *data, + int64_t length, + int64_t callback_index) { + if (length > 0) { + std::string redis_command = command + " %b %b"; + int status = redisAsyncCommand( + async_context_, + reinterpret_cast(&GlobalRedisCallback), + reinterpret_cast(callback_index), redis_command.c_str(), + id.data(), id.size(), data, length); + if (status == REDIS_ERR) { + return Status::RedisError(std::string(async_context_->errstr)); + } + } else { + std::string redis_command = command + " %b"; + int status = redisAsyncCommand( + async_context_, + reinterpret_cast(&GlobalRedisCallback), + reinterpret_cast(callback_index), redis_command.c_str(), + id.data(), id.size()); + if (status == REDIS_ERR) { + return Status::RedisError(std::string(async_context_->errstr)); + } + } + return Status::OK(); +} + +} // namespace gcs + +} // namespace ray diff --git a/src/ray/gcs/redis_context.h b/src/ray/gcs/redis_context.h new file mode 100644 index 000000000..b8842c45d --- /dev/null +++ b/src/ray/gcs/redis_context.h @@ -0,0 +1,63 @@ +#ifndef RAY_GCS_REDIS_CONTEXT_H +#define RAY_GCS_REDIS_CONTEXT_H + +#include +#include +#include + +#include "ray/id.h" +#include "ray/status.h" +#include "ray/util/logging.h" + +struct redisContext; +struct redisAsyncContext; +struct aeEventLoop; + +namespace ray { + +namespace gcs { + +class RedisCallbackManager { + public: + using RedisCallback = std::function; + + static RedisCallbackManager &instance() { + static RedisCallbackManager instance; + return instance; + } + + int64_t add(const RedisCallback &function); + + RedisCallback &get(int64_t callback_index); + + private: + RedisCallbackManager() : num_callbacks(0){}; + + ~RedisCallbackManager() { printf("shut down callback manager\n"); } + + int64_t num_callbacks; + std::unordered_map> callbacks_; +}; + +class RedisContext { + public: + RedisContext() {} + ~RedisContext(); + Status Connect(const std::string &address, int port); + Status AttachToEventLoop(aeEventLoop *loop); + Status RunAsync(const std::string &command, + const UniqueID &id, + uint8_t *data, + int64_t length, + int64_t callback_index); + + private: + redisContext *context_; + redisAsyncContext *async_context_; +}; + +} // namespace gcs + +} // namespace ray + +#endif // RAY_GCS_REDIS_CONTEXT_H diff --git a/src/ray/gcs/tables.cc b/src/ray/gcs/tables.cc new file mode 100644 index 000000000..19692dcd4 --- /dev/null +++ b/src/ray/gcs/tables.cc @@ -0,0 +1,7 @@ +#include "ray/gcs/tables.h" + +namespace ray { + +namespace gcs {} // namespace gcs + +} // namespace ray diff --git a/src/ray/gcs/tables.h b/src/ray/gcs/tables.h new file mode 100644 index 000000000..c93af861a --- /dev/null +++ b/src/ray/gcs/tables.h @@ -0,0 +1,195 @@ +#ifndef RAY_GCS_TABLES_H +#define RAY_GCS_TABLES_H + +#include +#include +#include + +#include "ray/constants.h" +#include "ray/id.h" +#include "ray/status.h" +#include "ray/util/logging.h" + +#include "ray/gcs/format/gcs_generated.h" +#include "ray/gcs/redis_context.h" + +struct redisAsyncContext; + +namespace ray { + +namespace gcs { + +class RedisContext; + +class AsyncGcsClient; + +template +class Table { + public: + using DataT = typename Data::NativeTableType; + using Callback = std::function< + void(AsyncGcsClient *client, const ID &id, std::shared_ptr data)>; + + struct CallbackData { + ID id; + std::shared_ptr data; + Callback callback; + Table *table; + AsyncGcsClient *client; + }; + + Table(const std::shared_ptr &context) : context_(context){}; + + /// Add an entry to the table + Status Add(const JobID &job_id, + const ID &id, + std::shared_ptr data, + const Callback &done) { + auto d = + std::shared_ptr(new CallbackData({id, data, done, this})); + int64_t callback_index = RedisCallbackManager::instance().add([d]( + const std::string &data) { (d->callback)(d->client, d->id, d->data); }); + flatbuffers::FlatBufferBuilder fbb; + fbb.Finish(Data::Pack(fbb, data.get())); + RETURN_NOT_OK(context_->RunAsync("RAY.TABLE_ADD", id, + fbb.GetBufferPointer(), fbb.GetSize(), + callback_index)); + return Status::OK(); + } + + /// Lookup an entry asynchronously + Status Lookup(const JobID &job_id, + const ID &id, + const Callback &lookup, + const Callback &done) { + auto d = std::shared_ptr( + new CallbackData({id, nullptr, done, this})); + int64_t callback_index = + RedisCallbackManager::instance().add([d](const std::string &data) { + auto result = std::make_shared(); + auto root = flatbuffers::GetRoot(data.data()); + root->UnPackTo(result.get()); + (d->callback)(d->client, d->id, result); + }); + std::vector nil; + RETURN_NOT_OK(context_->RunAsync("RAY.TABLE_LOOKUP", id, nil.data(), + nil.size(), callback_index)); + return Status::OK(); + } + + /// Subscribe to updates of this table + Status Subscribe(const JobID &job_id, + const ID &id, + const Callback &subscribe, + const Callback &done); + + /// Remove and entry from the table + Status Remove(const JobID &job_id, const ID &id, const Callback &done); + + private: + std::unordered_map, UniqueIDHasher> + callback_data_; + std::shared_ptr context_; +}; + +class ObjectTable : public Table { + public: + ObjectTable(const std::shared_ptr &context) : Table(context){}; + + /// Set up a client-specific channel for receiving notifications about + /// available + /// objects from the object table. The callback will be called once per + /// notification received on this channel. + /// + /// @param subscribe_all + /// @param object_available_callback Callback to be called when new object + /// becomes available. + /// @param done_callback Callback to be called when subscription is installed. + /// This is only used for the tests. + Status SubscribeToNotifications(const JobID &job_id, + bool subscribe_all, + const Callback &object_available, + const Callback &done); + + /// Request notifications about the availability of some objects from the + /// object + /// table. The notifications will be published to this client's object + /// notification channel, which was set up by the method + /// ObjectTableSubscribeToNotifications. + /// + /// @param object_ids The object IDs to receive notifications about. + Status RequestNotifications(const JobID &job_id, + const std::vector &object_ids); +}; + +using FunctionTable = Table; + +using ClassTable = Table; + +using ActorTable = Table; + +class TaskTable : public Table { + public: + TaskTable(const std::shared_ptr &context) : Table(context){}; + + using TestAndUpdateCallback = + std::function task)>; + using SubscribeToTaskCallback = + std::function task)>; + /// Update a task's scheduling information in the task table, if the current + /// value matches the given test value. If the update succeeds, it also + /// updates + /// the task entry's local scheduler ID with the ID of the client who called + /// this function. This assumes that the task spec already exists in the task + /// table entry. + /// + /// @param task_id The task ID of the task entry to update. + /// @param test_state_bitmask The bitmask to apply to the task entry's current + /// scheduling state. The update happens if and only if the current + /// scheduling state AND-ed with the bitmask is greater than 0. + /// @param update_state The value to update the task entry's scheduling state + /// with, if the current state matches test_state_bitmask. + /// @param callback Function to be called when database returns result. + Status TestAndUpdate(const JobID &job_id, + const TaskID &task_id, + int test_state_bitmask, + int updata_state, + const TaskTableData &data, + const TestAndUpdateCallback &callback); + + /// This has a separate signature from Subscribe in Table + /// Register a callback for a task event. An event is any update of a task in + /// the task table. + /// Events include changes to the task's scheduling state or changes to the + /// task's local scheduler ID. + /// + /// @param local_scheduler_id The db_client_id of the local scheduler whose + /// events we want to listen to. If you want to subscribe to updates + /// from + /// all local schedulers, pass in NIL_ID. + /// @param subscribe_callback Callback that will be called when the task table + /// is + /// updated. + /// @param state_filter Events we want to listen to. Can have values from the + /// enum "scheduling_state" in task.h. + /// TODO(pcm): Make it possible to combine these using flags like + /// TASK_STATUS_WAITING | TASK_STATUS_SCHEDULED. + /// @param callback Function to be called when database returns result. + Status SubscribeToTask(const JobID &job_id, + const DBClientID &local_scheduler_id, + int state_filter, + const SubscribeToTaskCallback &callback, + const Callback &done); +}; + +using ErrorTable = Table; + +using CustomSerializerTable = Table; + +using ConfigTable = Table; + +} // namespace gcs + +} // namespace ray + +#endif // RAY_GCS_TABLES_H diff --git a/src/ray/id.cc b/src/ray/id.cc new file mode 100644 index 000000000..f15a702b2 --- /dev/null +++ b/src/ray/id.cc @@ -0,0 +1,68 @@ +#include "ray/id.h" + +#include + +namespace ray { + +UniqueID UniqueID::from_random() { + UniqueID id; + uint8_t *data = id.mutable_data(); + std::random_device engine; + for (int i = 0; i < kUniqueIDSize; i++) { + data[i] = static_cast(engine()); + } + return id; +} + +UniqueID UniqueID::from_binary(const std::string &binary) { + UniqueID id; + std::memcpy(&id, binary.data(), sizeof(id)); + return id; +} + +const UniqueID UniqueID::nil() { + UniqueID result; + std::fill_n(result.id_, kUniqueIDSize, 255); + return result; +} + +bool is_nil(const UniqueID &rhs) { + int i = kUniqueIDSize; + const uint8_t *data = rhs.data(); + while (--i > 0 && data[i] == 255) { + } + return i != 0; +} + +const uint8_t *UniqueID::data() const { + return id_; +} + +uint8_t *UniqueID::mutable_data() { + return id_; +} + +size_t UniqueID::size() const { + return kUniqueIDSize; +} + +std::string UniqueID::binary() const { + return std::string(reinterpret_cast(id_), kUniqueIDSize); +} + +std::string UniqueID::hex() const { + constexpr char hex[] = "0123456789abcdef"; + std::string result; + for (int i = 0; i < kUniqueIDSize; i++) { + unsigned int val = id_[i]; + result.push_back(hex[val >> 4]); + result.push_back(hex[val & 0xf]); + } + return result; +} + +bool UniqueID::operator==(const UniqueID &rhs) const { + return std::memcmp(data(), rhs.data(), kUniqueIDSize) == 0; +} + +} // namespace ray diff --git a/src/ray/id.h b/src/ray/id.h new file mode 100644 index 000000000..299e5f006 --- /dev/null +++ b/src/ray/id.h @@ -0,0 +1,55 @@ +#ifndef RAY_ID_H_ +#define RAY_ID_H_ + +#include + +#include +#include + +#include "ray/constants.h" +#include "ray/util/visibility.h" + +namespace ray { + +class RAY_EXPORT UniqueID { + public: + static UniqueID from_random(); + static UniqueID from_binary(const std::string &binary); + static const UniqueID nil(); + bool is_nil(const UniqueID &rhs) const; + bool operator==(const UniqueID &rhs) const; + const uint8_t *data() const; + uint8_t *mutable_data(); + size_t size() const; + std::string binary() const; + std::string hex() const; + + private: + uint8_t id_[kUniqueIDSize]; +}; + +static_assert(std::is_pod::value, "UniqueID must be plain old data"); + +struct UniqueIDHasher { + // ID hashing function. + size_t operator()(const UniqueID &id) const { + size_t result; + std::memcpy(&result, id.data(), sizeof(size_t)); + return result; + } +}; + +typedef UniqueID TaskID; +typedef UniqueID JobID; +typedef UniqueID ObjectID; +typedef UniqueID FunctionID; +typedef UniqueID ClassID; +typedef UniqueID ActorID; +typedef UniqueID ActorHandleID; +typedef UniqueID WorkerID; +typedef UniqueID DBClientID; +typedef UniqueID ConfigID; + +} // namespace ray + +#endif // RAY_ID_H_ diff --git a/src/ray/ray.pc.in b/src/ray/ray.pc.in new file mode 100644 index 000000000..b8c9efbed --- /dev/null +++ b/src/ray/ray.pc.in @@ -0,0 +1,11 @@ +libdir=@CMAKE_INSTALL_FULL_LIBDIR@ +includedir=@CMAKE_INSTALL_FULL_INCLUDEDIR@ + +so_version=@RAY_SO_VERSION@ +abi_version=@RAY_ABI_VERSION@ + +Name: Ray +Description: Ray is a set of technologies for distributed execution. +Version: @RAY_VERSION@ +Libs: -L${libdir} -lray +Cflags: -I${includedir} diff --git a/src/ray/status.cc b/src/ray/status.cc new file mode 100644 index 000000000..4be0de442 --- /dev/null +++ b/src/ray/status.cc @@ -0,0 +1,88 @@ +// Copyright (c) 2011 The LevelDB Authors. All rights reserved. +// Use of this source code is governed by a BSD-style license that can be +// found in the LICENSE file. See the AUTHORS file for names of contributors. +// +// A Status encapsulates the result of an operation. It may indicate success, +// or it may indicate an error with an associated error message. +// +// Multiple threads can invoke const methods on a Status without +// external synchronization, but if any of the threads may call a +// non-const method, all threads accessing the same Status must use +// external synchronization. + +// Adapted from Apache Arrow, Apache Kudu, TensorFlow + +#include "ray/status.h" + +#include + +namespace ray { + +Status::Status(StatusCode code, const std::string &msg) { + assert(code != StatusCode::OK); + state_ = new State; + state_->code = code; + state_->msg = msg; +} + +void Status::CopyFrom(const State *state) { + delete state_; + if (state == nullptr) { + state_ = nullptr; + } else { + state_ = new State(*state); + } +} + +std::string Status::CodeAsString() const { + if (state_ == NULL) { + return "OK"; + } + + const char *type; + switch (code()) { + case StatusCode::OK: + type = "OK"; + break; + case StatusCode::OutOfMemory: + type = "Out of memory"; + break; + case StatusCode::KeyError: + type = "Key error"; + break; + case StatusCode::TypeError: + type = "Type error"; + break; + case StatusCode::Invalid: + type = "Invalid"; + break; + case StatusCode::IOError: + type = "IOError"; + break; + case StatusCode::UnknownError: + type = "Unknown error"; + break; + case StatusCode::NotImplemented: + type = "NotImplemented"; + break; + case StatusCode::RedisError: + type = "RedisError"; + break; + default: + type = "Unknown"; + break; + } + return std::string(type); +} + +std::string Status::ToString() const { + std::string result(CodeAsString()); + if (state_ == NULL) { + return result; + } + result += ": "; + result += state_->msg; + return result; +} + +} // namespace ray diff --git a/src/ray/status.h b/src/ray/status.h new file mode 100644 index 000000000..a8c3d2715 --- /dev/null +++ b/src/ray/status.h @@ -0,0 +1,184 @@ +// Copyright (c) 2011 The LevelDB Authors. All rights reserved. +// Use of this source code is governed by a BSD-style license that can be +// found in the LICENSE file. See the AUTHORS file for names of contributors. +// +// A Status encapsulates the result of an operation. It may indicate success, +// or it may indicate an error with an associated error message. +// +// Multiple threads can invoke const methods on a Status without +// external synchronization, but if any of the threads may call a +// non-const method, all threads accessing the same Status must use +// external synchronization. + +// Adapted from Apache Arrow, Apache Kudu, TensorFlow + +#ifndef RAY_STATUS_H_ +#define RAY_STATUS_H_ + +#include +#include +#include + +#include "ray/util/macros.h" +#include "ray/util/visibility.h" + +// Return the given status if it is not OK. +#define RAY_RETURN_NOT_OK(s) \ + do { \ + ::ray::Status _s = (s); \ + if (RAY_PREDICT_FALSE(!_s.ok())) { \ + return _s; \ + } \ + } while (0) + +// If 'to_call' returns a bad status, CHECK immediately with a logged message +// of 'msg' followed by the status. +#define RAY_CHECK_OK_PREPEND(to_call, msg) \ + do { \ + ::ray::Status _s = (to_call); \ + RAY_CHECK(_s.ok()) << (msg) << ": " << _s.ToString(); \ + } while (0) + +// If the status is bad, CHECK immediately, appending the status to the +// logged message. +#define RAY_CHECK_OK(s) RAY_CHECK_OK_PREPEND(s, "Bad status") + +namespace ray { + +#define RETURN_NOT_OK(s) \ + do { \ + Status _s = (s); \ + if (RAY_PREDICT_FALSE(!_s.ok())) { \ + return _s; \ + } \ + } while (0) + +#define RETURN_NOT_OK_ELSE(s, else_) \ + do { \ + Status _s = (s); \ + if (!_s.ok()) { \ + else_; \ + return _s; \ + } \ + } while (0) + +enum class StatusCode : char { + OK = 0, + OutOfMemory = 1, + KeyError = 2, + TypeError = 3, + Invalid = 4, + IOError = 5, + UnknownError = 9, + NotImplemented = 10, + RedisError = 11 +}; + +#if defined(__clang__) +// Only clang supports warn_unused_result as a type annotation. +class RAY_MUST_USE_RESULT RAY_EXPORT Status; +#endif + +class RAY_EXPORT Status { + public: + // Create a success status. + Status() : state_(NULL) {} + ~Status() { delete state_; } + + Status(StatusCode code, const std::string &msg); + + // Copy the specified status. + Status(const Status &s); + void operator=(const Status &s); + + // Return a success status. + static Status OK() { return Status(); } + + // Return error status of an appropriate type. + static Status OutOfMemory(const std::string &msg) { + return Status(StatusCode::OutOfMemory, msg); + } + + static Status KeyError(const std::string &msg) { + return Status(StatusCode::KeyError, msg); + } + + static Status TypeError(const std::string &msg) { + return Status(StatusCode::TypeError, msg); + } + + static Status UnknownError(const std::string &msg) { + return Status(StatusCode::UnknownError, msg); + } + + static Status NotImplemented(const std::string &msg) { + return Status(StatusCode::NotImplemented, msg); + } + + static Status Invalid(const std::string &msg) { + return Status(StatusCode::Invalid, msg); + } + + static Status IOError(const std::string &msg) { + return Status(StatusCode::IOError, msg); + } + + static Status RedisError(const std::string &msg) { + return Status(StatusCode::RedisError, msg); + } + + // Returns true iff the status indicates success. + bool ok() const { return (state_ == NULL); } + + bool IsOutOfMemory() const { return code() == StatusCode::OutOfMemory; } + bool IsKeyError() const { return code() == StatusCode::KeyError; } + bool IsInvalid() const { return code() == StatusCode::Invalid; } + bool IsIOError() const { return code() == StatusCode::IOError; } + bool IsTypeError() const { return code() == StatusCode::TypeError; } + bool IsUnknownError() const { return code() == StatusCode::UnknownError; } + bool IsNotImplemented() const { return code() == StatusCode::NotImplemented; } + bool IsRedisError() const { return code() == StatusCode::RedisError; } + + // Return a string representation of this status suitable for printing. + // Returns the string "OK" for success. + std::string ToString() const; + + // Return a string representation of the status code, without the message + // text or posix code information. + std::string CodeAsString() const; + + StatusCode code() const { return ok() ? StatusCode::OK : state_->code; } + + std::string message() const { return ok() ? "" : state_->msg; } + + private: + struct State { + StatusCode code; + std::string msg; + }; + // OK status has a `NULL` state_. Otherwise, `state_` points to + // a `State` structure containing the error code and message(s) + State *state_; + + void CopyFrom(const State *s); +}; + +static inline std::ostream &operator<<(std::ostream &os, const Status &x) { + os << x.ToString(); + return os; +} + +inline Status::Status(const Status &s) + : state_((s.state_ == NULL) ? NULL : new State(*s.state_)) {} + +inline void Status::operator=(const Status &s) { + // The following condition catches both aliasing (when this == &s), + // and the common case where both s and *this are ok. + if (state_ != s.state_) { + CopyFrom(s.state_); + } +} + +} // namespace ray + +#endif // RAY_STATUS_H_ diff --git a/src/ray/test/run_gcs_tests.sh b/src/ray/test/run_gcs_tests.sh new file mode 100644 index 000000000..624424e5e --- /dev/null +++ b/src/ray/test/run_gcs_tests.sh @@ -0,0 +1,14 @@ +#!/usr/bin/env bash + +# This needs to be run in the build tree, which is normally ray/python/ray/core + +# Cause the script to exit if a single command fails. +set -e + +# Start Redis. +./src/common/thirdparty/redis/src/redis-server --loglevel warning --loadmodule ./src/common/redis_module/libray_redis_module.so --port 6379 & +sleep 1s + +./src/ray/gcs/client_test + +./src/common/thirdparty/redis/src/redis-cli -p 6379 shutdown diff --git a/src/ray/util/CMakeLists.txt b/src/ray/util/CMakeLists.txt new file mode 100644 index 000000000..c4fd15c0d --- /dev/null +++ b/src/ray/util/CMakeLists.txt @@ -0,0 +1,6 @@ +install(FILES + logging.h + macros.h + visibility.h + DESTINATION "${CMAKE_INSTALL_INCLUDEDIR}/ray/util" +) diff --git a/src/ray/util/logging.h b/src/ray/util/logging.h new file mode 100644 index 000000000..5d6b6cc27 --- /dev/null +++ b/src/ray/util/logging.h @@ -0,0 +1,145 @@ +#ifndef RAY_UTIL_LOGGING_H +#define RAY_UTIL_LOGGING_H + +#ifndef _WIN32 +#include +#endif + +#include +#include + +#include "ray/util/macros.h" + +namespace ray { + +// Stubbed versions of macros defined in glog/logging.h, intended for +// environments where glog headers aren't available. +// +// Add more as needed. + +// Log levels. LOG ignores them, so their values are abitrary. + +#define RAY_DEBUG (-1) +#define RAY_INFO 0 +#define RAY_WARNING 1 +#define RAY_ERROR 2 +#define RAY_FATAL 3 + +#define RAY_LOG_INTERNAL(level) ::ray::internal::CerrLog(level) +#define RAY_LOG(level) RAY_LOG_INTERNAL(RAY_##level) +#define RAY_IGNORE_EXPR(expr) ((void) (expr)); + +#define RAY_CHECK(condition) \ + (condition) ? 0 : ::ray::internal::FatalLog(RAY_FATAL) \ + << __FILE__ << __LINE__ \ + << " Check failed: " #condition " " + +#ifdef NDEBUG +#define RAY_DFATAL RAY_WARNING + +#define DCHECK(condition) \ + RAY_IGNORE_EXPR(condition) \ + while (false) \ + ::ray::internal::NullLog() +#define DCHECK_EQ(val1, val2) \ + RAY_IGNORE_EXPR(val1) \ + while (false) \ + ::ray::internal::NullLog() +#define DCHECK_NE(val1, val2) \ + RAY_IGNORE_EXPR(val1) \ + while (false) \ + ::ray::internal::NullLog() +#define DCHECK_LE(val1, val2) \ + RAY_IGNORE_EXPR(val1) \ + while (false) \ + ::ray::internal::NullLog() +#define DCHECK_LT(val1, val2) \ + RAY_IGNORE_EXPR(val1) \ + while (false) \ + ::ray::internal::NullLog() +#define DCHECK_GE(val1, val2) \ + RAY_IGNORE_EXPR(val1) \ + while (false) \ + ::ray::internal::NullLog() +#define DCHECK_GT(val1, val2) \ + RAY_IGNORE_EXPR(val1) \ + while (false) \ + ::ray::internal::NullLog() + +#else +#define RAY_DFATAL RAY_FATAL + +#define DCHECK(condition) RAY_CHECK(condition) +#define DCHECK_EQ(val1, val2) RAY_CHECK((val1) == (val2)) +#define DCHECK_NE(val1, val2) RAY_CHECK((val1) != (val2)) +#define DCHECK_LE(val1, val2) RAY_CHECK((val1) <= (val2)) +#define DCHECK_LT(val1, val2) RAY_CHECK((val1) < (val2)) +#define DCHECK_GE(val1, val2) RAY_CHECK((val1) >= (val2)) +#define DCHECK_GT(val1, val2) RAY_CHECK((val1) > (val2)) + +#endif // NDEBUG + +namespace internal { + +class NullLog { + public: + template + NullLog &operator<<(const T &t) { + return *this; + } +}; + +class CerrLog { + public: + CerrLog(int severity) // NOLINT(runtime/explicit) + : severity_(severity), + has_logged_(false) {} + + virtual ~CerrLog() { + if (has_logged_) { + std::cerr << std::endl; + } + if (severity_ == RAY_FATAL) { + std::exit(1); + } + } + + template + CerrLog &operator<<(const T &t) { + if (severity_ != RAY_DEBUG) { + has_logged_ = true; + std::cerr << t; + } + return *this; + } + + protected: + const int severity_; + bool has_logged_; +}; + +// Clang-tidy isn't smart enough to determine that DCHECK using CerrLog doesn't +// return so we create a new class to give it a hint. +class FatalLog : public CerrLog { + public: + explicit FatalLog(int /* severity */) // NOLINT + : CerrLog(RAY_FATAL) {} // NOLINT + + RAY_NORETURN ~FatalLog() { + if (has_logged_) { + std::cerr << std::endl; +#if defined(_EXECINFO_H) || !defined(_WIN32) + void *buffer[255]; + const int calls = backtrace(buffer, sizeof(buffer) / sizeof(void *)); + backtrace_symbols_fd(buffer, calls, 1); +#endif + } + std::abort(); + } +}; + +} // namespace internal + +} // namespace ray + +#endif // RAY_UTIL_LOGGING_H diff --git a/src/ray/util/macros.h b/src/ray/util/macros.h new file mode 100644 index 000000000..dbf85fe39 --- /dev/null +++ b/src/ray/util/macros.h @@ -0,0 +1,44 @@ +#ifndef RAY_UTIL_MACROS_H +#define RAY_UTIL_MACROS_H + +// From Google gutil +#ifndef RAY_DISALLOW_COPY_AND_ASSIGN +#define RAY_DISALLOW_COPY_AND_ASSIGN(TypeName) \ + TypeName(const TypeName &) = delete; \ + void operator=(const TypeName &) = delete +#endif + +#define RAY_UNUSED(x) (void) x + +// +// GCC can be told that a certain branch is not likely to be taken (for +// instance, a CHECK failure), and use that information in static analysis. +// Giving it this information can help it optimize for the common case in +// the absence of better information (ie. -fprofile-arcs). +// +#if defined(__GNUC__) +#define RAY_PREDICT_FALSE(x) (__builtin_expect(x, 0)) +#define RAY_PREDICT_TRUE(x) (__builtin_expect(!!(x), 1)) +#define RAY_NORETURN __attribute__((noreturn)) +#define RAY_PREFETCH(addr) __builtin_prefetch(addr) +#elif defined(_MSC_VER) +#define RAY_NORETURN __declspec(noreturn) +#define RAY_PREDICT_FALSE(x) x +#define RAY_PREDICT_TRUE(x) x +#define RAY_PREFETCH(addr) +#else +#define RAY_NORETURN +#define RAY_PREDICT_FALSE(x) x +#define RAY_PREDICT_TRUE(x) x +#define RAY_PREFETCH(addr) +#endif + +#if (defined(__GNUC__) || defined(__APPLE__)) +#define RAY_MUST_USE_RESULT __attribute__((warn_unused_result)) +#elif defined(_MSC_VER) +#define RAY_MUST_USE_RESULT +#else +#define RAY_MUST_USE_RESULT +#endif + +#endif // RAY_UTIL_MACROS_H diff --git a/src/ray/util/visibility.h b/src/ray/util/visibility.h new file mode 100644 index 000000000..fa04ac3b2 --- /dev/null +++ b/src/ray/util/visibility.h @@ -0,0 +1,44 @@ +#ifndef RAY_UTIL_VISIBILITY_H +#define RAY_UTIL_VISIBILITY_H + +#if defined(_WIN32) || defined(__CYGWIN__) +#if defined(_MSC_VER) +#pragma warning(disable : 4251) +#else +#pragma GCC diagnostic ignored "-Wattributes" +#endif + +#ifdef RAY_EXPORTING +#define RAY_EXPORT __declspec(dllexport) +#else +#define RAY_EXPORT __declspec(dllimport) +#endif + +#define RAY_NO_EXPORT +#else // Not Windows +#ifndef RAY_EXPORT +#define RAY_EXPORT __attribute__((visibility("default"))) +#endif +#ifndef RAY_NO_EXPORT +#define RAY_NO_EXPORT __attribute__((visibility("hidden"))) +#endif +#endif // Non-Windows + +// gcc and clang disagree about how to handle template visibility when you have +// explicit specializations https://llvm.org/bugs/show_bug.cgi?id=24815 + +#if defined(__clang__) +#define RAY_EXTERN_TEMPLATE extern template class RAY_EXPORT +#else +#define RAY_EXTERN_TEMPLATE extern template class +#endif + +// This is a complicated topic, some reading on it: +// http://www.codesynthesis.com/~boris/blog/2010/01/18/dll-export-cxx-templates/ +#if defined(_MSC_VER) || defined(__clang__) +#define RAY_TEMPLATE_EXPORT RAY_EXPORT +#else +#define RAY_TEMPLATE_EXPORT +#endif + +#endif // RAY_UTIL_VISIBILITY_H