diff --git a/.travis.yml b/.travis.yml index f1f292c58..1888fa4ce 100644 --- a/.travis.yml +++ b/.travis.yml @@ -1,4 +1,6 @@ language: generic +# Use Ubuntu 16.04 +dist: xenial matrix: include: @@ -35,11 +37,8 @@ matrix: - os: linux env: LINT=1 PYTHONWARNINGS=ignore before_install: - # In case we ever want to use a different version of clang-format: - #- wget -O - http://apt.llvm.org/llvm-snapshot.gpg.key | sudo apt-key add - - #- echo "deb http://apt.llvm.org/trusty/ llvm-toolchain-trusty main" | sudo tee -a /etc/apt/sources.list > /dev/null - sudo apt-get update -qq - - sudo apt-get install -qq clang-format-3.8 + - sudo apt-get install -qq clang-format-6.0 install: [] script: - ./ci/travis/check-git-clang-format-output.sh diff --git a/ci/travis/check-git-clang-format-output.sh b/ci/travis/check-git-clang-format-output.sh index 4209811cd..6d83044c4 100755 --- a/ci/travis/check-git-clang-format-output.sh +++ b/ci/travis/check-git-clang-format-output.sh @@ -8,7 +8,7 @@ else base_commit="$TRAVIS_BRANCH" echo "Running clang-format against branch $base_commit, with hash $(git rev-parse $base_commit)" fi -output="$(ci/travis/git-clang-format --binary clang-format-3.8 --commit $base_commit --diff --exclude '(.*thirdparty/|.*redismodule.h|.*.js|.*.java)')" +output="$(ci/travis/git-clang-format --binary clang-format --commit $base_commit --diff --exclude '(.*thirdparty/|.*redismodule.h|.*.js|.*.java)')" if [ "$output" == "no modified files to format" ] || [ "$output" == "clang-format did not modify any files" ] ; then echo "clang-format passed." exit 0 diff --git a/src/ray/common/id.cc b/src/ray/common/id.cc index 57e41d97d..3928d4adf 100644 --- a/src/ray/common/id.cc +++ b/src/ray/common/id.cc @@ -105,8 +105,8 @@ ObjectID ObjectID::ForPut(const TaskID &task_id, int64_t put_index) { } ObjectID ObjectID::ForTaskReturn(const TaskID &task_id, int64_t return_index) { - RAY_CHECK(return_index >= 1 && return_index <= kMaxTaskReturns) << "index=" - << return_index; + RAY_CHECK(return_index >= 1 && return_index <= kMaxTaskReturns) + << "index=" << return_index; ObjectID object_id; std::memcpy(object_id.id_, task_id.Binary().c_str(), task_id.Size()); object_id.index_ = return_index; diff --git a/src/ray/core_worker/context.cc b/src/ray/core_worker/context.cc index 660330e5c..717c52e07 100644 --- a/src/ray/core_worker/context.cc +++ b/src/ray/core_worker/context.cc @@ -23,6 +23,7 @@ struct WorkerThreadContext { void SetCurrentTask(const raylet::TaskSpecification &spec) { SetCurrentTask(spec.TaskId()); } + private: /// The task ID for current task. TaskID current_task_id; diff --git a/src/ray/core_worker/mock_worker.cc b/src/ray/core_worker/mock_worker.cc index 205fcfce9..95d11bb25 100644 --- a/src/ray/core_worker/mock_worker.cc +++ b/src/ray/core_worker/mock_worker.cc @@ -25,7 +25,6 @@ class MockWorker { auto executor_func = [this](const RayFunction &ray_function, const std::vector> &args, const TaskID &task_id, int num_returns) { - // Note that this doesn't include dummy object id. RAY_CHECK(num_returns >= 0); diff --git a/src/ray/core_worker/task_interface.cc b/src/ray/core_worker/task_interface.cc index c19b1e23a..00f15237f 100644 --- a/src/ray/core_worker/task_interface.cc +++ b/src/ray/core_worker/task_interface.cc @@ -1,7 +1,6 @@ -#include "ray/raylet/task.h" +#include "ray/core_worker/task_interface.h" #include "ray/core_worker/context.h" #include "ray/core_worker/core_worker.h" -#include "ray/core_worker/task_interface.h" namespace ray { diff --git a/src/ray/core_worker/task_interface.h b/src/ray/core_worker/task_interface.h index 06bd5409a..2ec3b1329 100644 --- a/src/ray/core_worker/task_interface.h +++ b/src/ray/core_worker/task_interface.h @@ -7,6 +7,7 @@ #include "ray/common/id.h" #include "ray/common/status.h" #include "ray/core_worker/common.h" +#include "ray/raylet/task.h" namespace ray { diff --git a/src/ray/gcs/client.cc b/src/ray/gcs/client.cc index 3d1c66027..c9b1e1385 100644 --- a/src/ray/gcs/client.cc +++ b/src/ray/gcs/client.cc @@ -23,8 +23,8 @@ static void GetRedisShards(redisContext *context, std::vector &addr } RAY_CHECK(num_attempts < RayConfig::instance().redis_db_connect_retries()) << "No entry found for NumRedisShards"; - RAY_CHECK(reply->type == REDIS_REPLY_STRING) << "Expected string, found Redis type " - << reply->type << " for NumRedisShards"; + RAY_CHECK(reply->type == REDIS_REPLY_STRING) + << "Expected string, found Redis type " << reply->type << " for NumRedisShards"; int num_redis_shards = atoi(reply->str); RAY_CHECK(num_redis_shards >= 1) << "Expected at least one Redis shard, " << "found " << num_redis_shards; diff --git a/src/ray/gcs/client_test.cc b/src/ray/gcs/client_test.cc index 4eb34a953..c7dc02e50 100644 --- a/src/ray/gcs/client_test.cc +++ b/src/ray/gcs/client_test.cc @@ -150,8 +150,8 @@ void TestLogLookup(const DriverID &driver_id, // Check that lookup returns the added object entries. auto lookup_callback = [task_id, node_manager_ids]( - gcs::AsyncGcsClient *client, const TaskID &id, - const std::vector &data) { + gcs::AsyncGcsClient *client, const TaskID &id, + const std::vector &data) { ASSERT_EQ(id, task_id); for (const auto &entry : data) { ASSERT_EQ(entry.node_manager_id, node_manager_ids[test->NumCallbacks()]); @@ -241,8 +241,8 @@ void TestLogAppendAt(const DriverID &driver_id, /*done callback=*/nullptr, failure_callback, /*log_length=*/1)); auto lookup_callback = [node_manager_ids]( - gcs::AsyncGcsClient *client, const TaskID &id, - const std::vector &data) { + gcs::AsyncGcsClient *client, const TaskID &id, + const std::vector &data) { std::vector appended_managers; for (const auto &entry : data) { appended_managers.push_back(entry.node_manager_id); @@ -282,8 +282,8 @@ void TestSet(const DriverID &driver_id, std::shared_ptr cli // Check that lookup returns the added object entries. auto lookup_callback = [object_id, managers]( - gcs::AsyncGcsClient *client, const ObjectID &id, - const std::vector &data) { + gcs::AsyncGcsClient *client, const ObjectID &id, + const std::vector &data) { ASSERT_EQ(id, object_id); ASSERT_EQ(data.size(), managers.size()); test->IncrementNumCallbacks(); @@ -296,8 +296,9 @@ void TestSet(const DriverID &driver_id, std::shared_ptr cli auto data = std::make_shared(); data->manager = manager; // Check that we added the correct object entries. - auto remove_entry_callback = [object_id, data]( - gcs::AsyncGcsClient *client, const ObjectID &id, const ObjectTableDataT &d) { + auto remove_entry_callback = [object_id, data](gcs::AsyncGcsClient *client, + const ObjectID &id, + const ObjectTableDataT &d) { ASSERT_EQ(id, object_id); ASSERT_EQ(data->manager, d.manager); test->IncrementNumCallbacks(); @@ -308,8 +309,8 @@ void TestSet(const DriverID &driver_id, std::shared_ptr cli // Check that the entries are removed. auto lookup_callback2 = [object_id, managers]( - gcs::AsyncGcsClient *client, const ObjectID &id, - const std::vector &data) { + gcs::AsyncGcsClient *client, const ObjectID &id, + const std::vector &data) { ASSERT_EQ(id, object_id); ASSERT_EQ(data.size(), 0); test->IncrementNumCallbacks(); @@ -350,8 +351,8 @@ void TestDeleteKeysFromLog( for (const auto &task_id : ids) { // Check that lookup returns the added object entries. auto lookup_callback = [task_id, data_vector]( - gcs::AsyncGcsClient *client, const TaskID &id, - const std::vector &data) { + gcs::AsyncGcsClient *client, const TaskID &id, + const std::vector &data) { ASSERT_EQ(id, task_id); ASSERT_EQ(data.size(), 1); test->IncrementNumCallbacks(); @@ -445,8 +446,8 @@ void TestDeleteKeysFromSet(const DriverID &driver_id, for (const auto &object_id : ids) { // Check that lookup returns the added object entries. auto lookup_callback = [object_id, data_vector]( - gcs::AsyncGcsClient *client, const ObjectID &id, - const std::vector &data) { + gcs::AsyncGcsClient *client, const ObjectID &id, + const std::vector &data) { ASSERT_EQ(id, object_id); ASSERT_EQ(data.size(), 1); test->IncrementNumCallbacks(); @@ -657,8 +658,9 @@ void TestSetSubscribeAll(const DriverID &driver_id, // Callback for a notification. auto notification_callback = [object_ids, managers]( - gcs::AsyncGcsClient *client, const ObjectID &id, const GcsChangeMode change_mode, - const std::vector data) { + gcs::AsyncGcsClient *client, const ObjectID &id, + const GcsChangeMode change_mode, + const std::vector data) { if (test->NumCallbacks() < 3 * 3) { ASSERT_EQ(change_mode, GcsChangeMode::APPEND_OR_ADD); } else { @@ -736,8 +738,9 @@ void TestTableSubscribeId(const DriverID &driver_id, // The callback for a notification from the table. This should only be // received for keys that we requested notifications for. - auto notification_callback = [task_id2, task_specs2]( - gcs::AsyncGcsClient *client, const TaskID &id, const protocol::TaskT &data) { + auto notification_callback = [task_id2, task_specs2](gcs::AsyncGcsClient *client, + const TaskID &id, + const protocol::TaskT &data) { // Check that we only get notifications for the requested key. ASSERT_EQ(id, task_id2); // Check that we get notifications in the same order as the writes. @@ -751,7 +754,7 @@ void TestTableSubscribeId(const DriverID &driver_id, // The failure callback should be called once since both keys start as empty. bool failure_notification_received = false; auto failure_callback = [task_id2, &failure_notification_received]( - gcs::AsyncGcsClient *client, const TaskID &id) { + gcs::AsyncGcsClient *client, const TaskID &id) { ASSERT_EQ(id, task_id2); // The failure notification should be the first notification received. ASSERT_EQ(test->NumCallbacks(), 0); @@ -819,8 +822,8 @@ void TestLogSubscribeId(const DriverID &driver_id, // The callback for a notification from the table. This should only be // received for keys that we requested notifications for. auto notification_callback = [driver_id2, driver_ids2]( - gcs::AsyncGcsClient *client, const UniqueID &id, - const std::vector &data) { + gcs::AsyncGcsClient *client, const UniqueID &id, + const std::vector &data) { // Check that we only get notifications for the requested key. ASSERT_EQ(id, driver_id2); // Check that we get notifications in the same order as the writes. @@ -893,8 +896,9 @@ void TestSetSubscribeId(const DriverID &driver_id, // The callback for a notification from the table. This should only be // received for keys that we requested notifications for. auto notification_callback = [object_id2, managers2]( - gcs::AsyncGcsClient *client, const ObjectID &id, const GcsChangeMode change_mode, - const std::vector &data) { + gcs::AsyncGcsClient *client, const ObjectID &id, + const GcsChangeMode change_mode, + const std::vector &data) { ASSERT_EQ(change_mode, GcsChangeMode::APPEND_OR_ADD); // Check that we only get notifications for the requested key. ASSERT_EQ(id, object_id2); @@ -966,8 +970,9 @@ void TestTableSubscribeCancel(const DriverID &driver_id, // The callback for a notification from the table. This should only be // received for keys that we requested notifications for. - auto notification_callback = [task_id, task_specs]( - gcs::AsyncGcsClient *client, const TaskID &id, const protocol::TaskT &data) { + auto notification_callback = [task_id, task_specs](gcs::AsyncGcsClient *client, + const TaskID &id, + const protocol::TaskT &data) { ASSERT_EQ(id, task_id); // Check that we only get notifications for the first and last writes, // since notifications are canceled in between. @@ -1036,8 +1041,8 @@ void TestLogSubscribeCancel(const DriverID &driver_id, // The callback for a notification from the object table. This should only be // received for the object that we requested notifications for. auto notification_callback = [random_driver_id, driver_ids]( - gcs::AsyncGcsClient *client, const UniqueID &id, - const std::vector &data) { + gcs::AsyncGcsClient *client, const UniqueID &id, + const std::vector &data) { ASSERT_EQ(id, random_driver_id); // Check that we get a duplicate notification for the first write. We get a // duplicate notification because the log is append-only and notifications @@ -1109,8 +1114,9 @@ void TestSetSubscribeCancel(const DriverID &driver_id, // The callback for a notification from the object table. This should only be // received for the object that we requested notifications for. auto notification_callback = [object_id, managers]( - gcs::AsyncGcsClient *client, const ObjectID &id, const GcsChangeMode change_mode, - const std::vector &data) { + gcs::AsyncGcsClient *client, const ObjectID &id, + const GcsChangeMode change_mode, + const std::vector &data) { ASSERT_EQ(change_mode, GcsChangeMode::APPEND_OR_ADD); ASSERT_EQ(id, object_id); // Check that we get a duplicate notification for the first write. We get a @@ -1291,11 +1297,12 @@ void TestClientTableMarkDisconnected(const DriverID &driver_id, RAY_CHECK_OK(client->client_table().MarkDisconnected(dead_client_id)); // Make sure we only get a notification for the removal of the client we // marked as dead. - client->client_table().RegisterClientRemovedCallback([dead_client_id]( - gcs::AsyncGcsClient *client, const UniqueID &id, const ClientTableDataT &data) { - ASSERT_EQ(ClientID::FromBinary(data.client_id), dead_client_id); - test->Stop(); - }); + client->client_table().RegisterClientRemovedCallback( + [dead_client_id](gcs::AsyncGcsClient *client, const UniqueID &id, + const ClientTableDataT &data) { + ASSERT_EQ(ClientID::FromBinary(data.client_id), dead_client_id); + test->Stop(); + }); test->Start(); } @@ -1350,8 +1357,9 @@ void TestHashTable(const DriverID &driver_id, test->IncrementNumCallbacks(); }; auto notification_callback = [data_map1, data_map2, compare_test]( - AsyncGcsClient *client, const ClientID &id, const GcsChangeMode change_mode, - const DynamicResourceTable::DataMap &data) { + AsyncGcsClient *client, const ClientID &id, + const GcsChangeMode change_mode, + const DynamicResourceTable::DataMap &data) { if (change_mode == GcsChangeMode::REMOVE) { ASSERT_EQ(data.size(), 2); ASSERT_TRUE(data.find("GPU") != data.end()); @@ -1380,16 +1388,16 @@ void TestHashTable(const DriverID &driver_id, // Step 1: Add elements to the hash table. auto update_callback1 = [data_map1, compare_test]( - AsyncGcsClient *client, const ClientID &id, - const DynamicResourceTable::DataMap &callback_data) { + AsyncGcsClient *client, const ClientID &id, + const DynamicResourceTable::DataMap &callback_data) { compare_test(data_map1, callback_data); test->IncrementNumCallbacks(); }; RAY_CHECK_OK( client->resource_table().Update(driver_id, client_id, data_map1, update_callback1)); auto lookup_callback1 = [data_map1, compare_test]( - AsyncGcsClient *client, const ClientID &id, - const DynamicResourceTable::DataMap &callback_data) { + AsyncGcsClient *client, const ClientID &id, + const DynamicResourceTable::DataMap &callback_data) { compare_test(data_map1, callback_data); test->IncrementNumCallbacks(); }; @@ -1398,8 +1406,8 @@ void TestHashTable(const DriverID &driver_id, // Step 2: Decrease one element, increase one and add a new one. RAY_CHECK_OK(client->resource_table().Update(driver_id, client_id, data_map2, nullptr)); auto lookup_callback2 = [data_map2, compare_test]( - AsyncGcsClient *client, const ClientID &id, - const DynamicResourceTable::DataMap &callback_data) { + AsyncGcsClient *client, const ClientID &id, + const DynamicResourceTable::DataMap &callback_data) { compare_test(data_map2, callback_data); test->IncrementNumCallbacks(); }; @@ -1419,8 +1427,8 @@ void TestHashTable(const DriverID &driver_id, data_map3.erase("GPU"); data_map3.erase("CUSTOM"); auto lookup_callback3 = [data_map3, compare_test]( - AsyncGcsClient *client, const ClientID &id, - const DynamicResourceTable::DataMap &callback_data) { + AsyncGcsClient *client, const ClientID &id, + const DynamicResourceTable::DataMap &callback_data) { compare_test(data_map3, callback_data); test->IncrementNumCallbacks(); }; @@ -1430,8 +1438,8 @@ void TestHashTable(const DriverID &driver_id, RAY_CHECK_OK( client->resource_table().Update(driver_id, client_id, data_map1, update_callback1)); auto lookup_callback4 = [data_map1, compare_test]( - AsyncGcsClient *client, const ClientID &id, - const DynamicResourceTable::DataMap &callback_data) { + AsyncGcsClient *client, const ClientID &id, + const DynamicResourceTable::DataMap &callback_data) { compare_test(data_map1, callback_data); test->IncrementNumCallbacks(); }; diff --git a/src/ray/gcs/redis_context.cc b/src/ray/gcs/redis_context.cc index fe5ba3d1d..ae6cb6088 100644 --- a/src/ray/gcs/redis_context.cc +++ b/src/ray/gcs/redis_context.cc @@ -48,28 +48,28 @@ namespace gcs { CallbackReply::CallbackReply(redisReply *redis_reply) { RAY_CHECK(nullptr != redis_reply); - RAY_CHECK(redis_reply->type != REDIS_REPLY_ERROR) << "Got an error in redis reply: " - << redis_reply->str; + RAY_CHECK(redis_reply->type != REDIS_REPLY_ERROR) + << "Got an error in redis reply: " << redis_reply->str; this->redis_reply_ = redis_reply; } bool CallbackReply::IsNil() const { return REDIS_REPLY_NIL == redis_reply_->type; } int64_t CallbackReply::ReadAsInteger() const { - RAY_CHECK(REDIS_REPLY_INTEGER == redis_reply_->type) << "Unexpected type: " - << redis_reply_->type; + RAY_CHECK(REDIS_REPLY_INTEGER == redis_reply_->type) + << "Unexpected type: " << redis_reply_->type; return static_cast(redis_reply_->integer); } std::string CallbackReply::ReadAsString() const { - RAY_CHECK(REDIS_REPLY_STRING == redis_reply_->type) << "Unexpected type: " - << redis_reply_->type; + RAY_CHECK(REDIS_REPLY_STRING == redis_reply_->type) + << "Unexpected type: " << redis_reply_->type; return std::string(redis_reply_->str, redis_reply_->len); } Status CallbackReply::ReadAsStatus() const { - RAY_CHECK(REDIS_REPLY_STATUS == redis_reply_->type) << "Unexpected type: " - << redis_reply_->type; + RAY_CHECK(REDIS_REPLY_STATUS == redis_reply_->type) + << "Unexpected type: " << redis_reply_->type; const std::string status_str(redis_reply_->str, redis_reply_->len); if ("OK" == status_str) { return Status::OK(); @@ -79,8 +79,8 @@ Status CallbackReply::ReadAsStatus() const { } std::string CallbackReply::ReadAsPubsubData() const { - RAY_CHECK(REDIS_REPLY_ARRAY == redis_reply_->type) << "Unexpected type: " - << redis_reply_->type; + RAY_CHECK(REDIS_REPLY_ARRAY == redis_reply_->type) + << "Unexpected type: " << redis_reply_->type; std::string data = ""; // Parse the published message. diff --git a/src/ray/gcs/redis_module/ray_redis_module.cc b/src/ray/gcs/redis_module/ray_redis_module.cc index e05978747..e291b7ffd 100644 --- a/src/ray/gcs/redis_module/ray_redis_module.cc +++ b/src/ray/gcs/redis_module/ray_redis_module.cc @@ -1004,7 +1004,7 @@ int DebugString_RedisCommand(RedisModuleCtx *ctx, RedisModuleString **argv, int std::string debug_string = DebugString(); return RedisModule_ReplyWithStringBuffer(ctx, debug_string.data(), debug_string.size()); } -}; +}; // namespace internal_redis_commands // Wrap all Redis commands with Redis' auto memory management. AUTO_MEMORY(TableAdd_RedisCommand); diff --git a/src/ray/gcs/redis_module/redismodule.h b/src/ray/gcs/redis_module/redismodule.h index 186e284c0..23721beef 100644 --- a/src/ray/gcs/redis_module/redismodule.h +++ b/src/ray/gcs/redis_module/redismodule.h @@ -1,9 +1,9 @@ #ifndef REDISMODULE_H #define REDISMODULE_H -#include #include #include +#include /* ---------------- Defines common between core and modules --------------- */ @@ -15,8 +15,8 @@ #define REDISMODULE_APIVER_1 1 /* API flags and constants */ -#define REDISMODULE_READ (1<<0) -#define REDISMODULE_WRITE (1<<1) +#define REDISMODULE_READ (1 << 0) +#define REDISMODULE_WRITE (1 << 1) #define REDISMODULE_LIST_HEAD 0 #define REDISMODULE_LIST_TAIL 1 @@ -45,30 +45,31 @@ #define REDISMODULE_NO_EXPIRE -1 /* Sorted set API flags. */ -#define REDISMODULE_ZADD_XX (1<<0) -#define REDISMODULE_ZADD_NX (1<<1) -#define REDISMODULE_ZADD_ADDED (1<<2) -#define REDISMODULE_ZADD_UPDATED (1<<3) -#define REDISMODULE_ZADD_NOP (1<<4) +#define REDISMODULE_ZADD_XX (1 << 0) +#define REDISMODULE_ZADD_NX (1 << 1) +#define REDISMODULE_ZADD_ADDED (1 << 2) +#define REDISMODULE_ZADD_UPDATED (1 << 3) +#define REDISMODULE_ZADD_NOP (1 << 4) /* Hash API flags. */ -#define REDISMODULE_HASH_NONE 0 -#define REDISMODULE_HASH_NX (1<<0) -#define REDISMODULE_HASH_XX (1<<1) -#define REDISMODULE_HASH_CFIELDS (1<<2) -#define REDISMODULE_HASH_EXISTS (1<<3) +#define REDISMODULE_HASH_NONE 0 +#define REDISMODULE_HASH_NX (1 << 0) +#define REDISMODULE_HASH_XX (1 << 1) +#define REDISMODULE_HASH_CFIELDS (1 << 2) +#define REDISMODULE_HASH_EXISTS (1 << 3) /* A special pointer that we can use between the core and the module to signal * field deletion, and that is impossible to be a valid pointer. */ -#define REDISMODULE_HASH_DELETE ((RedisModuleString*)(long)1) +#define REDISMODULE_HASH_DELETE ((RedisModuleString *)(long)1) /* Error messages. */ -#define REDISMODULE_ERRORMSG_WRONGTYPE "WRONGTYPE Operation against a key holding the wrong kind of value" +#define REDISMODULE_ERRORMSG_WRONGTYPE \ + "WRONGTYPE Operation against a key holding the wrong kind of value" -#define REDISMODULE_POSITIVE_INFINITE (1.0/0.0) -#define REDISMODULE_NEGATIVE_INFINITE (-1.0/0.0) +#define REDISMODULE_POSITIVE_INFINITE (1.0 / 0.0) +#define REDISMODULE_NEGATIVE_INFINITE (-1.0 / 0.0) -#define REDISMODULE_NOT_USED(V) ((void) V) +#define REDISMODULE_NOT_USED(V) ((void)V) /* ------------------------- End of common defines ------------------------ */ @@ -86,95 +87,142 @@ typedef struct RedisModuleType RedisModuleType; typedef struct RedisModuleDigest RedisModuleDigest; typedef struct RedisModuleBlockedClient RedisModuleBlockedClient; -typedef int (*RedisModuleCmdFunc) (RedisModuleCtx *ctx, RedisModuleString **argv, int argc); +typedef int (*RedisModuleCmdFunc)(RedisModuleCtx *ctx, RedisModuleString **argv, + int argc); typedef void *(*RedisModuleTypeLoadFunc)(RedisModuleIO *rdb, int encver); typedef void (*RedisModuleTypeSaveFunc)(RedisModuleIO *rdb, void *value); -typedef void (*RedisModuleTypeRewriteFunc)(RedisModuleIO *aof, RedisModuleString *key, void *value); +typedef void (*RedisModuleTypeRewriteFunc)(RedisModuleIO *aof, RedisModuleString *key, + void *value); typedef size_t (*RedisModuleTypeMemUsageFunc)(void *value); typedef void (*RedisModuleTypeDigestFunc)(RedisModuleDigest *digest, void *value); typedef void (*RedisModuleTypeFreeFunc)(void *value); #define REDISMODULE_TYPE_METHOD_VERSION 1 typedef struct RedisModuleTypeMethods { - uint64_t version; - RedisModuleTypeLoadFunc rdb_load; - RedisModuleTypeSaveFunc rdb_save; - RedisModuleTypeRewriteFunc aof_rewrite; - RedisModuleTypeMemUsageFunc mem_usage; - RedisModuleTypeDigestFunc digest; - RedisModuleTypeFreeFunc free; + uint64_t version; + RedisModuleTypeLoadFunc rdb_load; + RedisModuleTypeSaveFunc rdb_save; + RedisModuleTypeRewriteFunc aof_rewrite; + RedisModuleTypeMemUsageFunc mem_usage; + RedisModuleTypeDigestFunc digest; + RedisModuleTypeFreeFunc free; } RedisModuleTypeMethods; #define REDISMODULE_GET_API(name) \ - RedisModule_GetApi("RedisModule_" #name, ((void **)&RedisModule_ ## name)) + RedisModule_GetApi("RedisModule_" #name, ((void **)&RedisModule_##name)) #define REDISMODULE_API_FUNC(x) (*x) - void *REDISMODULE_API_FUNC(RedisModule_Alloc)(size_t bytes); void *REDISMODULE_API_FUNC(RedisModule_Realloc)(void *ptr, size_t bytes); void REDISMODULE_API_FUNC(RedisModule_Free)(void *ptr); void *REDISMODULE_API_FUNC(RedisModule_Calloc)(size_t nmemb, size_t size); char *REDISMODULE_API_FUNC(RedisModule_Strdup)(const char *str); int REDISMODULE_API_FUNC(RedisModule_GetApi)(const char *, void *); -int REDISMODULE_API_FUNC(RedisModule_CreateCommand)(RedisModuleCtx *ctx, const char *name, RedisModuleCmdFunc cmdfunc, const char *strflags, int firstkey, int lastkey, int keystep); -int REDISMODULE_API_FUNC(RedisModule_SetModuleAttribs)(RedisModuleCtx *ctx, const char *name, int ver, int apiver); +int REDISMODULE_API_FUNC(RedisModule_CreateCommand)(RedisModuleCtx *ctx, const char *name, + RedisModuleCmdFunc cmdfunc, + const char *strflags, int firstkey, + int lastkey, int keystep); +int REDISMODULE_API_FUNC(RedisModule_SetModuleAttribs)(RedisModuleCtx *ctx, + const char *name, int ver, + int apiver); int REDISMODULE_API_FUNC(RedisModule_WrongArity)(RedisModuleCtx *ctx); -int REDISMODULE_API_FUNC(RedisModule_ReplyWithLongLong)(RedisModuleCtx *ctx, long long ll); +int REDISMODULE_API_FUNC(RedisModule_ReplyWithLongLong)(RedisModuleCtx *ctx, + long long ll); int REDISMODULE_API_FUNC(RedisModule_GetSelectedDb)(RedisModuleCtx *ctx); int REDISMODULE_API_FUNC(RedisModule_SelectDb)(RedisModuleCtx *ctx, int newid); -void *REDISMODULE_API_FUNC(RedisModule_OpenKey)(RedisModuleCtx *ctx, RedisModuleString *keyname, int mode); +void *REDISMODULE_API_FUNC(RedisModule_OpenKey)(RedisModuleCtx *ctx, + RedisModuleString *keyname, int mode); void REDISMODULE_API_FUNC(RedisModule_CloseKey)(RedisModuleKey *kp); int REDISMODULE_API_FUNC(RedisModule_KeyType)(RedisModuleKey *kp); size_t REDISMODULE_API_FUNC(RedisModule_ValueLength)(RedisModuleKey *kp); -int REDISMODULE_API_FUNC(RedisModule_ListPush)(RedisModuleKey *kp, int where, RedisModuleString *ele); -RedisModuleString *REDISMODULE_API_FUNC(RedisModule_ListPop)(RedisModuleKey *key, int where); -RedisModuleCallReply *REDISMODULE_API_FUNC(RedisModule_Call)(RedisModuleCtx *ctx, const char *cmdname, const char *fmt, ...); -const char *REDISMODULE_API_FUNC(RedisModule_CallReplyProto)(RedisModuleCallReply *reply, size_t *len); +int REDISMODULE_API_FUNC(RedisModule_ListPush)(RedisModuleKey *kp, int where, + RedisModuleString *ele); +RedisModuleString *REDISMODULE_API_FUNC(RedisModule_ListPop)(RedisModuleKey *key, + int where); +RedisModuleCallReply *REDISMODULE_API_FUNC(RedisModule_Call)(RedisModuleCtx *ctx, + const char *cmdname, + const char *fmt, ...); +const char *REDISMODULE_API_FUNC(RedisModule_CallReplyProto)(RedisModuleCallReply *reply, + size_t *len); void REDISMODULE_API_FUNC(RedisModule_FreeCallReply)(RedisModuleCallReply *reply); int REDISMODULE_API_FUNC(RedisModule_CallReplyType)(RedisModuleCallReply *reply); long long REDISMODULE_API_FUNC(RedisModule_CallReplyInteger)(RedisModuleCallReply *reply); size_t REDISMODULE_API_FUNC(RedisModule_CallReplyLength)(RedisModuleCallReply *reply); -RedisModuleCallReply *REDISMODULE_API_FUNC(RedisModule_CallReplyArrayElement)(RedisModuleCallReply *reply, size_t idx); -RedisModuleString *REDISMODULE_API_FUNC(RedisModule_CreateString)(RedisModuleCtx *ctx, const char *ptr, size_t len); -RedisModuleString *REDISMODULE_API_FUNC(RedisModule_CreateStringFromLongLong)(RedisModuleCtx *ctx, long long ll); -RedisModuleString *REDISMODULE_API_FUNC(RedisModule_CreateStringFromString)(RedisModuleCtx *ctx, const RedisModuleString *str); -RedisModuleString *REDISMODULE_API_FUNC(RedisModule_CreateStringPrintf)(RedisModuleCtx *ctx, const char *fmt, ...); -void REDISMODULE_API_FUNC(RedisModule_FreeString)(RedisModuleCtx *ctx, RedisModuleString *str); -const char *REDISMODULE_API_FUNC(RedisModule_StringPtrLen)(const RedisModuleString *str, size_t *len); -int REDISMODULE_API_FUNC(RedisModule_ReplyWithError)(RedisModuleCtx *ctx, const char *err); -int REDISMODULE_API_FUNC(RedisModule_ReplyWithSimpleString)(RedisModuleCtx *ctx, const char *msg); +RedisModuleCallReply *REDISMODULE_API_FUNC(RedisModule_CallReplyArrayElement)( + RedisModuleCallReply *reply, size_t idx); +RedisModuleString *REDISMODULE_API_FUNC(RedisModule_CreateString)(RedisModuleCtx *ctx, + const char *ptr, + size_t len); +RedisModuleString *REDISMODULE_API_FUNC(RedisModule_CreateStringFromLongLong)( + RedisModuleCtx *ctx, long long ll); +RedisModuleString *REDISMODULE_API_FUNC(RedisModule_CreateStringFromString)( + RedisModuleCtx *ctx, const RedisModuleString *str); +RedisModuleString *REDISMODULE_API_FUNC(RedisModule_CreateStringPrintf)( + RedisModuleCtx *ctx, const char *fmt, ...); +void REDISMODULE_API_FUNC(RedisModule_FreeString)(RedisModuleCtx *ctx, + RedisModuleString *str); +const char *REDISMODULE_API_FUNC(RedisModule_StringPtrLen)(const RedisModuleString *str, + size_t *len); +int REDISMODULE_API_FUNC(RedisModule_ReplyWithError)(RedisModuleCtx *ctx, + const char *err); +int REDISMODULE_API_FUNC(RedisModule_ReplyWithSimpleString)(RedisModuleCtx *ctx, + const char *msg); int REDISMODULE_API_FUNC(RedisModule_ReplyWithArray)(RedisModuleCtx *ctx, long len); void REDISMODULE_API_FUNC(RedisModule_ReplySetArrayLength)(RedisModuleCtx *ctx, long len); -int REDISMODULE_API_FUNC(RedisModule_ReplyWithStringBuffer)(RedisModuleCtx *ctx, const char *buf, size_t len); -int REDISMODULE_API_FUNC(RedisModule_ReplyWithString)(RedisModuleCtx *ctx, RedisModuleString *str); +int REDISMODULE_API_FUNC(RedisModule_ReplyWithStringBuffer)(RedisModuleCtx *ctx, + const char *buf, size_t len); +int REDISMODULE_API_FUNC(RedisModule_ReplyWithString)(RedisModuleCtx *ctx, + RedisModuleString *str); int REDISMODULE_API_FUNC(RedisModule_ReplyWithNull)(RedisModuleCtx *ctx); int REDISMODULE_API_FUNC(RedisModule_ReplyWithDouble)(RedisModuleCtx *ctx, double d); -int REDISMODULE_API_FUNC(RedisModule_ReplyWithCallReply)(RedisModuleCtx *ctx, RedisModuleCallReply *reply); -int REDISMODULE_API_FUNC(RedisModule_StringToLongLong)(const RedisModuleString *str, long long *ll); -int REDISMODULE_API_FUNC(RedisModule_StringToDouble)(const RedisModuleString *str, double *d); +int REDISMODULE_API_FUNC(RedisModule_ReplyWithCallReply)(RedisModuleCtx *ctx, + RedisModuleCallReply *reply); +int REDISMODULE_API_FUNC(RedisModule_StringToLongLong)(const RedisModuleString *str, + long long *ll); +int REDISMODULE_API_FUNC(RedisModule_StringToDouble)(const RedisModuleString *str, + double *d); void REDISMODULE_API_FUNC(RedisModule_AutoMemory)(RedisModuleCtx *ctx); -int REDISMODULE_API_FUNC(RedisModule_Replicate)(RedisModuleCtx *ctx, const char *cmdname, const char *fmt, ...); +int REDISMODULE_API_FUNC(RedisModule_Replicate)(RedisModuleCtx *ctx, const char *cmdname, + const char *fmt, ...); int REDISMODULE_API_FUNC(RedisModule_ReplicateVerbatim)(RedisModuleCtx *ctx); -const char *REDISMODULE_API_FUNC(RedisModule_CallReplyStringPtr)(RedisModuleCallReply *reply, size_t *len); -RedisModuleString *REDISMODULE_API_FUNC(RedisModule_CreateStringFromCallReply)(RedisModuleCallReply *reply); +const char *REDISMODULE_API_FUNC(RedisModule_CallReplyStringPtr)( + RedisModuleCallReply *reply, size_t *len); +RedisModuleString *REDISMODULE_API_FUNC(RedisModule_CreateStringFromCallReply)( + RedisModuleCallReply *reply); int REDISMODULE_API_FUNC(RedisModule_DeleteKey)(RedisModuleKey *key); -int REDISMODULE_API_FUNC(RedisModule_StringSet)(RedisModuleKey *key, RedisModuleString *str); -char *REDISMODULE_API_FUNC(RedisModule_StringDMA)(RedisModuleKey *key, size_t *len, int mode); +int REDISMODULE_API_FUNC(RedisModule_StringSet)(RedisModuleKey *key, + RedisModuleString *str); +char *REDISMODULE_API_FUNC(RedisModule_StringDMA)(RedisModuleKey *key, size_t *len, + int mode); int REDISMODULE_API_FUNC(RedisModule_StringTruncate)(RedisModuleKey *key, size_t newlen); mstime_t REDISMODULE_API_FUNC(RedisModule_GetExpire)(RedisModuleKey *key); int REDISMODULE_API_FUNC(RedisModule_SetExpire)(RedisModuleKey *key, mstime_t expire); -int REDISMODULE_API_FUNC(RedisModule_ZsetAdd)(RedisModuleKey *key, double score, RedisModuleString *ele, int *flagsptr); -int REDISMODULE_API_FUNC(RedisModule_ZsetIncrby)(RedisModuleKey *key, double score, RedisModuleString *ele, int *flagsptr, double *newscore); -int REDISMODULE_API_FUNC(RedisModule_ZsetScore)(RedisModuleKey *key, RedisModuleString *ele, double *score); -int REDISMODULE_API_FUNC(RedisModule_ZsetRem)(RedisModuleKey *key, RedisModuleString *ele, int *deleted); +int REDISMODULE_API_FUNC(RedisModule_ZsetAdd)(RedisModuleKey *key, double score, + RedisModuleString *ele, int *flagsptr); +int REDISMODULE_API_FUNC(RedisModule_ZsetIncrby)(RedisModuleKey *key, double score, + RedisModuleString *ele, int *flagsptr, + double *newscore); +int REDISMODULE_API_FUNC(RedisModule_ZsetScore)(RedisModuleKey *key, + RedisModuleString *ele, double *score); +int REDISMODULE_API_FUNC(RedisModule_ZsetRem)(RedisModuleKey *key, RedisModuleString *ele, + int *deleted); void REDISMODULE_API_FUNC(RedisModule_ZsetRangeStop)(RedisModuleKey *key); -int REDISMODULE_API_FUNC(RedisModule_ZsetFirstInScoreRange)(RedisModuleKey *key, double min, double max, int minex, int maxex); -int REDISMODULE_API_FUNC(RedisModule_ZsetLastInScoreRange)(RedisModuleKey *key, double min, double max, int minex, int maxex); -int REDISMODULE_API_FUNC(RedisModule_ZsetFirstInLexRange)(RedisModuleKey *key, RedisModuleString *min, RedisModuleString *max); -int REDISMODULE_API_FUNC(RedisModule_ZsetLastInLexRange)(RedisModuleKey *key, RedisModuleString *min, RedisModuleString *max); -RedisModuleString *REDISMODULE_API_FUNC(RedisModule_ZsetRangeCurrentElement)(RedisModuleKey *key, double *score); +int REDISMODULE_API_FUNC(RedisModule_ZsetFirstInScoreRange)(RedisModuleKey *key, + double min, double max, + int minex, int maxex); +int REDISMODULE_API_FUNC(RedisModule_ZsetLastInScoreRange)(RedisModuleKey *key, + double min, double max, + int minex, int maxex); +int REDISMODULE_API_FUNC(RedisModule_ZsetFirstInLexRange)(RedisModuleKey *key, + RedisModuleString *min, + RedisModuleString *max); +int REDISMODULE_API_FUNC(RedisModule_ZsetLastInLexRange)(RedisModuleKey *key, + RedisModuleString *min, + RedisModuleString *max); +RedisModuleString *REDISMODULE_API_FUNC(RedisModule_ZsetRangeCurrentElement)( + RedisModuleKey *key, double *score); int REDISMODULE_API_FUNC(RedisModule_ZsetRangeNext)(RedisModuleKey *key); int REDISMODULE_API_FUNC(RedisModule_ZsetRangePrev)(RedisModuleKey *key); int REDISMODULE_API_FUNC(RedisModule_ZsetRangeEndReached)(RedisModuleKey *key); @@ -184,31 +232,49 @@ int REDISMODULE_API_FUNC(RedisModule_IsKeysPositionRequest)(RedisModuleCtx *ctx) void REDISMODULE_API_FUNC(RedisModule_KeyAtPos)(RedisModuleCtx *ctx, int pos); unsigned long long REDISMODULE_API_FUNC(RedisModule_GetClientId)(RedisModuleCtx *ctx); void *REDISMODULE_API_FUNC(RedisModule_PoolAlloc)(RedisModuleCtx *ctx, size_t bytes); -RedisModuleType *REDISMODULE_API_FUNC(RedisModule_CreateDataType)(RedisModuleCtx *ctx, const char *name, int encver, RedisModuleTypeMethods *typemethods); -int REDISMODULE_API_FUNC(RedisModule_ModuleTypeSetValue)(RedisModuleKey *key, RedisModuleType *mt, void *value); +RedisModuleType *REDISMODULE_API_FUNC(RedisModule_CreateDataType)( + RedisModuleCtx *ctx, const char *name, int encver, + RedisModuleTypeMethods *typemethods); +int REDISMODULE_API_FUNC(RedisModule_ModuleTypeSetValue)(RedisModuleKey *key, + RedisModuleType *mt, + void *value); RedisModuleType *REDISMODULE_API_FUNC(RedisModule_ModuleTypeGetType)(RedisModuleKey *key); void *REDISMODULE_API_FUNC(RedisModule_ModuleTypeGetValue)(RedisModuleKey *key); void REDISMODULE_API_FUNC(RedisModule_SaveUnsigned)(RedisModuleIO *io, uint64_t value); uint64_t REDISMODULE_API_FUNC(RedisModule_LoadUnsigned)(RedisModuleIO *io); void REDISMODULE_API_FUNC(RedisModule_SaveSigned)(RedisModuleIO *io, int64_t value); int64_t REDISMODULE_API_FUNC(RedisModule_LoadSigned)(RedisModuleIO *io); -void REDISMODULE_API_FUNC(RedisModule_EmitAOF)(RedisModuleIO *io, const char *cmdname, const char *fmt, ...); -void REDISMODULE_API_FUNC(RedisModule_SaveString)(RedisModuleIO *io, RedisModuleString *s); -void REDISMODULE_API_FUNC(RedisModule_SaveStringBuffer)(RedisModuleIO *io, const char *str, size_t len); +void REDISMODULE_API_FUNC(RedisModule_EmitAOF)(RedisModuleIO *io, const char *cmdname, + const char *fmt, ...); +void REDISMODULE_API_FUNC(RedisModule_SaveString)(RedisModuleIO *io, + RedisModuleString *s); +void REDISMODULE_API_FUNC(RedisModule_SaveStringBuffer)(RedisModuleIO *io, + const char *str, size_t len); RedisModuleString *REDISMODULE_API_FUNC(RedisModule_LoadString)(RedisModuleIO *io); -char *REDISMODULE_API_FUNC(RedisModule_LoadStringBuffer)(RedisModuleIO *io, size_t *lenptr); +char *REDISMODULE_API_FUNC(RedisModule_LoadStringBuffer)(RedisModuleIO *io, + size_t *lenptr); void REDISMODULE_API_FUNC(RedisModule_SaveDouble)(RedisModuleIO *io, double value); double REDISMODULE_API_FUNC(RedisModule_LoadDouble)(RedisModuleIO *io); void REDISMODULE_API_FUNC(RedisModule_SaveFloat)(RedisModuleIO *io, float value); float REDISMODULE_API_FUNC(RedisModule_LoadFloat)(RedisModuleIO *io); -void REDISMODULE_API_FUNC(RedisModule_Log)(RedisModuleCtx *ctx, const char *level, const char *fmt, ...); -void REDISMODULE_API_FUNC(RedisModule_LogIOError)(RedisModuleIO *io, const char *levelstr, const char *fmt, ...); -int REDISMODULE_API_FUNC(RedisModule_StringAppendBuffer)(RedisModuleCtx *ctx, RedisModuleString *str, const char *buf, size_t len); -void REDISMODULE_API_FUNC(RedisModule_RetainString)(RedisModuleCtx *ctx, RedisModuleString *str); -int REDISMODULE_API_FUNC(RedisModule_StringCompare)(RedisModuleString *a, RedisModuleString *b); +void REDISMODULE_API_FUNC(RedisModule_Log)(RedisModuleCtx *ctx, const char *level, + const char *fmt, ...); +void REDISMODULE_API_FUNC(RedisModule_LogIOError)(RedisModuleIO *io, const char *levelstr, + const char *fmt, ...); +int REDISMODULE_API_FUNC(RedisModule_StringAppendBuffer)(RedisModuleCtx *ctx, + RedisModuleString *str, + const char *buf, size_t len); +void REDISMODULE_API_FUNC(RedisModule_RetainString)(RedisModuleCtx *ctx, + RedisModuleString *str); +int REDISMODULE_API_FUNC(RedisModule_StringCompare)(RedisModuleString *a, + RedisModuleString *b); RedisModuleCtx *REDISMODULE_API_FUNC(RedisModule_GetContextFromIO)(RedisModuleIO *io); -RedisModuleBlockedClient *REDISMODULE_API_FUNC(RedisModule_BlockClient)(RedisModuleCtx *ctx, RedisModuleCmdFunc reply_callback, RedisModuleCmdFunc timeout_callback, void (*free_privdata)(void*), long long timeout_ms); -int REDISMODULE_API_FUNC(RedisModule_UnblockClient)(RedisModuleBlockedClient *bc, void *privdata); +RedisModuleBlockedClient *REDISMODULE_API_FUNC(RedisModule_BlockClient)( + RedisModuleCtx *ctx, RedisModuleCmdFunc reply_callback, + RedisModuleCmdFunc timeout_callback, void (*free_privdata)(void *), + long long timeout_ms); +int REDISMODULE_API_FUNC(RedisModule_UnblockClient)(RedisModuleBlockedClient *bc, + void *privdata); int REDISMODULE_API_FUNC(RedisModule_IsBlockedReplyRequest)(RedisModuleCtx *ctx); int REDISMODULE_API_FUNC(RedisModule_IsBlockedTimeoutRequest)(RedisModuleCtx *ctx); void *REDISMODULE_API_FUNC(RedisModule_GetBlockedClientPrivateData)(RedisModuleCtx *ctx); @@ -216,115 +282,116 @@ int REDISMODULE_API_FUNC(RedisModule_AbortBlock)(RedisModuleBlockedClient *bc); long long REDISMODULE_API_FUNC(RedisModule_Milliseconds)(void); /* This is included inline inside each Redis module. */ -static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int apiver) __attribute__((unused)); +static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int apiver) + __attribute__((unused)); static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int apiver) { - void *getapifuncptr = ((void**)ctx)[0]; - RedisModule_GetApi = (int (*)(const char *, void *)) (unsigned long)getapifuncptr; - REDISMODULE_GET_API(Alloc); - REDISMODULE_GET_API(Calloc); - REDISMODULE_GET_API(Free); - REDISMODULE_GET_API(Realloc); - REDISMODULE_GET_API(Strdup); - REDISMODULE_GET_API(CreateCommand); - REDISMODULE_GET_API(SetModuleAttribs); - REDISMODULE_GET_API(WrongArity); - REDISMODULE_GET_API(ReplyWithLongLong); - REDISMODULE_GET_API(ReplyWithError); - REDISMODULE_GET_API(ReplyWithSimpleString); - REDISMODULE_GET_API(ReplyWithArray); - REDISMODULE_GET_API(ReplySetArrayLength); - REDISMODULE_GET_API(ReplyWithStringBuffer); - REDISMODULE_GET_API(ReplyWithString); - REDISMODULE_GET_API(ReplyWithNull); - REDISMODULE_GET_API(ReplyWithCallReply); - REDISMODULE_GET_API(ReplyWithDouble); - REDISMODULE_GET_API(ReplySetArrayLength); - REDISMODULE_GET_API(GetSelectedDb); - REDISMODULE_GET_API(SelectDb); - REDISMODULE_GET_API(OpenKey); - REDISMODULE_GET_API(CloseKey); - REDISMODULE_GET_API(KeyType); - REDISMODULE_GET_API(ValueLength); - REDISMODULE_GET_API(ListPush); - REDISMODULE_GET_API(ListPop); - REDISMODULE_GET_API(StringToLongLong); - REDISMODULE_GET_API(StringToDouble); - REDISMODULE_GET_API(Call); - REDISMODULE_GET_API(CallReplyProto); - REDISMODULE_GET_API(FreeCallReply); - REDISMODULE_GET_API(CallReplyInteger); - REDISMODULE_GET_API(CallReplyType); - REDISMODULE_GET_API(CallReplyLength); - REDISMODULE_GET_API(CallReplyArrayElement); - REDISMODULE_GET_API(CallReplyStringPtr); - REDISMODULE_GET_API(CreateStringFromCallReply); - REDISMODULE_GET_API(CreateString); - REDISMODULE_GET_API(CreateStringFromLongLong); - REDISMODULE_GET_API(CreateStringFromString); - REDISMODULE_GET_API(CreateStringPrintf); - REDISMODULE_GET_API(FreeString); - REDISMODULE_GET_API(StringPtrLen); - REDISMODULE_GET_API(AutoMemory); - REDISMODULE_GET_API(Replicate); - REDISMODULE_GET_API(ReplicateVerbatim); - REDISMODULE_GET_API(DeleteKey); - REDISMODULE_GET_API(StringSet); - REDISMODULE_GET_API(StringDMA); - REDISMODULE_GET_API(StringTruncate); - REDISMODULE_GET_API(GetExpire); - REDISMODULE_GET_API(SetExpire); - REDISMODULE_GET_API(ZsetAdd); - REDISMODULE_GET_API(ZsetIncrby); - REDISMODULE_GET_API(ZsetScore); - REDISMODULE_GET_API(ZsetRem); - REDISMODULE_GET_API(ZsetRangeStop); - REDISMODULE_GET_API(ZsetFirstInScoreRange); - REDISMODULE_GET_API(ZsetLastInScoreRange); - REDISMODULE_GET_API(ZsetFirstInLexRange); - REDISMODULE_GET_API(ZsetLastInLexRange); - REDISMODULE_GET_API(ZsetRangeCurrentElement); - REDISMODULE_GET_API(ZsetRangeNext); - REDISMODULE_GET_API(ZsetRangePrev); - REDISMODULE_GET_API(ZsetRangeEndReached); - REDISMODULE_GET_API(HashSet); - REDISMODULE_GET_API(HashGet); - REDISMODULE_GET_API(IsKeysPositionRequest); - REDISMODULE_GET_API(KeyAtPos); - REDISMODULE_GET_API(GetClientId); - REDISMODULE_GET_API(PoolAlloc); - REDISMODULE_GET_API(CreateDataType); - REDISMODULE_GET_API(ModuleTypeSetValue); - REDISMODULE_GET_API(ModuleTypeGetType); - REDISMODULE_GET_API(ModuleTypeGetValue); - REDISMODULE_GET_API(SaveUnsigned); - REDISMODULE_GET_API(LoadUnsigned); - REDISMODULE_GET_API(SaveSigned); - REDISMODULE_GET_API(LoadSigned); - REDISMODULE_GET_API(SaveString); - REDISMODULE_GET_API(SaveStringBuffer); - REDISMODULE_GET_API(LoadString); - REDISMODULE_GET_API(LoadStringBuffer); - REDISMODULE_GET_API(SaveDouble); - REDISMODULE_GET_API(LoadDouble); - REDISMODULE_GET_API(SaveFloat); - REDISMODULE_GET_API(LoadFloat); - REDISMODULE_GET_API(EmitAOF); - REDISMODULE_GET_API(Log); - REDISMODULE_GET_API(LogIOError); - REDISMODULE_GET_API(StringAppendBuffer); - REDISMODULE_GET_API(RetainString); - REDISMODULE_GET_API(StringCompare); - REDISMODULE_GET_API(GetContextFromIO); - REDISMODULE_GET_API(BlockClient); - REDISMODULE_GET_API(UnblockClient); - REDISMODULE_GET_API(IsBlockedReplyRequest); - REDISMODULE_GET_API(IsBlockedTimeoutRequest); - REDISMODULE_GET_API(GetBlockedClientPrivateData); - REDISMODULE_GET_API(AbortBlock); - REDISMODULE_GET_API(Milliseconds); + void *getapifuncptr = ((void **)ctx)[0]; + RedisModule_GetApi = (int (*)(const char *, void *))(unsigned long)getapifuncptr; + REDISMODULE_GET_API(Alloc); + REDISMODULE_GET_API(Calloc); + REDISMODULE_GET_API(Free); + REDISMODULE_GET_API(Realloc); + REDISMODULE_GET_API(Strdup); + REDISMODULE_GET_API(CreateCommand); + REDISMODULE_GET_API(SetModuleAttribs); + REDISMODULE_GET_API(WrongArity); + REDISMODULE_GET_API(ReplyWithLongLong); + REDISMODULE_GET_API(ReplyWithError); + REDISMODULE_GET_API(ReplyWithSimpleString); + REDISMODULE_GET_API(ReplyWithArray); + REDISMODULE_GET_API(ReplySetArrayLength); + REDISMODULE_GET_API(ReplyWithStringBuffer); + REDISMODULE_GET_API(ReplyWithString); + REDISMODULE_GET_API(ReplyWithNull); + REDISMODULE_GET_API(ReplyWithCallReply); + REDISMODULE_GET_API(ReplyWithDouble); + REDISMODULE_GET_API(ReplySetArrayLength); + REDISMODULE_GET_API(GetSelectedDb); + REDISMODULE_GET_API(SelectDb); + REDISMODULE_GET_API(OpenKey); + REDISMODULE_GET_API(CloseKey); + REDISMODULE_GET_API(KeyType); + REDISMODULE_GET_API(ValueLength); + REDISMODULE_GET_API(ListPush); + REDISMODULE_GET_API(ListPop); + REDISMODULE_GET_API(StringToLongLong); + REDISMODULE_GET_API(StringToDouble); + REDISMODULE_GET_API(Call); + REDISMODULE_GET_API(CallReplyProto); + REDISMODULE_GET_API(FreeCallReply); + REDISMODULE_GET_API(CallReplyInteger); + REDISMODULE_GET_API(CallReplyType); + REDISMODULE_GET_API(CallReplyLength); + REDISMODULE_GET_API(CallReplyArrayElement); + REDISMODULE_GET_API(CallReplyStringPtr); + REDISMODULE_GET_API(CreateStringFromCallReply); + REDISMODULE_GET_API(CreateString); + REDISMODULE_GET_API(CreateStringFromLongLong); + REDISMODULE_GET_API(CreateStringFromString); + REDISMODULE_GET_API(CreateStringPrintf); + REDISMODULE_GET_API(FreeString); + REDISMODULE_GET_API(StringPtrLen); + REDISMODULE_GET_API(AutoMemory); + REDISMODULE_GET_API(Replicate); + REDISMODULE_GET_API(ReplicateVerbatim); + REDISMODULE_GET_API(DeleteKey); + REDISMODULE_GET_API(StringSet); + REDISMODULE_GET_API(StringDMA); + REDISMODULE_GET_API(StringTruncate); + REDISMODULE_GET_API(GetExpire); + REDISMODULE_GET_API(SetExpire); + REDISMODULE_GET_API(ZsetAdd); + REDISMODULE_GET_API(ZsetIncrby); + REDISMODULE_GET_API(ZsetScore); + REDISMODULE_GET_API(ZsetRem); + REDISMODULE_GET_API(ZsetRangeStop); + REDISMODULE_GET_API(ZsetFirstInScoreRange); + REDISMODULE_GET_API(ZsetLastInScoreRange); + REDISMODULE_GET_API(ZsetFirstInLexRange); + REDISMODULE_GET_API(ZsetLastInLexRange); + REDISMODULE_GET_API(ZsetRangeCurrentElement); + REDISMODULE_GET_API(ZsetRangeNext); + REDISMODULE_GET_API(ZsetRangePrev); + REDISMODULE_GET_API(ZsetRangeEndReached); + REDISMODULE_GET_API(HashSet); + REDISMODULE_GET_API(HashGet); + REDISMODULE_GET_API(IsKeysPositionRequest); + REDISMODULE_GET_API(KeyAtPos); + REDISMODULE_GET_API(GetClientId); + REDISMODULE_GET_API(PoolAlloc); + REDISMODULE_GET_API(CreateDataType); + REDISMODULE_GET_API(ModuleTypeSetValue); + REDISMODULE_GET_API(ModuleTypeGetType); + REDISMODULE_GET_API(ModuleTypeGetValue); + REDISMODULE_GET_API(SaveUnsigned); + REDISMODULE_GET_API(LoadUnsigned); + REDISMODULE_GET_API(SaveSigned); + REDISMODULE_GET_API(LoadSigned); + REDISMODULE_GET_API(SaveString); + REDISMODULE_GET_API(SaveStringBuffer); + REDISMODULE_GET_API(LoadString); + REDISMODULE_GET_API(LoadStringBuffer); + REDISMODULE_GET_API(SaveDouble); + REDISMODULE_GET_API(LoadDouble); + REDISMODULE_GET_API(SaveFloat); + REDISMODULE_GET_API(LoadFloat); + REDISMODULE_GET_API(EmitAOF); + REDISMODULE_GET_API(Log); + REDISMODULE_GET_API(LogIOError); + REDISMODULE_GET_API(StringAppendBuffer); + REDISMODULE_GET_API(RetainString); + REDISMODULE_GET_API(StringCompare); + REDISMODULE_GET_API(GetContextFromIO); + REDISMODULE_GET_API(BlockClient); + REDISMODULE_GET_API(UnblockClient); + REDISMODULE_GET_API(IsBlockedReplyRequest); + REDISMODULE_GET_API(IsBlockedTimeoutRequest); + REDISMODULE_GET_API(GetBlockedClientPrivateData); + REDISMODULE_GET_API(AbortBlock); + REDISMODULE_GET_API(Milliseconds); - RedisModule_SetModuleAttribs(ctx,name,ver,apiver); - return REDISMODULE_OK; + RedisModule_SetModuleAttribs(ctx, name, ver, apiver); + return REDISMODULE_OK; } #else diff --git a/src/ray/gcs/tables.cc b/src/ray/gcs/tables.cc index e20384a04..33f161558 100644 --- a/src/ray/gcs/tables.cc +++ b/src/ray/gcs/tables.cc @@ -674,8 +674,8 @@ void ClientTable::HandleNotification(AsyncGcsClient *client, void ClientTable::HandleConnected(AsyncGcsClient *client, const ClientTableDataT &data) { auto connected_client_id = ClientID::FromBinary(data.client_id); - RAY_CHECK(client_id_ == connected_client_id) << connected_client_id << " " - << client_id_; + RAY_CHECK(client_id_ == connected_client_id) + << connected_client_id << " " << client_id_; } const ClientID &ClientTable::GetLocalClientId() const { return client_id_; } @@ -704,8 +704,8 @@ Status ClientTable::Connect(const ClientTableDataT &local_client) { // Callback for a notification from the client table. auto notification_callback = [this]( - AsyncGcsClient *client, const UniqueID &log_key, - const std::vector ¬ifications) { + AsyncGcsClient *client, const UniqueID &log_key, + const std::vector ¬ifications) { RAY_CHECK(log_key == client_log_key_); std::unordered_map connected_nodes; std::unordered_map disconnected_nodes; @@ -797,8 +797,8 @@ Status ActorCheckpointIdTable::AddCheckpointId(const DriverID &driver_id, const ActorID &actor_id, const ActorCheckpointID &checkpoint_id) { auto lookup_callback = [this, checkpoint_id, driver_id, actor_id]( - ray::gcs::AsyncGcsClient *client, const UniqueID &id, - const ActorCheckpointIdDataT &data) { + ray::gcs::AsyncGcsClient *client, const UniqueID &id, + const ActorCheckpointIdDataT &data) { std::shared_ptr copy = std::make_shared(data); copy->timestamps.push_back(current_sys_time_ms()); @@ -817,7 +817,7 @@ Status ActorCheckpointIdTable::AddCheckpointId(const DriverID &driver_id, RAY_CHECK_OK(Add(driver_id, actor_id, copy, nullptr)); }; auto failure_callback = [this, checkpoint_id, driver_id, actor_id]( - ray::gcs::AsyncGcsClient *client, const UniqueID &id) { + ray::gcs::AsyncGcsClient *client, const UniqueID &id) { std::shared_ptr data = std::make_shared(); data->actor_id = id.Binary(); diff --git a/src/ray/object_manager/object_directory.cc b/src/ray/object_manager/object_directory.cc index d2496dceb..5b6794a50 100644 --- a/src/ray/object_manager/object_directory.cc +++ b/src/ray/object_manager/object_directory.cc @@ -39,36 +39,36 @@ void UpdateObjectLocations(const GcsChangeMode change_mode, } // namespace void ObjectDirectory::RegisterBackend() { - auto object_notification_callback = [this]( - gcs::AsyncGcsClient *client, const ObjectID &object_id, - const GcsChangeMode change_mode, - const std::vector &location_updates) { - // Objects are added to this map in SubscribeObjectLocations. - auto it = listeners_.find(object_id); - // Do nothing for objects we are not listening for. - if (it == listeners_.end()) { - return; - } + auto object_notification_callback = + [this](gcs::AsyncGcsClient *client, const ObjectID &object_id, + const GcsChangeMode change_mode, + const std::vector &location_updates) { + // Objects are added to this map in SubscribeObjectLocations. + auto it = listeners_.find(object_id); + // Do nothing for objects we are not listening for. + if (it == listeners_.end()) { + return; + } - // Once this flag is set to true, it should never go back to false. - it->second.subscribed = true; + // Once this flag is set to true, it should never go back to false. + it->second.subscribed = true; - // Update entries for this object. - UpdateObjectLocations(change_mode, location_updates, gcs_client_->client_table(), - &it->second.current_object_locations); - // Copy the callbacks so that the callbacks can unsubscribe without interrupting - // looping over the callbacks. - auto callbacks = it->second.callbacks; - // Call all callbacks associated with the object id locations we have - // received. This notifies the client even if the list of locations is - // empty, since this may indicate that the objects have been evicted from - // all nodes. - for (const auto &callback_pair : callbacks) { - // It is safe to call the callback directly since this is already running - // in the subscription callback stack. - callback_pair.second(object_id, it->second.current_object_locations); - } - }; + // Update entries for this object. + UpdateObjectLocations(change_mode, location_updates, gcs_client_->client_table(), + &it->second.current_object_locations); + // Copy the callbacks so that the callbacks can unsubscribe without interrupting + // looping over the callbacks. + auto callbacks = it->second.callbacks; + // Call all callbacks associated with the object id locations we have + // received. This notifies the client even if the list of locations is + // empty, since this may indicate that the objects have been evicted from + // all nodes. + for (const auto &callback_pair : callbacks) { + // It is safe to call the callback directly since this is already running + // in the subscription callback stack. + callback_pair.second(object_id, it->second.current_object_locations); + } + }; RAY_CHECK_OK(gcs_client_->object_table().Subscribe( DriverID::Nil(), gcs_client_->client_table().GetLocalClientId(), object_notification_callback, nullptr)); diff --git a/src/ray/object_manager/test/object_manager_stress_test.cc b/src/ray/object_manager/test/object_manager_stress_test.cc index f11696051..55aa59124 100644 --- a/src/ray/object_manager/test/object_manager_stress_test.cc +++ b/src/ray/object_manager/test/object_manager_stress_test.cc @@ -70,11 +70,11 @@ class MockServer { void HandleAcceptObjectManager(const boost::system::error_code &error) { ClientHandler client_handler = [this](TcpClientConnection &client) { object_manager_.ProcessNewClient(client); }; - MessageHandler message_handler = [this]( - std::shared_ptr client, int64_t message_type, - const uint8_t *message) { - object_manager_.ProcessClientMessage(client, message_type, message); - }; + MessageHandler message_handler = + [this](std::shared_ptr client, int64_t message_type, + const uint8_t *message) { + object_manager_.ProcessClientMessage(client, message_type, message); + }; // Accept a new local client and dispatch it to the node manager. auto new_connection = TcpClientConnection::Create( client_handler, message_handler, std::move(object_manager_socket_), @@ -240,16 +240,17 @@ class StressTestObjectManager : public TestObjectManagerBase { void WaitConnections() { client_id_1 = gcs_client_1->client_table().GetLocalClientId(); client_id_2 = gcs_client_2->client_table().GetLocalClientId(); - gcs_client_1->client_table().RegisterClientAddedCallback([this]( - gcs::AsyncGcsClient *client, const ClientID &id, const ClientTableDataT &data) { - ClientID parsed_id = ClientID::FromBinary(data.client_id); - if (parsed_id == client_id_1 || parsed_id == client_id_2) { - num_connected_clients += 1; - } - if (num_connected_clients == 2) { - StartTests(); - } - }); + gcs_client_1->client_table().RegisterClientAddedCallback( + [this](gcs::AsyncGcsClient *client, const ClientID &id, + const ClientTableDataT &data) { + ClientID parsed_id = ClientID::FromBinary(data.client_id); + if (parsed_id == client_id_1 || parsed_id == client_id_2) { + num_connected_clients += 1; + } + if (num_connected_clients == 2) { + StartTests(); + } + }); } void StartTests() { diff --git a/src/ray/object_manager/test/object_manager_test.cc b/src/ray/object_manager/test/object_manager_test.cc index 012c30693..ee6c78d8e 100644 --- a/src/ray/object_manager/test/object_manager_test.cc +++ b/src/ray/object_manager/test/object_manager_test.cc @@ -64,11 +64,11 @@ class MockServer { void HandleAcceptObjectManager(const boost::system::error_code &error) { ClientHandler client_handler = [this](TcpClientConnection &client) { object_manager_.ProcessNewClient(client); }; - MessageHandler message_handler = [this]( - std::shared_ptr client, int64_t message_type, - const uint8_t *message) { - object_manager_.ProcessClientMessage(client, message_type, message); - }; + MessageHandler message_handler = + [this](std::shared_ptr client, int64_t message_type, + const uint8_t *message) { + object_manager_.ProcessClientMessage(client, message_type, message); + }; // Accept a new local client and dispatch it to the node manager. auto new_connection = TcpClientConnection::Create( client_handler, message_handler, std::move(object_manager_socket_), @@ -219,16 +219,17 @@ class TestObjectManager : public TestObjectManagerBase { void WaitConnections() { client_id_1 = gcs_client_1->client_table().GetLocalClientId(); client_id_2 = gcs_client_2->client_table().GetLocalClientId(); - gcs_client_1->client_table().RegisterClientAddedCallback([this]( - gcs::AsyncGcsClient *client, const ClientID &id, const ClientTableDataT &data) { - ClientID parsed_id = ClientID::FromBinary(data.client_id); - if (parsed_id == client_id_1 || parsed_id == client_id_2) { - num_connected_clients += 1; - } - if (num_connected_clients == 2) { - StartTests(); - } - }); + gcs_client_1->client_table().RegisterClientAddedCallback( + [this](gcs::AsyncGcsClient *client, const ClientID &id, + const ClientTableDataT &data) { + ClientID parsed_id = ClientID::FromBinary(data.client_id); + if (parsed_id == client_id_1 || parsed_id == client_id_2) { + num_connected_clients += 1; + } + if (num_connected_clients == 2) { + StartTests(); + } + }); } void StartTests() { @@ -291,9 +292,10 @@ class TestObjectManager : public TestObjectManagerBase { UniqueID sub_id = ray::UniqueID::FromRandom(); RAY_CHECK_OK(server1->object_manager_.object_directory_->SubscribeObjectLocations( - sub_id, object_1, [this, sub_id, object_1, object_2]( - const ray::ObjectID &object_id, - const std::unordered_set &clients) { + sub_id, object_1, + [this, sub_id, object_1, object_2]( + const ray::ObjectID &object_id, + const std::unordered_set &clients) { if (!clients.empty()) { TestWaitWhileSubscribed(sub_id, object_1, object_2); } diff --git a/src/ray/raylet/client_connection_test.cc b/src/ray/raylet/client_connection_test.cc index 952088bb2..d6359ae28 100644 --- a/src/ray/raylet/client_connection_test.cc +++ b/src/ray/raylet/client_connection_test.cc @@ -73,9 +73,9 @@ TEST_F(ClientConnectionTest, SimpleAsyncWrite) { ClientHandler client_handler = [](LocalClientConnection &client) {}; - MessageHandler noop_handler = []( - std::shared_ptr client, int64_t message_type, - const uint8_t *message) {}; + MessageHandler noop_handler = + [](std::shared_ptr client, int64_t message_type, + const uint8_t *message) {}; std::shared_ptr reader = NULL; @@ -120,9 +120,9 @@ TEST_F(ClientConnectionTest, SimpleAsyncError) { ClientHandler client_handler = [](LocalClientConnection &client) {}; - MessageHandler noop_handler = []( - std::shared_ptr client, int64_t message_type, - const uint8_t *message) {}; + MessageHandler noop_handler = + [](std::shared_ptr client, int64_t message_type, + const uint8_t *message) {}; auto writer = LocalClientConnection::Create( client_handler, noop_handler, std::move(in_), "writer", {}, error_message_type_); @@ -142,9 +142,9 @@ TEST_F(ClientConnectionTest, CallbackWithSharedRefDoesNotLeakConnection) { ClientHandler client_handler = [](LocalClientConnection &client) {}; - MessageHandler noop_handler = []( - std::shared_ptr client, int64_t message_type, - const uint8_t *message) {}; + MessageHandler noop_handler = + [](std::shared_ptr client, int64_t message_type, + const uint8_t *message) {}; auto writer = LocalClientConnection::Create( client_handler, noop_handler, std::move(in_), "writer", {}, error_message_type_); diff --git a/src/ray/raylet/lib/java/org_ray_runtime_raylet_RayletClientImpl.cc b/src/ray/raylet/lib/java/org_ray_runtime_raylet_RayletClientImpl.cc index 2afcba18c..319d29d4a 100644 --- a/src/ray/raylet/lib/java/org_ray_runtime_raylet_RayletClientImpl.cc +++ b/src/ray/raylet/lib/java/org_ray_runtime_raylet_RayletClientImpl.cc @@ -307,15 +307,15 @@ Java_org_ray_runtime_raylet_RayletClientImpl_nativeNotifyActorResumedFromCheckpo * Method: nativeSetResource * Signature: (JLjava/lang/String;D[B)V */ -JNIEXPORT void JNICALL -Java_org_ray_runtime_raylet_RayletClientImpl_nativeSetResource(JNIEnv *env, jclass, - jlong client, jstring resourceName, jdouble capacity, jbyteArray nodeId) { +JNIEXPORT void JNICALL Java_org_ray_runtime_raylet_RayletClientImpl_nativeSetResource( + JNIEnv *env, jclass, jlong client, jstring resourceName, jdouble capacity, + jbyteArray nodeId) { auto raylet_client = reinterpret_cast(client); UniqueIdFromJByteArray node_id(env, nodeId); const char *native_resource_name = env->GetStringUTFChars(resourceName, JNI_FALSE); - auto status = raylet_client->SetResource(native_resource_name, - static_cast(capacity), node_id.GetId()); + auto status = raylet_client->SetResource( + native_resource_name, static_cast(capacity), node_id.GetId()); env->ReleaseStringUTFChars(resourceName, native_resource_name); ThrowRayExceptionIfNotOK(env, status); } diff --git a/src/ray/raylet/lineage_cache.cc b/src/ray/raylet/lineage_cache.cc index fcae3f8e0..32dddada5 100644 --- a/src/ray/raylet/lineage_cache.cc +++ b/src/ray/raylet/lineage_cache.cc @@ -290,10 +290,9 @@ void LineageCache::FlushTask(const TaskID &task_id) { RAY_CHECK(entry); RAY_CHECK(entry->GetStatus() < GcsStatus::COMMITTING); - gcs::raylet::TaskTable::WriteCallback task_callback = [this]( - ray::gcs::AsyncGcsClient *client, const TaskID &id, const protocol::TaskT &data) { - HandleEntryCommitted(id); - }; + gcs::raylet::TaskTable::WriteCallback task_callback = + [this](ray::gcs::AsyncGcsClient *client, const TaskID &id, + const protocol::TaskT &data) { HandleEntryCommitted(id); }; auto task = lineage_.GetEntry(task_id); // TODO(swang): Make this better... flatbuffers::FlatBufferBuilder fbb; diff --git a/src/ray/raylet/main.cc b/src/ray/raylet/main.cc index c6e581cec..eca282a53 100644 --- a/src/ray/raylet/main.cc +++ b/src/ray/raylet/main.cc @@ -174,7 +174,7 @@ int main(int argc, char *argv[]) { // instead of returning immediately. // We should stop the service and remove the local socket file. auto handler = [&main_service, &raylet_socket_name, &server, &gcs_client]( - const boost::system::error_code &error, int signal_number) { + const boost::system::error_code &error, int signal_number) { auto shutdown_callback = [&server, &main_service]() { server.reset(); main_service.stop(); diff --git a/src/ray/raylet/monitor.cc b/src/ray/raylet/monitor.cc index a87257cad..62ecb00b8 100644 --- a/src/ray/raylet/monitor.cc +++ b/src/ray/raylet/monitor.cc @@ -48,8 +48,8 @@ void Monitor::Tick() { auto client_id = it->first; RAY_LOG(WARNING) << "Client timed out: " << client_id; auto lookup_callback = [this, client_id]( - gcs::AsyncGcsClient *client, const ClientID &id, - const std::vector &all_data) { + gcs::AsyncGcsClient *client, const ClientID &id, + const std::vector &all_data) { bool marked = false; for (const auto &data : all_data) { if (client_id.Binary() == data.client_id && diff --git a/src/ray/raylet/node_manager.cc b/src/ray/raylet/node_manager.cc index 75377a13c..671a7a798 100644 --- a/src/ray/raylet/node_manager.cc +++ b/src/ray/raylet/node_manager.cc @@ -179,45 +179,42 @@ ray::Status NodeManager::RegisterGcs() { }; gcs_client_->client_table().RegisterClientAddedCallback(node_manager_client_added); // Register a callback on the client table for removed clients. - auto node_manager_client_removed = [this]( - gcs::AsyncGcsClient *client, const UniqueID &id, const ClientTableDataT &data) { - ClientRemoved(data); - }; + auto node_manager_client_removed = + [this](gcs::AsyncGcsClient *client, const UniqueID &id, + const ClientTableDataT &data) { ClientRemoved(data); }; gcs_client_->client_table().RegisterClientRemovedCallback(node_manager_client_removed); // Register a callback on the client table for resource create/update requests - auto node_manager_resource_createupdated = [this]( - gcs::AsyncGcsClient *client, const UniqueID &id, const ClientTableDataT &data) { - ResourceCreateUpdated(data); - }; + auto node_manager_resource_createupdated = + [this](gcs::AsyncGcsClient *client, const UniqueID &id, + const ClientTableDataT &data) { ResourceCreateUpdated(data); }; gcs_client_->client_table().RegisterResourceCreateUpdatedCallback( node_manager_resource_createupdated); // Register a callback on the client table for resource delete requests - auto node_manager_resource_deleted = [this]( - gcs::AsyncGcsClient *client, const UniqueID &id, const ClientTableDataT &data) { - ResourceDeleted(data); - }; + auto node_manager_resource_deleted = + [this](gcs::AsyncGcsClient *client, const UniqueID &id, + const ClientTableDataT &data) { ResourceDeleted(data); }; gcs_client_->client_table().RegisterResourceDeletedCallback( node_manager_resource_deleted); // Subscribe to heartbeat batches from the monitor. - const auto &heartbeat_batch_added = [this]( - gcs::AsyncGcsClient *client, const ClientID &id, - const HeartbeatBatchTableDataT &heartbeat_batch) { - HeartbeatBatchAdded(heartbeat_batch); - }; + const auto &heartbeat_batch_added = + [this](gcs::AsyncGcsClient *client, const ClientID &id, + const HeartbeatBatchTableDataT &heartbeat_batch) { + HeartbeatBatchAdded(heartbeat_batch); + }; RAY_RETURN_NOT_OK(gcs_client_->heartbeat_batch_table().Subscribe( DriverID::Nil(), ClientID::Nil(), heartbeat_batch_added, /*subscribe_callback=*/nullptr, /*done_callback=*/nullptr)); // Subscribe to driver table updates. - const auto driver_table_handler = [this]( - gcs::AsyncGcsClient *client, const DriverID &client_id, - const std::vector &driver_data) { - HandleDriverTableUpdate(client_id, driver_data); - }; + const auto driver_table_handler = + [this](gcs::AsyncGcsClient *client, const DriverID &client_id, + const std::vector &driver_data) { + HandleDriverTableUpdate(client_id, driver_data); + }; RAY_RETURN_NOT_OK(gcs_client_->driver_table().Subscribe( DriverID::Nil(), ClientID::Nil(), driver_table_handler, nullptr)); @@ -2202,53 +2199,55 @@ void NodeManager::ForwardTaskOrResubmit(const Task &task, /// TODO(rkn): Should we check that the node manager is remote and not local? /// TODO(rkn): Should we check if the remote node manager is known to be dead? // Attempt to forward the task. - ForwardTask(task, node_manager_id, [this, node_manager_id](ray::Status error, - const Task &task) { - const TaskID task_id = task.GetTaskSpecification().TaskId(); - RAY_LOG(INFO) << "Failed to forward task " << task_id << " to node manager " - << node_manager_id; + ForwardTask( + task, node_manager_id, + [this, node_manager_id](ray::Status error, const Task &task) { + const TaskID task_id = task.GetTaskSpecification().TaskId(); + RAY_LOG(INFO) << "Failed to forward task " << task_id << " to node manager " + << node_manager_id; - // Mark the failed task as pending to let other raylets know that we still - // have the task. TaskDependencyManager::TaskPending() is assumed to be - // idempotent. - task_dependency_manager_.TaskPending(task); + // Mark the failed task as pending to let other raylets know that we still + // have the task. TaskDependencyManager::TaskPending() is assumed to be + // idempotent. + task_dependency_manager_.TaskPending(task); - // Actor tasks can only be executed at the actor's location, so they are - // retried after a timeout. All other tasks that fail to be forwarded are - // deemed to be placeable again. - if (task.GetTaskSpecification().IsActorTask()) { - // The task is for an actor on another node. Create a timer to resubmit - // the task in a little bit. TODO(rkn): Really this should be a - // unique_ptr instead of a shared_ptr. However, it's a little harder to - // move unique_ptrs into lambdas. - auto retry_timer = std::make_shared(io_service_); - auto retry_duration = boost::posix_time::milliseconds( - RayConfig::instance().node_manager_forward_task_retry_timeout_milliseconds()); - retry_timer->expires_from_now(retry_duration); - retry_timer->async_wait( - [this, task_id, retry_timer](const boost::system::error_code &error) { - // Timer killing will receive the boost::asio::error::operation_aborted, - // we only handle the timeout event. - RAY_CHECK(!error); - RAY_LOG(INFO) << "Resubmitting task " << task_id - << " because ForwardTask failed."; - // Remove the RESUBMITTED task from the SWAP queue. - TaskState state; - const auto task = local_queues_.RemoveTask(task_id, &state); - RAY_CHECK(state == TaskState::SWAP); - // Submit the task again. - SubmitTask(task, Lineage()); - }); - // Temporarily move the RESUBMITTED task to the SWAP queue while the - // timer is active. - local_queues_.QueueTasks({task}, TaskState::SWAP); - } else { - // The task is not for an actor and may therefore be placed on another - // node immediately. Send it to the scheduling policy to be placed again. - local_queues_.QueueTasks({task}, TaskState::PLACEABLE); - ScheduleTasks(cluster_resource_map_); - } - }); + // Actor tasks can only be executed at the actor's location, so they are + // retried after a timeout. All other tasks that fail to be forwarded are + // deemed to be placeable again. + if (task.GetTaskSpecification().IsActorTask()) { + // The task is for an actor on another node. Create a timer to resubmit + // the task in a little bit. TODO(rkn): Really this should be a + // unique_ptr instead of a shared_ptr. However, it's a little harder to + // move unique_ptrs into lambdas. + auto retry_timer = std::make_shared(io_service_); + auto retry_duration = boost::posix_time::milliseconds( + RayConfig::instance() + .node_manager_forward_task_retry_timeout_milliseconds()); + retry_timer->expires_from_now(retry_duration); + retry_timer->async_wait( + [this, task_id, retry_timer](const boost::system::error_code &error) { + // Timer killing will receive the boost::asio::error::operation_aborted, + // we only handle the timeout event. + RAY_CHECK(!error); + RAY_LOG(INFO) << "Resubmitting task " << task_id + << " because ForwardTask failed."; + // Remove the RESUBMITTED task from the SWAP queue. + TaskState state; + const auto task = local_queues_.RemoveTask(task_id, &state); + RAY_CHECK(state == TaskState::SWAP); + // Submit the task again. + SubmitTask(task, Lineage()); + }); + // Temporarily move the RESUBMITTED task to the SWAP queue while the + // timer is active. + local_queues_.QueueTasks({task}, TaskState::SWAP); + } else { + // The task is not for an actor and may therefore be placed on another + // node immediately. Send it to the scheduling policy to be placed again. + local_queues_.QueueTasks({task}, TaskState::PLACEABLE); + ScheduleTasks(cluster_resource_map_); + } + }); } void NodeManager::ForwardTask( diff --git a/src/ray/raylet/object_manager_integration_test.cc b/src/ray/raylet/object_manager_integration_test.cc index 1b043ca58..0f411e8c5 100644 --- a/src/ray/raylet/object_manager_integration_test.cc +++ b/src/ray/raylet/object_manager_integration_test.cc @@ -136,16 +136,17 @@ class TestObjectManagerIntegration : public TestObjectManagerBase { void WaitConnections() { client_id_1 = gcs_client_1->client_table().GetLocalClientId(); client_id_2 = gcs_client_2->client_table().GetLocalClientId(); - gcs_client_1->client_table().RegisterClientAddedCallback([this]( - gcs::AsyncGcsClient *client, const ClientID &id, const ClientTableDataT &data) { - ClientID parsed_id = ClientID::FromBinary(data.client_id); - if (parsed_id == client_id_1 || parsed_id == client_id_2) { - num_connected_clients += 1; - } - if (num_connected_clients == 2) { - StartTests(); - } - }); + gcs_client_1->client_table().RegisterClientAddedCallback( + [this](gcs::AsyncGcsClient *client, const ClientID &id, + const ClientTableDataT &data) { + ClientID parsed_id = ClientID::FromBinary(data.client_id); + if (parsed_id == client_id_1 || parsed_id == client_id_2) { + num_connected_clients += 1; + } + if (num_connected_clients == 2) { + StartTests(); + } + }); } void StartTests() { diff --git a/src/ray/raylet/raylet.cc b/src/ray/raylet/raylet.cc index dd9e5fac3..80630d372 100644 --- a/src/ray/raylet/raylet.cc +++ b/src/ray/raylet/raylet.cc @@ -128,13 +128,15 @@ void Raylet::DoAcceptNodeManager() { void Raylet::HandleAcceptNodeManager(const boost::system::error_code &error) { if (!error) { - ClientHandler client_handler = [this]( - TcpClientConnection &client) { node_manager_.ProcessNewNodeManager(client); }; - MessageHandler message_handler = [this]( - std::shared_ptr client, int64_t message_type, - const uint8_t *message) { - node_manager_.ProcessNodeManagerMessage(*client, message_type, message); - }; + ClientHandler client_handler = + [this](TcpClientConnection &client) { + node_manager_.ProcessNewNodeManager(client); + }; + MessageHandler message_handler = + [this](std::shared_ptr client, int64_t message_type, + const uint8_t *message) { + node_manager_.ProcessNodeManagerMessage(*client, message_type, message); + }; // Accept a new TCP client and dispatch it to the node manager. auto new_connection = TcpClientConnection::Create( client_handler, message_handler, std::move(node_manager_socket_), "node manager", @@ -154,11 +156,11 @@ void Raylet::DoAcceptObjectManager() { void Raylet::HandleAcceptObjectManager(const boost::system::error_code &error) { ClientHandler client_handler = [this](TcpClientConnection &client) { object_manager_.ProcessNewClient(client); }; - MessageHandler message_handler = [this]( - std::shared_ptr client, int64_t message_type, - const uint8_t *message) { - object_manager_.ProcessClientMessage(client, message_type, message); - }; + MessageHandler message_handler = + [this](std::shared_ptr client, int64_t message_type, + const uint8_t *message) { + object_manager_.ProcessClientMessage(client, message_type, message); + }; // Accept a new TCP client and dispatch it to the node manager. auto new_connection = TcpClientConnection::Create( client_handler, message_handler, std::move(object_manager_socket_), @@ -177,11 +179,11 @@ void Raylet::HandleAccept(const boost::system::error_code &error) { // TODO: typedef these handlers. ClientHandler client_handler = [this](LocalClientConnection &client) { node_manager_.ProcessNewClient(client); }; - MessageHandler message_handler = [this]( - std::shared_ptr client, int64_t message_type, - const uint8_t *message) { - node_manager_.ProcessClientMessage(client, message_type, message); - }; + MessageHandler message_handler = + [this](std::shared_ptr client, int64_t message_type, + const uint8_t *message) { + node_manager_.ProcessClientMessage(client, message_type, message); + }; // Accept a new local client and dispatch it to the node manager. auto new_connection = LocalClientConnection::Create( client_handler, message_handler, std::move(socket_), "worker", diff --git a/src/ray/raylet/scheduling_queue.cc b/src/ray/raylet/scheduling_queue.cc index 85295e403..73f0e2ef8 100644 --- a/src/ray/raylet/scheduling_queue.cc +++ b/src/ray/raylet/scheduling_queue.cc @@ -233,8 +233,13 @@ std::vector SchedulingQueue::RemoveTasks(std::unordered_set &task_ std::vector removed_tasks; // Try to find the tasks to remove from the queues. for (const auto &task_state : { - TaskState::PLACEABLE, TaskState::WAITING, TaskState::READY, TaskState::RUNNING, - TaskState::INFEASIBLE, TaskState::WAITING_FOR_ACTOR_CREATION, TaskState::SWAP, + TaskState::PLACEABLE, + TaskState::WAITING, + TaskState::READY, + TaskState::RUNNING, + TaskState::INFEASIBLE, + TaskState::WAITING_FOR_ACTOR_CREATION, + TaskState::SWAP, }) { RemoveTasksFromQueue(task_state, task_ids, &removed_tasks); } @@ -248,8 +253,13 @@ Task SchedulingQueue::RemoveTask(const TaskID &task_id, TaskState *removed_task_ std::unordered_set task_id_set = {task_id}; // Try to find the task to remove in the queues. for (const auto &task_state : { - TaskState::PLACEABLE, TaskState::WAITING, TaskState::READY, TaskState::RUNNING, - TaskState::INFEASIBLE, TaskState::WAITING_FOR_ACTOR_CREATION, TaskState::SWAP, + TaskState::PLACEABLE, + TaskState::WAITING, + TaskState::READY, + TaskState::RUNNING, + TaskState::INFEASIBLE, + TaskState::WAITING_FOR_ACTOR_CREATION, + TaskState::SWAP, }) { RemoveTasksFromQueue(task_state, task_id_set, &removed_tasks); if (task_id_set.empty()) { diff --git a/src/ray/raylet/scheduling_queue.h b/src/ray/raylet/scheduling_queue.h index 4fd07e5ca..465f2a434 100644 --- a/src/ray/raylet/scheduling_queue.h +++ b/src/ray/raylet/scheduling_queue.h @@ -149,9 +149,13 @@ class SchedulingQueue { /// Create a scheduling queue. SchedulingQueue() : ready_queue_(std::make_shared()) { for (const auto &task_state : { - TaskState::PLACEABLE, TaskState::WAITING, TaskState::READY, - TaskState::RUNNING, TaskState::INFEASIBLE, - TaskState::WAITING_FOR_ACTOR_CREATION, TaskState::SWAP, + TaskState::PLACEABLE, + TaskState::WAITING, + TaskState::READY, + TaskState::RUNNING, + TaskState::INFEASIBLE, + TaskState::WAITING_FOR_ACTOR_CREATION, + TaskState::SWAP, }) { if (task_state == TaskState::READY) { task_queues_[static_cast(task_state)] = ready_queue_; diff --git a/src/ray/raylet/scheduling_resources.cc b/src/ray/raylet/scheduling_resources.cc index 923e6aad9..c80282601 100644 --- a/src/ray/raylet/scheduling_resources.cc +++ b/src/ray/raylet/scheduling_resources.cc @@ -15,8 +15,8 @@ FractionalResourceQuantity::FractionalResourceQuantity(double resource_quantity) // We check for nonnegativeity due to the implicit conversion to // FractionalResourceQuantity from ints/doubles when we do logical // comparisons. - RAY_CHECK(resource_quantity >= 0) << "Resource capacity, " << resource_quantity - << ", should be nonnegative."; + RAY_CHECK(resource_quantity >= 0) + << "Resource capacity, " << resource_quantity << ", should be nonnegative."; resource_quantity_ = static_cast(resource_quantity * kResourceConversionFactor); diff --git a/src/ray/util/logging.cc b/src/ray/util/logging.cc index 97c871d8c..1e2a95408 100644 --- a/src/ray/util/logging.cc +++ b/src/ray/util/logging.cc @@ -186,8 +186,7 @@ bool RayLog::IsLevelEnabled(RayLogLevel log_level) { RayLog::RayLog(const char *file_name, int line_number, RayLogLevel severity) // glog does not have DEBUG level, we can handle it using is_enabled_. - : logging_provider_(nullptr), - is_enabled_(severity >= severity_threshold_) { + : logging_provider_(nullptr), is_enabled_(severity >= severity_threshold_) { #ifdef RAY_USE_GLOG if (is_enabled_) { logging_provider_ = diff --git a/src/ray/util/logging.h b/src/ray/util/logging.h index d37ab9a73..39428eba9 100644 --- a/src/ray/util/logging.h +++ b/src/ray/util/logging.h @@ -16,19 +16,19 @@ enum class RayLogLevel { DEBUG = -1, INFO = 0, WARNING = 1, ERROR = 2, FATAL = 3 #define RAY_IGNORE_EXPR(expr) ((void)(expr)) -#define RAY_CHECK(condition) \ - (condition) ? RAY_IGNORE_EXPR(0) \ - : ::ray::Voidify() & \ - ::ray::RayLog(__FILE__, __LINE__, ray::RayLogLevel::FATAL) \ - << " Check failed: " #condition " " +#define RAY_CHECK(condition) \ + (condition) \ + ? RAY_IGNORE_EXPR(0) \ + : ::ray::Voidify() & ::ray::RayLog(__FILE__, __LINE__, ray::RayLogLevel::FATAL) \ + << " Check failed: " #condition " " #ifdef NDEBUG -#define RAY_DCHECK(condition) \ - (condition) ? RAY_IGNORE_EXPR(0) \ - : ::ray::Voidify() & \ - ::ray::RayLog(__FILE__, __LINE__, ray::RayLogLevel::ERROR) \ - << " Debug check failed: " #condition " " +#define RAY_DCHECK(condition) \ + (condition) \ + ? RAY_IGNORE_EXPR(0) \ + : ::ray::Voidify() & ::ray::RayLog(__FILE__, __LINE__, ray::RayLogLevel::ERROR) \ + << " Debug check failed: " #condition " " #else #define RAY_DCHECK(condition) RAY_CHECK(condition) diff --git a/src/ray/util/macros.h b/src/ray/util/macros.h index dbf85fe39..f105c4bd2 100644 --- a/src/ray/util/macros.h +++ b/src/ray/util/macros.h @@ -8,7 +8,7 @@ void operator=(const TypeName &) = delete #endif -#define RAY_UNUSED(x) (void) x +#define RAY_UNUSED(x) (void)x // // GCC can be told that a certain branch is not likely to be taken (for