wassname/ray
1
0
Fork 0
mirror of https://github.com/wassname/ray.git synced 2026-07-10 15:19:12 +08:00
Code Issues Packages Projects Releases Wiki Activity

Remove git submodules and C++ files.

Browse Source
This commit is contained in:
Robert Nishihara
2016-10-25 12:51:30 -07:00
parent 0a44145906
commit 7c1b2f702f
18 changed files with 0 additions and 3982 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitmodules
-14
View File
@@ -1,17 +1,3 @@
[submodule "thirdparty/grpc"]
path = thirdparty/grpc
url = https://github.com/grpc/grpc
ignore = dirty
[submodule "thirdparty/numbuf"]
path = thirdparty/numbuf
url = https://github.com/ray-project/numbuf.git
[submodule "thirdparty/arrow"]
path = thirdparty/arrow
url = https://github.com/ray-project/arrow.git
[submodule "thirdparty/python"]
path = thirdparty/python
url = https://github.com/austinsc/python.git
ignore = dirty
[submodule "thirdparty/hiredis"]
path = thirdparty/hiredis
url = https://github.com/redis/hiredis.git
src/computation_graph.cc
-27
View File
@@ -1,27 +0,0 @@
#include "computation_graph.h"
OperationId ComputationGraph::add_operation(std::unique_ptr<Operation> operation) {
OperationId operationid = operations_.size();
OperationId creator_operationid = operation->creator_operationid();
RAY_CHECK_EQ(spawned_operations_.size(), operationid, "ComputationGraph is attempting to call add_operation, but spawned_operations_.size() != operationid.");
operations_.emplace_back(std::move(operation));
if (creator_operationid != NO_OPERATION && creator_operationid != ROOT_OPERATION) {
spawned_operations_[creator_operationid].push_back(operationid);
}
spawned_operations_.push_back(std::vector<OperationId>());
return operationid;
}
const Task& ComputationGraph::get_task(OperationId operationid) {
RAY_CHECK_NEQ(operationid, ROOT_OPERATION, "ComputationGraph attempting to get_task with operationid == ROOT_OPERATION");
RAY_CHECK_NEQ(operationid, NO_OPERATION, "ComputationGraph attempting to get_task with operationid == NO_OPERATION");
RAY_CHECK_LT(operationid, operations_.size(), "ComputationGraph attempting to get_task with operationid " << operationid << ", but operationid >= operations_.size().");
RAY_CHECK(operations_[operationid]->has_task(), "Calling get_task with operationid " << operationid << ", but this corresponds to a put not a task.");
return operations_[operationid]->task();
}
void ComputationGraph::to_protobuf(CompGraph* computation_graph) {
for (OperationId id = 0; id < operations_.size(); ++id) {
computation_graph->add_operation()->CopyFrom(*operations_[id]);
}
}
src/computation_graph.h
-35
View File
@@ -1,35 +0,0 @@
#ifndef RAY_COMPUTATIONGRAPH_H
#define RAY_COMPUTATIONGRAPH_H
#include <iostream>
#include <limits>
#include "ray/ray.h"
#include "graph.pb.h"
#include "types.pb.h"
// used to represent the root operation (that is, the driver code)
const OperationId ROOT_OPERATION = std::numeric_limits<OperationId>::max();
// used to represent the absence of an operation
const OperationId NO_OPERATION = std::numeric_limits<OperationId>::max() - 1;
class ComputationGraph {
public:
// Add an operation to the computation graph, this returns the OperationId for
// the new operation. This method takes ownership over operation.
OperationId add_operation(std::unique_ptr<Operation> operation);
// Return the task corresponding to a particular OperationId. If operationid
// corresponds to a put, then fail.
const Task& get_task(OperationId operationid);
// Serialize the computation graph to ProtoBuf and store it in computation_graph
void to_protobuf(CompGraph* computation_graph);
private:
// maps an OperationId to the corresponding task or put
std::vector<std::unique_ptr<Operation> > operations_;
// spawned_operations_[operationid] is a vector of the OperationIds of the
// operations spawned by the task with OperationId operationid
std::vector<std::vector<OperationId> > spawned_operations_;
};
#endif
src/ipc.cc
-202
View File
@@ -1,202 +0,0 @@
#include "ipc.h"
#if defined(__unix__) || defined(__linux__)
#include <sys/statvfs.h>
#endif
#include <stdlib.h>
#include "ray/ray.h"
#include "utils.h"
ObjHandle::ObjHandle(SegmentId segmentid, size_t size, IpcPointer ipcpointer, size_t metadata_offset)
: segmentid_(segmentid), size_(size), ipcpointer_(ipcpointer), metadata_offset_(metadata_offset)
{}
MessageQueue<>::MessageQueue() : create_(false) { }
MessageQueue<>::~MessageQueue() {
if (!name_.empty() && create_) {
// Only remove the message queue if we created it.
RAY_LOG(RAY_DEBUG, "Removing message queue " << name_.c_str() << ", create = " << create_);
bip::message_queue::remove(name_.c_str());
}
}
MessageQueue<>::MessageQueue(MessageQueue&& other) {
*this = std::move(other);
}
MessageQueue<>& MessageQueue<>::operator=(MessageQueue&& other) {
name_ = std::move(other.name_);
create_ = other.create_;
queue_ = std::move(other.queue_);
other.name_.clear(); // It is unclear if this is guaranteed, but we need it to hold for the destructor. See: https://stackoverflow.com/a/17735913
return *this;
}
bool MessageQueue<>::connect(const std::string& name, bool create, size_t message_size, size_t message_capacity) {
name_ = name;
name_.insert(0, "ray-{BC200A09-2465-431D-AEC7-2F8530B04535}-");
#if defined(WIN32) || defined(_WIN32)
std::replace(name_.begin(), name_.end(), ':', '-');
#endif
try {
if (create) {
bip::message_queue::remove(name_.c_str()); // remove queue if it has not been properly removed from last run
queue_ = std::unique_ptr<bip::message_queue>(new bip::message_queue(bip::create_only, name_.c_str(), message_capacity, message_size));
create_ = true; // Only set create_ = true on success.
}
else {
queue_ = std::unique_ptr<bip::message_queue>(new bip::message_queue(bip::open_only, name_.c_str()));
}
}
catch (bip::interprocess_exception &ex) {
RAY_CHECK(false, "name = " << name_ << ", create = " << create << ", boost::interprocess exception: " << ex.what());
}
return true;
}
bool MessageQueue<>::connected() {
return queue_ != NULL;
}
bool MessageQueue<>::send(const void * object, size_t size) {
bool succeeded;
try {
// This will return true if the message was successfully sent and false if
// the message queue is full.
succeeded = queue_->try_send(object, size, 0);
}
catch (bip::interprocess_exception &ex) {
RAY_CHECK(false, "boost::interprocess exception: " << ex.what());
}
return succeeded;
}
bool MessageQueue<>::receive(void * object, size_t size) {
unsigned int priority;
bip::message_queue::size_type recvd_size;
try {
queue_->receive(object, size, recvd_size, priority);
}
catch (bip::interprocess_exception &ex) {
RAY_CHECK(false, "boost::interprocess exception: " << ex.what());
}
return true;
}
MemorySegmentPool::MemorySegmentPool(ObjStoreId objstoreid, std::string& objstore_address, bool create) : objstoreid_(objstoreid), objstore_address_(objstore_address), create_mode_(create) {
std::string::iterator split_point = split_ip_address(objstore_address);
objstore_port_.assign(split_point, objstore_address.end());
}
// creates a memory segment if it is not already there; if the pool is in create mode,
// space is allocated, if it is in open mode, the shared memory is mapped into the process
void MemorySegmentPool::open_segment(SegmentId segmentid, size_t size) {
RAY_LOG(RAY_DEBUG, "Opening segmentid " << segmentid << " on object store " << objstoreid_ << " with port " << objstore_port_ << " with create_mode_ = " << create_mode_);
RAY_CHECK(segmentid == segments_.size() || !create_mode_, "Object store " << objstoreid_ << " with port " << objstore_port_ << " is attempting to open segmentid " << segmentid << " on the object store, but segments_.size() = " << segments_.size());
if (segmentid >= segments_.size()) { // resize and initialize segments_
int current_size = segments_.size();
segments_.resize(segmentid + 1);
for (int i = current_size; i < segments_.size(); ++i) {
segments_[i].first = nullptr;
segments_[i].second = SegmentStatusType::UNOPENED;
}
}
if (segments_[segmentid].second == SegmentStatusType::OPENED) {
return;
}
RAY_CHECK_NEQ(segments_[segmentid].second, SegmentStatusType::CLOSED, "Attempting to open segmentid " << segmentid << ", but segments_[segmentid].second == SegmentStatusType::CLOSED.");
std::string segment_name = get_segment_name(segmentid);
if (create_mode_) {
assert(size > 0);
bip::shared_memory_object::remove(segment_name.c_str()); // remove segment if it has not been properly removed from last run
size_t new_size = (size / page_size_ + 2) * page_size_; // additional room for boost's bookkeeping
segments_[segmentid] = std::make_pair(std::unique_ptr<bip::managed_shared_memory>(new bip::managed_shared_memory(bip::create_only, segment_name.c_str(), new_size)), SegmentStatusType::OPENED);
} else {
segments_[segmentid] = std::make_pair(std::unique_ptr<bip::managed_shared_memory>(new bip::managed_shared_memory(bip::open_only, segment_name.c_str())), SegmentStatusType::OPENED);
}
}
void MemorySegmentPool::unmap_segment(SegmentId segmentid) {
segments_[segmentid].first.reset();
segments_[segmentid].second = SegmentStatusType::UNOPENED;
}
void MemorySegmentPool::close_segment(SegmentId segmentid) {
RAY_LOG(RAY_DEBUG, "closing segmentid " << segmentid);
std::string segment_name = get_segment_name(segmentid);
bip::shared_memory_object::remove(segment_name.c_str());
segments_[segmentid].first.reset();
segments_[segmentid].second = SegmentStatusType::CLOSED;
}
ObjHandle MemorySegmentPool::allocate(size_t size) {
RAY_CHECK(create_mode_, "Attempting to call allocate, but create_mode_ is false");
// TODO(pcm): at the moment, this always creates a new segment, this will be changed
SegmentId segmentid = segments_.size();
open_segment(segmentid, size);
objstore_memcheck(size);
void* ptr = segments_[segmentid].first->allocate(size);
auto handle = segments_[segmentid].first->get_handle_from_address(ptr);
return ObjHandle(segmentid, size, handle);
}
void MemorySegmentPool::deallocate(ObjHandle pointer) {
SegmentId segmentid = pointer.segmentid();
void* ptr = segments_[segmentid].first->get_address_from_handle(pointer.ipcpointer());
segments_[segmentid].first->deallocate(ptr);
close_segment(segmentid);
}
// returns address of the object refered to by the handle, needs to be called on
// the process that will use the address
uint8_t* MemorySegmentPool::get_address(ObjHandle pointer) {
RAY_CHECK(!create_mode_ || segments_[pointer.segmentid()].second == SegmentStatusType::OPENED, "Object store " << objstoreid_ << " is attempting to call get_address on segmentid " << pointer.segmentid() << ", which has not been opened yet.");
if (!create_mode_) {
open_segment(pointer.segmentid());
}
bip::managed_shared_memory* segment = segments_[pointer.segmentid()].first.get();
return static_cast<uint8_t*>(segment->get_address_from_handle(pointer.ipcpointer()));
}
// returns the name of the segment
std::string MemorySegmentPool::get_segment_name(SegmentId segmentid) {
return std::string("ray-{BC200A09-2465-431D-AEC7-2F8530B04535}-objstore-") + std::to_string(objstoreid_) + "-" + objstore_port_ + std::string("-segment-") + std::to_string(segmentid);
}
MemorySegmentPool::~MemorySegmentPool() {
destroy_segments();
}
void MemorySegmentPool::objstore_memcheck(int64_t size) {
#if defined(__unix__) || defined(__linux__)
struct statvfs buffer;
statvfs("/dev/shm/", &buffer);
if (size + 100 > buffer.f_bsize * buffer.f_bavail) {
MemorySegmentPool::destroy_segments();
RAY_LOG(RAY_FATAL, "Not enough memory for allocating object in objectstore.");
}
#endif
}
void MemorySegmentPool::destroy_segments() {
for (size_t segmentid = 0; segmentid < segments_.size(); ++segmentid) {
std::string segment_name = get_segment_name(segmentid);
segments_[segmentid].first.reset();
bip::shared_memory_object::remove(segment_name.c_str());
}
}
#if defined(WIN32) || defined(_WIN32)
namespace boost {
namespace interprocess {
namespace ipcdetail {
windows_bootstamp windows_intermodule_singleton<windows_bootstamp>::get() {
// HACK: Only do this for Windows as there seems to be no better workaround. Possibly undefined behavior!
return reinterpret_cast<windows_bootstamp const &>(std::string("BOOTSTAMP"));
}
}
}
}
#endif
src/ipc.h
-142
View File
@@ -1,142 +0,0 @@
#ifndef RAY_IPC_H
#define RAY_IPC_H
#include <iostream>
#include <limits>
#if defined(WIN32) || defined(_WIN32)
#include <boost/interprocess/detail/windows_intermodule_singleton.hpp>
namespace boost {
namespace interprocess {
namespace ipcdetail {
struct windows_bootstamp;
template<>
class windows_intermodule_singleton<windows_bootstamp> {
public:
static windows_bootstamp get();
};
}
}
}
#endif
#include <boost/interprocess/managed_shared_memory.hpp>
#include <boost/interprocess/ipc/message_queue.hpp>
#include "ray/ray.h"
namespace bip = boost::interprocess;
// Methods for inter process communication (abstracts from the shared memory implementation)
// Message Queues: Exchanging objects of type T between processes on a node
template<typename T = void>
class MessageQueue;
template<>
class MessageQueue<> {
public:
~MessageQueue();
MessageQueue();
MessageQueue(MessageQueue&& other);
MessageQueue& operator=(MessageQueue&& other);
bool connected();
protected:
bool connect(const std::string& name, bool create, size_t message_size, size_t message_capacity);
bool send(const void* object, size_t size);;
bool receive(void* object, size_t size);
private:
std::string name_;
bool create_;
std::unique_ptr<bip::message_queue> queue_;
};
template<typename T>
class MessageQueue : public MessageQueue<> {
public:
bool connect(const std::string& name, bool create, size_t capacity = 1000) { return MessageQueue<>::connect(name, create, sizeof(T), capacity); }
bool send(const T* object) { return MessageQueue<>::send(object, sizeof(*object)); };
bool receive(T* object) { return MessageQueue<>::receive(object, sizeof(*object)); }
};
// Object Queues
// For communicating between object store and workers, the following
// messages can be sent:
// ALLOC: workerid, objectid, size -> objhandle:
// worker requests an allocation from the object store
// GET: workerid, objectid -> objhandle:
// worker requests an object from the object store
// WORKER_DONE: workerid, objectid -> ():
// worker tells the object store that an object has been finalized
// ALIAS_DONE: objectid -> ():
// objstore tells itself that it has finalized something (perhaps an alias)
enum ObjRequestType {ALLOC = 0, GET = 1, WORKER_DONE = 2, ALIAS_DONE = 3};
struct ObjRequest {
WorkerId workerid; // worker that sends the request
ObjRequestType type; // do we want to allocate a new object or get a handle?
ObjectID objectid; // object ID of the object to be returned/allocated
int64_t size; // if allocate, that's the size of the object
int64_t metadata_offset; // if sending 'WORKER_DONE', that's the location of the metadata relative to the beginning of the object
};
typedef size_t SegmentId; // index into a memory segment table
typedef bip::managed_shared_memory::handle_t IpcPointer;
// Object handle: Handle to object that can be passed around between processes
// that are connected to the same object store
class ObjHandle {
public:
ObjHandle(SegmentId segmentid = 0, size_t size = 0, IpcPointer ipcpointer = IpcPointer(), size_t metadata_offset = 0);
SegmentId segmentid() { return segmentid_; }
size_t size() { return size_; }
IpcPointer ipcpointer() { return ipcpointer_; }
size_t metadata_offset() { return metadata_offset_; }
void set_metadata_offset(size_t metadata_offset) {metadata_offset_ = metadata_offset; }
private:
SegmentId segmentid_; // which shared memory file the object is stored in
IpcPointer ipcpointer_; // pointer to the beginning of the object, exchangeable between processes
size_t size_; // total size of the object
size_t metadata_offset_; // offset of the metadata that describes this object
};
// Memory segment pool: A collection of shared memory segments
// used in two modes:
// \item on the object store it is used with create = true, in this case the
// segments are allocated
// \item on the worker it is used in open mode, with create = false, in this case
// the segments, which have been created by the object store, are just mapped
// into memory
enum SegmentStatusType {UNOPENED = 0, OPENED = 1, CLOSED = 2};
class MemorySegmentPool {
public:
MemorySegmentPool(ObjStoreId objstoreid, std::string& objstore_address, bool create); // can be used in two modes: create mode and open mode (see above)
~MemorySegmentPool();
ObjHandle allocate(size_t nbytes); // allocate memory, potentially creating a new segment (only run on object store)
void deallocate(ObjHandle pointer); // deallocate object, potentially deallocating a new segment (only run on object store)
uint8_t* get_address(ObjHandle pointer); // get address of shared object
std::string get_segment_name(SegmentId segmentid); // get the name of a segment
void unmap_segment(SegmentId segmentid); // unmap a memory segment from a client (only to be called by clients)
void destroy_segments();
void objstore_memcheck(int64_t size);
private:
void open_segment(SegmentId segmentid, size_t size = 0); // create a segment or map an existing one into memory
void close_segment(SegmentId segmentid); // close a segment
bool create_mode_; // true in the object stores, false on the workers
ObjStoreId objstoreid_; // the identity of the associated object store
// The address of the object store.
std::string objstore_address_;
// The port of the object store. This is used to help avoid name collisions.
std::string objstore_port_;
size_t page_size_ = bip::mapped_region::get_page_size();
std::vector<std::pair<std::unique_ptr<bip::managed_shared_memory>, SegmentStatusType> > segments_;
};
#endif
src/objstore.cc
-375
View File
@@ -1,375 +0,0 @@
#include "objstore.h"
#include <chrono>
#include "utils.h"
const size_t ObjStoreService::CHUNK_SIZE = 8 * 1024;
// this method needs to be protected by a objstore_lock_
// TODO(rkn): Make sure that we do not in fact need the objstore_lock_. We want multiple deliveries to be able to happen simultaneously.
void ObjStoreService::get_data_from(ObjectID objectid, ObjStore::Stub& stub) {
RAY_LOG(RAY_DEBUG, "Objstore " << objstoreid_ << " is beginning to get objectid " << objectid);
ObjChunk chunk;
ClientContext context;
StreamObjToRequest stream_request;
stream_request.set_objectid(objectid);
std::unique_ptr<ClientReader<ObjChunk> > reader(stub.StreamObjTo(&context, stream_request));
size_t total_size = 0;
ObjHandle handle;
if (reader->Read(&chunk)) {
total_size = chunk.total_size();
handle = alloc(objectid, total_size);
}
size_t num_bytes = 0;
segmentpool_lock_.lock();
uint8_t* data = segmentpool_->get_address(handle);
segmentpool_lock_.unlock();
do {
RAY_CHECK_LE(num_bytes + chunk.data().size(), total_size, "The reader attempted to stream too many bytes.");
std::memcpy(data, chunk.data().c_str(), chunk.data().size());
data += chunk.data().size();
num_bytes += chunk.data().size();
} while (reader->Read(&chunk));
RAY_CHECK_GRPC(reader->Finish());
// finalize object
RAY_CHECK_EQ(num_bytes, total_size, "Streamed objectid " << objectid << ", but num_bytes != total_size");
object_ready(objectid, chunk.metadata_offset());
RAY_LOG(RAY_DEBUG, "finished streaming data, objectid was " << objectid << " and size was " << num_bytes);
}
ObjStoreService::ObjStoreService(std::shared_ptr<Channel> scheduler_channel)
: scheduler_stub_(Scheduler::NewStub(scheduler_channel)) {
}
void ObjStoreService::register_objstore(const std::string& objstore_address, const std::string& recv_queue_name) {
// Create the queue that will be used by workers to send requests to the
// object store.
RAY_LOG(RAY_INFO, "Object store is creating queue with name " << recv_queue_name);
RAY_CHECK(recv_queue_.connect(recv_queue_name, true), "error connecting recv_queue_");
objstore_address_ = objstore_address;
// Register the object store with the scheduler.
ClientContext context;
RegisterObjStoreRequest request;
request.set_objstore_address(objstore_address);
RegisterObjStoreReply reply;
RAY_CHECK_GRPC(scheduler_stub_->RegisterObjStore(&context, request, &reply));
objstoreid_ = reply.objstoreid();
segmentpool_ = std::make_shared<MemorySegmentPool>(objstoreid_, objstore_address_, true);
}
// this method needs to be protected by a objstores_lock_
ObjStore::Stub& ObjStoreService::get_objstore_stub(const std::string& objstore_address) {
auto iter = objstores_.find(objstore_address);
if (iter != objstores_.end())
return *(iter->second);
auto channel = grpc::CreateChannel(objstore_address, grpc::InsecureChannelCredentials());
objstores_.emplace(objstore_address, ObjStore::NewStub(channel));
return *objstores_[objstore_address];
}
Status ObjStoreService::StartDelivery(ServerContext* context, const StartDeliveryRequest* request, AckReply* reply) {
// TODO(rkn): We're pushing the delivery task onto a new thread so that this method can return immediately. This matters
// because the scheduler holds a lock while DeliverObj is being called. The correct solution is to make DeliverObj
// an asynchronous call (and similarly with the rest of the object store service methods).
std::string address = request->objstore_address();
ObjectID objectid = request->objectid();
{
std::lock_guard<std::mutex> memory_lock(memory_lock_);
if (objectid >= memory_.size()) {
memory_.resize(objectid + 1, std::make_pair(ObjHandle(), MemoryStatusType::NOT_PRESENT));
}
if (memory_[objectid].second == MemoryStatusType::NOT_PRESENT) {
}
else {
RAY_CHECK_NEQ(memory_[objectid].second, MemoryStatusType::DEALLOCATED, "Objstore " << objstoreid_ << " is attempting to get objectid " << objectid << ", but memory_[objectid] == DEALLOCATED.");
RAY_LOG(RAY_DEBUG, "Objstore " << objstoreid_ << " already has objectid " << objectid << " or it is already being shipped, so no need to get it again.");
return Status::OK;
}
memory_[objectid].second = MemoryStatusType::PRE_ALLOCED;
}
delivery_threads_.push_back(std::make_shared<std::thread>([this, address, objectid]() {
std::lock_guard<std::mutex> objstores_lock(objstores_lock_);
ObjStore::Stub& stub = get_objstore_stub(address);
get_data_from(objectid, stub);
}));
return Status::OK;
}
Status ObjStoreService::ObjStoreInfo(ServerContext* context, const ObjStoreInfoRequest* request, ObjStoreInfoReply* reply) {
std::lock_guard<std::mutex> memory_lock(memory_lock_);
for (size_t i = 0; i < memory_.size(); ++i) {
if (memory_[i].second == MemoryStatusType::READY) { // is the object available?
reply->add_objectid(i);
}
}
/*
for (int i = 0; i < request->objectid_size(); ++i) {
ObjectID objectid = request->objectid(i);
Obj* obj = new Obj();
std::string data(memory_[objectid].ptr.data, memory_[objectid].ptr.len); // copies, but for debugging should be ok
obj->ParseFromString(data);
reply->mutable_obj()->AddAllocated(obj);
}
*/
return Status::OK;
}
Status ObjStoreService::StreamObjTo(ServerContext* context, const StreamObjToRequest* request, ServerWriter<ObjChunk>* writer) {
RAY_LOG(RAY_DEBUG, "begin to stream data from object store " << objstoreid_);
ObjChunk chunk;
ObjectID objectid = request->objectid();
memory_lock_.lock();
RAY_CHECK_LT(objectid, memory_.size(), "Objstore " << objstoreid_ << " is attempting to use objectid " << objectid << " in StreamObjTo, but this objectid is not present in the object store.");
RAY_CHECK_EQ(memory_[objectid].second, MemoryStatusType::READY, "Objstore " << objstoreid_ << " is attempting to stream objectid " << objectid << ", but memory_[objectid].second != MemoryStatusType::READY.");
ObjHandle handle = memory_[objectid].first;
memory_lock_.unlock(); // TODO(rkn): Make sure we don't still need to hold on to this lock.
segmentpool_lock_.lock();
const uint8_t* head = segmentpool_->get_address(handle);
segmentpool_lock_.unlock();
size_t size = handle.size();
for (size_t i = 0; i < size; i += CHUNK_SIZE) {
chunk.set_metadata_offset(handle.metadata_offset());
chunk.set_total_size(size);
chunk.set_data(head + i, std::min(CHUNK_SIZE, size - i));
RAY_CHECK(writer->Write(chunk), "stream connection prematurely closed")
}
return Status::OK;
}
Status ObjStoreService::NotifyAlias(ServerContext* context, const NotifyAliasRequest* request, AckReply* reply) {
// NotifyAlias assumes that the objstore already holds canonical_objectid
ObjectID alias_objectid = request->alias_objectid();
ObjectID canonical_objectid = request->canonical_objectid();
RAY_LOG(RAY_DEBUG, "Aliasing objectid " << alias_objectid << " with objectid " << canonical_objectid);
{
std::lock_guard<std::mutex> memory_lock(memory_lock_);
RAY_CHECK_LT(canonical_objectid, memory_.size(), "Attempting to alias objectid " << alias_objectid << " with objectid " << canonical_objectid << ", but objectid " << canonical_objectid << " is not in the objstore.")
RAY_CHECK_NEQ(memory_[canonical_objectid].second, MemoryStatusType::NOT_READY, "Attempting to alias objectid " << alias_objectid << " with objectid " << canonical_objectid << ", but objectid " << canonical_objectid << " is not ready yet in the objstore.")
RAY_CHECK_NEQ(memory_[canonical_objectid].second, MemoryStatusType::NOT_PRESENT, "Attempting to alias objectid " << alias_objectid << " with objectid " << canonical_objectid << ", but objectid " << canonical_objectid << " is not present in the objstore.")
RAY_CHECK_NEQ(memory_[canonical_objectid].second, MemoryStatusType::DEALLOCATED, "Attempting to alias objectid " << alias_objectid << " with objectid " << canonical_objectid << ", but objectid " << canonical_objectid << " has already been deallocated.")
if (alias_objectid >= memory_.size()) {
memory_.resize(alias_objectid + 1, std::make_pair(ObjHandle(), MemoryStatusType::NOT_PRESENT));
}
memory_[alias_objectid].first = memory_[canonical_objectid].first;
memory_[alias_objectid].second = MemoryStatusType::READY;
}
ObjRequest done_request;
done_request.type = ObjRequestType::ALIAS_DONE;
done_request.objectid = alias_objectid;
RAY_CHECK(recv_queue_.send(&done_request), "Failed to send message from the object store to itself because the message queue was full.");
return Status::OK;
}
Status ObjStoreService::DeallocateObject(ServerContext* context, const DeallocateObjectRequest* request, AckReply* reply) {
ObjectID canonical_objectid = request->canonical_objectid();
RAY_LOG(RAY_INFO, "Deallocating canonical_objectid " << canonical_objectid);
std::lock_guard<std::mutex> memory_lock(memory_lock_);
RAY_CHECK_EQ(memory_[canonical_objectid].second, MemoryStatusType::READY, "Attempting to deallocate canonical_objectid " << canonical_objectid << ", but memory_[canonical_objectid].second = " << memory_[canonical_objectid].second);
RAY_CHECK_LT(canonical_objectid, memory_.size(), "Attempting to deallocate canonical_objectid " << canonical_objectid << ", but it is not in the objstore.");
segmentpool_lock_.lock();
segmentpool_->deallocate(memory_[canonical_objectid].first);
segmentpool_lock_.unlock();
memory_[canonical_objectid].second = MemoryStatusType::DEALLOCATED;
return Status::OK;
}
// This table describes how the memory status changes in response to requests.
//
// MemoryStatus | ObjRequest | New MemoryStatus | action performed
// -------------+-------------+------------------+----------------------------
// NOT_PRESENT | ALLOC | NOT_READY | allocate object
// NOT_READY | WORKER_DONE | READY | send ObjReady to scheduler
// NOT_READY | GET | NOT_READY | add to get queue
// READY | GET | READY | return handle
// READY | DEALLOC | DEALLOCATED | deallocate
// -------------+-------------+------------------+----------------------------
void ObjStoreService::process_objstore_request(const ObjRequest request) {
switch (request.type) {
case ObjRequestType::ALIAS_DONE: {
process_gets_for_objectid(request.objectid);
}
break;
default: {
RAY_CHECK(false, "Attempting to process request of type " << request.type << ". This code should be unreachable.");
}
}
}
void ObjStoreService::process_worker_request(const ObjRequest request) {
if (request.workerid >= send_queues_.size()) {
send_queues_.resize(request.workerid + 1);
}
if (!send_queues_[request.workerid].connected()) {
std::string queue_name = std::string("queue:") + objstore_address_ + std::string(":worker:") + std::to_string(request.workerid) + std::string(":obj");
RAY_CHECK(send_queues_[request.workerid].connect(queue_name, false), "error connecting receive_queue_");
}
{
std::lock_guard<std::mutex> memory_lock(memory_lock_);
if (request.objectid >= memory_.size()) {
memory_.resize(request.objectid + 1, std::make_pair(ObjHandle(), MemoryStatusType::NOT_PRESENT));
}
}
switch (request.type) {
case ObjRequestType::ALLOC: {
ObjHandle handle = alloc(request.objectid, request.size); // This method acquires memory_lock_
RAY_CHECK(send_queues_[request.workerid].send(&handle), "Failed to send message from the object store to the worker with id " << request.workerid << " because the message queue was full.");
}
break;
case ObjRequestType::GET: {
std::lock_guard<std::mutex> memory_lock(memory_lock_);
std::pair<ObjHandle, MemoryStatusType>& item = memory_[request.objectid];
if (item.second == MemoryStatusType::READY) {
RAY_LOG(RAY_DEBUG, "Responding to GET request: returning objectid " << request.objectid);
RAY_CHECK(send_queues_[request.workerid].send(&item.first), "Failed to send message from the object store to the worker with id " << request.workerid << " because the message queue was full.");
} else if (item.second == MemoryStatusType::NOT_READY || item.second == MemoryStatusType::NOT_PRESENT || item.second == MemoryStatusType::PRE_ALLOCED) {
std::lock_guard<std::mutex> lock(get_queue_lock_);
get_queue_.push_back(std::make_pair(request.workerid, request.objectid));
} else {
RAY_CHECK(false, "A worker requested objectid " << request.objectid << ", but memory_[objectid].second = " << memory_[request.objectid].second);
}
}
break;
case ObjRequestType::WORKER_DONE: {
object_ready(request.objectid, request.metadata_offset); // This method acquires memory_lock_
}
break;
default: {
RAY_CHECK(false, "Attempting to process request of type " << request.type << ". This code should be unreachable.");
}
}
}
void ObjStoreService::process_requests() {
// TODO(rkn): Should memory_lock_ be used in this method?
ObjRequest request;
while (true) {
RAY_CHECK(recv_queue_.receive(&request), "error receiving over IPC");
switch (request.type) {
case ObjRequestType::ALLOC: {
RAY_LOG(RAY_VERBOSE, "Request (worker " << request.workerid << " to objstore " << objstoreid_ << "): Allocate object with objectid " << request.objectid << " and size " << request.size);
process_worker_request(request);
}
break;
case ObjRequestType::GET: {
RAY_LOG(RAY_VERBOSE, "Request (worker " << request.workerid << " to objstore " << objstoreid_ << "): Get object with objectid " << request.objectid);
process_worker_request(request);
}
break;
case ObjRequestType::WORKER_DONE: {
RAY_LOG(RAY_VERBOSE, "Request (worker " << request.workerid << " to objstore " << objstoreid_ << "): Finalize object with objectid " << request.objectid);
process_worker_request(request);
}
break;
case ObjRequestType::ALIAS_DONE: {
process_objstore_request(request);
}
break;
default: {
RAY_CHECK(false, "Attempting to process request of type " << request.type << ". This code should be unreachable.");
}
}
}
}
void ObjStoreService::process_gets_for_objectid(ObjectID objectid) {
std::pair<ObjHandle, MemoryStatusType>& item = memory_[objectid];
std::lock_guard<std::mutex> get_queue_lock(get_queue_lock_);
for (size_t i = 0; i < get_queue_.size(); ++i) {
if (get_queue_[i].second == objectid) {
ObjHandle& elem = memory_[objectid].first;
RAY_CHECK(send_queues_[get_queue_[i].first].send(&item.first), "Failed to send message from the object store to the worker with id " << get_queue_[i].first << " because the message queue was full.");
// Remove the get task from the queue
std::swap(get_queue_[i], get_queue_[get_queue_.size() - 1]);
get_queue_.pop_back();
i -= 1;
}
}
}
ObjHandle ObjStoreService::alloc(ObjectID objectid, size_t size) {
segmentpool_lock_.lock();
ObjHandle handle = segmentpool_->allocate(size);
segmentpool_lock_.unlock();
std::lock_guard<std::mutex> memory_lock(memory_lock_);
RAY_LOG(RAY_VERBOSE, "Allocating space for objectid " << objectid << " on object store " << objstoreid_);
RAY_CHECK(memory_[objectid].second == MemoryStatusType::NOT_PRESENT || memory_[objectid].second == MemoryStatusType::PRE_ALLOCED, "Attempting to allocate space for objectid " << objectid << ", but memory_[objectid].second = " << memory_[objectid].second);
memory_[objectid].first = handle;
memory_[objectid].second = MemoryStatusType::NOT_READY;
return handle;
}
void ObjStoreService::object_ready(ObjectID objectid, size_t metadata_offset) {
{
RAY_LOG(RAY_INFO, "Object with ObjectID " << objectid << " is ready.");
std::lock_guard<std::mutex> memory_lock(memory_lock_);
std::pair<ObjHandle, MemoryStatusType>& item = memory_[objectid];
RAY_CHECK_EQ(item.second, MemoryStatusType::NOT_READY, "A worker notified the object store that objectid " << objectid << " has been written to the object store, but memory_[objectid].second != NOT_READY.");
item.first.set_metadata_offset(metadata_offset);
item.second = MemoryStatusType::READY;
}
process_gets_for_objectid(objectid);
// Tell the scheduler that the object arrived
// TODO(pcm): put this in a separate thread so we don't have to pay the latency here
ClientContext objready_context;
ObjReadyRequest objready_request;
objready_request.set_objectid(objectid);
objready_request.set_objstoreid(objstoreid_);
AckReply objready_reply;
RAY_CHECK_GRPC(scheduler_stub_->ObjReady(&objready_context, objready_request, &objready_reply));
}
void ObjStoreService::start_objstore_service() {
communicator_thread_ = std::thread([this]() {
RAY_LOG(RAY_INFO, "started object store communicator server");
process_requests();
});
}
void start_objstore(const char* scheduler_addr, const char* node_ip_address) {
RAY_LOG(RAY_INFO, "Starting an object store on node " << std::string(node_ip_address));
auto scheduler_channel = grpc::CreateChannel(scheduler_addr, grpc::InsecureChannelCredentials());
RAY_LOG(RAY_INFO, "Object store connected to scheduler " << scheduler_addr);
ObjStoreService service(scheduler_channel);
ServerBuilder builder;
// Get GRPC to assign an unused port.
int port;
builder.AddListeningPort(std::string("0.0.0.0:0"), grpc::InsecureServerCredentials(), &port);
builder.RegisterService(&service);
std::unique_ptr<Server> server(builder.BuildAndStart());
if (server == nullptr) {
RAY_CHECK(false, "Failed to create the object store service.");
}
std::string objstore_address = std::string(node_ip_address) + ":" + std::to_string(port);
RAY_LOG(RAY_INFO, "This object store has address " << objstore_address);
std::string recv_queue_name = std::string("queue:") + objstore_address + std::string(":obj");
service.register_objstore(objstore_address, recv_queue_name);
service.start_objstore_service();
// Process incoming GRPC calls. These may come from the scheduler or from
// other object stores. This method does not return.
server->Wait();
}
RayConfig global_ray_config;
int main(int argc, char** argv) {
RAY_CHECK_GE(argc, 3, "object store: expected at least two arguments (scheduler ip address and object store ip address)");
if (argc > 3) {
const char* log_file_name = get_cmd_option(argv, argv + argc, "--log-file-name");
if (log_file_name) {
std::cout << "object store: writing to log file " << log_file_name << std::endl;
create_log_dir_or_die(log_file_name);
global_ray_config.log_to_file = true;
global_ray_config.logfile.open(log_file_name);
} else {
std::cout << "object store: writing logs to stdout; you can change this by passing --log-file-name <filename> to ./scheduler" << std::endl;
global_ray_config.log_to_file = false;
}
}
start_objstore(argv[1], argv[2]);
return 0;
}
src/objstore.h
-81
View File
@@ -1,81 +0,0 @@
#ifndef RAY_OBJSTORE_H
#define RAY_OBJSTORE_H
#include <unordered_map>
#include <memory>
#include <thread>
#include <iostream>
#include <grpc++/grpc++.h>
#include "ray/ray.h"
#include "ray.grpc.pb.h"
#include "types.pb.h"
#include "ipc.h"
using grpc::Server;
using grpc::ServerBuilder;
using grpc::ServerReader;
using grpc::ServerContext;
using grpc::ClientContext;
using grpc::ServerWriter;
using grpc::ClientReader;
using grpc::Status;
using grpc::Channel;
// READY: This is used to indicate that the object has been copied from a
// worker and is ready to be used.
// NOT_READY: This is used to indicate that memory has been allocated for the
// object, but the object hasn't been copied from a worker yet.
// DEALLOCATED: This is used to indicate that the object has been deallocated.
// NOT_PRESENT: This is used to indicate that space has not been allocated for
// this object in this object store.
// PRE_ALLOCED: This is used to indicate that the memory has not yet been
// alloced, but it will be alloced soon. This is set when we call
// StartDelivery.
enum MemoryStatusType {READY = 0, NOT_READY = 1, DEALLOCATED = 2, NOT_PRESENT = 3, PRE_ALLOCED = 4};
class ObjStoreService final : public ObjStore::Service {
public:
ObjStoreService(std::shared_ptr<Channel> scheduler_channel);
Status StartDelivery(ServerContext* context, const StartDeliveryRequest* request, AckReply* reply) override;
Status StreamObjTo(ServerContext* context, const StreamObjToRequest* request, ServerWriter<ObjChunk>* writer) override;
Status NotifyAlias(ServerContext* context, const NotifyAliasRequest* request, AckReply* reply) override;
Status DeallocateObject(ServerContext* context, const DeallocateObjectRequest* request, AckReply* reply) override;
Status ObjStoreInfo(ServerContext* context, const ObjStoreInfoRequest* request, ObjStoreInfoReply* reply) override;
void start_objstore_service();
void register_objstore(const std::string& objstore_address, const std::string& recv_queue_name);
private:
void get_data_from(ObjectID objectid, ObjStore::Stub& stub);
// check if we already connected to the other objstore, if yes, return reference to connection, otherwise connect
ObjStore::Stub& get_objstore_stub(const std::string& objstore_address);
void process_worker_request(const ObjRequest request);
void process_objstore_request(const ObjRequest request);
void process_requests();
void process_gets_for_objectid(ObjectID objectid);
ObjHandle alloc(ObjectID objectid, size_t size);
void object_ready(ObjectID objectid, size_t metadata_offset);
static const size_t CHUNK_SIZE;
std::string objstore_address_;
ObjStoreId objstoreid_; // id of this objectstore in the scheduler object store table
std::shared_ptr<MemorySegmentPool> segmentpool_;
std::mutex segmentpool_lock_;
std::vector<std::pair<ObjHandle, MemoryStatusType> > memory_; // object ID -> (memory address, memory status)
std::mutex memory_lock_;
std::unordered_map<std::string, std::unique_ptr<ObjStore::Stub>> objstores_;
std::mutex objstores_lock_;
std::unique_ptr<Scheduler::Stub> scheduler_stub_;
std::vector<std::pair<WorkerId, ObjectID> > get_queue_;
std::mutex get_queue_lock_;
MessageQueue<ObjRequest> recv_queue_; // This queue is used by workers to send tasks to the object store.
std::vector<MessageQueue<ObjHandle> > send_queues_; // This maps workerid -> queue. The object store uses these queues to send replies to the relevant workers.
std::thread communicator_thread_;
std::vector<std::shared_ptr<std::thread> > delivery_threads_; // TODO(rkn): document
// TODO(rkn): possibly add lock, and properly remove these threads from the delivery_threads_ when the deliveries are done
};
#endif
src/raylib.cc
-870
View File
@@ -1,870 +0,0 @@
// TODO: - Implement other datatypes for ndarray
#define NPY_NO_DEPRECATED_API NPY_1_7_API_VERSION
#include <Python.h>
#include <structmember.h>
#define PY_ARRAY_UNIQUE_SYMBOL RAYLIB_ARRAY_API
#include <numpy/arrayobject.h>
#include <iostream>
#include "types.pb.h"
#include "worker.h"
#include "utils.h"
RayConfig global_ray_config;
extern "C" {
static int PyObjectToWorker(PyObject* object, Worker **worker);
// Object references
typedef struct {
PyObject_HEAD
ObjectID id;
// We give the PyObjectID object a reference to the worker capsule object to
// make sure that the worker capsule does not go out of scope until all of the
// object references have gone out of scope. The reason for this is that the
// worker capsule destructor destroys the worker object. If the worker object
// has been destroyed, then when the object reference tries to call
// worker->decrement_reference_count, we can get a segfault.
PyObject* worker_capsule;
} PyObjectID;
static void PyObjectID_dealloc(PyObjectID *self) {
Worker* worker;
PyObjectToWorker(self->worker_capsule, &worker);
std::vector<ObjectID> objectids;
objectids.push_back(self->id);
RAY_LOG(RAY_REFCOUNT, "In PyObjectID_dealloc, calling decrement_reference_count for objectid " << self->id);
worker->decrement_reference_count(objectids);
Py_DECREF(self->worker_capsule); // The corresponding increment happens in PyObjectID_init.
self->ob_type->tp_free((PyObject*) self);
}
static PyObject* PyObjectID_new(PyTypeObject *type, PyObject *args, PyObject *kwds) {
PyObjectID* self = (PyObjectID*) type->tp_alloc(type, 0);
if (self != NULL) {
self->id = 0;
}
return (PyObject*) self;
}
static int PyObjectID_init(PyObjectID *self, PyObject *args, PyObject *kwds) {
if (!PyArg_ParseTuple(args, "iO", &self->id, &self->worker_capsule)) {
return -1;
}
Worker* worker;
PyObjectToWorker(self->worker_capsule, &worker);
Py_INCREF(self->worker_capsule); // The corresponding decrement happens in PyObjectID_dealloc.
std::vector<ObjectID> objectids;
objectids.push_back(self->id);
RAY_LOG(RAY_REFCOUNT, "In PyObjectID_init, calling increment_reference_count for objectid " << objectids[0]);
worker->increment_reference_count(objectids);
return 0;
};
static int PyObjectID_compare(PyObject* a, PyObject* b) {
PyObjectID* A = (PyObjectID*) a;
PyObjectID* B = (PyObjectID*) b;
if (A->id < B->id) {
return -1;
}
if (A->id > B->id) {
return 1;
}
return 0;
}
static long PyObjectID_hash(PyObject* a) {
PyObjectID* A = (PyObjectID*) a;
PyObject* tuple = PyTuple_New(1);
PyTuple_SetItem(tuple, 0, PyInt_FromLong(A->id));
long hash = PyObject_Hash(tuple);
Py_XDECREF(tuple);
return hash;
}
char RAY_ID_LITERAL[] = "id";
char RAY_OBJECT_ID_LITERAL[] = "object id";
static PyMemberDef PyObjectID_members[] = {
{RAY_ID_LITERAL, T_INT, offsetof(PyObjectID, id), 0, RAY_OBJECT_ID_LITERAL},
{NULL}
};
static PyTypeObject PyObjectIDType = {
PyObject_HEAD_INIT(NULL)
0, /* ob_size */
"ray.ObjectID", /* tp_name */
sizeof(PyObjectID), /* tp_basicsize */
0, /* tp_itemsize */
(destructor)PyObjectID_dealloc, /* tp_dealloc */
0, /* tp_print */
0, /* tp_getattr */
0, /* tp_setattr */
PyObjectID_compare, /* tp_compare */
0, /* tp_repr */
0, /* tp_as_number */
0, /* tp_as_sequence */
0, /* tp_as_mapping */
PyObjectID_hash, /* tp_hash */
0, /* tp_call */
0, /* tp_str */
0, /* tp_getattro */
0, /* tp_setattro */
0, /* tp_as_buffer */
Py_TPFLAGS_DEFAULT, /* tp_flags */
"Ray objects", /* tp_doc */
0, /* tp_traverse */
0, /* tp_clear */
0, /* tp_richcompare */
0, /* tp_weaklistoffset */
0, /* tp_iter */
0, /* tp_iternext */
0, /* tp_methods */
PyObjectID_members, /* tp_members */
0, /* tp_getset */
0, /* tp_base */
0, /* tp_dict */
0, /* tp_descr_get */
0, /* tp_descr_set */
0, /* tp_dictoffset */
(initproc)PyObjectID_init, /* tp_init */
0, /* tp_alloc */
PyObjectID_new, /* tp_new */
};
// create PyObjectID from C++ (could be made more efficient if neccessary)
PyObject* make_pyobjectid(PyObject* worker_capsule, ObjectID objectid) {
PyObject* arglist = Py_BuildValue("(iO)", objectid, worker_capsule);
PyObject* result = PyObject_CallObject((PyObject*) &PyObjectIDType, arglist);
Py_DECREF(arglist);
return result;
}
// Error handling
static PyObject *RayError;
static PyObject *RaySizeError;
// Pass arguments from Python to C++
static int PyObjectToTask(PyObject* object, Task **task) {
if (PyCapsule_IsValid(object, "task")) {
*task = static_cast<Task*>(PyCapsule_GetPointer(object, "task"));
return 1;
} else {
PyErr_SetString(PyExc_TypeError, "must be a 'task' capsule");
return 0;
}
}
static int PyObjectToObj(PyObject* object, Obj **obj) {
if (PyCapsule_IsValid(object, "obj")) {
*obj = static_cast<Obj*>(PyCapsule_GetPointer(object, "obj"));
return 1;
} else {
PyErr_SetString(PyExc_TypeError, "must be a 'obj' capsule");
return 0;
}
}
static int PyObjectToWorker(PyObject* object, Worker **worker) {
if (PyCapsule_IsValid(object, "worker")) {
*worker = static_cast<Worker*>(PyCapsule_GetPointer(object, "worker"));
return 1;
} else {
PyErr_SetString(PyExc_TypeError, "must be a 'worker' capsule");
return 0;
}
}
static int PyObjectToObjectID(PyObject* object, ObjectID *objectid) {
if (PyObject_IsInstance(object, (PyObject*)&PyObjectIDType)) {
*objectid = ((PyObjectID*) object)->id;
return 1;
} else {
PyErr_SetString(PyExc_TypeError, "must be an object reference");
return 0;
}
}
// Destructors
static void ObjCapsule_Destructor(PyObject* capsule) {
Obj* obj = static_cast<Obj*>(PyCapsule_GetPointer(capsule, "obj"));
delete obj;
}
static void WorkerCapsule_Destructor(PyObject* capsule) {
Worker* obj = static_cast<Worker*>(PyCapsule_GetPointer(capsule, "worker"));
delete obj;
}
static void TaskCapsule_Destructor(PyObject* capsule) {
Task* obj = static_cast<Task*>(PyCapsule_GetPointer(capsule, "task"));
delete obj;
}
// Helper methods
// Pass ownership of both the key and the value to the PyDict.
// This is only required for PyDicts, not for PyLists or PyTuples, compare
// https://docs.python.org/2/c-api/dict.html
// https://docs.python.org/2/c-api/list.html
// https://docs.python.org/2/c-api/tuple.html
void set_dict_item_and_transfer_ownership(PyObject* dict, PyObject* key, PyObject* val) {
PyDict_SetItem(dict, key, val);
Py_XDECREF(key);
Py_XDECREF(val);
}
// This converts an Python ObjectID to an Python integer.
static PyObject* serialize_objectid(PyObject* self, PyObject* args) {
Worker* worker;
ObjectID objectid;
if (!PyArg_ParseTuple(args, "O&O&", &PyObjectToWorker, &worker, &PyObjectToObjectID, &objectid)) {
return NULL;
}
return PyInt_FromLong(objectid);
}
// This converts a Python integer to a Python ObjectID.
static PyObject* deserialize_objectid(PyObject* self, PyObject* args) {
PyObject* worker_capsule;
int objectid;
if (!PyArg_ParseTuple(args, "Oi", &worker_capsule, &objectid)) {
return NULL;
}
return make_pyobjectid(worker_capsule, static_cast<ObjectID>(objectid));
}
static PyObject* allocate_buffer(PyObject* self, PyObject* args) {
Worker* worker;
ObjectID objectid;
SegmentId segmentid;
long size;
if (!PyArg_ParseTuple(args, "O&O&l", &PyObjectToWorker, &worker, &PyObjectToObjectID, &objectid, &size)) {
return NULL;
}
void* address = reinterpret_cast<void*>(const_cast<char*>(worker->allocate_buffer(objectid, size, segmentid)));
std::vector<npy_intp> dim({size});
PyObject* t = PyTuple_New(2);
PyTuple_SetItem(t, 0, PyArray_SimpleNewFromData(1, dim.data(), NPY_BYTE, address));
PyTuple_SetItem(t, 1, PyInt_FromLong(segmentid));
return t;
}
static PyObject* finish_buffer(PyObject* self, PyObject* args) {
Worker* worker;
ObjectID objectid;
long segmentid;
long metadata_offset;
if (!PyArg_ParseTuple(args, "O&O&ll", &PyObjectToWorker, &worker, &PyObjectToObjectID, &objectid, &segmentid, &metadata_offset)) {
return NULL;
}
return worker->finish_buffer(objectid, segmentid, metadata_offset);
}
static PyObject* get_buffer(PyObject* self, PyObject* args) {
Worker* worker;
ObjectID objectid;
int64_t size;
SegmentId segmentid;
int64_t metadata_offset;
if (!PyArg_ParseTuple(args, "O&O&", &PyObjectToWorker, &worker, &PyObjectToObjectID, &objectid)) {
return NULL;
}
void* address = reinterpret_cast<void*>(const_cast<char*>(worker->get_buffer(objectid, size, segmentid, metadata_offset)));
std::vector<npy_intp> dim({static_cast<npy_intp>(size)});
PyObject* t = PyTuple_New(3);
PyTuple_SetItem(t, 0, PyArray_SimpleNewFromData(1, dim.data(), NPY_BYTE, address));
PyTuple_SetItem(t, 1, PyInt_FromLong(segmentid));
PyTuple_SetItem(t, 2, PyInt_FromLong(metadata_offset));
return t;
}
static PyObject* is_arrow(PyObject* self, PyObject* args) {
Worker* worker;
ObjectID objectid;
if (!PyArg_ParseTuple(args, "O&O&", &PyObjectToWorker, &worker, &PyObjectToObjectID, &objectid)) {
return NULL;
}
if (worker->is_arrow(objectid))
Py_RETURN_TRUE;
else
Py_RETURN_FALSE;
}
static PyObject* unmap_object(PyObject* self, PyObject* args) {
Worker* worker;
int segmentid;
if (!PyArg_ParseTuple(args, "O&i", &PyObjectToWorker, &worker, &segmentid)) {
return NULL;
}
worker->unmap_object(segmentid);
Py_RETURN_NONE;
}
static PyObject* serialize_task(PyObject* self, PyObject* args) {
PyObject* worker_capsule;
Task* task = new Task(); // TODO: to be freed in capsule destructor
char* name;
int len;
PyObject* arguments;
if (!PyArg_ParseTuple(args, "Os#O", &worker_capsule, &name, &len, &arguments)) {
return NULL;
}
task->set_name(std::string(name, len));
std::vector<ObjectID> objectids; // This is a vector of all the objectids that are serialized in this task, including objectids that are contained in Python objects that are passed by value.
if (PyList_Check(arguments)) {
for (size_t i = 0, size = PyList_Size(arguments); i < size; ++i) {
PyObject* element = PyList_GetItem(arguments, i);
if (PyObject_IsInstance(element, (PyObject*)&PyObjectIDType)) {
// Handle the case where the argument to the task is an ObjectID.
ObjectID objectid = ((PyObjectID*) element)->id;
task->add_arg()->set_objectid(objectid);
objectids.push_back(objectid);
} else if (PyString_CheckExact(element)) {
// Handle the case where the argument to the task is being passed by
// value and we receive an argument serialized as a string here.
char* buffer;
Py_ssize_t length;
PyString_AsStringAndSize(element, &buffer, &length);
task->add_arg()->set_serialized_arg(std::string(buffer, length));
} else {
RAY_CHECK(false, "This code should be unreachable.");
}
}
} else {
PyErr_SetString(RayError, "serialize_task: second argument needs to be a list");
return NULL;
}
Worker* worker;
PyObjectToWorker(worker_capsule, &worker);
if (objectids.size() > 0) {
RAY_LOG(RAY_REFCOUNT, "In serialize_task, calling increment_reference_count for contained objectids");
worker->increment_reference_count(objectids);
}
std::string output;
task->SerializeToString(&output);
int task_size = output.length();
return PyCapsule_New(static_cast<void*>(task), "task", &TaskCapsule_Destructor);
}
static PyObject* deserialize_task(PyObject* worker_capsule, const Task& task) {
std::vector<ObjectID> objectids; // This is a vector of all the objectids that were serialized in this task, including objectids that are contained in Python objects that are passed by value.
PyObject* string = PyString_FromStringAndSize(task.name().c_str(), task.name().size());
int argsize = task.arg_size();
PyObject* arglist = PyList_New(argsize);
for (int i = 0; i < argsize; ++i) {
if (task.arg(i).serialized_arg().empty()) {
PyList_SetItem(arglist, i, make_pyobjectid(worker_capsule, task.arg(i).objectid()));
objectids.push_back(task.arg(i).objectid());
} else {
PyObject* serialized_arg = PyString_FromStringAndSize(task.arg(i).serialized_arg().data(), task.arg(i).serialized_arg().size());
PyList_SetItem(arglist, i, serialized_arg);
}
}
Worker* worker;
PyObjectToWorker(worker_capsule, &worker);
worker->decrement_reference_count(objectids);
int resultsize = task.result_size();
std::vector<ObjectID> result_objectids;
PyObject* resultlist = PyList_New(resultsize);
for (int i = 0; i < resultsize; ++i) {
PyList_SetItem(resultlist, i, make_pyobjectid(worker_capsule, task.result(i)));
result_objectids.push_back(task.result(i));
}
worker->decrement_reference_count(result_objectids); // The corresponding increment is done in SubmitTask in the scheduler.
PyObject* t = PyTuple_New(3); // We set the items of the tuple using PyTuple_SetItem, because that transfers ownership to the tuple.
PyTuple_SetItem(t, 0, string);
PyTuple_SetItem(t, 1, arglist);
PyTuple_SetItem(t, 2, resultlist);
return t;
}
// Ray Python API
static PyObject* create_worker(PyObject* self, PyObject* args) {
const char* node_ip_address;
const char* scheduler_address;
// The object store address can be the empty string, in which case the
// scheduler will choose the object store address.
const char* objstore_address;
int mode;
const char* log_file_name;
if (!PyArg_ParseTuple(args, "sssis", &node_ip_address, &scheduler_address, &objstore_address, &mode, &log_file_name)) {
return NULL;
}
// Set the logging file.
create_log_dir_or_die(log_file_name);
global_ray_config.log_to_file = true;
global_ray_config.logfile.open(log_file_name);
// Create the worker.
bool is_driver = (mode != Mode::WORKER_MODE);
Worker* worker = new Worker(std::string(node_ip_address), std::string(scheduler_address), static_cast<Mode>(mode));
// Register the worker.
worker->register_worker(std::string(node_ip_address), std::string(objstore_address), is_driver);
PyObject* t = PyTuple_New(2);
PyObject* worker_capsule = PyCapsule_New(static_cast<void*>(worker), "worker", &WorkerCapsule_Destructor);
PyTuple_SetItem(t, 0, worker_capsule);
PyTuple_SetItem(t, 1, PyString_FromString(worker->get_worker_address()));
return t;
}
static PyObject* disconnect(PyObject* self, PyObject* args) {
Worker* worker;
if (!PyArg_ParseTuple(args, "O&", &PyObjectToWorker, &worker)) {
return NULL;
}
worker->disconnect();
Py_RETURN_NONE;
}
static PyObject* connected(PyObject* self, PyObject* args) {
Worker* worker;
if (!PyArg_ParseTuple(args, "O&", &PyObjectToWorker, &worker)) {
return NULL;
}
if (worker->connected()) {
Py_RETURN_TRUE;
}
Py_RETURN_FALSE;
}
static PyObject* wait_for_next_message(PyObject* self, PyObject* args) {
PyObject* worker_capsule;
if (!PyArg_ParseTuple(args, "O", &worker_capsule)) {
return NULL;
}
Worker* worker;
PyObjectToWorker(worker_capsule, &worker);
if (std::unique_ptr<WorkerMessage> message = worker->receive_next_message()) {
bool task_present = !message->task().name().empty();
bool function_present = !message->function().implementation().empty();
bool reusable_variable_present = !message->reusable_variable().name().empty();
bool function_to_run_present = !message->function_to_run().implementation().empty();
RAY_CHECK(task_present + function_present + reusable_variable_present + function_to_run_present <= 1, "The worker message should contain at most one item.");
PyObject* t = PyTuple_New(2);
if (task_present) {
PyTuple_SetItem(t, 0, PyString_FromString("task"));
PyTuple_SetItem(t, 1, deserialize_task(worker_capsule, message->task()));
} else if (function_present) {
PyTuple_SetItem(t, 0, PyString_FromString("function"));
PyObject* remote_function_data = PyTuple_New(2);
PyTuple_SetItem(remote_function_data, 0, PyString_FromStringAndSize(message->function().name().data(), static_cast<ssize_t>(message->function().name().size())));
PyTuple_SetItem(remote_function_data, 1, PyString_FromStringAndSize(message->function().implementation().data(), static_cast<ssize_t>(message->function().implementation().size())));
PyTuple_SetItem(t, 1, remote_function_data);
} else if (reusable_variable_present) {
PyTuple_SetItem(t, 0, PyString_FromString("reusable_variable"));
PyObject* reusable_variable = PyTuple_New(3);
PyTuple_SetItem(reusable_variable, 0, PyString_FromStringAndSize(message->reusable_variable().name().data(), static_cast<ssize_t>(message->reusable_variable().name().size())));
PyTuple_SetItem(reusable_variable, 1, PyString_FromStringAndSize(message->reusable_variable().initializer().implementation().data(), static_cast<ssize_t>(message->reusable_variable().initializer().implementation().size())));
PyTuple_SetItem(reusable_variable, 2, PyString_FromStringAndSize(message->reusable_variable().reinitializer().implementation().data(), static_cast<ssize_t>(message->reusable_variable().reinitializer().implementation().size())));
PyTuple_SetItem(t, 1, reusable_variable);
} else if (function_to_run_present) {
PyTuple_SetItem(t, 0, PyString_FromString("function_to_run"));
PyTuple_SetItem(t, 1, PyString_FromStringAndSize(message->function_to_run().implementation().data(), static_cast<ssize_t>(message->function_to_run().implementation().size())));
} else {
PyTuple_SetItem(t, 0, PyString_FromString("die"));
Py_INCREF(Py_None);
PyTuple_SetItem(t, 1, Py_None);
}
return t;
}
RAY_CHECK(false, "This code should be unreachable.");
Py_RETURN_NONE;
}
static PyObject* run_function_on_all_workers(PyObject* self, PyObject* args) {
Worker* worker;
const char* function;
int function_size;
if (!PyArg_ParseTuple(args, "O&s#", &PyObjectToWorker, &worker, &function, &function_size)) {
return NULL;
}
worker->run_function_on_all_workers(std::string(function, static_cast<size_t>(function_size)));
Py_RETURN_NONE;
}
static PyObject* export_remote_function(PyObject* self, PyObject* args) {
Worker* worker;
const char* function_name;
const char* function;
int function_size;
if (!PyArg_ParseTuple(args, "O&ss#", &PyObjectToWorker, &worker, &function_name, &function, &function_size)) {
return NULL;
}
if (worker->export_remote_function(std::string(function_name), std::string(function, static_cast<size_t>(function_size)))) {
Py_RETURN_TRUE;
} else {
Py_RETURN_FALSE;
}
}
static PyObject* export_reusable_variable(PyObject* self, PyObject* args) {
Worker* worker;
const char* name;
int name_size;
const char* initializer;
int initializer_size;
const char* reinitializer;
int reinitializer_size;
if (!PyArg_ParseTuple(args, "O&s#s#s#", &PyObjectToWorker, &worker, &name, &name_size, &initializer, &initializer_size, &reinitializer, &reinitializer_size)) {
return NULL;
}
std::string name_str(name, static_cast<size_t>(name_size));
std::string initializer_str(initializer, static_cast<size_t>(initializer_size));
std::string reinitializer_str(reinitializer, static_cast<size_t>(reinitializer_size));
worker->export_reusable_variable(name_str, initializer_str, reinitializer_str);
Py_RETURN_NONE;
}
static PyObject* submit_task(PyObject* self, PyObject* args) {
PyObject* worker_capsule;
Task* task;
if (!PyArg_ParseTuple(args, "OO&", &worker_capsule, &PyObjectToTask, &task)) {
return NULL;
}
Worker* worker;
PyObjectToWorker(worker_capsule, &worker);
SubmitTaskRequest request;
request.set_allocated_task(task);
SubmitTaskReply reply = worker->submit_task(&request);
request.release_task(); // TODO: Make sure that task is not moved, otherwise capsule pointer needs to be updated
if (reply.no_workers()) {
PyErr_SetString(RayError, "No workers have registered with the scheduler, so this function cannot be run.");
return NULL;
}
if (!reply.function_registered()) {
PyErr_SetString(RayError, "No worker has registered this function with the scheduler.");
return NULL;
}
int size = reply.result_size();
PyObject* list = PyList_New(size);
std::vector<ObjectID> result_objectids;
for (int i = 0; i < size; ++i) {
PyList_SetItem(list, i, make_pyobjectid(worker_capsule, reply.result(i)));
result_objectids.push_back(reply.result(i));
}
worker->decrement_reference_count(result_objectids); // The corresponding increment is done in SubmitTask in the scheduler.
return list;
}
static PyObject* ready_for_new_task(PyObject* self, PyObject* args) {
Worker* worker;
if (!PyArg_ParseTuple(args, "O&", &PyObjectToWorker, &worker)) {
return NULL;
}
worker->ready_for_new_task();
Py_RETURN_NONE;
}
static PyObject* register_remote_function(PyObject* self, PyObject* args) {
Worker* worker;
const char* function_name;
int num_return_vals;
if (!PyArg_ParseTuple(args, "O&si", &PyObjectToWorker, &worker, &function_name, &num_return_vals)) {
return NULL;
}
worker->register_remote_function(std::string(function_name), num_return_vals);
Py_RETURN_NONE;
}
static PyObject* notify_failure(PyObject* self, PyObject* args) {
Worker* worker;
const char* name;
const char* error_message;
int type;
if (!PyArg_ParseTuple(args, "O&ssi", &PyObjectToWorker, &worker, &name, &error_message, &type)) {
return NULL;
}
worker->notify_failure(static_cast<FailedType>(type), std::string(name), std::string(error_message));
Py_RETURN_NONE;
}
static PyObject* get_objectid(PyObject* self, PyObject* args) {
PyObject* worker_capsule;
if (!PyArg_ParseTuple(args, "O", &worker_capsule)) {
return NULL;
}
Worker* worker;
PyObjectToWorker(worker_capsule, &worker);
ObjectID objectid = worker->get_objectid();
return make_pyobjectid(worker_capsule, objectid);
}
static PyObject* add_contained_objectids(PyObject* self, PyObject* args) {
Worker* worker;
ObjectID objectid;
PyObject* contained_objectids;
if (!PyArg_ParseTuple(args, "O&O&O", &PyObjectToWorker, &worker, &PyObjectToObjectID, &objectid, &contained_objectids)) {
return NULL;
}
RAY_CHECK(PyList_Check(contained_objectids), "The contained_objectids argument must be a list.")
std::vector<ObjectID> vec_contained_objectids;
size_t size = PyList_Size(contained_objectids);
for (size_t i = 0; i < size; ++i) {
ObjectID contained_objectid;
PyObjectToObjectID(PyList_GetItem(contained_objectids, i), &contained_objectid);
vec_contained_objectids.push_back(contained_objectid);
}
worker->add_contained_objectids(objectid, vec_contained_objectids);
Py_RETURN_NONE;
}
static PyObject* request_object(PyObject* self, PyObject* args) {
Worker* worker;
ObjectID objectid;
if (!PyArg_ParseTuple(args, "O&O&", &PyObjectToWorker, &worker, &PyObjectToObjectID, &objectid)) {
return NULL;
}
worker->request_object(objectid);
Py_RETURN_NONE;
}
static PyObject* wait(PyObject* self, PyObject* args) {
Worker* worker;
PyObject* objectids;
if (!PyArg_ParseTuple(args, "O&O", &PyObjectToWorker, &worker, &objectids)) {
return NULL;
}
std::vector<ObjectID> objectids_vec;
for (size_t i = 0; i < PyList_Size(objectids); ++i) {
ObjectID objectid;
PyObjectToObjectID(PyList_GetItem(objectids, i), &objectid);
objectids_vec.push_back(objectid);
}
std::vector<int> indices = worker->wait(objectids_vec);
PyObject* result = PyList_New(indices.size());
for (size_t i = 0; i < indices.size(); ++i) {
PyList_SetItem(result, i, PyInt_FromLong(indices[i]));
}
return result;
}
static PyObject* alias_objectids(PyObject* self, PyObject* args) {
Worker* worker;
ObjectID alias_objectid;
ObjectID target_objectid;
if (!PyArg_ParseTuple(args, "O&O&O&", &PyObjectToWorker, &worker, &PyObjectToObjectID, &alias_objectid, &PyObjectToObjectID, &target_objectid)) {
return NULL;
}
worker->alias_objectids(alias_objectid, target_objectid);
Py_RETURN_NONE;
}
static PyObject* scheduler_info(PyObject* self, PyObject* args) {
Worker* worker;
if (!PyArg_ParseTuple(args, "O&", &PyObjectToWorker, &worker)) {
return NULL;
}
ClientContext context;
SchedulerInfoRequest request;
SchedulerInfoReply reply;
worker->scheduler_info(context, request, reply);
// Unpack the target object reference information.
PyObject* target_objectid_list = PyList_New(reply.target_objectid_size());
for (size_t i = 0; i < reply.target_objectid_size(); ++i) {
PyList_SetItem(target_objectid_list, i, PyInt_FromLong(reply.target_objectid(i)));
}
// Unpack the reference count information.
PyObject* reference_count_list = PyList_New(reply.reference_count_size());
for (size_t i = 0; i < reply.reference_count_size(); ++i) {
PyList_SetItem(reference_count_list, i, PyInt_FromLong(reply.reference_count(i)));
}
// Unpack the available worker information.
PyObject* available_worker_list = PyList_New(reply.avail_worker_size());
for (size_t i = 0; i < reply.avail_worker_size(); ++i) {
PyList_SetItem(available_worker_list, i, PyInt_FromLong(reply.avail_worker(i)));
}
// Unpack the object store information.
PyObject* objstore_list = PyList_New(reply.objstore_size());
for (size_t i = 0; i < reply.objstore_size(); ++i) {
PyObject* objstore_data = PyDict_New();
set_dict_item_and_transfer_ownership(objstore_data, PyString_FromString("objstoreid"), PyInt_FromLong(reply.objstore(i).objstoreid()));
set_dict_item_and_transfer_ownership(objstore_data, PyString_FromString("address"), PyString_FromStringAndSize(reply.objstore(i).address().data(), reply.objstore(i).address().size()));
PyList_SetItem(objstore_list, i, objstore_data);
}
// Store the unpacked values in a dictionary to return.
PyObject* dict = PyDict_New();
set_dict_item_and_transfer_ownership(dict, PyString_FromString("target_objectids"), target_objectid_list);
set_dict_item_and_transfer_ownership(dict, PyString_FromString("reference_counts"), reference_count_list);
set_dict_item_and_transfer_ownership(dict, PyString_FromString("available_workers"), available_worker_list);
set_dict_item_and_transfer_ownership(dict, PyString_FromString("objstores"), objstore_list);
return dict;
}
static PyObject* failure_to_dict(const Failure& failure) {
PyObject* failure_dict = PyDict_New();
set_dict_item_and_transfer_ownership(failure_dict, PyString_FromString("workerid"), PyInt_FromLong(failure.workerid()));
set_dict_item_and_transfer_ownership(failure_dict, PyString_FromString("worker_address"), PyString_FromStringAndSize(failure.worker_address().data(), failure.worker_address().size()));
set_dict_item_and_transfer_ownership(failure_dict, PyString_FromString("function_name"), PyString_FromStringAndSize(failure.name().data(), failure.name().size()));
set_dict_item_and_transfer_ownership(failure_dict, PyString_FromString("error_message"), PyString_FromStringAndSize(failure.error_message().data(), failure.error_message().size()));
return failure_dict;
}
static PyObject* task_info(PyObject* self, PyObject* args) {
Worker* worker;
if (!PyArg_ParseTuple(args, "O&", &PyObjectToWorker, &worker)) {
return NULL;
}
ClientContext context;
TaskInfoRequest request;
TaskInfoReply reply;
worker->task_info(context, request, reply);
PyObject* failed_tasks_list = PyList_New(reply.failed_task_size());
for (size_t i = 0; i < reply.failed_task_size(); ++i) {
const TaskStatus& info = reply.failed_task(i);
PyObject* info_dict = PyDict_New();
set_dict_item_and_transfer_ownership(info_dict, PyString_FromString("worker_address"), PyString_FromStringAndSize(info.worker_address().data(), info.worker_address().size()));
set_dict_item_and_transfer_ownership(info_dict, PyString_FromString("function_name"), PyString_FromStringAndSize(info.function_name().data(), info.function_name().size()));
set_dict_item_and_transfer_ownership(info_dict, PyString_FromString("operationid"), PyInt_FromLong(info.operationid()));
set_dict_item_and_transfer_ownership(info_dict, PyString_FromString("error_message"), PyString_FromStringAndSize(info.error_message().data(), info.error_message().size()));
PyList_SetItem(failed_tasks_list, i, info_dict);
}
PyObject* running_tasks_list = PyList_New(reply.running_task_size());
for (size_t i = 0; i < reply.running_task_size(); ++i) {
const TaskStatus& info = reply.running_task(i);
PyObject* info_dict = PyDict_New();
set_dict_item_and_transfer_ownership(info_dict, PyString_FromString("worker_address"), PyString_FromStringAndSize(info.worker_address().data(), info.worker_address().size()));
set_dict_item_and_transfer_ownership(info_dict, PyString_FromString("function_name"), PyString_FromStringAndSize(info.function_name().data(), info.function_name().size()));
set_dict_item_and_transfer_ownership(info_dict, PyString_FromString("operationid"), PyInt_FromLong(info.operationid()));
PyList_SetItem(running_tasks_list, i, info_dict);
}
PyObject* failed_remote_function_imports = PyList_New(reply.failed_remote_function_import_size());
for (size_t i = 0; i < reply.failed_remote_function_import_size(); ++i) {
PyList_SetItem(failed_remote_function_imports, i, failure_to_dict(reply.failed_remote_function_import(i)));
}
PyObject* failed_reusable_variable_imports = PyList_New(reply.failed_reusable_variable_import_size());
for (size_t i = 0; i < reply.failed_reusable_variable_import_size(); ++i) {
PyList_SetItem(failed_reusable_variable_imports, i, failure_to_dict(reply.failed_reusable_variable_import(i)));
}
PyObject* failed_reinitialize_reusable_variables = PyList_New(reply.failed_reinitialize_reusable_variable_size());
for (size_t i = 0; i < reply.failed_reinitialize_reusable_variable_size(); ++i) {
PyList_SetItem(failed_reinitialize_reusable_variables, i, failure_to_dict(reply.failed_reinitialize_reusable_variable(i)));
}
PyObject* failed_function_to_runs = PyList_New(reply.failed_function_to_run_size());
for (size_t i = 0; i < reply.failed_function_to_run_size(); ++i) {
PyList_SetItem(failed_function_to_runs, i, failure_to_dict(reply.failed_function_to_run(i)));
}
PyObject* dict = PyDict_New();
set_dict_item_and_transfer_ownership(dict, PyString_FromString("failed_tasks"), failed_tasks_list);
set_dict_item_and_transfer_ownership(dict, PyString_FromString("running_tasks"), running_tasks_list);
set_dict_item_and_transfer_ownership(dict, PyString_FromString("failed_remote_function_imports"), failed_remote_function_imports);
set_dict_item_and_transfer_ownership(dict, PyString_FromString("failed_reusable_variable_imports"), failed_reusable_variable_imports);
set_dict_item_and_transfer_ownership(dict, PyString_FromString("failed_reinitialize_reusable_variables"), failed_reinitialize_reusable_variables);
set_dict_item_and_transfer_ownership(dict, PyString_FromString("failed_function_to_runs"), failed_function_to_runs);
return dict;
}
static PyObject* dump_computation_graph(PyObject* self, PyObject* args) {
Worker* worker;
const char* output_file_name;
if (!PyArg_ParseTuple(args, "O&s", &PyObjectToWorker, &worker, &output_file_name)) {
return NULL;
}
ClientContext context;
SchedulerInfoRequest request;
SchedulerInfoReply reply;
worker->scheduler_info(context, request, reply);
std::fstream output(output_file_name, std::ios::out | std::ios::trunc | std::ios::binary);
RAY_CHECK(reply.computation_graph().SerializeToOstream(&output), "Cannot dump computation graph to file " << output_file_name);
Py_RETURN_NONE;
}
static PyObject* kill_workers(PyObject* self, PyObject* args) {
Worker* worker;
if (!PyArg_ParseTuple(args, "O&", &PyObjectToWorker, &worker)) {
return NULL;
}
ClientContext context;
if (worker->kill_workers(context)) {
Py_RETURN_TRUE;
} else {
Py_RETURN_FALSE;
}
}
static PyMethodDef RayLibMethods[] = {
{ "serialize_objectid", serialize_objectid, METH_VARARGS, "serialize an object id" },
{ "deserialize_objectid", deserialize_objectid, METH_VARARGS, "deserialize an object id" },
{ "allocate_buffer", allocate_buffer, METH_VARARGS, "Allocates and returns buffer for objectid."},
{ "finish_buffer", finish_buffer, METH_VARARGS, "Makes the buffer immutable and closes memory segment of objectid."},
{ "get_buffer", get_buffer, METH_VARARGS, "Gets buffer for objectid"},
{ "is_arrow", is_arrow, METH_VARARGS, "is the object in the local object store an arrow object?"},
{ "unmap_object", unmap_object, METH_VARARGS, "unmap the object from the client's shared memory pool"},
{ "serialize_task", serialize_task, METH_VARARGS, "serialize a task to protocol buffers" },
{ "create_worker", create_worker, METH_VARARGS, "connect to the scheduler and the object store" },
{ "disconnect", disconnect, METH_VARARGS, "disconnect the worker from the scheduler and the object store" },
{ "connected", connected, METH_VARARGS, "check if the worker is connected to the scheduler and the object store" },
{ "register_remote_function", register_remote_function, METH_VARARGS, "register a function with the scheduler" },
{ "notify_failure", notify_failure, METH_VARARGS, "notify the scheduler of a failure" },
{ "add_contained_objectids", add_contained_objectids, METH_VARARGS, "notify the scheduler about the object IDs contained in a remote object" },
{ "get_objectid", get_objectid, METH_VARARGS, "register a new object reference with the scheduler" },
{ "request_object" , request_object, METH_VARARGS, "request an object to be delivered to the local object store" },
{ "wait" , wait, METH_VARARGS, "checks the scheduler to see if a object can be gotten" },
{ "alias_objectids", alias_objectids, METH_VARARGS, "make two objectids refer to the same object" },
{ "wait_for_next_message", wait_for_next_message, METH_VARARGS, "get next message from scheduler (blocking)" },
{ "submit_task", submit_task, METH_VARARGS, "call a remote function" },
{ "ready_for_new_task", ready_for_new_task, METH_VARARGS, "notify the scheduler that the worker is ready for a new task" },
{ "scheduler_info", scheduler_info, METH_VARARGS, "get info about scheduler state" },
{ "task_info", task_info, METH_VARARGS, "get information about task statuses and failures" },
{ "run_function_on_all_workers", run_function_on_all_workers, METH_VARARGS, "run an arbitrary function on all workers" },
{ "export_remote_function", export_remote_function, METH_VARARGS, "export a remote function to workers" },
{ "export_reusable_variable", export_reusable_variable, METH_VARARGS, "export a reusable variable to the workers" },
{ "dump_computation_graph", dump_computation_graph, METH_VARARGS, "dump the current computation graph to a file" },
{ "kill_workers", kill_workers, METH_VARARGS, "kills all of the workers" },
{ NULL, NULL, 0, NULL }
};
PyMODINIT_FUNC initlibraylib(void) {
PyObject* m;
PyObjectIDType.tp_new = PyType_GenericNew;
if (PyType_Ready(&PyObjectIDType) < 0) {
return;
}
m = Py_InitModule3("libraylib", RayLibMethods, "Python C Extension for Ray");
Py_INCREF(&PyObjectIDType);
PyModule_AddObject(m, "ObjectID", (PyObject *)&PyObjectIDType);
char ray_error[] = "ray.error";
char ray_size_error[] = "ray_size.error";
RayError = PyErr_NewException(ray_error, NULL, NULL);
RaySizeError = PyErr_NewException(ray_size_error, NULL, NULL);
Py_INCREF(RayError);
Py_INCREF(RaySizeError);
PyModule_AddObject(m, "ray_error", RayError);
PyModule_AddObject(m, "ray_size_error", RaySizeError);
import_array();
// Export constants used for the worker mode types so they can be accessed
// from Python. The Mode enum is defined in worker.h.
PyModule_AddIntConstant(m, "SCRIPT_MODE", Mode::SCRIPT_MODE);
PyModule_AddIntConstant(m, "WORKER_MODE", Mode::WORKER_MODE);
PyModule_AddIntConstant(m, "PYTHON_MODE", Mode::PYTHON_MODE);
PyModule_AddIntConstant(m, "SILENT_MODE", Mode::SILENT_MODE);
// Export constants for the failure types so they can be accessed from Python.
// The FailedType enum is defined in types.proto.
PyModule_AddIntConstant(m, "FailedTask", FailedType::FailedTask);
PyModule_AddIntConstant(m, "FailedRemoteFunctionImport", FailedType::FailedRemoteFunctionImport);
PyModule_AddIntConstant(m, "FailedReusableVariableImport", FailedType::FailedReusableVariableImport);
PyModule_AddIntConstant(m, "FailedReinitializeReusableVariable", FailedType::FailedReinitializeReusableVariable);
PyModule_AddIntConstant(m, "FailedFunctionToRun", FailedType::FailedFunctionToRun);
}
}
src/scheduler.cc
-1187
View File
File diff suppressed because it is too large Load Diff
src/scheduler.h
-237
View File
@@ -1,237 +0,0 @@
#ifndef RAY_SCHEDULER_H
#define RAY_SCHEDULER_H
#include <deque>
#include <memory>
#include <algorithm>
#include <iostream>
#include <limits>
#include <grpc++/grpc++.h>
#include "ray/ray.h"
#include "ray.grpc.pb.h"
#include "types.pb.h"
#include "utils.h"
#include "computation_graph.h"
using grpc::Server;
using grpc::ServerBuilder;
using grpc::ServerReader;
using grpc::ServerContext;
using grpc::Status;
using grpc::ClientContext;
using grpc::Channel;
typedef size_t RefCount;
const ObjectID UNITIALIZED_ALIAS = std::numeric_limits<ObjectID>::max();
const RefCount DEALLOCATED = std::numeric_limits<RefCount>::max();
struct WorkerHandle {
std::shared_ptr<Channel> channel;
std::unique_ptr<WorkerService::Stub> worker_stub; // If null, the worker has died
ObjStoreId objstoreid;
std::string worker_address;
// This field is initialized to false, and it is set to true after all of the
// initial exports have been shipped to this worker.
bool initial_exports_done;
OperationId current_task;
};
struct ObjStoreHandle {
std::shared_ptr<Channel> channel;
std::unique_ptr<ObjStore::Stub> objstore_stub;
std::string address;
};
enum SchedulingAlgorithmType {
SCHEDULING_ALGORITHM_NAIVE = 0,
SCHEDULING_ALGORITHM_LOCALITY_AWARE = 1
};
class SchedulerService : public Scheduler::Service {
public:
SchedulerService(SchedulingAlgorithmType scheduling_algorithm);
Status SubmitTask(ServerContext* context, const SubmitTaskRequest* request, SubmitTaskReply* reply) override;
Status PutObj(ServerContext* context, const PutObjRequest* request, PutObjReply* reply) override;
Status RequestObj(ServerContext* context, const RequestObjRequest* request, AckReply* reply) override;
Status AliasObjectIDs(ServerContext* context, const AliasObjectIDsRequest* request, AckReply* reply) override;
Status RegisterObjStore(ServerContext* context, const RegisterObjStoreRequest* request, RegisterObjStoreReply* reply) override;
Status RegisterWorker(ServerContext* context, const RegisterWorkerRequest* request, RegisterWorkerReply* reply) override;
Status RegisterRemoteFunction(ServerContext* context, const RegisterRemoteFunctionRequest* request, AckReply* reply) override;
Status ObjReady(ServerContext* context, const ObjReadyRequest* request, AckReply* reply) override;
Status ReadyForNewTask(ServerContext* context, const ReadyForNewTaskRequest* request, AckReply* reply) override;
Status IncrementRefCount(ServerContext* context, const IncrementRefCountRequest* request, AckReply* reply) override;
Status DecrementRefCount(ServerContext* context, const DecrementRefCountRequest* request, AckReply* reply) override;
Status AddContainedObjectIDs(ServerContext* context, const AddContainedObjectIDsRequest* request, AckReply* reply) override;
Status SchedulerInfo(ServerContext* context, const SchedulerInfoRequest* request, SchedulerInfoReply* reply) override;
Status TaskInfo(ServerContext* context, const TaskInfoRequest* request, TaskInfoReply* reply) override;
Status KillWorkers(ServerContext* context, const KillWorkersRequest* request, KillWorkersReply* reply) override;
Status RunFunctionOnAllWorkers(ServerContext* context, const RunFunctionOnAllWorkersRequest* request, AckReply* reply) override;
Status ExportRemoteFunction(ServerContext* context, const ExportRemoteFunctionRequest* request, AckReply* reply) override;
Status ExportReusableVariable(ServerContext* context, const ExportReusableVariableRequest* request, AckReply* reply) override;
Status NotifyFailure(ServerContext*, const NotifyFailureRequest* request, AckReply* reply) override;
Status Wait(ServerContext*, const WaitRequest* request, WaitReply* reply) override;
#ifdef NDEBUG
// If we've disabled assertions, then just use regular SynchronizedPtr to skip lock checking.
template<class T>
using MySynchronizedPtr = SynchronizedPtr<T>;
#else
// A SynchronizedPtr specialized for this class to dynamically check that locks are obtained in the correct order (in the order of field declarations).
template<class T>
class MySynchronizedPtr;
#endif
// This will ask an object store to send an object to another object store if
// the object is not already present in that object store and is not already
// being transmitted.
void deliver_object_async_if_necessary(ObjectID objectid, ObjStoreId from, ObjStoreId to);
// ask an object store to send object to another object store
void deliver_object_async(ObjectID objectid, ObjStoreId from, ObjStoreId to);
// assign a task to a worker
void schedule();
// execute a task on a worker and ship required object IDs
void assign_task(OperationId operationid, WorkerId workerid, const MySynchronizedPtr<ComputationGraph> &computation_graph);
// checks if the dependencies of the task are met
bool can_run(const Task& task);
// register a new object with the scheduler and return its object ID
ObjectID register_new_object();
// register the location of the object ID in the object table
void add_location(ObjectID objectid, ObjStoreId objstoreid);
// indicate that objectid is a canonical objectid
void add_canonical_objectid(ObjectID objectid);
// get object store associated with a workerid
ObjStoreId get_store(WorkerId workerid);
// register a function with the scheduler
void register_function(const std::string& name, WorkerId workerid, size_t num_return_vals);
// get information about the scheduler state
void get_info(const SchedulerInfoRequest& request, SchedulerInfoReply* reply);
private:
// pick an objectstore that holds a given object (needs protection by objects_lock_)
ObjStoreId pick_objstore(ObjectID objectid);
// checks if objectid is a canonical objectid
bool is_canonical(ObjectID objectid);
// Perform all queued up gets that can be performed.
void perform_gets();
// schedule tasks using the naive algorithm
void schedule_tasks_naively();
// schedule tasks using a scheduling algorithm that takes into account data locality
void schedule_tasks_location_aware();
void perform_notify_aliases();
// checks if aliasing for objectid has been completed
bool has_canonical_objectid(ObjectID objectid);
// get the canonical objectid for an objectid
ObjectID get_canonical_objectid(ObjectID objectid);
// attempt to notify the objstore about potential objectid aliasing, returns true if successful, if false then retry later
bool attempt_notify_alias(ObjStoreId objstoreid, ObjectID alias_objectid, ObjectID canonical_objectid);
// tell all of the objstores holding canonical_objectid to deallocate it, the
// data structures are passed into ensure that the appropriate locks are held.
void deallocate_object(ObjectID canonical_objectid, const MySynchronizedPtr<std::vector<RefCount> > &reference_counts, const MySynchronizedPtr<std::vector<std::vector<ObjectID> > > &contained_objectids);
// increment the ref counts for the object IDs in objectids, the data
// structures are passed into ensure that the appropriate locks are held.
void increment_ref_count(const std::vector<ObjectID> &objectids, const MySynchronizedPtr<std::vector<RefCount> > &reference_count);
// decrement the ref counts for the object IDs in objectids, the data
// structures are passed into ensure that the appropriate locks are held.
void decrement_ref_count(const std::vector<ObjectID> &objectids, const MySynchronizedPtr<std::vector<RefCount> > &reference_count, const MySynchronizedPtr<std::vector<std::vector<ObjectID> > > &contained_objectids);
// Find all of the object IDs which are upstream of objectid (including objectid itself). That is, you can get from everything in objectids to objectid by repeatedly indexing in target_objectids_.
void upstream_objectids(ObjectID objectid, std::vector<ObjectID> &objectids, const MySynchronizedPtr<std::vector<std::vector<ObjectID> > > &reverse_target_objectids);
// Find all of the object IDs that refer to the same object as objectid (as best as we can determine at the moment). The information may be incomplete because not all of the aliases may be known.
void get_equivalent_objectids(ObjectID objectid, std::vector<ObjectID> &equivalent_objectids);
// Export a function to run to a worker.
void export_function_to_run_to_worker(WorkerId workerid, int function_index, MySynchronizedPtr<std::vector<WorkerHandle> > &workers, const MySynchronizedPtr<std::vector<std::unique_ptr<Function> > > &exported_functions_to_run);
// Export a remote function to a worker.
void export_remote_function_to_worker(WorkerId workerid, int function_index, MySynchronizedPtr<std::vector<WorkerHandle> > &workers, const MySynchronizedPtr<std::vector<std::unique_ptr<Function> > > &exported_remote_functions);
// Export a reusable variable to a worker
void export_reusable_variable_to_worker(WorkerId workerid, int reusable_variable_index, MySynchronizedPtr<std::vector<WorkerHandle> > &workers, const MySynchronizedPtr<std::vector<std::unique_ptr<ReusableVar> > > &exported_reusable_variables);
// Export all exports to all workers that need them. This happens the first
// time any export would be exported to a worker or when a worker first calls
// ReadyForNewTask.
void export_everything_to_all_workers_if_necessary(MySynchronizedPtr<std::vector<WorkerHandle> > &workers);
template<class T>
MySynchronizedPtr<T> get(Synchronized<T>& my_field, const char* name,unsigned int line_number);
template<class T>
MySynchronizedPtr<const T> get(const Synchronized<T>& my_field, const char* name,unsigned int line_number) const;
// Preferably keep this as the first field to distinguish it from the rest
// Maps every thread to an identifier of a lock it is holding, as well the name of the lock.
// Internally, the identifier for each lock is the offset of the field being locked.
// When we lock, we set the field offset and store the difference; the difference should always be positive. If not, we throw.
// When we unlock, we subtract back the field offset to restore it to the previous field that was locked.
mutable Synchronized<std::vector<std::pair<unsigned long long, std::pair<size_t, const char*> > > > lock_orders_;
// List of failed tasks
Synchronized<std::vector<TaskStatus> > failed_tasks_;
// A list of remote functions import failures.
Synchronized<std::vector<Failure> > failed_remote_function_imports_;
// A list of reusable variables import failures.
Synchronized<std::vector<Failure> > failed_reusable_variable_imports_;
// A list of reusable variables reinitialization failures.
Synchronized<std::vector<Failure> > failed_reinitialize_reusable_variables_;
// A list of function to run failures.
Synchronized<std::vector<Failure> > failed_function_to_runs_;
// List of pending get calls.
Synchronized<std::vector<std::pair<WorkerId, ObjectID> > > get_queue_;
// The computation graph tracks the operations that have been submitted to the
// scheduler and is mostly used for fault tolerance.
Synchronized<ComputationGraph> computation_graph_;
// Hash map from function names to workers where the function is registered.
Synchronized<FnTable> fntable_;
// Vector of all workers that are currently idle.
Synchronized<std::vector<WorkerId> > avail_workers_;
// List of pending tasks.
Synchronized<std::deque<OperationId> > task_queue_;
// Reference counts. Currently, reference_counts_[objectid] is the number of
// existing references held to objectid. This is done for all objectids, not just
// canonical_objectids. This data structure completely ignores aliasing. If the
// object corresponding to objectid has been deallocated, then
// reference_counts[objectid] will equal DEALLOCATED.
Synchronized<std::vector<RefCount> > reference_counts_;
// contained_objectids_[objectid] is a vector of all of the objectids contained inside the object referred to by objectid
Synchronized<std::vector<std::vector<ObjectID> > > contained_objectids_;
// Vector of all workers registered in the system. Their index in this vector
// is the workerid.
Synchronized<std::vector<WorkerHandle> > workers_;
// List of pending alias notifications. Each element consists of (objstoreid, (alias_objectid, canonical_objectid)).
Synchronized<std::vector<std::pair<ObjStoreId, std::pair<ObjectID, ObjectID> > > > alias_notification_queue_;
// Mapping from canonical objectid to list of object stores where the object is stored. Non-canonical (aliased) objectids should not be used to index objtable_.
Synchronized<ObjTable> objtable_; // This lock protects objtable_ and objects_in_transit_
// Vector of all object stores registered in the system. Their index in this
// vector is the objstoreid.
Synchronized<std::vector<ObjStoreHandle> > objstores_;
// Mapping from an aliased objectid to the objectid it is aliased with. If an
// objectid is a canonical objectid (meaning it is not aliased), then
// target_objectids_[objectid] == objectid. For each objectid, target_objectids_[objectid]
// is initialized to UNITIALIZED_ALIAS and the correct value is filled later
// when it is known.
Synchronized<std::vector<ObjectID> > target_objectids_;
// This data structure maps an objectid to all of the objectids that alias it (there could be multiple such objectids).
Synchronized<std::vector<std::vector<ObjectID> > > reverse_target_objectids_;
// For each object store objstoreid, objects_in_transit_[objstoreid] is a
// vector of the canonical object IDs that are being streamed to that
// object store but are not yet present. object IDs are added to this
// in deliver_object_async_if_necessary (to ensure that we do not attempt to deliver
// the same object to a given object store twice), and object IDs are
// removed when add_location is called (from ObjReady), and they are moved to
// the objtable_. Note that objects_in_transit_ and objtable_ share the same
// lock (objects_lock_). // TODO(rkn): Consider making this part of the
// objtable data structure.
std::vector<std::vector<ObjectID> > objects_in_transit_;
// All of the functions that have been exported to the workers to run.
Synchronized<std::vector<std::unique_ptr<Function> > > exported_functions_to_run_;
// All of the remote functions that have been exported to the workers.
Synchronized<std::vector<std::unique_ptr<Function> > > exported_remote_functions_;
// All of the reusable variables that have been exported to the workers.
Synchronized<std::vector<std::unique_ptr<ReusableVar> > > exported_reusable_variables_;
// the scheduling algorithm that will be used
SchedulingAlgorithmType scheduling_algorithm_;
};
#endif
src/utils.cc
-69
View File
@@ -1,69 +0,0 @@
#include "utils.h"
#include "ray/ray.h"
#include <sys/stat.h>
#ifdef _S_IREAD // Visual C++ runtime?
#include <direct.h> // _mkdir
#else
namespace {
int _mkdir(char const* path) {
return mkdir(path, S_IRWXU | S_IRWXG | S_IRWXO);
}
}
#endif
std::string::iterator split_ip_address(std::string& ip_address) {
if (ip_address[0] == '[') { // IPv6
auto split_end = std::find(ip_address.begin() + 1, ip_address.end(), ']');
if(split_end != ip_address.end()) {
split_end++;
}
if(split_end != ip_address.end() && *split_end == ':') {
return split_end;
}
RAY_CHECK(false, "ip address should contain a port number");
} else { // IPv4
auto split_point = std::find(ip_address.rbegin(), ip_address.rend(), ':').base();
RAY_CHECK_NEQ(split_point, ip_address.begin(), "ip address should contain a port number");
return split_point;
}
}
const char* get_cmd_option(char** begin, char** end, const std::string& option) {
char** it = std::find(begin, end, option);
if (it != end && ++it != end) {
return *it;
}
return 0;
}
void create_directories(const char* log_file_name) {
bool success = _mkdir(log_file_name) != -1 || errno == EEXIST;
if (!success) {
// If we couldn't create it directly and it didn't already exist, then try to create it from the root...
// Note that we keep going until the end even if creating the root fails, because we don't necessarily have access to the root
bool stop = false;
size_t i = 0;
do {
stop = log_file_name[i] == '\0';
bool delimiter = stop || log_file_name[i] == '/' || log_file_name[i] == '\\';
if (!stop) {
++i;
}
if (delimiter) {
std::string ancestor(log_file_name, i);
success = _mkdir(ancestor.c_str()) != -1 || errno == EEXIST;
}
} while (!stop);
}
RAY_CHECK(success, "Failed to create directory for " << log_file_name);
}
void create_log_dir_or_die(const char* log_file_name) {
std::string dirname = log_file_name;
while (!dirname.empty() && dirname.back() != '/' && dirname.back() != '\\') {
dirname.pop_back();
}
return create_directories(dirname.c_str());
}
src/utils.h
-97
View File
@@ -1,97 +0,0 @@
#ifndef RAY_UTILS_H
#define RAY_UTILS_H
#include <mutex>
#include <string>
template<class T = void, class Mutex = std::mutex>
class Synchronized;
template<class T, class Mutex>
class Synchronized<const T, Mutex>; // Prevent use of const T; it doesn't make sense
template<class T, class Mutex> struct SynchronizedSource { typedef Synchronized<T, Mutex> type; };
template<class T, class Mutex> struct SynchronizedSource<const T, Mutex> { typedef const Synchronized<T, Mutex> type; };
template<class T, class Mutex> struct SynchronizedSource<volatile T, Mutex> { typedef volatile Synchronized<T, Mutex> type; };
template<class T, class Mutex> struct SynchronizedSource<const volatile T, Mutex> { typedef const Synchronized<T, Mutex> type; };
template<class T>
class SynchronizedPtr : public std::unique_lock<typename SynchronizedSource<T, void>::type> {
protected:
typedef std::unique_lock<typename SynchronizedSource<T, void>::type> base_type;
// Make these private; they don't make much sense externally...
using base_type::mutex;
public:
typedef T value_type;
SynchronizedPtr(typename base_type::mutex_type& value) : base_type(value) { }
value_type& operator*() const { return *mutex()->unsafe_get(); }
value_type* operator->() const { return mutex() ? mutex()->unsafe_get() : NULL; }
};
template<class T>
class Synchronized<T, void> {
T value_;
public:
typedef T element_type;
template<class... U>
Synchronized(U&&... args) : value_(std::forward<U>(args)...) { }
Synchronized(const Synchronized& other) : value_((std::lock_guard<Synchronized>(other), other.value_)) { }
Synchronized(Synchronized&& other) : value_((std::lock_guard<Synchronized>(other), std::move(other.value_))) { }
Synchronized& operator =(const Synchronized& other)
{
if (this != &other)
{
std::lock_guard<Synchronized> guard_this(*this);
std::lock_guard<Synchronized> guard_other(other);
value_ = other.value_;
}
return *this;
}
Synchronized& operator =(Synchronized&& other)
{
if (this != &other)
{
std::lock_guard<Synchronized> guard_this(*this);
std::lock_guard<Synchronized> guard_other(other);
value_ = std::move(other.value_);
}
return *this;
}
virtual void lock() const = 0;
virtual void unlock() const = 0;
virtual bool try_lock() const = 0;
element_type* unsafe_get() { return &value_; }
const element_type* unsafe_get() const { return &value_; }
};
template<class Mutex>
class Synchronized<void, Mutex> {
mutable Mutex mutex_;
public:
typedef Mutex mutex_type;
void lock() const { return mutex_.lock(); }
void unlock() const { return mutex_.unlock(); }
bool try_lock() const { return mutex_.try_lock(); }
};
template<class T, class Mutex>
class Synchronized : public Synchronized<T, void>, public Synchronized<void, Mutex> {
typedef Synchronized<T, void> base1_type;
typedef Synchronized<void, Mutex> base2_type;
public:
template<class... U>
Synchronized(U&&... args) : base1_type(std::forward<U>(args)...), base2_type() { }
SynchronizedPtr<T> unchecked_get() { return *this; }
SynchronizedPtr<const T> unchecked_get() const { return *this; }
void lock() const override { return base2_type::lock(); }
void unlock() const override { return base2_type::unlock(); }
bool try_lock() const override { return base2_type::try_lock(); }
};
std::string::iterator split_ip_address(std::string& ip_address);
const char* get_cmd_option(char** begin, char** end, const std::string& option);
void create_log_dir_or_die(const char* log_file_name);
#endif
src/worker.cc
-497
View File
@@ -1,497 +0,0 @@
#include "worker.h"
#include <atomic>
#include <random>
#include <chrono>
#include <thread>
#include "utils.h"
extern "C" {
static PyObject *RayError;
}
inline WorkerServiceImpl::WorkerServiceImpl(const std::string& send_queue_name, Mode mode)
: mode_(mode) {
RAY_LOG(RAY_INFO, "Worker service connecting to queue " << send_queue_name);
RAY_CHECK(send_queue_.connect(send_queue_name, false), "error connecting send_queue_");
}
Status WorkerServiceImpl::ExecuteTask(ServerContext* context, const ExecuteTaskRequest* request, AckReply* reply) {
RAY_CHECK(mode_ == Mode::WORKER_MODE, "ExecuteTask can only be called on workers.");
RAY_LOG(RAY_INFO, "invoked task " << request->task().name());
std::unique_ptr<WorkerMessage> message(new WorkerMessage());
message->mutable_task()->CopyFrom(request->task());
{
WorkerMessage* message_ptr = message.get();
RAY_CHECK(send_queue_.send(&message_ptr), "Failed to send message from the worker service to the worker because the message queue was full.");
}
// The message will get deleted in receive_next_message().
message.release();
return Status::OK;
}
Status WorkerServiceImpl::RunFunctionOnWorker(ServerContext* context, const RunFunctionOnWorkerRequest* request, AckReply* reply) {
RAY_CHECK(mode_ == Mode::WORKER_MODE, "RunFunctionOnWorker can only be called on workers.");
std::unique_ptr<WorkerMessage> message(new WorkerMessage());
message->mutable_function_to_run()->CopyFrom(request->function());
RAY_LOG(RAY_INFO, "Running function on worker.");
{
WorkerMessage* message_ptr = message.get();
RAY_CHECK(send_queue_.send(&message_ptr), "Failed to send message from the worker service to the worker because the message queue was full.");
}
// The message will get deleted in receive_next_message().
message.release();
return Status::OK;
}
Status WorkerServiceImpl::ImportRemoteFunction(ServerContext* context, const ImportRemoteFunctionRequest* request, AckReply* reply) {
RAY_CHECK(mode_ == Mode::WORKER_MODE, "ImportRemoteFunction can only be called on workers.");
std::unique_ptr<WorkerMessage> message(new WorkerMessage());
message->mutable_function()->CopyFrom(request->function());
RAY_LOG(RAY_INFO, "importing function");
{
WorkerMessage* message_ptr = message.get();
RAY_CHECK(send_queue_.send(&message_ptr), "Failed to send message from the worker service to the worker because the message queue was full.");
}
// The message will get deleted in receive_next_message().
message.release();
return Status::OK;
}
Status WorkerServiceImpl::ImportReusableVariable(ServerContext* context, const ImportReusableVariableRequest* request, AckReply* reply) {
RAY_CHECK(mode_ == Mode::WORKER_MODE, "ImportReusableVariable can only be called on workers.");
std::unique_ptr<WorkerMessage> message(new WorkerMessage());
message->mutable_reusable_variable()->CopyFrom(request->reusable_variable());
RAY_LOG(RAY_INFO, "importing reusable variable");
{
WorkerMessage* message_ptr = message.get();
RAY_CHECK(send_queue_.send(&message_ptr), "Failed to send message from the worker service to the worker because the message queue was full.");
}
// The message will get deleted in receive_next_message().
message.release();
return Status::OK;
}
Status WorkerServiceImpl::Die(ServerContext* context, const DieRequest* request, AckReply* reply) {
RAY_CHECK(mode_ == Mode::WORKER_MODE, "Die can only be called on workers.");
WorkerMessage* message_ptr = NULL;
RAY_CHECK(send_queue_.send(&message_ptr), "Failed to send message from the worker service to the worker because the message queue was full.");
return Status::OK;
}
Status WorkerServiceImpl::PrintErrorMessage(ServerContext* context, const PrintErrorMessageRequest* request, AckReply* reply) {
RAY_CHECK(mode_ != Mode::WORKER_MODE, "PrintErrorMessage can only be called on drivers.");
if (mode_ == Mode::SILENT_MODE) {
// Do not log error messages in this case. This is just used for the tests.
return Status::OK;
}
const Failure failure = request->failure();
WorkerId workerid = failure.workerid();
if (failure.type() == FailedType::FailedTask) {
// A task threw an exception while executing.
std::cout << "Error: Worker " << workerid << " failed to execute function " << failure.name() << ". Failed with error message:\n" << failure.error_message() << std::endl;
} else if (failure.type() == FailedType::FailedRemoteFunctionImport) {
// An exception was thrown while a remote function was being imported.
std::cout << "Error: Worker " << workerid << " failed to import remote function " << failure.name() << ", failed with error message:\n" << failure.error_message() << std::endl;
} else if (failure.type() == FailedType::FailedReusableVariableImport) {
// An exception was thrown while a reusable variable was being imported.
std::cout << "Error: Worker " << workerid << " failed to import reusable variable " << failure.name() << ", failed with error message:\n" << failure.error_message() << std::endl;
} else if (failure.type() == FailedType::FailedReinitializeReusableVariable) {
// An exception was thrown while a reusable variable was being reinitialized.
std::cout << "Error: Worker " << workerid << " failed to reinitialize a reusable variable after running remote function " << failure.name() << ", failed with error message:\n" << failure.error_message() << std::endl;
} else if (failure.type() == FailedType::FailedFunctionToRun) {
// An exception was thrown while a function was being run on all workers.
std::cout << "Error: Worker " << workerid << " failed to run function " << failure.name() << " on all workers, failed with error message:\n" << failure.error_message() << std::endl;
} else {
RAY_CHECK(false, "This code should be unreachable.")
}
return Status::OK;
}
Worker::Worker(const std::string& node_ip_address, const std::string& scheduler_address, Mode mode)
: scheduler_address_(scheduler_address),
node_ip_address_(node_ip_address),
mode_(mode) {
auto scheduler_channel = grpc::CreateChannel(scheduler_address, grpc::InsecureChannelCredentials());
scheduler_stub_ = Scheduler::NewStub(scheduler_channel);
// Generate a random string to use for naming the message queue to avoid
// collisions with message queues created by other workers.
std::random_device rd;
std::mt19937 rng(rd());
std::uniform_int_distribution<int> queue_name_generator(0, 10000000);
receive_queue_name_ = "worker_receive_queue:" + std::to_string(queue_name_generator(rng));
RAY_LOG(RAY_INFO, "Worker creating queue " << receive_queue_name_);
RAY_CHECK(receive_queue_.connect(receive_queue_name_, true), "error connecting receive_queue_");
}
SubmitTaskReply Worker::submit_task(SubmitTaskRequest* request, int max_retries, int retry_wait_milliseconds) {
RAY_CHECK(connected_, "Attempted to perform submit_task but failed.");
SubmitTaskReply reply;
request->set_workerid(workerid_);
for (int i = 0; i < 1 + max_retries; ++i) {
ClientContext context;
RAY_CHECK_GRPC(scheduler_stub_->SubmitTask(&context, *request, &reply));
if (reply.function_registered()) {
break;
}
RAY_LOG(RAY_INFO, "The function " << request->task().name() << " was not registered, so attempting to resubmit the task.");
std::this_thread::sleep_for(std::chrono::milliseconds(retry_wait_milliseconds));
}
return reply;
}
bool Worker::kill_workers(ClientContext &context) {
KillWorkersRequest request;
KillWorkersReply reply;
RAY_CHECK_GRPC(scheduler_stub_->KillWorkers(&context, request, &reply));
return reply.success();
}
void Worker::register_worker(const std::string& node_ip_address, const std::string& objstore_address, bool is_driver) {
if (mode_ == Mode::WORKER_MODE) {
start_worker_service(mode_);
RAY_CHECK(!worker_address_.empty(), "The worker address is empty. This should be initialized by start_worker_service, so it is possible that the thread synchronization failed.")
}
unsigned int retry_wait_milliseconds = 20;
RegisterWorkerRequest request;
request.set_node_ip_address(node_ip_address);
request.set_worker_address(worker_address_);
// The object store address can be the empty string, in which case the
// scheduler will assign an object store address.
request.set_objstore_address(objstore_address);
request.set_is_driver(is_driver);
RegisterWorkerReply reply;
Status status;
// TODO: HACK: retrying is a hack
for (int i = 0; i < 5; ++i) {
ClientContext context;
status = scheduler_stub_->RegisterWorker(&context, request, &reply);
if (status.error_code() != grpc::UNAVAILABLE) {
break;
}
// Note that each pass through the loop may take substantially longer than
// retry_wait_milliseconds because grpc may do its own retrying.
std::this_thread::sleep_for(std::chrono::milliseconds(retry_wait_milliseconds));
}
RAY_CHECK_GRPC(status);
workerid_ = reply.workerid();
objstoreid_ = reply.objstoreid();
objstore_address_ = reply.objstore_address();
segmentpool_ = std::make_shared<MemorySegmentPool>(objstoreid_, objstore_address_, false);
// Connect to the queue for sending requests to the object store.
std::string request_obj_queue_name = std::string("queue:") + objstore_address_ + std::string(":obj");
RAY_LOG(RAY_INFO, "Worker connecting to queue with name " << request_obj_queue_name << " to send requests to the object store.");
RAY_CHECK(request_obj_queue_.connect(request_obj_queue_name, false), "error connecting request_obj_queue_");
// Create a queue for receiving messages from the object store.
std::string receive_obj_queue_name = std::string("queue:") + objstore_address_ + std::string(":worker:") + std::to_string(workerid_) + std::string(":obj");
RAY_LOG(RAY_INFO, "Worker creating queue with name " << receive_obj_queue_name << " to receive messages from the object store.");
RAY_CHECK(receive_obj_queue_.connect(receive_obj_queue_name, true), "error connecting receive_obj_queue_");
connected_ = true;
return;
}
void Worker::request_object(ObjectID objectid) {
RAY_CHECK(connected_, "Attempted to perform request_object but failed.");
RequestObjRequest request;
request.set_workerid(workerid_);
request.set_objectid(objectid);
AckReply reply;
ClientContext context;
RAY_CHECK_GRPC(scheduler_stub_->RequestObj(&context, request, &reply));
return;
}
ObjectID Worker::get_objectid() {
// first get objectid for the new object
RAY_CHECK(connected_, "Attempted to perform get_objectid but failed.");
PutObjRequest request;
request.set_workerid(workerid_);
PutObjReply reply;
ClientContext context;
RAY_CHECK_GRPC(scheduler_stub_->PutObj(&context, request, &reply));
return reply.objectid();
}
void Worker::add_contained_objectids(ObjectID objectid, std::vector<ObjectID> &contained_objectids) {
RAY_CHECK(connected_, "Attempted to perform add_contained_objectids but failed.");
if (contained_objectids.size() > 0) {
RAY_LOG(RAY_REFCOUNT, "In add_contained_objectids, calling increment_reference_count for contained objectids");
// Notify the scheduler that some object references are serialized in the
// objstore. The corresponding decrement happens when the object
// corresponding to objectid is deallocated.
increment_reference_count(contained_objectids);
// Notify the scheduler about the objectids that we are serializing in the objstore.
AddContainedObjectIDsRequest contained_objectids_request;
contained_objectids_request.set_objectid(objectid);
for (int i = 0; i < contained_objectids.size(); ++i) {
contained_objectids_request.add_contained_objectid(contained_objectids[i]); // TODO(rkn): The naming here is bad
}
AckReply reply;
ClientContext context;
RAY_CHECK_GRPC(scheduler_stub_->AddContainedObjectIDs(&context, contained_objectids_request, &reply));
}
}
#define CHECK_ARROW_STATUS(s, msg) \
do { \
arrow::Status _s = (s); \
if (!_s.ok()) { \
std::string _errmsg = std::string(msg) + _s.ToString(); \
PyErr_SetString(RayError, _errmsg.c_str()); \
return NULL; \
} \
} while (0);
const char* Worker::allocate_buffer(ObjectID objectid, int64_t size, SegmentId& segmentid) {
RAY_CHECK(connected_, "Attempted to perform put_arrow but failed.");
ObjRequest request;
request.workerid = workerid_;
request.type = ObjRequestType::ALLOC;
request.objectid = objectid;
request.size = size;
RAY_CHECK(request_obj_queue_.send(&request), "Failed to send request from the worker to the object store because the message queue was full.");
ObjHandle result;
RAY_CHECK(receive_obj_queue_.receive(&result), "error receiving over IPC");
const char* address = reinterpret_cast<const char*>(segmentpool_->get_address(result));
segmentid = result.segmentid();
return address;
}
PyObject* Worker::finish_buffer(ObjectID objectid, SegmentId segmentid, int64_t metadata_offset) {
segmentpool_->unmap_segment(segmentid);
ObjRequest request;
request.workerid = workerid_;
request.objectid = objectid;
request.type = ObjRequestType::WORKER_DONE;
request.metadata_offset = metadata_offset;
RAY_CHECK(request_obj_queue_.send(&request), "Failed to send request from the worker to the object store because the message queue was full.");
Py_RETURN_NONE;
}
const char* Worker::get_buffer(ObjectID objectid, int64_t &size, SegmentId& segmentid, int64_t& metadata_offset) {
RAY_CHECK(connected_, "Attempted to perform get_arrow but failed.");
ObjRequest request;
request.workerid = workerid_;
request.type = ObjRequestType::GET;
request.objectid = objectid;
RAY_CHECK(request_obj_queue_.send(&request), "Failed to send request from the worker to the object store because the message queue was full.");
ObjHandle result;
RAY_CHECK(receive_obj_queue_.receive(&result), "error receiving over IPC");
const char* address = reinterpret_cast<const char*>(segmentpool_->get_address(result));
size = result.size();
segmentid = result.segmentid();
metadata_offset = result.metadata_offset();
return address;
}
bool Worker::is_arrow(ObjectID objectid) {
RAY_CHECK(connected_, "Attempted to perform is_arrow but failed.");
ObjRequest request;
request.workerid = workerid_;
request.type = ObjRequestType::GET;
request.objectid = objectid;
RAY_CHECK(request_obj_queue_.send(&request), "Failed to send request from the worker to the object store because the message queue was full.");
ObjHandle result;
RAY_CHECK(receive_obj_queue_.receive(&result), "error receiving over IPC");
return result.metadata_offset() != 0;
}
void Worker::unmap_object(ObjectID objectid) {
if (!connected_) {
RAY_LOG(RAY_DEBUG, "Attempted to perform unmap_object but failed.");
return;
}
segmentpool_->unmap_segment(objectid);
}
void Worker::alias_objectids(ObjectID alias_objectid, ObjectID target_objectid) {
RAY_CHECK(connected_, "Attempted to perform alias_objectids but failed.");
ClientContext context;
AliasObjectIDsRequest request;
request.set_alias_objectid(alias_objectid);
request.set_target_objectid(target_objectid);
AckReply reply;
RAY_CHECK_GRPC(scheduler_stub_->AliasObjectIDs(&context, request, &reply));
}
void Worker::increment_reference_count(std::vector<ObjectID> &objectids) {
if (!connected_) {
RAY_LOG(RAY_DEBUG, "Attempting to increment_reference_count for objectids, but connected_ = " << connected_ << " so returning instead.");
return;
}
if (objectids.size() > 0) {
ClientContext context;
IncrementRefCountRequest request;
for (int i = 0; i < objectids.size(); ++i) {
RAY_LOG(RAY_REFCOUNT, "Incrementing reference count for objectid " << objectids[i]);
request.add_objectid(objectids[i]);
}
AckReply reply;
RAY_CHECK_GRPC(scheduler_stub_->IncrementRefCount(&context, request, &reply));
}
}
void Worker::decrement_reference_count(std::vector<ObjectID> &objectids) {
if (!connected_) {
RAY_LOG(RAY_DEBUG, "Attempting to decrement_reference_count, but connected_ = " << connected_ << " so returning instead.");
return;
}
if (objectids.size() > 0) {
ClientContext context;
DecrementRefCountRequest request;
for (int i = 0; i < objectids.size(); ++i) {
RAY_LOG(RAY_REFCOUNT, "Decrementing reference count for objectid " << objectids[i]);
request.add_objectid(objectids[i]);
}
AckReply reply;
RAY_CHECK_GRPC(scheduler_stub_->DecrementRefCount(&context, request, &reply));
}
}
void Worker::register_remote_function(const std::string& name, size_t num_return_vals) {
RAY_CHECK(connected_, "Attempted to perform register_function but failed.");
ClientContext context;
RegisterRemoteFunctionRequest request;
request.set_workerid(workerid_);
request.set_function_name(name);
request.set_num_return_vals(num_return_vals);
AckReply reply;
RAY_CHECK_GRPC(scheduler_stub_->RegisterRemoteFunction(&context, request, &reply));
}
void Worker::notify_failure(FailedType type, const std::string& name, const std::string& error_message) {
RAY_CHECK(connected_, "Attempted to perform notify_failure but failed.");
ClientContext context;
NotifyFailureRequest request;
request.mutable_failure()->set_type(type);
request.mutable_failure()->set_workerid(workerid_);
request.mutable_failure()->set_worker_address(worker_address_);
request.mutable_failure()->set_name(name);
request.mutable_failure()->set_error_message(error_message);
AckReply reply;
RAY_CHECK_GRPC(scheduler_stub_->NotifyFailure(&context, request, &reply));
}
std::unique_ptr<WorkerMessage> Worker::receive_next_message() {
WorkerMessage* message_ptr;
RAY_CHECK(receive_queue_.receive(&message_ptr), "error receiving over IPC");
return std::unique_ptr<WorkerMessage>(message_ptr);
}
void Worker::ready_for_new_task() {
RAY_CHECK(connected_, "Attempted to perform ready_for_new_task but failed.");
ClientContext context;
ReadyForNewTaskRequest request;
request.set_workerid(workerid_);
AckReply reply;
RAY_CHECK_GRPC(scheduler_stub_->ReadyForNewTask(&context, request, &reply));
}
void Worker::disconnect() {
connected_ = false;
// Shut down the worker service. This will cause the call to server->Wait() to
// return.
// server_ptr_->Shutdown();
// Wait for the thread that launched the worker service to return.
// worker_server_thread_.join();
}
// TODO(rkn): Should we be using pointers or references? And should they be const?
void Worker::scheduler_info(ClientContext &context, SchedulerInfoRequest &request, SchedulerInfoReply &reply) {
RAY_CHECK(connected_, "Attempted to get scheduler info but failed.");
RAY_CHECK_GRPC(scheduler_stub_->SchedulerInfo(&context, request, &reply));
}
void Worker::task_info(ClientContext &context, TaskInfoRequest &request, TaskInfoReply &reply) {
RAY_CHECK(connected_, "Attempted to get worker info but failed.");
RAY_CHECK_GRPC(scheduler_stub_->TaskInfo(&context, request, &reply));
}
std::vector<int> Worker::wait(std::vector<ObjectID>& objectids) {
RAY_CHECK(connected_, "Attempted to test if object was ready but failed.");
ClientContext context;
WaitRequest request;
WaitReply reply;
for (int i = 0; i < objectids.size(); ++i) {
request.add_objectids(objectids[i]);
}
RAY_CHECK_GRPC(scheduler_stub_->Wait(&context, request, &reply));
std::vector<int> result;
for (int i = 0; i < reply.indices_size(); ++i) {
result.push_back(reply.indices(i));
}
return result;
}
void Worker::run_function_on_all_workers(const std::string& function) {
RAY_CHECK(connected_, "Attempted to run function on all workers but failed.");
ClientContext context;
RunFunctionOnAllWorkersRequest request;
request.mutable_function()->set_implementation(function);
AckReply reply;
RAY_CHECK_GRPC(scheduler_stub_->RunFunctionOnAllWorkers(&context, request, &reply));
}
bool Worker::export_remote_function(const std::string& function_name, const std::string& function) {
RAY_CHECK(connected_, "Attempted to export function but failed.");
ClientContext context;
ExportRemoteFunctionRequest request;
request.mutable_function()->set_name(function_name);
request.mutable_function()->set_implementation(function);
AckReply reply;
RAY_CHECK_GRPC(scheduler_stub_->ExportRemoteFunction(&context, request, &reply));
return true;
}
void Worker::export_reusable_variable(const std::string& name, const std::string& initializer, const std::string& reinitializer) {
RAY_CHECK(connected_, "Attempted to export reusable variable but failed.");
ClientContext context;
ExportReusableVariableRequest request;
request.mutable_reusable_variable()->set_name(name);
request.mutable_reusable_variable()->mutable_initializer()->set_implementation(initializer);
request.mutable_reusable_variable()->mutable_reinitializer()->set_implementation(reinitializer);
AckReply reply;
RAY_CHECK_GRPC(scheduler_stub_->ExportReusableVariable(&context, request, &reply));
}
// Communication between the WorkerServer and the Worker happens via a message
// queue. This is because the Python interpreter needs to be single threaded
// (in our case running in the main thread), whereas the WorkerService will
// run in a separate thread and potentially utilize multiple threads.
void Worker::start_worker_service(Mode mode) {
// Use atomics so the worker service thread can signal the outside thread that
// the worker service has been started.
std::atomic_bool worker_service_started;
worker_service_started.store(false);
// Launch a new thread for running the worker service. We store this as a
// field so that we can clean it up when we disconnect the worker.
worker_server_thread_ = std::thread([this, mode, &worker_service_started]() {
// Create the worker service.
WorkerServiceImpl service(receive_queue_name_, mode);
ServerBuilder builder;
// Let GRPC choose an unused port.
int port;
builder.AddListeningPort(std::string("0.0.0.0:0"), grpc::InsecureServerCredentials(), &port);
builder.RegisterService(&service);
std::unique_ptr<Server> server(builder.BuildAndStart());
if (server == nullptr) {
RAY_CHECK(false, "Failed to create the worker service.");
}
worker_address_ = node_ip_address_ + ":" + std::to_string(port);
server_ptr_ = server.get();
RAY_LOG(RAY_INFO, "worker server listening at " << worker_address_);
worker_service_started.store(true);
// Wait for work and process work. This method does not return until
// Shutdown is called from a different thread.
server->Wait();
RAY_LOG(RAY_INFO, "Worker service thread returning.")
});
// Wait for the worker service to start. This essentially implements a
// condition variable using atomics, but that failed on Mac OS X on Travis.
while (!worker_service_started.load()) {
RAY_LOG(RAY_INFO, "Looping while waiting for the worker service to start.");
std::this_thread::sleep_for(std::chrono::milliseconds(100));
}
}
src/worker.h
-145
View File
@@ -1,145 +0,0 @@
#ifndef RAY_WORKER_H
#define RAY_WORKER_H
#include <iostream>
#include <memory>
#include <string>
#include <thread>
#include <grpc++/grpc++.h>
#include <Python.h>
using grpc::Server;
using grpc::ServerBuilder;
using grpc::ServerContext;
using grpc::Status;
#include "ray.grpc.pb.h"
#include "ray/ray.h"
#include "ipc.h"
using grpc::Channel;
using grpc::ClientContext;
using grpc::ClientWriter;
// These three constants are used to define the mode that a worker is running
// in. Right now, this is mostly used for determining how to print information
// about task failures.
enum Mode {SCRIPT_MODE, WORKER_MODE, PYTHON_MODE, SILENT_MODE};
class WorkerServiceImpl final : public WorkerService::Service {
public:
WorkerServiceImpl(const std::string& worker_address, Mode mode);
Status ExecuteTask(ServerContext* context, const ExecuteTaskRequest* request, AckReply* reply) override;
Status RunFunctionOnWorker(ServerContext* context, const RunFunctionOnWorkerRequest* request, AckReply* reply) override;
Status ImportRemoteFunction(ServerContext* context, const ImportRemoteFunctionRequest* request, AckReply* reply) override;
Status Die(ServerContext* context, const DieRequest* request, AckReply* reply) override;
Status ImportReusableVariable(ServerContext* context, const ImportReusableVariableRequest* request, AckReply* reply) override;
Status PrintErrorMessage(ServerContext* context, const PrintErrorMessageRequest* request, AckReply* reply) override;
private:
// The queue used to send commands from the worker service to the worker. This
// corresponds to the receive_queue_ in the worker.
MessageQueue<WorkerMessage*> send_queue_;
// This is true if the worker service is part of a driver process and false
// if it is part of a worker process.
Mode mode_;
};
class Worker {
public:
Worker(const std::string& node_ip_address, const std::string& scheduler_address, Mode mode);
// Submit a remote task to the scheduler. If the function in the task is not
// registered with the scheduler, we will sleep for retry_wait_milliseconds
// and try to resubmit the task to the scheduler up to max_retries more times.
SubmitTaskReply submit_task(SubmitTaskRequest* request, int max_retries = 10, int retry_wait_milliseconds = 500);
// Requests the scheduler to kill workers
bool kill_workers(ClientContext &context);
// send request to the scheduler to register this worker
void register_worker(const std::string& ip_address, const std::string& objstore_address, bool is_driver);
// get a new object ID that is registered with the scheduler
ObjectID get_objectid();
// request an object to be delivered to the local object store
void request_object(ObjectID objectid);
// Notify the scheduler about the object IDs contained within a remote object.
void add_contained_objectids(ObjectID objectid, std::vector<ObjectID> &contained_objectids);
// Allocates buffer for objectid with size of size
const char* allocate_buffer(ObjectID objectid, int64_t size, SegmentId& segmentid);
// Finishes buffer with segmentid and an offset of metadata_ofset
PyObject* finish_buffer(ObjectID objectid, SegmentId segmentid, int64_t metadata_offset);
// Gets the buffer for objectid
const char* get_buffer(ObjectID objectid, int64_t& size, SegmentId& segmentid, int64_t& metadata_offset);
// determine if the object stored in objectid is an arrow object // TODO(pcm): more general mechanism for this?
bool is_arrow(ObjectID objectid);
// unmap the segment containing an object from the local address space
void unmap_object(ObjectID objectid);
// make `alias_objectid` refer to the same object that `target_objectid` refers to
void alias_objectids(ObjectID alias_objectid, ObjectID target_objectid);
// increment the reference count for objectid
void increment_reference_count(std::vector<ObjectID> &objectid);
// decrement the reference count for objectid
void decrement_reference_count(std::vector<ObjectID> &objectid);
// Notify the scheduler that a remote function has been imported successfully.
void register_remote_function(const std::string& name, size_t num_return_vals);
// Notify the scheduler that a failure has occurred.
void notify_failure(FailedType type, const std::string& name, const std::string& error_message);
// Start the worker server which accepts commands from the scheduler. For
// workers, these commands are stored in the message queue, which is read by
// the Python interpreter. For drivers, these commands are only for printing
// error messages.
void start_worker_service(Mode mode);
// wait for next task from the RPC system. If null, it means there are no more tasks and the worker should shut down.
std::unique_ptr<WorkerMessage> receive_next_message();
// Tell the scheduler that the worker is ready for a new task.
void ready_for_new_task();
// disconnect the worker
void disconnect();
// return connected_
bool connected() { return connected_; }
// get info about scheduler state
void scheduler_info(ClientContext &context, SchedulerInfoRequest &request, SchedulerInfoReply &reply);
// get task statuses from scheduler
void task_info(ClientContext &context, TaskInfoRequest &request, TaskInfoReply &reply);
// gets indices of available objects
std::vector<int> wait(std::vector<ObjectID>& objectids);
// Export a function to be run on all workers.
void run_function_on_all_workers(const std::string& function);
// export function to workers
bool export_remote_function(const std::string& function_name, const std::string& function);
// export reusable variable to workers
void export_reusable_variable(const std::string& name, const std::string& initializer, const std::string& reinitializer);
// return the worker address
const char* get_worker_address() { return worker_address_.c_str(); }
private:
Mode mode_;
bool connected_;
const size_t CHUNK_SIZE = 8 * 1024;
std::unique_ptr<Scheduler::Stub> scheduler_stub_;
Server* server_ptr_;
std::thread worker_server_thread_;
bip::managed_shared_memory segment_;
WorkerId workerid_;
ObjStoreId objstoreid_;
std::string scheduler_address_;
std::string objstore_address_;
std::string worker_address_;
std::string node_ip_address_;
// The queue used to send commands from the worker service to the worker.
// This queue is created by the worker. This corresponds to the send_queue_ in
// the worker service.
MessageQueue<WorkerMessage*> receive_queue_;
// The name of the receive queue.
std::string receive_queue_name_;
// The queue used to send requests to the object store. There is a single
// queue shared by all workers sending requests to the object store, and this
// queue is created by the object store.
MessageQueue<ObjRequest> request_obj_queue_;
// The queue used to receive object addresses from the object store. This
// queue is created by this worker.
MessageQueue<ObjHandle> receive_obj_queue_;
std::shared_ptr<MemorySegmentPool> segmentpool_;
};
#endif
thirdparty/grpc
Vendored
-1
Submodule thirdparty/grpc deleted from 2a69139aa7
thirdparty/hiredis
Vendored
-1
Submodule thirdparty/hiredis deleted from 5f98e1d35d
thirdparty/numbuf
Vendored
-1
Submodule thirdparty/numbuf deleted from 7055c6f793
thirdparty/python
Vendored
-1
Submodule thirdparty/python deleted from 3f8fa00528