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

Switch to updated Plasma API and consolidate wait and fetch implementations. (#116)

Browse Source 
* Consolidate wait implementations.

* Consolidate fetch implementations.

* Share callback between wait and fetch to address issue in which only one callback can be run for a given subscribe channel.

* Reactivate manager tests.

* Remove more code.

* Add some documentation.
This commit is contained in:
Robert Nishihara
2016-12-10 21:22:05 -08:00
committed by Philipp Moritz
parent 86973059de
commit 9474d03912
12 changed files with 287 additions and 1524 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/python/ray/worker.py
+1 -1
View File
@@ -437,7 +437,7 @@ class Worker(object):
Args:
objectid (object_id.ObjectID): The object ID of the value to retrieve.
"""
self.plasma_client.fetch2([objectid.id()])
self.plasma_client.fetch([objectid.id()])
buff = self.plasma_client.get(objectid.id())
metadata = self.plasma_client.get_metadata(objectid.id())
metadata_size = len(metadata)
src/photon/photon_algorithm.c
+2 -2
View File
@@ -123,7 +123,7 @@ bool can_run(scheduling_algorithm_state *algorithm_state, task_spec *task) {
int fetch_object_timeout_handler(event_loop *loop, timer_id id, void *context) {
fetch_object_request *fetch_req = (fetch_object_request *) context;
object_id object_ids[1] = {fetch_req->object_id};
plasma_fetch2(fetch_req->state->plasma_conn, 1, object_ids);
plasma_fetch(fetch_req->state->plasma_conn, 1, object_ids);
return LOCAL_SCHEDULER_FETCH_TIMEOUT_MILLISECONDS;
}
@@ -140,7 +140,7 @@ void fetch_missing_dependencies(local_scheduler_state *state,
if (entry == NULL) {
/* The object is not present locally, fetch the object. */
object_id object_ids[1] = {obj_id};
plasma_fetch2(state->plasma_conn, 1, object_ids);
plasma_fetch(state->plasma_conn, 1, object_ids);
/* Create a fetch request and add a timer to the event loop to ensure
* that the fetch actually happens. */
fetch_object_request *fetch_req = malloc(sizeof(fetch_object_request));
src/plasma/lib/python/plasma.py
+1 -9
View File
@@ -188,20 +188,12 @@ class PlasmaClient(object):
return libplasma.transfer(self.conn, object_id, addr, port)
def fetch(self, object_ids):
"""Fetch the object with id object_id from another plasma manager instance.
Args:
object_id (str): A string used to identify an object.
"""
return libplasma.fetch(self.conn, object_ids)
def fetch2(self, object_ids):
"""Fetch the objects with the given IDs from other plasma manager instances.
Args:
object_ids (List[str]): A list of strings used to identify the objects.
"""
return libplasma.fetch2(self.conn, object_ids)
return libplasma.fetch(self.conn, object_ids)
def wait(self, object_ids, timeout=PLASMA_WAIT_TIMEOUT, num_returns=1):
"""Wait until num_returns objects in object_ids are ready.
src/plasma/plasma.h
+1 -9
View File
@@ -120,21 +120,13 @@ enum plasma_message_type {
/** Header for sending data. */
PLASMA_DATA,
/** Request a fetch of an object in another store. Non-blocking call. */
PLASMA_FETCH_REMOTE,
/** Request a fetch of an object in another store. Blocking call. */
PLASMA_FETCH,
/** Request a fetch of an object in another store. Non-blocking call. */
PLASMA_FETCH2,
/** Request status of an object, i.e., whether the object is stored in the
* local Plasma Store, in a remote Plasma Store, in transfer, or doesn't
* exist in the system. */
PLASMA_STATUS,
/** Wait until an object becomes available. */
PLASMA_WAIT,
/** Wait until an object becomes available. */
PLASMA_WAIT1,
/** Wait until an object becomes available. */
PLASMA_WAIT2
PLASMA_WAIT
};
typedef struct {
src/plasma/plasma_client.c
+29 -174
View File
@@ -552,97 +552,20 @@ void plasma_transfer(plasma_connection *conn,
void plasma_fetch(plasma_connection *conn,
int num_object_ids,
object_id object_ids[],
int is_fetched[]) {
CHECK(conn->manager_conn >= 0);
/* Make sure that there are no duplicated object IDs. TODO(rkn): we should
* allow this case in the future. */
CHECK(plasma_object_ids_distinct(num_object_ids, object_ids));
plasma_request *req = plasma_alloc_request(num_object_ids);
for (int i = 0; i < num_object_ids; ++i) {
req->object_requests[i].object_id = object_ids[i];
}
LOG_DEBUG("Requesting fetch");
CHECK(plasma_send_request(conn->manager_conn, PLASMA_FETCH, req) >= 0);
free(req);
plasma_reply reply;
int success;
for (int received = 0; received < num_object_ids; ++received) {
CHECK(plasma_receive_reply(conn->manager_conn, sizeof(reply), &reply) >= 0);
CHECK(reply.num_object_ids == 1);
success = reply.has_object;
/* Update the correct index in is_fetched. */
int i = 0;
for (; i < num_object_ids; ++i) {
if (object_ids_equal(object_ids[i], reply.object_requests[0].object_id) &&
!is_fetched[i]) {
is_fetched[i] = success;
break;
}
}
CHECKM(i != num_object_ids,
"Received an unexpected object ID from manager during fetch or the "
"object ID was received multiple times.");
}
}
void plasma_fetch2(plasma_connection *conn,
int num_object_ids,
object_id object_ids[]) {
object_id object_ids[]) {
CHECK(conn != NULL);
CHECK(conn->manager_conn >= 0);
plasma_request *req = plasma_alloc_request(num_object_ids);
for (int i = 0; i < num_object_ids; ++i) {
req->object_requests[i].object_id = object_ids[i];
}
CHECK(plasma_send_request(conn->manager_conn, PLASMA_FETCH2, req) >= 0);
}
int plasma_wait(plasma_connection *conn,
int num_object_ids,
object_id object_ids[],
uint64_t timeout,
int num_returns,
object_id return_object_ids[]) {
CHECK(conn->manager_conn >= 0);
plasma_request *req = plasma_alloc_request(num_object_ids);
for (int i = 0; i < num_object_ids; ++i) {
req->object_requests[i].object_id = object_ids[i];
}
req->num_ready_objects = num_returns;
req->timeout = timeout;
CHECK(plasma_send_request(conn->manager_conn, PLASMA_WAIT, req) >= 0);
plasma_free_request(req);
int64_t return_size = plasma_reply_size(num_returns);
plasma_reply *reply = malloc(return_size);
CHECK(plasma_receive_reply(conn->manager_conn, return_size, reply) >= 0);
for (int i = 0; i < num_returns; ++i) {
return_object_ids[i] = reply->object_requests[i].object_id;
}
int num_objects_returned = reply->num_objects_returned;
free(reply);
return num_objects_returned;
CHECK(plasma_send_request(conn->manager_conn, PLASMA_FETCH, req) >= 0);
}
int get_manager_fd(plasma_connection *conn) {
return conn->manager_conn;
}
/** === ALTERNATE PLASMA CLIENT API ===
* This API simplifies the previous one in two ways. First if factors out
* object (re)construction from the Plasma Manager. Second, except for
* plasma_wait_for_objects() all other functions are non-blocking.
*
* TODO:
* - plasma_info() not implemented yet, but not needed at this point.
* - assume new implementation of object_table_subscribe() which returns
* if object is in the Local Store (check with jpm).
* - need to phase out old API and drope *1 from the names of the functions
* once the old ones are dropped.
*/
bool plasma_get_local(plasma_connection *conn,
object_id object_id,
object_buffer *object_buffer) {
@@ -692,20 +615,6 @@ bool plasma_get_local(plasma_connection *conn,
return true;
}
int plasma_fetch_remote(plasma_connection *conn, object_id object_id) {
CHECK(conn != NULL);
CHECK(conn->manager_conn >= 0);
plasma_request req = plasma_make_request(object_id);
CHECK(plasma_send_request(conn->manager_conn, PLASMA_FETCH_REMOTE, &req) >=
0);
plasma_reply reply;
CHECK(plasma_receive_reply(conn->manager_conn, sizeof(reply), &reply) >= 0);
return reply.object_status;
}
int plasma_status(plasma_connection *conn, object_id object_id) {
CHECK(conn != NULL);
CHECK(conn->manager_conn >= 0);
@@ -719,57 +628,11 @@ int plasma_status(plasma_connection *conn, object_id object_id) {
return reply.object_status;
}
int plasma_wait_for_objects(plasma_connection *conn,
int num_object_requests,
object_request object_requests[],
int num_ready_objects,
uint64_t timeout_ms) {
CHECK(conn != NULL);
CHECK(conn->manager_conn >= 0);
CHECK(num_object_requests > 0);
plasma_request *req = plasma_alloc_request(num_object_requests);
for (int i = 0; i < num_object_requests; ++i) {
req->object_requests[i] = object_requests[i];
}
req->num_ready_objects = num_ready_objects;
req->timeout = timeout_ms;
CHECK(plasma_send_request(conn->manager_conn, PLASMA_WAIT1, req) >= 0);
free(req);
plasma_reply *reply = plasma_alloc_reply(num_object_requests);
CHECK(plasma_receive_reply(conn->manager_conn,
plasma_reply_size(num_object_requests),
reply) >= 0);
int num_objects_ready = 0;
for (int i = 0; i < num_object_requests; ++i) {
int type, status;
object_requests[i].object_id = reply->object_requests[i].object_id;
type = reply->object_requests[i].type;
object_requests[i].type = type;
status = reply->object_requests[i].status;
object_requests[i].status = status;
if (type == PLASMA_QUERY_LOCAL) {
if (status == PLASMA_OBJECT_LOCAL) {
num_objects_ready += 1;
}
} else {
CHECK(type == PLASMA_QUERY_ANYWHERE);
if (status == PLASMA_OBJECT_LOCAL || status == PLASMA_OBJECT_REMOTE) {
num_objects_ready += 1;
}
}
}
free(reply);
return num_objects_ready;
}
int plasma_wait_for_objects2(plasma_connection *conn,
int num_object_requests,
object_request object_requests[],
int num_ready_objects,
uint64_t timeout_ms) {
int plasma_wait(plasma_connection *conn,
int num_object_requests,
object_request object_requests[],
int num_ready_objects,
uint64_t timeout_ms) {
CHECK(conn != NULL);
CHECK(conn->manager_conn >= 0);
CHECK(num_object_requests > 0);
@@ -784,7 +647,7 @@ int plasma_wait_for_objects2(plasma_connection *conn,
}
req->num_ready_objects = num_ready_objects;
req->timeout = timeout_ms;
CHECK(plasma_send_request(conn->manager_conn, PLASMA_WAIT2, req) >= 0);
CHECK(plasma_send_request(conn->manager_conn, PLASMA_WAIT, req) >= 0);
free(req);
plasma_reply *reply = plasma_alloc_reply(num_object_requests);
@@ -827,35 +690,37 @@ int plasma_wait_for_objects2(plasma_connection *conn,
*/
void plasma_client_get(plasma_connection *conn,
object_id object_id,
object_id obj_id,
object_buffer *object_buffer) {
CHECK(conn != NULL);
CHECK(conn->manager_conn >= 0);
object_request request;
request.object_id = object_id;
request.object_id = obj_id;
while (true) {
if (plasma_get_local(conn, object_id, object_buffer)) {
if (plasma_get_local(conn, obj_id, object_buffer)) {
/* Object is in the local Plasma Store, and it is sealed. */
return;
}
switch (plasma_fetch_remote(conn, object_id)) {
object_id object_ids[1] = {obj_id};
plasma_fetch(conn, 1, object_ids);
switch (plasma_status(conn, obj_id)) {
case PLASMA_OBJECT_LOCAL:
/* Object has finished being transfered just after calling
* plasma_get_local(), and it is now in the local Plasma Store. Loop again
* to call plasma_get_local() and eventually return. */
continue;
case PLASMA_OBJECT_REMOTE:
/* A fetch request has been already scheduled for object_id, so wait for
/* A fetch request has been already scheduled for obj_id, so wait for
* it to complete. */
request.type = PLASMA_QUERY_LOCAL;
break;
case PLASMA_OBJECT_NONEXISTENT:
/* Object doesn’t exist in the system so ask local scheduler to create it.
*/
/* TODO: scheduler_create_object(object_id); */
/* TODO: scheduler_create_object(obj_id); */
/* Wait for the object to be (re)constructed and sealed either in the
* local Plasma Store or remotely. */
request.type = PLASMA_QUERY_ANYWHERE;
@@ -865,9 +730,9 @@ void plasma_client_get(plasma_connection *conn,
}
/*
* Wait for object_id to (1) be transferred and sealed in the local
* Wait for obj_id to (1) be transferred and sealed in the local
* Plasma Store, if available remotely, or (2) be (re)constructued either
* locally or remotely, if object_id didn't exist in the system.
* locally or remotely, if obj_id didn't exist in the system.
* - if timeout, next iteration will retry plasma_fetch() or
* scheduler_create_object()
* - if request.status == PLASMA_OBJECT_LOCAL, next iteration
@@ -878,7 +743,7 @@ void plasma_client_get(plasma_connection *conn,
* will call scheduler_create_object()
*/
#define TIMEOUT_WAIT_MS 200
plasma_wait_for_objects(conn, 1, &request, 1, TIMEOUT_WAIT_MS);
plasma_wait(conn, 1, &request, 1, TIMEOUT_WAIT_MS);
}
}
@@ -908,8 +773,8 @@ int plasma_client_wait(plasma_connection *conn,
struct timeval start, end;
gettimeofday(&start, NULL);
int n = plasma_wait_for_objects(conn, num_object_ids, requests, num_returns,
MIN(remaining_timeout, TIMEOUT_WAIT_MS));
int n = plasma_wait(conn, num_object_ids, requests, num_returns,
MIN(remaining_timeout, TIMEOUT_WAIT_MS));
gettimeofday(&end, NULL);
float diff_ms = (end.tv_sec - start.tv_sec);
@@ -961,30 +826,20 @@ void plasma_client_multiget(plasma_connection *conn,
while (true) {
int n;
/* Wait to get all objects in the system. The reason we call
* plasma_wait_for_objects() here instead of iterating over
* plasma_client_get() is to increase concurrency as plasma_client_get() is
* blocking. */
n = plasma_wait_for_objects(conn, num_object_ids, requests, num_object_ids,
TIMEOUT_WAIT_MS);
/* Issue a fetch command so the object IDs end up locally. */
plasma_fetch(conn, num_object_ids, object_ids);
/* Wait to get all objects in the system. The reason we call plasma_wait()
* here instead of iterating over plasma_client_get() is to increase
* concurrency as plasma_client_get() is blocking. */
n = plasma_wait(conn, num_object_ids, requests, num_object_ids,
TIMEOUT_WAIT_MS);
if (n == num_object_ids) {
/* All objects are in the system either on the local or a remote Plasma
* store, so we are done. */
break;
}
for (int i = 0; i < num_object_ids; ++i) {
if (requests[i].status == PLASMA_OBJECT_REMOTE) {
plasma_fetch_remote(conn, requests[i].object_id);
} else {
if (requests[i].status == PLASMA_OBJECT_NONEXISTENT) {
/* Object doesn’t exist so ask local scheduler to create it. */
/* TODO: scheduler_create_object(requests[i].object_id); */
printf("XXX Need to schedule object -- not implemented yet!\n");
}
}
}
}
/* Now get the data for every object. */
src/plasma/plasma_client.h
+18 -126
View File
@@ -184,38 +184,16 @@ void plasma_delete(plasma_connection *conn, object_id object_id);
*/
int64_t plasma_evict(plasma_connection *conn, int64_t num_bytes);
/**
* Fetch objects from remote plasma stores that have the
* objects stored.
*
* @param manager A file descriptor for the socket connection
* to the local manager.
* @param object_id_count The number of object IDs requested.
* @param object_ids[] The vector of object IDs requested. Length must be at
* least num_object_ids.
* @param is_fetched[] The vector in which to return the success
* of each object's fetch operation, in the same order as
* object_ids. Length must be at least num_object_ids.
* @return Void.
*/
void plasma_fetch(plasma_connection *conn,
int num_object_ids,
object_id object_ids[],
int is_fetched[]);
/**
* Attempt to initiate the transfer of some objects from remote Plasma Stores.
* This method does not guarantee that the fetched objects will arrive locally.
*
* For an object that is available in the local Plasma Store, this method will
* not do anything. For an object that is not available locally, it will check
* if the object are already being fetched. If so, it will not do anything. If
* not, it will query the object table for a list of Plasma Managers that have
* the object. If that list is non-empty, it will attempt to initiate transfers
* from one of those Plasma Managers. If the list is empty, it will set a
* callback to initiate a transfer when the list becomes non-empty.
*
* TODO(rkn): Setting the callback for when the list becomes non-empty is not
* implemented.
* the object. The object table will return a non-empty list, and this Plasma
* Manager will attempt to initiate transfers from one of those Plasma Managers.
*
* This function is non-blocking.
*
@@ -227,9 +205,9 @@ void plasma_fetch(plasma_connection *conn,
* @param object_ids The IDs of the objects that fetch is being called on.
* @return Void.
*/
void plasma_fetch2(plasma_connection *conn,
int num_object_ids,
object_id object_ids[]);
void plasma_fetch(plasma_connection *conn,
int num_object_ids,
object_id object_ids[]);
/**
* Transfer local object to a different plasma manager.
@@ -246,28 +224,6 @@ void plasma_transfer(plasma_connection *conn,
int port,
object_id object_id);
/**
* Wait for objects to be created (right now, wait for local objects).
*
* @param conn The object containing the connection state.
* @param num_object_ids Number of object IDs wait is called on.
* @param object_ids Object IDs wait is called on.
* @param timeout Wait will time out and return after this number of ms.
* @param num_returns Number of object IDs wait will return if it doesn't time
* out.
* @param return_object_ids Out parameter for the object IDs returned by wait.
* This is an array of size num_returns. If the number of objects that
* are ready when we time out, the objects will be stored in the last
* slots of the array and the number of objects is returned.
* @return Number of objects that are actually ready.
*/
int plasma_wait(plasma_connection *conn,
int num_object_ids,
object_id object_ids[],
uint64_t timeout,
int num_returns,
object_id return_object_ids[]);
/**
* 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
@@ -288,8 +244,6 @@ int plasma_subscribe(plasma_connection *conn);
*/
int get_manager_fd(plasma_connection *conn);
/* === ALTERNATE PLASMA CLIENT API === */
/**
* Object buffer data structure.
*/
@@ -319,38 +273,7 @@ bool plasma_get_local(plasma_connection *conn,
object_buffer *object_buffer);
/**
* Initiates the fetch (transfer) of an object from a remote Plasma Store.
*
* If the object is stored in the local Plasma Store, tell the caller.
*
* If not, check whether the object is stored on a remote Plasma Store. If yes,
* and if a transfer for the object has either been scheduled or is in progress,
* then return. Otherwise schedule a transfer for the object.
*
* If the object is not available locally or remotely, the client has to tell
* local scheduler to (re)create the object.
*
* This function is non-blocking.
*
* @param conn The object containing the connection state.
* @param object_id The ID of the object we want to transfer.
* @return Status as returned by the get_status() function. Status can take the
* following values.
* - PLASMA_CLIENT_LOCAL, if the object is stored in the local Plasma
* Store.
* - PLASMA_CLIENT_TRANSFER, if the object is either currently being
* transferred or the transfer has been scheduled.
* - PLASMA_CLIENT_REMOTE, if the object is stored at a remote Plasma
* Store.
* - PLASMA_CLIENT_DOES_NOT_EXIST, if the object doesn’t exist in the
* system.
*/
int plasma_fetch_remote(plasma_connection *conn, object_id object_id);
/**
* Return the status of a given object. This function is similar to
* plasma_fetch_remote() with the only difference that plamsa_fetch_remote()
* also schedules the obejct transfer, if not local.
* Return the status of a given object. This method may query the object table.
*
* @param conn The object containing the connection state.
* @param object_id The ID of the object whose status we query.
@@ -393,7 +316,9 @@ int plasma_info(plasma_connection *conn,
* "type" field.
* - A PLASMA_QUERY_LOCAL request is satisfied when object_id becomes
* available in the local Plasma Store. In this case, this function
* sets the "status" field to PLASMA_OBJECT_LOCAL.
* sets the "status" field to PLASMA_OBJECT_LOCAL. Note, if the status
* is not PLASMA_OBJECT_LOCAL, it will be PLASMA_OBJECT_NONEXISTENT,
* but it may exist elsewhere in the system.
* - A PLASMA_QUERY_ANYWHERE request is satisfied when object_id becomes
* available either at the local Plasma Store or on a remote Plasma
* Store. In this case, the functions sets the "status" field to
@@ -401,51 +326,18 @@ int plasma_info(plasma_connection *conn,
* @param num_ready_objects The number of requests in object_requests array that
* must be satisfied before the function returns, unless it timeouts.
* The num_ready_objects should be no larger than num_object_requests.
* @param timeout_ms Timeout value in milliseconds. If this timeout expires
* @param timeout_ms Timeout value in milliseconds. If this timeout expires
* before min_num_ready_objects of requests are satisfied, the function
* returns.
* @return Number of satisfied requests in the object_requests list. If the
* returned number is less than min_num_ready_objects this means that
* timeout expired.
*/
int plasma_wait_for_objects(plasma_connection *conn,
int num_object_requests,
object_request object_requests[],
int num_ready_objects,
uint64_t timeout_ms);
/**
* Wait for (1) a specified number of objects to be available (sealed) in the
* local Plasma Store or in a remote Plasma Store, or (2) for a timeout to
* expire. This is a blocking call.
*
* @param conn The object containing the connection state.
* @param num_object_requests Size of the object_requests array.
* @param object_requests Object event array. Each element contains a request
* for a particular object_id. The type of request is specified in the
* "type" field.
* - A PLASMA_QUERY_LOCAL request is satisfied when object_id becomes
* available in the local Plasma Store. In this case, this function
* sets the "status" field to PLASMA_OBJECT_LOCAL.
* - A PLASMA_QUERY_ANYWHERE request is satisfied when object_id becomes
* available either at the local Plasma Store or on a remote Plasma
* Store. In this case, the functions sets the "status" field to
* PLASMA_OBJECT_LOCAL or PLASMA_OBJECT_REMOTE.
* @param num_ready_objects The number of requests in object_requests array that
* must be satisfied before the function returns, unless it timeouts.
* The num_ready_objects should be no larger than num_object_requests.
* @param timeout_ms Timeout value in milliseconds. If this timeout expires
* before min_num_ready_objects of requests are satisfied, the function
* returns.
* @return Number of satisfied requests in the object_requests list. If the
* returned number is less than min_num_ready_objects this means that
* timeout expired.
*/
int plasma_wait_for_objects2(plasma_connection *conn,
int num_object_requests,
object_request object_requests[],
int num_ready_objects,
uint64_t timeout_ms);
int plasma_wait(plasma_connection *conn,
int num_object_requests,
object_request object_requests[],
int num_ready_objects,
uint64_t timeout_ms);
/**
* TODO: maybe move the plasma_client_* functions in another file.
@@ -498,8 +390,8 @@ int plasma_client_wait(plasma_connection *conn,
* @param num_object_ids The number of objects in the array to be returned.
* @param object_ids The array of object IDs to be returned.
* @param object_buffers The array of data structure where the information of
* the return objects will be stored. The objects appear
* in the same order as their IDs in the object_ids array,
* the return objects will be stored. The objects appear in the same
* order as their IDs in the object_ids array,
* @return Void.
*/
void plasma_client_multiget(plasma_connection *conn,
src/plasma/plasma_extension.c
+4 -109
View File
@@ -170,49 +170,7 @@ PyObject *PyPlasma_fetch(PyObject *self, PyObject *args) {
for (int i = 0; i < n; ++i) {
PyObjectToUniqueID(PyList_GetItem(object_id_list, i), &object_ids[i]);
}
/* Check that there are no duplicate object IDs. TODO(rkn): we should allow
* this in the future. */
if (!plasma_object_ids_distinct(n, object_ids)) {
PyErr_SetString(PyExc_RuntimeError,
"The same object ID is used multiple times in this call to "
"fetch.");
return NULL;
}
int *success_array = malloc(sizeof(int) * n);
memset(success_array, 0, sizeof(int) * n);
plasma_fetch(conn, (int) n, object_ids, success_array);
PyObject *success_list = PyList_New(n);
for (int i = 0; i < n; ++i) {
if (success_array[i]) {
Py_INCREF(Py_True);
PyList_SetItem(success_list, i, Py_True);
} else {
Py_INCREF(Py_False);
PyList_SetItem(success_list, i, Py_False);
}
}
free(object_ids);
free(success_array);
return success_list;
}
PyObject *PyPlasma_fetch2(PyObject *self, PyObject *args) {
plasma_connection *conn;
PyObject *object_id_list;
if (!PyArg_ParseTuple(args, "O&O", PyObjectToPlasmaConnection, &conn,
&object_id_list)) {
return NULL;
}
if (!plasma_manager_is_connected(conn)) {
PyErr_SetString(PyExc_RuntimeError, "Not connected to the plasma manager");
return NULL;
}
Py_ssize_t n = PyList_Size(object_id_list);
object_id *object_ids = malloc(sizeof(object_id) * n);
for (int i = 0; i < n; ++i) {
PyObjectToUniqueID(PyList_GetItem(object_id_list, i), &object_ids[i]);
}
plasma_fetch2(conn, (int) n, object_ids);
plasma_fetch(conn, (int) n, object_ids);
free(object_ids);
Py_RETURN_NONE;
}
@@ -250,67 +208,6 @@ PyObject *PyPlasma_wait(PyObject *self, PyObject *args) {
return NULL;
}
object_id *object_ids = malloc(sizeof(object_id) * n);
for (int i = 0; i < n; ++i) {
PyObjectToUniqueID(PyList_GetItem(object_id_list, i), &object_ids[i]);
}
object_id *return_ids = malloc(sizeof(object_id) * num_returns);
/* Drop the global interpreter lock while we are waiting, so other threads can
* run. */
int num_return_objects;
Py_BEGIN_ALLOW_THREADS;
num_return_objects = plasma_wait(conn, (int) n, object_ids,
(uint64_t) timeout, num_returns, return_ids);
Py_END_ALLOW_THREADS;
PyObject *ready_ids = PyList_New(num_return_objects);
PyObject *waiting_ids = PySet_New(object_id_list);
for (int i = num_returns - num_return_objects; i < num_returns; ++i) {
PyObject *ready =
PyString_FromStringAndSize((char *) return_ids[i].id, UNIQUE_ID_SIZE);
PyList_SetItem(ready_ids, i - (num_returns - num_return_objects), ready);
PySet_Discard(waiting_ids, ready);
}
PyObject *t = PyTuple_New(2);
PyTuple_SetItem(t, 0, ready_ids);
PyTuple_SetItem(t, 1, waiting_ids);
return t;
}
PyObject *PyPlasma_wait2(PyObject *self, PyObject *args) {
plasma_connection *conn;
PyObject *object_id_list;
long long timeout;
int num_returns;
if (!PyArg_ParseTuple(args, "O&OLi", PyObjectToPlasmaConnection, &conn,
&object_id_list, &timeout, &num_returns)) {
return NULL;
}
Py_ssize_t n = PyList_Size(object_id_list);
if (!plasma_manager_is_connected(conn)) {
PyErr_SetString(PyExc_RuntimeError, "Not connected to the plasma manager");
return NULL;
}
if (num_returns < 0) {
PyErr_SetString(PyExc_RuntimeError,
"The argument num_returns cannot be less than zero.");
return NULL;
}
if (num_returns > n) {
PyErr_SetString(
PyExc_RuntimeError,
"The argument num_returns cannot be greater than len(object_ids)");
return NULL;
}
int64_t threshold = 1 << 30;
if (timeout > threshold) {
PyErr_SetString(PyExc_RuntimeError,
"The argument timeout cannot be greater than 2 ** 30.");
return NULL;
}
object_request *object_requests = malloc(sizeof(object_request) * n);
for (int i = 0; i < n; ++i) {
PyObjectToUniqueID(PyList_GetItem(object_id_list, i),
@@ -321,8 +218,8 @@ PyObject *PyPlasma_wait2(PyObject *self, PyObject *args) {
* run. */
int num_return_objects;
Py_BEGIN_ALLOW_THREADS;
num_return_objects = plasma_wait_for_objects2(
conn, (int) n, object_requests, num_returns, (uint64_t) timeout);
num_return_objects = plasma_wait(conn, (int) n, object_requests, num_returns,
(uint64_t) timeout);
Py_END_ALLOW_THREADS;
int num_to_return = MIN(num_return_objects, num_returns);
@@ -444,9 +341,7 @@ static PyMethodDef plasma_methods[] = {
"Does the plasma store contain this plasma object?"},
{"fetch", PyPlasma_fetch, METH_VARARGS,
"Fetch the object from another plasma manager instance."},
{"fetch2", PyPlasma_fetch2, METH_VARARGS,
"Fetch the object from another plasma manager instance."},
{"wait", PyPlasma_wait2, METH_VARARGS,
{"wait", PyPlasma_wait, METH_VARARGS,
"Wait until num_returns objects in object_ids are ready."},
{"evict", PyPlasma_evict, METH_VARARGS,
"Evict some objects until we recover some number of bytes."},
src/plasma/plasma_manager.c
+176 -831
View File
File diff suppressed because it is too large Load Diff
src/plasma/plasma_manager.h
+13 -125
View File
@@ -111,58 +111,6 @@ void process_data_chunk(event_loop *loop,
void *context,
int events);
/**
* Process a fetch request. The fetch request tries:
* 1) If there is no connection to the database, return faliure to the client.
* 2) If the object is available locally, return success to the client.
* 3) Query the database for plasma managers that the object might be on.
* 4) Request a transfer from each of the managers that the object might be on
* until we receive the data, or until we timeout.
* 5) Returns success or failure to the client depending on whether we received
* the data or not.
*
* @param client_conn The connection context for the client that made the
* request.
* @param object_id The object ID requested.
* @return Void.
*/
void process_fetch_request(client_connection *client_conn, object_id object_id);
/**
* Process a fetch request for multiple objects. The success of each object
* will be written back individually to the socket connected to the client that
* made the request in a plasma_reply. See documentation for
* process_fetch_request for the sequence of operations per object.
*
* @param client_conn The connection context for the client that made the
* request.
* @param num_object_ids The number of object IDs requested.
* @param object_requests[] The object requests fetch is called on.
* @return Void.
*/
void process_fetch_requests(client_connection *client_conn,
int num_object_ids,
object_request object_requests[]);
/**
* Process a wait request from a client.
*
* @param client_conn The connection context for the client that made the
* request.
* @param num_object_ids Number of object IDs wait is called on.
* @param object_requests The object requests wait is called on.
* @param timeout Wait will time out and return after this number of
* milliseconds.
* @param num_returns Number of object IDs wait will return if it doesn't time
* out.
* @return Void.
*/
void process_wait_request(client_connection *client_conn,
int num_object_ids,
object_request object_requests[],
uint64_t timeout,
int num_returns);
/**
* Callback that will be called when a new object becomes available.
*
@@ -232,36 +180,21 @@ struct plasma_request_buffer {
};
/**
* Create a new context for the given object ID with the given
* client connection and register it with the manager's
* outstanding fetch or wait requests and the client
* connection's active object contexts.
* Call the request_transfer method, which well attempt to get an object from
* a remote Plasma manager. If it is unable to get it from another Plasma
* manager, it will cycle through a list of Plasma managers that have the
* object.
*
* @param client_conn The client connection context.
* @param object_id The object ID whose context we want to
* create.
* @return A pointer to the newly created object context.
* @param object_id The object ID of the object to transfer.
* @param manager_count The number of managers that have the object.
* @param manager_vector The Plasma managers that have the object.
* @param context The plasma manager state.
* @return Void.
*/
client_object_request *add_object_request(client_connection *client_conn,
object_id object_id);
/**
* Given an object ID and the managers it can be found on, start requesting a
* transfer from the managers.
*
* @param object_id The object ID we want to request a transfer of.
* @param manager_count The number of managers the object can be found on.
* @param manager_vector A vector of the IP addresses of the managers that the
* object can be found on.
* @param context The context for the connection to this client.
*
* Initializes a new context for this client and object. Managers are tried in
* order until we receive the data or we timeout and run out of retries.
*/
void request_transfer(object_id object_id,
int manager_count,
const char *manager_vector[],
void *context);
void call_request_transfer(object_id object_id,
int manager_count,
const char *manager_vector[],
void *context);
/**
* Clean up and free an active object context. Deregister it from the
@@ -326,49 +259,4 @@ event_loop *get_event_loop(plasma_manager_state *state);
*/
int get_client_sock(client_connection *conn);
/**
* Process a wait request from a client.
*
* @param client_conn The connection context for the client that made the
* request.
* @param num_object_requests Number of object requests wait is called on.
* @param object_requests The array of bject requests wait is called on.
* @param timeout Wait will time out and return after this number of
* milliseconds.
* @param num_returns Number of object requests that will be satsified before
* wait will retunr, unless it timeouts.
* @return Void.
*/
void process_wait_request1(client_connection *client_conn,
int num_object_requests,
object_request object_requests[],
uint64_t timeout,
int num_ready_objects);
/**
* Callback to be invoked when object_id entry is changed in the
* Object Table. We assume that the change means the object is available.
*
* @param object_id ID of the object becoming available locally or remotely.
* @param user_context This is the client connection on which the wait has been
* called.
* @return Void.
*/
void wait_object_available_callback(object_id object_id,
int manager_count,
const char *manager_vector[],
void *user_context);
/**
* Object is available (sealed) in the local Object Store. This is part of
* executing wait operation.
*
* @param client_conn The client conection.
* @param user_context This is the client connection on which the wait has been
* called.
* @return Void.
*/
void wait_process_object_available_local(client_connection *client_conn,
object_id object_id);
#endif /* PLASMA_MANAGER_H */
src/plasma/test/client_tests.c
+22 -19
View File
@@ -44,7 +44,7 @@ TEST plasma_status_tests(void) {
PASS();
}
TEST plasma_fetch_remote_tests(void) {
TEST plasma_fetch_tests(void) {
plasma_connection *plasma_conn1 = plasma_connect(
"/tmp/store1", "/tmp/manager1", PLASMA_DEFAULT_RELEASE_DELAY);
plasma_connection *plasma_conn2 = plasma_connect(
@@ -55,7 +55,7 @@ TEST plasma_fetch_remote_tests(void) {
int status;
/* No object in the system */
status = plasma_fetch_remote(plasma_conn1, oid1);
status = plasma_status(plasma_conn1, oid1);
ASSERT(status == PLASMA_OBJECT_NONEXISTENT);
/* Test for the object being in local Plasma store. */
@@ -70,23 +70,26 @@ TEST plasma_fetch_remote_tests(void) {
/* Object with ID oid1 has been just inserted. On the next fetch we might
* either find the object or not, depending on whether the Plasma Manager has
* received the notification from the Plasma Store or not. */
status = plasma_fetch_remote(plasma_conn1, oid1);
object_id oid_array1[1] = {oid1};
plasma_fetch(plasma_conn1, 1, oid_array1);
status = plasma_status(plasma_conn1, oid1);
ASSERT((status == PLASMA_OBJECT_LOCAL) ||
(status == PLASMA_OBJECT_NONEXISTENT));
/* Sleep to make sure Plasma Manager got the notification. */
sleep(1);
status = plasma_fetch_remote(plasma_conn1, oid1);
status = plasma_status(plasma_conn1, oid1);
ASSERT(status == PLASMA_OBJECT_LOCAL);
/* Test for object being remote. */
status = plasma_fetch_remote(plasma_conn2, oid1);
status = plasma_status(plasma_conn2, oid1);
ASSERT(status == PLASMA_OBJECT_REMOTE);
/* Sleep to make sure the object has been fetched and it is now stored in the
* local Plasma Store. */
plasma_fetch(plasma_conn2, 1, oid_array1);
sleep(1);
status = plasma_fetch_remote(plasma_conn2, oid1);
status = plasma_status(plasma_conn2, oid1);
ASSERT(status == PLASMA_OBJECT_LOCAL);
sleep(1);
@@ -160,8 +163,8 @@ TEST plasma_wait_for_objects_tests(void) {
struct timeval start, end;
gettimeofday(&start, NULL);
int n = plasma_wait_for_objects2(plasma_conn1, NUM_OBJ_REQUEST, obj_requests,
NUM_OBJ_REQUEST, WAIT_TIMEOUT_MS);
int n = plasma_wait(plasma_conn1, NUM_OBJ_REQUEST, obj_requests,
NUM_OBJ_REQUEST, WAIT_TIMEOUT_MS);
ASSERT(n == 0);
gettimeofday(&end, NULL);
float diff_ms = (end.tv_sec - start.tv_sec);
@@ -177,30 +180,30 @@ TEST plasma_wait_for_objects_tests(void) {
plasma_create(plasma_conn1, oid1, data_size, metadata, metadata_size, &data);
plasma_seal(plasma_conn1, oid1);
n = plasma_wait_for_objects2(plasma_conn1, NUM_OBJ_REQUEST, obj_requests,
NUM_OBJ_REQUEST, WAIT_TIMEOUT_MS);
n = plasma_wait(plasma_conn1, NUM_OBJ_REQUEST, obj_requests, NUM_OBJ_REQUEST,
WAIT_TIMEOUT_MS);
ASSERT(n == 1);
/* Create and insert an object in plasma_conn2. */
plasma_create(plasma_conn2, oid2, data_size, metadata, metadata_size, &data);
plasma_seal(plasma_conn2, oid2);
n = plasma_wait_for_objects2(plasma_conn1, NUM_OBJ_REQUEST, obj_requests,
NUM_OBJ_REQUEST, WAIT_TIMEOUT_MS);
n = plasma_wait(plasma_conn1, NUM_OBJ_REQUEST, obj_requests, NUM_OBJ_REQUEST,
WAIT_TIMEOUT_MS);
ASSERT(n == 2);
n = plasma_wait_for_objects2(plasma_conn2, NUM_OBJ_REQUEST, obj_requests,
NUM_OBJ_REQUEST, WAIT_TIMEOUT_MS);
n = plasma_wait(plasma_conn2, NUM_OBJ_REQUEST, obj_requests, NUM_OBJ_REQUEST,
WAIT_TIMEOUT_MS);
ASSERT(n == 2);
obj_requests[0].type = PLASMA_QUERY_LOCAL;
obj_requests[1].type = PLASMA_QUERY_LOCAL;
n = plasma_wait_for_objects2(plasma_conn1, NUM_OBJ_REQUEST, obj_requests,
NUM_OBJ_REQUEST, WAIT_TIMEOUT_MS);
n = plasma_wait(plasma_conn1, NUM_OBJ_REQUEST, obj_requests, NUM_OBJ_REQUEST,
WAIT_TIMEOUT_MS);
ASSERT(n == 1);
n = plasma_wait_for_objects2(plasma_conn2, NUM_OBJ_REQUEST, obj_requests,
NUM_OBJ_REQUEST, WAIT_TIMEOUT_MS);
n = plasma_wait(plasma_conn2, NUM_OBJ_REQUEST, obj_requests, NUM_OBJ_REQUEST,
WAIT_TIMEOUT_MS);
ASSERT(n == 1);
plasma_disconnect(plasma_conn1);
@@ -360,7 +363,7 @@ TEST plasma_multiget_tests(void) {
SUITE(plasma_client_tests) {
RUN_TEST(plasma_status_tests);
RUN_TEST(plasma_fetch_remote_tests);
RUN_TEST(plasma_fetch_tests);
RUN_TEST(plasma_get_local_tests);
RUN_TEST(plasma_wait_for_objects_tests);
RUN_TEST(plasma_get_tests);
src/plasma/test/manager_tests.c
+2 -42
View File
@@ -157,7 +157,7 @@ TEST request_transfer_test(void) {
utstring_new(addr);
utstring_printf(addr, "127.0.0.1:%d", remote_mock->port);
manager_vector[0] = utstring_body(addr);
request_transfer(oid, 1, manager_vector, local_mock->client_conn);
call_request_transfer(oid, 1, manager_vector, local_mock->state);
free(manager_vector);
event_loop_add_timer(local_mock->loop, MANAGER_TIMEOUT, test_done_handler,
local_mock->state);
@@ -203,7 +203,7 @@ TEST request_transfer_retry_test(void) {
utstring_new(addr1);
utstring_printf(addr1, "127.0.0.1:%d", remote_mock2->port);
manager_vector[1] = utstring_body(addr1);
request_transfer(oid, 2, manager_vector, local_mock->client_conn);
call_request_transfer(oid, 2, manager_vector, local_mock->state);
free(manager_vector);
event_loop_add_timer(local_mock->loop, MANAGER_TIMEOUT * 2, test_done_handler,
local_mock->state);
@@ -227,45 +227,6 @@ TEST request_transfer_retry_test(void) {
PASS();
}
/**
* This test checks correct behavior of request_transfer in a failure scenario.
* Specifically, when one plasma manager calls request_transfer, and the remote
* manager that holds the object is unreachable, the client should receive the
* failure message after all the retries have timed out.
* - Buffer a transfer request for the remote manager.
* - Start and stop the event loop after NUM_RETRIES timeouts to make sure that
* we trigger all the retries.
* - Expect to see a response on the plasma client saying that the object
* wasn't fetched.
*/
TEST request_transfer_timeout_test(void) {
plasma_mock *local_mock = init_plasma_mock(NULL);
plasma_mock *remote_mock = init_plasma_mock(local_mock);
const char **manager_vector = malloc(sizeof(char *));
UT_string *addr = NULL;
utstring_new(addr);
utstring_printf(addr, "127.0.0.1:%d", remote_mock->port);
manager_vector[0] = utstring_body(addr);
request_transfer(oid, 1, manager_vector, local_mock->client_conn);
free(manager_vector);
event_loop_add_timer(local_mock->loop, MANAGER_TIMEOUT * (NUM_RETRIES + 2),
test_done_handler, local_mock->state);
event_loop_run(local_mock->loop);
plasma_reply reply;
int manager_fd = get_manager_fd(local_mock->plasma_conn);
int nbytes = recv(manager_fd, (uint8_t *) &reply, sizeof(reply), MSG_WAITALL);
ASSERT_EQ(nbytes, sizeof(reply));
ASSERT_EQ(reply.num_object_ids, 1);
ASSERT(object_ids_equal(oid, reply.object_requests[0].object_id));
ASSERT_EQ(reply.has_object, 0);
/* Clean up. */
utstring_free(addr);
destroy_plasma_mock(remote_mock);
destroy_plasma_mock(local_mock);
PASS();
}
/**
* This test checks correct behavior of reading and writing an object chunk
* from one manager to another.
@@ -317,7 +278,6 @@ SUITE(plasma_manager_tests) {
memset(&oid, 1, sizeof(oid));
RUN_TEST(request_transfer_test);
RUN_TEST(request_transfer_retry_test);
RUN_TEST(request_transfer_timeout_test);
RUN_TEST(read_write_object_chunk_test);
}
src/plasma/test/test.py
+18 -77
View File
@@ -401,85 +401,26 @@ class TestPlasmaManager(unittest.TestCase):
self.redis_process.kill()
def test_fetch(self):
if self.redis_process is None:
print("Cannot test fetch without a running redis instance.")
self.assertTrue(False)
for _ in range(100):
# Create an object.
object_id1, memory_buffer1, metadata1 = create_object(self.client1, 2000, 2000)
# Fetch the object from the other plasma store.
# TODO(swang): This line is a hack! It makes sure that the entry will be
# in the object table once we call the fetch operation. Remove once
# retries are implemented by Ray common.
time.sleep(0.1)
successes = self.client2.fetch([object_id1])
self.assertEqual(successes, [True])
# Compare the two buffers.
assert_get_object_equal(self, self.client1, self.client2, object_id1,
memory_buffer=memory_buffer1, metadata=metadata1)
# Fetch in the other direction. These should return quickly because
# client1 already has the object.
successes = self.client1.fetch([object_id1])
self.assertEqual(successes, [True])
assert_get_object_equal(self, self.client2, self.client1, object_id1,
memory_buffer=memory_buffer1, metadata=metadata1)
def test_fetch_multiple(self):
if self.redis_process is None:
print("Cannot test fetch without a running redis instance.")
self.assertTrue(False)
for _ in range(20):
# Create two objects and a third fake one that doesn't exist.
object_id1, memory_buffer1, metadata1 = create_object(self.client1, 2000, 2000)
missing_object_id = random_object_id()
object_id2, memory_buffer2, metadata2 = create_object(self.client1, 2000, 2000)
object_ids = [object_id1, missing_object_id, object_id2]
# Fetch the objects from the other plasma store. The second object ID
# should timeout since it does not exist.
# TODO(swang): This line is a hack! It makes sure that the entry will be
# in the object table once we call the fetch operation. Remove once
# retries are implemented by Ray common.
time.sleep(0.1)
successes = self.client2.fetch(object_ids)
self.assertEqual(successes, [True, False, True])
# Compare the buffers of the objects that do exist.
assert_get_object_equal(self, self.client1, self.client2, object_id1,
memory_buffer=memory_buffer1, metadata=metadata1)
assert_get_object_equal(self, self.client1, self.client2, object_id2,
memory_buffer=memory_buffer2, metadata=metadata2)
# Fetch in the other direction. The fake object still does not exist.
successes = self.client1.fetch(object_ids)
self.assertEqual(successes, [True, False, True])
assert_get_object_equal(self, self.client2, self.client1, object_id1,
memory_buffer=memory_buffer1, metadata=metadata1)
assert_get_object_equal(self, self.client2, self.client1, object_id2,
memory_buffer=memory_buffer2, metadata=metadata2)
# Check that calling fetch with the same object ID fails.
object_id = random_object_id()
self.assertRaises(Exception, lambda : self.client1.fetch([object_id, object_id]))
def test_fetch2(self):
if self.redis_process is None:
print("Cannot test fetch without a running redis instance.")
self.assertTrue(False)
for _ in range(10):
# Create an object.
object_id1, memory_buffer1, metadata1 = create_object(self.client1, 2000, 2000)
self.client1.fetch2([object_id1])
self.client1.fetch([object_id1])
self.assertEqual(self.client1.contains(object_id1), True)
self.assertEqual(self.client2.contains(object_id1), False)
# Fetch the object from the other plasma manager.
# TODO(rkn): Right now we must wait for the object table to be updated.
while not self.client2.contains(object_id1):
self.client2.fetch2([object_id1])
self.client2.fetch([object_id1])
# Compare the two buffers.
assert_get_object_equal(self, self.client1, self.client2, object_id1,
memory_buffer=memory_buffer1, metadata=metadata1)
# Test that we can call fetch on object IDs that don't exist yet.
object_id2 = random_object_id()
self.client1.fetch2([object_id2])
self.client1.fetch([object_id2])
self.assertEqual(self.client1.contains(object_id2), False)
memory_buffer2, metadata2 = create_object_with_id(self.client2, object_id2, 2000, 2000)
# # Check that the object has been fetched.
@@ -493,19 +434,19 @@ class TestPlasmaManager(unittest.TestCase):
self.assertEqual(self.client1.contains(object_id3), False)
self.assertEqual(self.client2.contains(object_id3), False)
for _ in range(10):
self.client1.fetch2([object_id3])
self.client2.fetch2([object_id3])
self.client1.fetch([object_id3])
self.client2.fetch([object_id3])
memory_buffer3, metadata3 = create_object_with_id(self.client1, object_id3, 2000, 2000)
for _ in range(10):
self.client1.fetch2([object_id3])
self.client2.fetch2([object_id3])
self.client1.fetch([object_id3])
self.client2.fetch([object_id3])
#TODO(rkn): Right now we must wait for the object table to be updated.
while not self.client2.contains(object_id3):
self.client2.fetch2([object_id3])
self.client2.fetch([object_id3])
assert_get_object_equal(self, self.client1, self.client2, object_id3,
memory_buffer=memory_buffer3, metadata=metadata3)
def test_fetch2_multiple(self):
def test_fetch_multiple(self):
if self.redis_process is None:
print("Cannot test fetch without a running redis instance.")
self.assertTrue(False)
@@ -519,14 +460,14 @@ class TestPlasmaManager(unittest.TestCase):
# should timeout since it does not exist.
# TODO(rkn): Right now we must wait for the object table to be updated.
while (not self.client2.contains(object_id1)) or (not self.client2.contains(object_id2)):
self.client2.fetch2(object_ids)
self.client2.fetch(object_ids)
# Compare the buffers of the objects that do exist.
assert_get_object_equal(self, self.client1, self.client2, object_id1,
memory_buffer=memory_buffer1, metadata=metadata1)
assert_get_object_equal(self, self.client1, self.client2, object_id2,
memory_buffer=memory_buffer2, metadata=metadata2)
# Fetch in the other direction. The fake object still does not exist.
self.client1.fetch2(object_ids)
self.client1.fetch(object_ids)
assert_get_object_equal(self, self.client2, self.client1, object_id1,
memory_buffer=memory_buffer1, metadata=metadata1)
assert_get_object_equal(self, self.client2, self.client1, object_id2,
@@ -534,12 +475,12 @@ class TestPlasmaManager(unittest.TestCase):
# Check that we can call fetch with duplicated object IDs.
object_id3 = random_object_id()
self.client1.fetch2([object_id3, object_id3])
self.client1.fetch([object_id3, object_id3])
object_id4, memory_buffer4, metadata4 = create_object(self.client1, 2000, 2000)
time.sleep(0.1)
# TODO(rkn): Right now we must wait for the object table to be updated.
while not self.client2.contains(object_id4):
self.client2.fetch2([object_id3, object_id3, object_id4, object_id4])
self.client2.fetch([object_id3, object_id3, object_id4, object_id4])
assert_get_object_equal(self, self.client2, self.client1, object_id4,
memory_buffer=memory_buffer4, metadata=metadata4)
@@ -673,16 +614,16 @@ class TestPlasmaManager(unittest.TestCase):
self.client2.seal(object_id)
# Give the second manager some time to complete the seal, then make sure it
# exited.
time_left = 10
time_left = 100
while time_left > 0:
if self.p5.poll() != None:
self.processes_to_kill.remove(self.p5)
break
time_left -= 0.2
time.sleep(0.2)
time_left -= 0.1
time.sleep(0.1)
print("Time waiting for plasma manager to fail = {:.2}".format(10 - time_left))
self.assertNotEqual(self.p5.returncode, None)
print("Time waiting for plasma manager to fail = {:.2}".format(100 - time_left))
self.assertNotEqual(self.p5.poll(), None)
def test_illegal_functionality(self):
# Create an object id string.