From d61d92afc7321525b77cfd094b28d40ab14a292f Mon Sep 17 00:00:00 2001 From: "Siyuan (Ryans) Zhuang" Date: Thu, 16 Jul 2020 14:52:14 -0700 Subject: [PATCH] Cleanup Plasma Store (hash utilities) (#9524) --- src/ray/object_manager/plasma/client.cc | 123 ------------------ src/ray/object_manager/plasma/client.h | 8 -- .../test/object_manager_stress_test.cc | 17 --- 3 files changed, 148 deletions(-) diff --git a/src/ray/object_manager/plasma/client.cc b/src/ray/object_manager/plasma/client.cc index 30f7e9a61..8a2da049e 100644 --- a/src/ray/object_manager/plasma/client.cc +++ b/src/ray/object_manager/plasma/client.cc @@ -56,16 +56,6 @@ #include "ray/object_manager/plasma/plasma.h" #include "ray/object_manager/plasma/protocol.h" -// This macro is used to replace the "ARROW_CHECK_OK_PREPEND" macro. -#define RAY_ARROW_CHECK_OK_PREPEND(to_call, msg) \ - do { \ - ::arrow::Status _s = (to_call); \ - RAY_CHECK(_s.ok()) << (msg) << ": " << _s.ToString(); \ - } while (0) - -// This macro is used to replace the "ARROW_CHECK_OK" macro. -#define RAY_ARROW_CHECK_OK(s) RAY_ARROW_CHECK_OK_PREPEND(s, "Bad status") - #ifdef PLASMA_CUDA #include "arrow/gpu/cuda_api.h" @@ -75,12 +65,6 @@ using arrow::cuda::CudaContext; using arrow::cuda::CudaDeviceManager; #endif -#define XXH_INLINE_ALL 1 -#define XXH_NAMESPACE plasma_client_ -#include "arrow/vendored/xxhash.h" - -#define XXH64_DEFAULT_SEED 0 - namespace fb = plasma::flatbuf; namespace plasma { @@ -90,12 +74,6 @@ using fb::PlasmaError; using arrow::MutableBuffer; -typedef struct XXH64_state_s XXH64_state_t; - -// Number of threads used for hash computations. -constexpr int64_t kHashingConcurrency = 8; -constexpr int64_t kBytesInMB = 1 << 20; - // ---------------------------------------------------------------------- // GPU support @@ -273,8 +251,6 @@ class PlasmaClient::Impl : public std::enable_shared_from_this& object_ids); - Status Hash(const ObjectID& object_id, uint8_t* digest); - Status Subscribe(int* fd); Status GetNotification(int fd, ObjectID* object_id, int64_t* data_size, @@ -320,14 +296,6 @@ class PlasmaClient::Impl : public std::enable_shared_from_this store_conn_; /// Table of dlmalloc buffer files that have been memory mapped so far. This @@ -728,76 +696,6 @@ Status PlasmaClient::Impl::Contains(const ObjectID& object_id, bool* has_object) return Status::OK(); } -static void ComputeBlockHash(const unsigned char* data, int64_t nbytes, uint64_t* hash) { - XXH64_state_t hash_state; - XXH64_reset(&hash_state, XXH64_DEFAULT_SEED); - XXH64_update(&hash_state, data, nbytes); - *hash = XXH64_digest(&hash_state); -} - -bool PlasmaClient::Impl::ComputeObjectHashParallel(XXH64_state_t* hash_state, - const unsigned char* data, - int64_t nbytes) { - // Note that this function will likely be faster if the address of data is - // aligned on a 64-byte boundary. - auto pool = arrow::internal::GetCpuThreadPool(); - - const int num_threads = kHashingConcurrency; - uint64_t threadhash[num_threads + 1]; - const uint64_t data_address = reinterpret_cast(data); - const uint64_t num_blocks = nbytes / kBlockSize; - const uint64_t chunk_size = (num_blocks / num_threads) * kBlockSize; - const uint64_t right_address = data_address + chunk_size * num_threads; - const uint64_t suffix = (data_address + nbytes) - right_address; - // Now the data layout is | k * num_threads * block_size | suffix | == - // | num_threads * chunk_size | suffix |, where chunk_size = k * block_size. - // Each thread gets a "chunk" of k blocks, except the suffix thread. - - std::vector> futures; - for (int i = 0; i < num_threads; i++) { - futures.push_back(*pool->Submit( - ComputeBlockHash, reinterpret_cast(data_address) + i * chunk_size, - chunk_size, &threadhash[i])); - } - ComputeBlockHash(reinterpret_cast(right_address), suffix, - &threadhash[num_threads]); - - for (auto& fut : futures) { - RAY_ARROW_CHECK_OK(fut.status()); - } - - XXH64_update(hash_state, reinterpret_cast(threadhash), - sizeof(threadhash)); - return true; -} - -uint64_t PlasmaClient::Impl::ComputeObjectHash(const ObjectBuffer& obj_buffer) { - if (obj_buffer.device_num != 0) { - // TODO(wap): Create cuda program to hash data on gpu. - return 0; - } - return ComputeObjectHashCPU(obj_buffer.data->data(), obj_buffer.data->size(), - obj_buffer.metadata->data(), obj_buffer.metadata->size()); -} - -uint64_t PlasmaClient::Impl::ComputeObjectHashCPU(const uint8_t* data, int64_t data_size, - const uint8_t* metadata, - int64_t metadata_size) { - RAY_DCHECK(metadata); - RAY_DCHECK(data); - XXH64_state_t hash_state; - XXH64_reset(&hash_state, XXH64_DEFAULT_SEED); - if (data_size >= kBytesInMB) { - ComputeObjectHashParallel(&hash_state, reinterpret_cast(data), - data_size); - } else { - XXH64_update(&hash_state, reinterpret_cast(data), data_size); - } - XXH64_update(&hash_state, reinterpret_cast(metadata), - metadata_size); - return XXH64_digest(&hash_state); -} - Status PlasmaClient::Impl::Seal(const ObjectID& object_id) { std::lock_guard guard(client_mutex_); @@ -913,23 +811,6 @@ Status PlasmaClient::Impl::Refresh(const std::vector& object_ids) { return ReadRefreshLRUReply(buffer.data(), buffer.size()); } -Status PlasmaClient::Impl::Hash(const ObjectID& object_id, uint8_t* digest) { - std::lock_guard guard(client_mutex_); - - // Get the plasma object data. We pass in a timeout of 0 to indicate that - // the operation should timeout immediately. - std::vector object_buffers; - RAY_RETURN_NOT_OK(Get({object_id}, 0, &object_buffers)); - // If the object was not retrieved, return false. - if (!object_buffers[0].data) { - return Status::ObjectNotFound("Object not found"); - } - // Compute the hash. - uint64_t hash = ComputeObjectHash(object_buffers[0]); - memcpy(digest, &hash, sizeof(hash)); - return Status::OK(); -} - Status PlasmaClient::Impl::Subscribe(int* fd) { std::lock_guard guard(client_mutex_); @@ -1143,10 +1024,6 @@ Status PlasmaClient::Refresh(const std::vector& object_ids) { return impl_->Refresh(object_ids); } -Status PlasmaClient::Hash(const ObjectID& object_id, uint8_t* digest) { - return impl_->Hash(object_id, digest); -} - Status PlasmaClient::Subscribe(int* fd) { return impl_->Subscribe(fd); } Status PlasmaClient::GetNotification(int fd, ObjectID* object_id, int64_t* data_size, diff --git a/src/ray/object_manager/plasma/client.h b/src/ray/object_manager/plasma/client.h index 6ef994380..34bc6d4ab 100644 --- a/src/ray/object_manager/plasma/client.h +++ b/src/ray/object_manager/plasma/client.h @@ -205,14 +205,6 @@ class RAY_EXPORT PlasmaClient { /// \return The return status. Status Refresh(const std::vector& object_ids); - /// Compute the hash of an object in the object store. - /// - /// \param object_id The ID of the object we want to hash. - /// \param digest A pointer at which to return the hash digest of the object. - /// The pointer must have at least kDigestSize bytes allocated. - /// \return The return status. - Status Hash(const ObjectID& object_id, uint8_t* digest); - /// Subscribe to notifications when objects are sealed in the object store. /// Whenever an object is sealed, a message will be written to the client /// socket that is returned by this method. 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 4b8376113..3062595b3 100644 --- a/src/ray/object_manager/test/object_manager_stress_test.cc +++ b/src/ray/object_manager/test/object_manager_stress_test.cc @@ -271,13 +271,6 @@ class StressTestObjectManager : public TestObjectManagerBase { return object_buffer; } - static unsigned char *GetDigest(plasma::PlasmaClient &client, ObjectID &object_id) { - const int64_t size = sizeof(uint64_t); - static unsigned char digest_1[size]; - RAY_CHECK_OK(client.Hash(object_id, &digest_1[0])); - return digest_1; - } - void CompareObjects(ObjectID &object_id_1, ObjectID &object_id_2) { plasma::ObjectBuffer object_buffer_1 = GetObject(client1, object_id_1); plasma::ObjectBuffer object_buffer_2 = GetObject(client2, object_id_2); @@ -292,15 +285,6 @@ class StressTestObjectManager : public TestObjectManagerBase { } } - void CompareHashes(ObjectID &object_id_1, ObjectID &object_id_2) { - const int64_t size = sizeof(uint64_t); - static unsigned char *digest_1 = GetDigest(client1, object_id_1); - static unsigned char *digest_2 = GetDigest(client2, object_id_2); - for (int i = -1; ++i < size;) { - ASSERT_TRUE(digest_1[i] == digest_2[i]); - } - } - void TransferTestComplete() { int64_t elapsed = current_time_ms() - start_time; RAY_LOG(INFO) << "TransferTestComplete: " @@ -316,7 +300,6 @@ class StressTestObjectManager : public TestObjectManagerBase { ObjectID object_id_2 = v2[i]; ObjectID object_id_1 = v1[std::distance(v1.begin(), std::find(v1.begin(), v1.end(), v2[i]))]; - CompareHashes(object_id_1, object_id_2); CompareObjects(object_id_1, object_id_2); }