wassname/ray
1
0
Fork 0
mirror of https://github.com/wassname/ray.git synced 2026-07-27 11:26:41 +08:00
Code Issues Packages Projects Releases Wiki Activity

Cleanup Plasma Store (hash utilities) (#9524)

Browse Source
This commit is contained in:
Siyuan (Ryans) Zhuang
2020-07-16 14:52:14 -07:00
committed by GitHub
parent 94fcd43593
commit d61d92afc7
3 changed files with 0 additions and 148 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ray/object_manager/plasma/client.cc
-123
View File
@@ -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<PlasmaClient::Imp
Status Refresh(const std::vector<ObjectID>& 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<PlasmaClient::Imp
void IncrementObjectCount(const ObjectID& object_id, PlasmaObject* object,
bool is_sealed);
bool ComputeObjectHashParallel(XXH64_state_t* hash_state, const unsigned char* data,
int64_t nbytes);
uint64_t ComputeObjectHash(const ObjectBuffer& obj_buffer);
uint64_t ComputeObjectHashCPU(const uint8_t* data, int64_t data_size,
const uint8_t* metadata, int64_t metadata_size);
/// File descriptor of the Unix domain socket that connects to the store.
std::shared_ptr<StoreConn> 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<uint64_t>(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<arrow::Future<void>> futures;
for (int i = 0; i < num_threads; i++) {
futures.push_back(*pool->Submit(
ComputeBlockHash, reinterpret_cast<uint8_t*>(data_address) + i * chunk_size,
chunk_size, &threadhash[i]));
}
ComputeBlockHash(reinterpret_cast<uint8_t*>(right_address), suffix,
&threadhash[num_threads]);
for (auto& fut : futures) {
RAY_ARROW_CHECK_OK(fut.status());
}
XXH64_update(hash_state, reinterpret_cast<unsigned char*>(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<const unsigned char*>(data),
data_size);
} else {
XXH64_update(&hash_state, reinterpret_cast<const unsigned char*>(data), data_size);
}
XXH64_update(&hash_state, reinterpret_cast<const unsigned char*>(metadata),
metadata_size);
return XXH64_digest(&hash_state);
}
Status PlasmaClient::Impl::Seal(const ObjectID& object_id) {
std::lock_guard<std::recursive_mutex> guard(client_mutex_);
@@ -913,23 +811,6 @@ Status PlasmaClient::Impl::Refresh(const std::vector<ObjectID>& object_ids) {
return ReadRefreshLRUReply(buffer.data(), buffer.size());
}
Status PlasmaClient::Impl::Hash(const ObjectID& object_id, uint8_t* digest) {
std::lock_guard<std::recursive_mutex> 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<ObjectBuffer> 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<std::recursive_mutex> guard(client_mutex_);
@@ -1143,10 +1024,6 @@ Status PlasmaClient::Refresh(const std::vector<ObjectID>& 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,
src/ray/object_manager/plasma/client.h
-8
View File
@@ -205,14 +205,6 @@ class RAY_EXPORT PlasmaClient {
/// \return The return status.
Status Refresh(const std::vector<ObjectID>& 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.
src/ray/object_manager/test/object_manager_stress_test.cc
-17
View File
@@ -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);
}