mirror of
https://github.com/vale981/ray
synced 2025-03-06 02:21:39 -05:00
Remove legacy C++ code (#9459)
This commit is contained in:
Siyuan (Ryans) Zhuang
GitHub
parent
deba082cb4
commit
d57ff5e2af
5 changed files with 0 additions and 637 deletions
25
BUILD.bazel
25
BUILD.bazel
|
|
@ -320,21 +320,6 @@ cc_library(
|
||||||
],
|
],
|
||||||
)
|
)
|
||||||
|
|
||||||
cc_binary(
|
|
||||||
name = "libplasma_java.so",
|
|
||||||
srcs = [
|
|
||||||
"src/ray/object_manager/plasma/lib/java/org_apache_arrow_plasma_PlasmaClientJNI.cc",
|
|
||||||
"src/ray/object_manager/plasma/lib/java/org_apache_arrow_plasma_PlasmaClientJNI.h",
|
|
||||||
":jni.h",
|
|
||||||
":jni_md.h",
|
|
||||||
],
|
|
||||||
copts = PLASMA_COPTS,
|
|
||||||
includes = ["src/ray/object_manager/plasma/lib/java"],
|
|
||||||
linkshared = 1,
|
|
||||||
linkstatic = 1,
|
|
||||||
deps = [":plasma_client"],
|
|
||||||
)
|
|
||||||
|
|
||||||
copy_file(
|
copy_file(
|
||||||
name = "copy_jni_h",
|
name = "copy_jni_h",
|
||||||
src = "@bazel_tools//tools/jdk:jni_header",
|
src = "@bazel_tools//tools/jdk:jni_header",
|
||||||
|
|
@ -351,14 +336,6 @@ copy_file(
|
||||||
out = "jni_md.h",
|
out = "jni_md.h",
|
||||||
)
|
)
|
||||||
|
|
||||||
genrule(
|
|
||||||
name = "plasma-jni-darwin-compat",
|
|
||||||
srcs = [":libplasma_java.so"],
|
|
||||||
outs = ["libplasma_java.dylib"],
|
|
||||||
cmd = "cp $< $@",
|
|
||||||
output_to_bindir = 1,
|
|
||||||
)
|
|
||||||
|
|
||||||
cc_library(
|
cc_library(
|
||||||
name = "ae",
|
name = "ae",
|
||||||
srcs = [
|
srcs = [
|
||||||
|
|
@ -618,8 +595,6 @@ cc_library(
|
||||||
"src/ray/raylet/*.cc",
|
"src/ray/raylet/*.cc",
|
||||||
],
|
],
|
||||||
exclude = [
|
exclude = [
|
||||||
"src/ray/raylet/mock_gcs_client.cc",
|
|
||||||
"src/ray/raylet/monitor*.cc",
|
|
||||||
"src/ray/raylet/*_test.cc",
|
"src/ray/raylet/*_test.cc",
|
||||||
"src/ray/raylet/main.cc",
|
"src/ray/raylet/main.cc",
|
||||||
],
|
],
|
||||||
|
|
|
||||||
|
|
@ -1,245 +0,0 @@
|
||||||
// Licensed to the Apache Software Foundation (ASF) under one
|
|
||||||
// or more contributor license agreements. See the NOTICE file
|
|
||||||
// distributed with this work for additional information
|
|
||||||
// regarding copyright ownership. The ASF licenses this file
|
|
||||||
// to you under the Apache License, Version 2.0 (the
|
|
||||||
// "License"); you may not use this file except in compliance
|
|
||||||
// with the License. You may obtain a copy of the License at
|
|
||||||
//
|
|
||||||
// http://www.apache.org/licenses/LICENSE-2.0
|
|
||||||
//
|
|
||||||
// Unless required by applicable law or agreed to in writing,
|
|
||||||
// software distributed under the License is distributed on an
|
|
||||||
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
|
|
||||||
// KIND, either express or implied. See the License for the
|
|
||||||
// specific language governing permissions and limitations
|
|
||||||
// under the License.
|
|
||||||
|
|
||||||
#include "org_apache_arrow_plasma_PlasmaClientJNI.h"
|
|
||||||
|
|
||||||
#include <pthread.h>
|
|
||||||
#include <stdlib.h>
|
|
||||||
#include <unistd.h>
|
|
||||||
|
|
||||||
#include <algorithm>
|
|
||||||
#include <cstring>
|
|
||||||
#include <iostream>
|
|
||||||
#include <memory>
|
|
||||||
#include <string>
|
|
||||||
#include <vector>
|
|
||||||
|
|
||||||
#include "ray/common/status.h"
|
|
||||||
#include "ray/object_manager/plasma/client.h"
|
|
||||||
|
|
||||||
using ray::ObjectID;
|
|
||||||
using ray::Status;
|
|
||||||
|
|
||||||
constexpr jsize OBJECT_ID_SIZE = sizeof(ObjectID) / sizeof(jbyte);
|
|
||||||
|
|
||||||
inline void jbyteArray_to_object_id(JNIEnv* env, jbyteArray a, ObjectID* oid) {
|
|
||||||
env->GetByteArrayRegion(a, 0, OBJECT_ID_SIZE, reinterpret_cast<jbyte*>(oid));
|
|
||||||
}
|
|
||||||
|
|
||||||
inline void object_id_to_jbyteArray(JNIEnv* env, jbyteArray a, ObjectID* oid) {
|
|
||||||
env->SetByteArrayRegion(a, 0, OBJECT_ID_SIZE, reinterpret_cast<jbyte*>(oid));
|
|
||||||
}
|
|
||||||
|
|
||||||
inline void throw_exception_if_not_OK(JNIEnv* env, const Status& status) {
|
|
||||||
if (!status.ok()) {
|
|
||||||
jclass Exception =
|
|
||||||
env->FindClass("org/apache/arrow/plasma/exceptions/PlasmaClientException");
|
|
||||||
env->ThrowNew(Exception, status.message().c_str());
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
class JByteArrayGetter {
|
|
||||||
private:
|
|
||||||
JNIEnv* _env;
|
|
||||||
jbyteArray _a;
|
|
||||||
jbyte* bp;
|
|
||||||
|
|
||||||
public:
|
|
||||||
JByteArrayGetter(JNIEnv* env, jbyteArray a, jbyte** out) {
|
|
||||||
_env = env;
|
|
||||||
_a = a;
|
|
||||||
|
|
||||||
bp = _env->GetByteArrayElements(_a, nullptr);
|
|
||||||
*out = bp;
|
|
||||||
}
|
|
||||||
|
|
||||||
~JByteArrayGetter() { _env->ReleaseByteArrayElements(_a, bp, 0); }
|
|
||||||
};
|
|
||||||
|
|
||||||
JNIEXPORT jlong JNICALL Java_org_apache_arrow_plasma_PlasmaClientJNI_connect(
|
|
||||||
JNIEnv* env, jclass cls, jstring store_socket_name, jstring manager_socket_name,
|
|
||||||
jint release_delay) {
|
|
||||||
const char* s_name = env->GetStringUTFChars(store_socket_name, nullptr);
|
|
||||||
const char* m_name = env->GetStringUTFChars(manager_socket_name, nullptr);
|
|
||||||
|
|
||||||
plasma::PlasmaClient* client = new plasma::PlasmaClient();
|
|
||||||
throw_exception_if_not_OK(env, client->Connect(s_name, m_name, release_delay));
|
|
||||||
|
|
||||||
env->ReleaseStringUTFChars(store_socket_name, s_name);
|
|
||||||
env->ReleaseStringUTFChars(manager_socket_name, m_name);
|
|
||||||
return reinterpret_cast<int64_t>(client);
|
|
||||||
}
|
|
||||||
|
|
||||||
JNIEXPORT void JNICALL Java_org_apache_arrow_plasma_PlasmaClientJNI_disconnect(
|
|
||||||
JNIEnv* env, jclass cls, jlong conn) {
|
|
||||||
plasma::PlasmaClient* client = reinterpret_cast<plasma::PlasmaClient*>(conn);
|
|
||||||
|
|
||||||
throw_exception_if_not_OK(env, client->Disconnect());
|
|
||||||
delete client;
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
|
|
||||||
JNIEXPORT jobject JNICALL Java_org_apache_arrow_plasma_PlasmaClientJNI_create(
|
|
||||||
JNIEnv* env, jclass cls, jlong conn, jbyteArray object_id, jint size,
|
|
||||||
jbyteArray metadata) {
|
|
||||||
plasma::PlasmaClient* client = reinterpret_cast<plasma::PlasmaClient*>(conn);
|
|
||||||
ObjectID oid;
|
|
||||||
jbyteArray_to_object_id(env, object_id, &oid);
|
|
||||||
|
|
||||||
// prepare metadata buffer
|
|
||||||
uint8_t* md = nullptr;
|
|
||||||
jsize md_size = 0;
|
|
||||||
std::unique_ptr<JByteArrayGetter> md_getter;
|
|
||||||
if (metadata != nullptr) {
|
|
||||||
md_size = env->GetArrayLength(metadata);
|
|
||||||
}
|
|
||||||
if (md_size > 0) {
|
|
||||||
md_getter.reset(new JByteArrayGetter(env, metadata, reinterpret_cast<jbyte**>(&md)));
|
|
||||||
}
|
|
||||||
|
|
||||||
std::shared_ptr<Buffer> data;
|
|
||||||
Status s = client->Create(oid, size, md, md_size, &data);
|
|
||||||
if (s.IsObjectExists()) {
|
|
||||||
jclass exceptionClass =
|
|
||||||
env->FindClass("org/apache/arrow/plasma/exceptions/DuplicateObjectException");
|
|
||||||
env->ThrowNew(exceptionClass, oid.Hex().c_str());
|
|
||||||
return nullptr;
|
|
||||||
}
|
|
||||||
if (s.IsObjectStoreFull()) {
|
|
||||||
jclass exceptionClass =
|
|
||||||
env->FindClass("org/apache/arrow/plasma/exceptions/PlasmaOutOfMemoryException");
|
|
||||||
env->ThrowNew(exceptionClass, "");
|
|
||||||
return nullptr;
|
|
||||||
}
|
|
||||||
throw_exception_if_not_OK(env, s);
|
|
||||||
|
|
||||||
return env->NewDirectByteBuffer(data->mutable_data(), size);
|
|
||||||
}
|
|
||||||
|
|
||||||
JNIEXPORT jbyteArray JNICALL Java_org_apache_arrow_plasma_PlasmaClientJNI_hash(
|
|
||||||
JNIEnv* env, jclass cls, jlong conn, jbyteArray object_id) {
|
|
||||||
plasma::PlasmaClient* client = reinterpret_cast<plasma::PlasmaClient*>(conn);
|
|
||||||
ObjectID oid;
|
|
||||||
jbyteArray_to_object_id(env, object_id, &oid);
|
|
||||||
|
|
||||||
unsigned char digest[plasma::kDigestSize];
|
|
||||||
bool success = client->Hash(oid, digest).ok();
|
|
||||||
|
|
||||||
if (success) {
|
|
||||||
jbyteArray ret = env->NewByteArray(plasma::kDigestSize);
|
|
||||||
env->SetByteArrayRegion(ret, 0, plasma::kDigestSize,
|
|
||||||
reinterpret_cast<jbyte*>(digest));
|
|
||||||
return ret;
|
|
||||||
} else {
|
|
||||||
return nullptr;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
JNIEXPORT void JNICALL Java_org_apache_arrow_plasma_PlasmaClientJNI_seal(
|
|
||||||
JNIEnv* env, jclass cls, jlong conn, jbyteArray object_id) {
|
|
||||||
plasma::PlasmaClient* client = reinterpret_cast<plasma::PlasmaClient*>(conn);
|
|
||||||
ObjectID oid;
|
|
||||||
jbyteArray_to_object_id(env, object_id, &oid);
|
|
||||||
|
|
||||||
throw_exception_if_not_OK(env, client->Seal(oid));
|
|
||||||
}
|
|
||||||
|
|
||||||
JNIEXPORT void JNICALL Java_org_apache_arrow_plasma_PlasmaClientJNI_release(
|
|
||||||
JNIEnv* env, jclass cls, jlong conn, jbyteArray object_id) {
|
|
||||||
plasma::PlasmaClient* client = reinterpret_cast<plasma::PlasmaClient*>(conn);
|
|
||||||
ObjectID oid;
|
|
||||||
jbyteArray_to_object_id(env, object_id, &oid);
|
|
||||||
|
|
||||||
throw_exception_if_not_OK(env, client->Release(oid));
|
|
||||||
}
|
|
||||||
|
|
||||||
JNIEXPORT void JNICALL Java_org_apache_arrow_plasma_PlasmaClientJNI_delete(
|
|
||||||
JNIEnv* env, jclass cls, jlong conn, jbyteArray object_id) {
|
|
||||||
plasma::PlasmaClient* client = reinterpret_cast<plasma::PlasmaClient*>(conn);
|
|
||||||
ObjectID oid;
|
|
||||||
jbyteArray_to_object_id(env, object_id, &oid);
|
|
||||||
|
|
||||||
throw_exception_if_not_OK(env, client->Delete(oid));
|
|
||||||
}
|
|
||||||
|
|
||||||
JNIEXPORT jobjectArray JNICALL Java_org_apache_arrow_plasma_PlasmaClientJNI_get(
|
|
||||||
JNIEnv* env, jclass cls, jlong conn, jobjectArray object_ids, jint timeout_ms) {
|
|
||||||
plasma::PlasmaClient* client = reinterpret_cast<plasma::PlasmaClient*>(conn);
|
|
||||||
|
|
||||||
jsize num_oids = env->GetArrayLength(object_ids);
|
|
||||||
std::vector<ObjectID> oids(num_oids);
|
|
||||||
std::vector<plasma::ObjectBuffer> obufs(num_oids);
|
|
||||||
for (int i = 0; i < num_oids; ++i) {
|
|
||||||
jbyteArray_to_object_id(
|
|
||||||
env, reinterpret_cast<jbyteArray>(env->GetObjectArrayElement(object_ids, i)),
|
|
||||||
&oids[i]);
|
|
||||||
}
|
|
||||||
// TODO: may be blocked. consider to add the thread support
|
|
||||||
throw_exception_if_not_OK(env,
|
|
||||||
client->Get(oids.data(), num_oids, timeout_ms, obufs.data()));
|
|
||||||
|
|
||||||
jclass clsByteBuffer = env->FindClass("java/nio/ByteBuffer");
|
|
||||||
jclass clsByteBufferArray = env->FindClass("[Ljava/nio/ByteBuffer;");
|
|
||||||
|
|
||||||
jobjectArray ret = env->NewObjectArray(num_oids, clsByteBufferArray, nullptr);
|
|
||||||
jobjectArray o = nullptr;
|
|
||||||
jobject dataBuf, metadataBuf;
|
|
||||||
for (int i = 0; i < num_oids; ++i) {
|
|
||||||
o = env->NewObjectArray(2, clsByteBuffer, nullptr);
|
|
||||||
if (obufs[i].data && obufs[i].data->size() != -1) {
|
|
||||||
dataBuf = env->NewDirectByteBuffer(const_cast<uint8_t*>(obufs[i].data->data()),
|
|
||||||
obufs[i].data->size());
|
|
||||||
if (obufs[i].metadata && obufs[i].metadata->size() > 0) {
|
|
||||||
metadataBuf = env->NewDirectByteBuffer(
|
|
||||||
const_cast<uint8_t*>(obufs[i].metadata->data()), obufs[i].metadata->size());
|
|
||||||
} else {
|
|
||||||
metadataBuf = nullptr;
|
|
||||||
}
|
|
||||||
} else {
|
|
||||||
dataBuf = nullptr;
|
|
||||||
metadataBuf = nullptr;
|
|
||||||
}
|
|
||||||
|
|
||||||
env->SetObjectArrayElement(o, 0, dataBuf);
|
|
||||||
env->SetObjectArrayElement(o, 1, metadataBuf);
|
|
||||||
env->SetObjectArrayElement(ret, i, o);
|
|
||||||
}
|
|
||||||
return ret;
|
|
||||||
}
|
|
||||||
|
|
||||||
JNIEXPORT jboolean JNICALL Java_org_apache_arrow_plasma_PlasmaClientJNI_contains(
|
|
||||||
JNIEnv* env, jclass cls, jlong conn, jbyteArray object_id) {
|
|
||||||
plasma::PlasmaClient* client = reinterpret_cast<plasma::PlasmaClient*>(conn);
|
|
||||||
ObjectID oid;
|
|
||||||
jbyteArray_to_object_id(env, object_id, &oid);
|
|
||||||
|
|
||||||
bool has_object;
|
|
||||||
throw_exception_if_not_OK(env, client->Contains(oid, &has_object));
|
|
||||||
|
|
||||||
return has_object;
|
|
||||||
}
|
|
||||||
|
|
||||||
JNIEXPORT jlong JNICALL Java_org_apache_arrow_plasma_PlasmaClientJNI_evict(
|
|
||||||
JNIEnv* env, jclass cls, jlong conn, jlong num_bytes) {
|
|
||||||
plasma::PlasmaClient* client = reinterpret_cast<plasma::PlasmaClient*>(conn);
|
|
||||||
|
|
||||||
int64_t evicted_bytes;
|
|
||||||
throw_exception_if_not_OK(
|
|
||||||
env, client->Evict(static_cast<int64_t>(num_bytes), evicted_bytes));
|
|
||||||
|
|
||||||
return static_cast<jlong>(evicted_bytes);
|
|
||||||
}
|
|
||||||
|
|
@ -1,132 +0,0 @@
|
||||||
// Licensed to the Apache Software Foundation (ASF) under one
|
|
||||||
// or more contributor license agreements. See the NOTICE file
|
|
||||||
// distributed with this work for additional information
|
|
||||||
// regarding copyright ownership. The ASF licenses this file
|
|
||||||
// to you under the Apache License, Version 2.0 (the
|
|
||||||
// "License"); you may not use this file except in compliance
|
|
||||||
// with the License. You may obtain a copy of the License at
|
|
||||||
//
|
|
||||||
// http://www.apache.org/licenses/LICENSE-2.0
|
|
||||||
//
|
|
||||||
// Unless required by applicable law or agreed to in writing,
|
|
||||||
// software distributed under the License is distributed on an
|
|
||||||
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
|
|
||||||
// KIND, either express or implied. See the License for the
|
|
||||||
// specific language governing permissions and limitations
|
|
||||||
// under the License.
|
|
||||||
|
|
||||||
/* DO NOT EDIT THIS FILE - it is machine generated */
|
|
||||||
#include "jni.h"
|
|
||||||
/* Header for class org_apache_arrow_plasma_PlasmaClientJNI */
|
|
||||||
|
|
||||||
#ifndef _Included_org_apache_arrow_plasma_PlasmaClientJNI
|
|
||||||
#define _Included_org_apache_arrow_plasma_PlasmaClientJNI
|
|
||||||
#ifdef __cplusplus
|
|
||||||
extern "C" {
|
|
||||||
#endif
|
|
||||||
/*
|
|
||||||
* Class: org_apache_arrow_plasma_PlasmaClientJNI
|
|
||||||
* Method: connect
|
|
||||||
* Signature: (Ljava/lang/String;Ljava/lang/String;I)J
|
|
||||||
*/
|
|
||||||
JNIEXPORT jlong JNICALL Java_org_apache_arrow_plasma_PlasmaClientJNI_connect(
|
|
||||||
JNIEnv*, jclass, jstring, jstring, jint);
|
|
||||||
|
|
||||||
/*
|
|
||||||
* Class: org_apache_arrow_plasma_PlasmaClientJNI
|
|
||||||
* Method: disconnect
|
|
||||||
* Signature: (J)V
|
|
||||||
*/
|
|
||||||
JNIEXPORT void JNICALL Java_org_apache_arrow_plasma_PlasmaClientJNI_disconnect(JNIEnv*,
|
|
||||||
jclass,
|
|
||||||
jlong);
|
|
||||||
|
|
||||||
/*
|
|
||||||
* Class: org_apache_arrow_plasma_PlasmaClientJNI
|
|
||||||
* Method: create
|
|
||||||
* Signature: (J[BI[B)Ljava/nio/ByteBuffer;
|
|
||||||
*/
|
|
||||||
JNIEXPORT jobject JNICALL Java_org_apache_arrow_plasma_PlasmaClientJNI_create(
|
|
||||||
JNIEnv*, jclass, jlong, jbyteArray, jint, jbyteArray);
|
|
||||||
|
|
||||||
/*
|
|
||||||
* Class: org_apache_arrow_plasma_PlasmaClientJNI
|
|
||||||
* Method: hash
|
|
||||||
* Signature: (J[B)[B
|
|
||||||
*/
|
|
||||||
JNIEXPORT jbyteArray JNICALL
|
|
||||||
Java_org_apache_arrow_plasma_PlasmaClientJNI_hash(JNIEnv*, jclass, jlong, jbyteArray);
|
|
||||||
|
|
||||||
/*
|
|
||||||
* Class: org_apache_arrow_plasma_PlasmaClientJNI
|
|
||||||
* Method: seal
|
|
||||||
* Signature: (J[B)V
|
|
||||||
*/
|
|
||||||
JNIEXPORT void JNICALL Java_org_apache_arrow_plasma_PlasmaClientJNI_seal(JNIEnv*, jclass,
|
|
||||||
jlong,
|
|
||||||
jbyteArray);
|
|
||||||
|
|
||||||
/*
|
|
||||||
* Class: org_apache_arrow_plasma_PlasmaClientJNI
|
|
||||||
* Method: release
|
|
||||||
* Signature: (J[B)V
|
|
||||||
*/
|
|
||||||
JNIEXPORT void JNICALL Java_org_apache_arrow_plasma_PlasmaClientJNI_release(JNIEnv*,
|
|
||||||
jclass, jlong,
|
|
||||||
jbyteArray);
|
|
||||||
|
|
||||||
/*
|
|
||||||
* Class: org_apache_arrow_plasma_PlasmaClientJNI
|
|
||||||
* Method: delete
|
|
||||||
* Signature: (J[B)V
|
|
||||||
*/
|
|
||||||
JNIEXPORT void JNICALL Java_org_apache_arrow_plasma_PlasmaClientJNI_delete(JNIEnv*,
|
|
||||||
jclass, jlong,
|
|
||||||
jbyteArray);
|
|
||||||
|
|
||||||
/*
|
|
||||||
* Class: org_apache_arrow_plasma_PlasmaClientJNI
|
|
||||||
* Method: get
|
|
||||||
* Signature: (J[[BI)[[Ljava/nio/ByteBuffer;
|
|
||||||
*/
|
|
||||||
JNIEXPORT jobjectArray JNICALL Java_org_apache_arrow_plasma_PlasmaClientJNI_get(
|
|
||||||
JNIEnv*, jclass, jlong, jobjectArray, jint);
|
|
||||||
|
|
||||||
/*
|
|
||||||
* Class: org_apache_arrow_plasma_PlasmaClientJNI
|
|
||||||
* Method: contains
|
|
||||||
* Signature: (J[B)Z
|
|
||||||
*/
|
|
||||||
JNIEXPORT jboolean JNICALL
|
|
||||||
Java_org_apache_arrow_plasma_PlasmaClientJNI_contains(JNIEnv*, jclass, jlong, jbyteArray);
|
|
||||||
|
|
||||||
/*
|
|
||||||
* Class: org_apache_arrow_plasma_PlasmaClientJNI
|
|
||||||
* Method: fetch
|
|
||||||
* Signature: (J[[B)V
|
|
||||||
*/
|
|
||||||
JNIEXPORT void JNICALL Java_org_apache_arrow_plasma_PlasmaClientJNI_fetch(JNIEnv*, jclass,
|
|
||||||
jlong,
|
|
||||||
jobjectArray);
|
|
||||||
|
|
||||||
/*
|
|
||||||
* Class: org_apache_arrow_plasma_PlasmaClientJNI
|
|
||||||
* Method: wait
|
|
||||||
* Signature: (J[[BII)[[B
|
|
||||||
*/
|
|
||||||
JNIEXPORT jobjectArray JNICALL Java_org_apache_arrow_plasma_PlasmaClientJNI_wait(
|
|
||||||
JNIEnv*, jclass, jlong, jobjectArray, jint, jint);
|
|
||||||
|
|
||||||
/*
|
|
||||||
* Class: org_apache_arrow_plasma_PlasmaClientJNI
|
|
||||||
* Method: evict
|
|
||||||
* Signature: (JJ)J
|
|
||||||
*/
|
|
||||||
JNIEXPORT jlong JNICALL Java_org_apache_arrow_plasma_PlasmaClientJNI_evict(JNIEnv*,
|
|
||||||
jclass, jlong,
|
|
||||||
jlong);
|
|
||||||
|
|
||||||
#ifdef __cplusplus
|
|
||||||
}
|
|
||||||
#endif
|
|
||||||
#endif
|
|
||||||
|
|
@ -1,133 +0,0 @@
|
||||||
// Copyright 2017 The Ray Authors.
|
|
||||||
//
|
|
||||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
|
||||||
// you may not use this file except in compliance with the License.
|
|
||||||
// You may obtain a copy of the License at
|
|
||||||
//
|
|
||||||
// http://www.apache.org/licenses/LICENSE-2.0
|
|
||||||
//
|
|
||||||
// Unless required by applicable law or agreed to in writing, software
|
|
||||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
|
||||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
||||||
// See the License for the specific language governing permissions and
|
|
||||||
// limitations under the License.
|
|
||||||
|
|
||||||
#include <functional>
|
|
||||||
#include <iostream>
|
|
||||||
|
|
||||||
#include "ray/raylet/mock_gcs_client.h"
|
|
||||||
|
|
||||||
namespace ray {
|
|
||||||
|
|
||||||
ray::Status ObjectTable::GetObjectClientIDs(const ray::ObjectID &object_id,
|
|
||||||
const ClientIDsCallback &success,
|
|
||||||
const FailCallback &fail) {
|
|
||||||
RAY_LOG(DEBUG) << "GetObjectClientIDs " << object_id;
|
|
||||||
if (client_lookup.count(object_id) > 0) {
|
|
||||||
if (!client_lookup[object_id].empty()) {
|
|
||||||
std::vector<ClientID> v;
|
|
||||||
for (auto client_id : client_lookup[object_id]) {
|
|
||||||
v.push_back(client_id);
|
|
||||||
}
|
|
||||||
success(std::move(v));
|
|
||||||
return Status::OK();
|
|
||||||
} else {
|
|
||||||
fail(Status::KeyError("ObjectID has no clients."));
|
|
||||||
return Status::OK();
|
|
||||||
}
|
|
||||||
} else {
|
|
||||||
fail(Status::KeyError("ObjectID doesn't exist."));
|
|
||||||
return Status::OK();
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
ray::Status ObjectTable::Add(const ObjectID &object_id, const ClientID &client_id,
|
|
||||||
const DoneCallback &done_callback) {
|
|
||||||
if (client_lookup.count(object_id) == 0) {
|
|
||||||
RAY_LOG(DEBUG) << "Add ObjectID set " << object_id;
|
|
||||||
client_lookup[object_id] = std::unordered_set<ClientID>();
|
|
||||||
} else if (client_lookup[object_id].count(client_id) != 0) {
|
|
||||||
return ray::Status::KeyError("ClientID already exists.");
|
|
||||||
}
|
|
||||||
RAY_LOG(DEBUG) << "Insert ClientID " << client_id;
|
|
||||||
client_lookup[object_id].insert(client_id);
|
|
||||||
done_callback();
|
|
||||||
return ray::Status::OK();
|
|
||||||
}
|
|
||||||
|
|
||||||
ray::Status ObjectTable::Remove(const ObjectID &object_id, const ClientID &client_id,
|
|
||||||
const DoneCallback &done_callback) {
|
|
||||||
if (client_lookup.count(object_id) == 0) {
|
|
||||||
return ray::Status::KeyError("ObjectID doesn't exist.");
|
|
||||||
} else if (client_lookup[object_id].count(client_id) == 0) {
|
|
||||||
return ray::Status::KeyError("ClientID doesn't exist.");
|
|
||||||
}
|
|
||||||
client_lookup[object_id].erase(client_id);
|
|
||||||
done_callback();
|
|
||||||
return ray::Status::OK();
|
|
||||||
}
|
|
||||||
|
|
||||||
ray::Status ClientTable::GetClientIds(ClientIDsCallback callback) {
|
|
||||||
std::vector<ClientID> keys;
|
|
||||||
keys.reserve(info_lookup.size());
|
|
||||||
for (auto kv : info_lookup) {
|
|
||||||
keys.push_back(kv.first);
|
|
||||||
}
|
|
||||||
callback(keys);
|
|
||||||
return Status::OK();
|
|
||||||
}
|
|
||||||
|
|
||||||
void ClientTable::GetClientInformationSet(const std::vector<ClientID> &client_ids,
|
|
||||||
ManyInfoCallback callback,
|
|
||||||
FailCallback failcb) {
|
|
||||||
std::vector<ClientInformation> info_vec;
|
|
||||||
for (const auto &client_id : client_ids) {
|
|
||||||
if (info_lookup.count(client_id) != 0) {
|
|
||||||
info_vec.push_back(info_lookup.at(client_id));
|
|
||||||
}
|
|
||||||
}
|
|
||||||
if (info_vec.empty()) {
|
|
||||||
failcb(Status::KeyError("ClientID not found."));
|
|
||||||
} else {
|
|
||||||
callback(info_vec);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
void ClientTable::GetClientInformation(const ClientID &client_id,
|
|
||||||
SingleInfoCallback callback, FailCallback failcb) {
|
|
||||||
if (info_lookup.count(client_id) == 0) {
|
|
||||||
failcb(ray::Status::KeyError("CleintID not found."));
|
|
||||||
} else {
|
|
||||||
callback(info_lookup.at(client_id));
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
ray::Status ClientTable::Add(const ClientID &client_id, const std::string &ip,
|
|
||||||
uint16_t port, DoneCallback done_callback) {
|
|
||||||
if (info_lookup.count(client_id) != 0) {
|
|
||||||
return ray::Status::KeyError("ClientID already exists.");
|
|
||||||
}
|
|
||||||
info_lookup.emplace(client_id, ClientInformation(client_id, ip, port));
|
|
||||||
done_callback();
|
|
||||||
return ray::Status::OK();
|
|
||||||
}
|
|
||||||
|
|
||||||
ray::Status ClientTable::Remove(const ClientID &client_id, DoneCallback done_callback) {
|
|
||||||
if (info_lookup.count(client_id) == 0) {
|
|
||||||
return ray::Status::KeyError("ClientID doesn't exist.");
|
|
||||||
}
|
|
||||||
info_lookup.erase(client_id);
|
|
||||||
done_callback();
|
|
||||||
return ray::Status::OK();
|
|
||||||
}
|
|
||||||
|
|
||||||
ClientID GcsClient::Register(const std::string &ip, uint16_t port) {
|
|
||||||
ClientID client_id = ClientID::FromRandom();
|
|
||||||
// TODO: handle client registration failure.
|
|
||||||
ray::Status status = client_table().Add(client_id, ip, port, []() {});
|
|
||||||
return client_id;
|
|
||||||
}
|
|
||||||
|
|
||||||
ClientTable &GcsClient::client_table() { return *client_table_; }
|
|
||||||
|
|
||||||
} // namespace ray
|
|
||||||
|
|
@ -1,102 +0,0 @@
|
||||||
// Copyright 2017 The Ray Authors.
|
|
||||||
//
|
|
||||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
|
||||||
// you may not use this file except in compliance with the License.
|
|
||||||
// You may obtain a copy of the License at
|
|
||||||
//
|
|
||||||
// http://www.apache.org/licenses/LICENSE-2.0
|
|
||||||
//
|
|
||||||
// Unless required by applicable law or agreed to in writing, software
|
|
||||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
|
||||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
||||||
// See the License for the specific language governing permissions and
|
|
||||||
// limitations under the License.
|
|
||||||
|
|
||||||
#pragma once
|
|
||||||
|
|
||||||
#include <cstdint>
|
|
||||||
#include <functional>
|
|
||||||
#include <map>
|
|
||||||
#include <unordered_map>
|
|
||||||
#include <unordered_set>
|
|
||||||
#include <vector>
|
|
||||||
|
|
||||||
#include <boost/asio.hpp>
|
|
||||||
#include <boost/asio/error.hpp>
|
|
||||||
#include <boost/bind.hpp>
|
|
||||||
|
|
||||||
#include "ray/common/id.h"
|
|
||||||
#include "ray/common/status.h"
|
|
||||||
|
|
||||||
namespace ray {
|
|
||||||
|
|
||||||
class ObjectTable {
|
|
||||||
public:
|
|
||||||
using DoneCallback = std::function<void()>;
|
|
||||||
using ClientIDsCallback = std::function<void(const std::vector<ray::ClientID> &)>;
|
|
||||||
using FailCallback = std::function<void(const ray::Status &)>;
|
|
||||||
ray::Status Add(const ObjectID &object_id, const ClientID &client_id,
|
|
||||||
const DoneCallback &done);
|
|
||||||
ray::Status Remove(const ObjectID &object_id, const ClientID &client_id,
|
|
||||||
const DoneCallback &done);
|
|
||||||
ray::Status GetObjectClientIDs(const ObjectID &object_id, const ClientIDsCallback &,
|
|
||||||
const FailCallback &);
|
|
||||||
|
|
||||||
private:
|
|
||||||
std::vector<ClientID> empty_set_;
|
|
||||||
std::unordered_map<ObjectID, std::unordered_set<ClientID>> client_lookup;
|
|
||||||
};
|
|
||||||
|
|
||||||
class ClientInformation {
|
|
||||||
public:
|
|
||||||
ClientInformation(const ClientID &client_id, const std::string &ip_address,
|
|
||||||
uint16_t port)
|
|
||||||
: client_id_(client_id), ip_address_(ip_address), port_(port) {}
|
|
||||||
const ClientID &GetClientId() const { return client_id_; }
|
|
||||||
const std::string &GetIp() const { return ip_address_; }
|
|
||||||
const uint16_t &GetPort() const { return port_; }
|
|
||||||
|
|
||||||
private:
|
|
||||||
ClientID client_id_;
|
|
||||||
std::string ip_address_;
|
|
||||||
uint16_t port_;
|
|
||||||
};
|
|
||||||
|
|
||||||
class ClientTable {
|
|
||||||
public:
|
|
||||||
typedef std::unordered_map<ClientID, ClientInformation> info_type;
|
|
||||||
|
|
||||||
using ClientIDsCallback = std::function<void(std::vector<ray::ClientID>)>;
|
|
||||||
using SingleInfoCallback = std::function<void(ClientInformation info)>;
|
|
||||||
using ManyInfoCallback = std::function<void(std::vector<ClientInformation> info_vec)>;
|
|
||||||
using DoneCallback = std::function<void()>;
|
|
||||||
using FailCallback = std::function<void(ray::Status)>;
|
|
||||||
|
|
||||||
ray::Status GetClientIds(ClientIDsCallback cb);
|
|
||||||
void GetClientInformationSet(const std::vector<ClientID> &client_ids,
|
|
||||||
ManyInfoCallback cb, FailCallback failcb);
|
|
||||||
void GetClientInformation(const ClientID &client_id, SingleInfoCallback callback,
|
|
||||||
FailCallback failcb);
|
|
||||||
ray::Status Add(const ClientID &client_id, const std::string &ip, uint16_t port,
|
|
||||||
DoneCallback cb);
|
|
||||||
ray::Status Remove(const ClientID &client_id, DoneCallback done);
|
|
||||||
|
|
||||||
private:
|
|
||||||
info_type info_lookup;
|
|
||||||
};
|
|
||||||
|
|
||||||
class GcsClient {
|
|
||||||
public:
|
|
||||||
GcsClient() {
|
|
||||||
this->object_table_.reset(new ObjectTable());
|
|
||||||
this->client_table_.reset(new ClientTable());
|
|
||||||
}
|
|
||||||
// Register the ip and port of the connecting client.
|
|
||||||
ClientID Register(const std::string &ip, uint16_t port);
|
|
||||||
ClientTable &client_table();
|
|
||||||
|
|
||||||
private:
|
|
||||||
std::unique_ptr<ObjectTable> object_table_;
|
|
||||||
std::unique_ptr<ClientTable> client_table_;
|
|
||||||
};
|
|
||||||
} // namespace ray
|
|
||||||
Loading…
Add table
Reference in a new issue