From 3420ec2ca3ee7ff2c97e7014fa73f43444c18800 Mon Sep 17 00:00:00 2001 From: Garry Boyer Date: Tue, 22 May 2007 04:50:06 +0000 Subject: [PATCH] hi --- fastlib/u/garryb/nbr/dfs.h | 44 +++--- fastlib/u/garryb/nbr/rpc.cc | 6 + fastlib/u/garryb/nbr/rpc.h | 295 +++++++++++++++++++++++++++++++++++ fastlib/u/garryb/nbr/tkde.cc | 2 +- 4 files changed, 322 insertions(+), 25 deletions(-) create mode 100644 fastlib/u/garryb/nbr/rpc.cc create mode 100644 fastlib/u/garryb/nbr/rpc.h diff --git a/fastlib/u/garryb/nbr/dfs.h b/fastlib/u/garryb/nbr/dfs.h index 79199fbdfc..9539e962e0 100644 --- a/fastlib/u/garryb/nbr/dfs.h +++ b/fastlib/u/garryb/nbr/dfs.h @@ -66,15 +66,18 @@ class DualTreeDepthFirst { } private: - void Pair_(index_t q_node_i, index_t r_node_i, + void Pair_( + const typename GNP::QNode *q_node, + const typename GNP::RNode *r_node, const typename GNP::Delta& delta, - const typename GNP::QMassResult& exclusive_unvisited); + const typename GNP::QMassResult& exclusive_unvisited, + QMutableInfo *q_node_mut); void BaseCase_( const typename GNP::QNode *q_node, const typename GNP::RNode *r_node, const typename GNP::QMassResult& exclusive_unvisited, QMutableInfo *q_node_mut); - void PushDown_(index_t q_node_i); + void PushDown_(index_t q_node_i, QMutableInfo *q_node_mut); }; template @@ -149,8 +152,8 @@ void DualTreeDepthFirst::Begin() { if (do_naive_) { BaseCase_(q_root, r_root_, empty_mass_result, q_root_mut); } else { - Pair_(0, 0, delta, empty_mass_result); - PushDown_(0); + Pair_(q_root, r_root_, delta, empty_mass_result, q_root_mut); + PushDown_(0, q_root_mut); } } @@ -178,9 +181,9 @@ void DualTreeDepthFirst::Begin() { } template -void DualTreeDepthFirst::PushDown_(index_t q_node_i) { +void DualTreeDepthFirst::PushDown_( + index_t q_node_i, QMutableInfo *q_node_mut) { const typename GNP::QNode *q_node = q_nodes_.StartRead(q_node_i); - QMutableInfo *q_node_mut = q_mutables_.StartWrite(q_node_i); if (q_node->is_leaf()) { for (index_t q_i = q_node->begin(); q_i < q_node->end(); q_i++) { @@ -202,24 +205,21 @@ void DualTreeDepthFirst::PushDown_(index_t q_node_i) { q_child_mut->postponed.ApplyPostponed(param_, q_node_mut->postponed); + PushDown_(q_child_i, q_child_mut); q_mutables_.StopWrite(q_child_i); - - PushDown_(q_child_i); } } q_nodes_.StopRead(q_node_i); - q_mutables_.StopWrite(q_node_i); } template -void DualTreeDepthFirst::Pair_(index_t q_node_i, index_t r_node_i, +void DualTreeDepthFirst::Pair_( + const typename GNP::QNode *q_node, + const typename GNP::RNode *r_node, const typename GNP::Delta& delta, - const typename GNP::QMassResult& exclusive_unvisited) { - const typename GNP::RNode *r_node = r_nodes_.StartRead(r_node_i); - const typename GNP::QNode *q_node = q_nodes_.StartRead(q_node_i); - QMutableInfo *q_node_mut = q_mutables_.StartWrite(q_node_i); - + const typename GNP::QMassResult& exclusive_unvisited, + QMutableInfo *q_node_mut) { DEBUG_MSG(1.0, "Checking (%d,%d) x (%d,%d)", q_node->begin(), q_node->end(), r_node->begin(), r_node->end()); @@ -232,7 +232,7 @@ void DualTreeDepthFirst::Pair_(index_t q_node_i, index_t r_node_i, mu.ApplyDelta(param_, delta); if (!GNP::Algorithm::ConsiderQueryTermination( - param_, *q_node, mu, global_result_, &q_node_mut->postponed)) { + param_, *q_node, mu, global_result_, &q_node_mut->postponed)) { q_node_mut->mass_result.ApplyDelta(param_, delta); DEBUG_MSG(1.0, "Termination prune"); } else if (!GNP::Algorithm::ConsiderPairExtrinsic( @@ -263,7 +263,7 @@ void DualTreeDepthFirst::Pair_(index_t q_node_i, index_t r_node_i, if (GNP::Algorithm::ConsiderPairIntrinsic( param_, *q_child, *r_node, &child_delta, &global_result_, &q_child_mut->postponed)) { - Pair_(q_child_i, r_node_i, delta, exclusive_unvisited); + Pair_(q_child, r_node, delta, exclusive_unvisited, q_child_mut); } // We must VERY carefully apply both the horizontal and vertical join @@ -318,20 +318,16 @@ void DualTreeDepthFirst::Pair_(index_t q_node_i, index_t r_node_i, if (do_r2) { exclusive_unvisited_for_r1.ApplyDelta(param_, delta2); } - Pair_(q_node_i, r_child1_i, delta1, exclusive_unvisited_for_r1); + Pair_(q_node, r_child1, delta1, exclusive_unvisited_for_r1, q_node_mut); } if (do_r2) { - Pair_(q_node_i, r_child2_i, delta2, exclusive_unvisited); + Pair_(q_node, r_child2, delta2, exclusive_unvisited, q_node_mut); } r_nodes_.StopRead(r_child1_i); r_nodes_.StopRead(r_child2_i); } } - - r_nodes_.StopRead(r_node_i); - q_nodes_.StopRead(q_node_i); - q_mutables_.StopWrite(q_node_i); } template diff --git a/fastlib/u/garryb/nbr/rpc.cc b/fastlib/u/garryb/nbr/rpc.cc new file mode 100644 index 0000000000..edfe9f8902 --- /dev/null +++ b/fastlib/u/garryb/nbr/rpc.cc @@ -0,0 +1,6 @@ +#include "rpc.h" + +Mutex global_mpi_lock; + + + diff --git a/fastlib/u/garryb/nbr/rpc.h b/fastlib/u/garryb/nbr/rpc.h new file mode 100644 index 0000000000..647eda03af --- /dev/null +++ b/fastlib/u/garryb/nbr/rpc.h @@ -0,0 +1,295 @@ +#ifndef NBR_RPC_H +#define NBR_RPC_H + +extern Mutex global_mpi_lock; + +/** + * This class is your interface to an object that's somewhere else. + */ +template +class RemoteObjectStub { + FORBID_COPY(RemoteObjectStub); + + private: + ArrayList data_; + int destination_; + int channel_; + Mutex mutex_; +#ifdef DEBUG + bool locked_; // for debug mode +#endif + + public: + void Init(int channel_in, int destination_in) { + channel_ = channel_in; + destination_ = destination_in; + } + + const Response *Request(const Request& request) { + DEBUG_ASSERT(locked_ == true); + + global_mpi_lock.Lock(); + + data_.Resize(ot::PointerFrozenSize(request)); + ot::PointerFreeze(request, data_.begin()); + MPI_Send(data_.begin(), data_.size(), MPI_CHAR, + destination_, channel_, MPI_COMM_WORLD); + MPI_Status status; + MPI_Probe(MPI_ANY_SOURCE, channel_, MPI_COMM_WORLD, &status); + data_.Resize(status.MPI_LENGTH); + MPI_Recv(data_.begin(), data_.size(), MPI_CHAR, + destination_, channel_, MPI_COMM_WORLD, &status); + + global_mpi_lock.Unlock(); + + return ot::PointerThaw(data_.begin()); + } + + void Lock() { + mutex_.Lock(); + DEBUG_ONLY(locked_ = true); + } + + void Unlock() { + DEBUG_ONLY(locked_ = false); + mutex_.Unlock(); + } +}; + +class RawRemoteObjectBackend { + private: + int channel_; + + public: + virtual ~RawRemoteObjectBackend() {} + + void RemoteObjectInit(int channel_in) { + channel_ = channel_in; + } + + virtual void HandleRequestRaw(ArrayList *request, + ArrayList *response); + + int channel() const { + return channel_; + } +}; + +/** + * This is how you define the network object on the server. + */ +template +class RemoteObjectBackend + : public RawRemoteObjectBackend { + public: + virtual ~RemoteObjectBackend() {} + + virtual void HandleRequestRaw(ArrayList *raw_request, + ArrayList *raw_response) { + const Request* real_request = ot::PointerThaw(raw_request->begin()); + Response real_response; + HandleRequest(*real_request, &real_response); + raw_response->Resize(ot::PointerFrozenSize(real_response)); + ot::PointerFreeze(real_response, raw_response->begin()); + } + + virtual void HandleRequest(const Request& request, Response *response) = 0; +}; + +class RemoteObjectServer { + FORBID_COPY(RemoteObjectServer); + + public: + ArrayList channels_; + int last_tag_; + + public: + const int TAG_BORN = 0; + const int TAG_DONE = 1; + const int TAG_FIRST_AVAILABLE = 2; + + public: + void Init() { + last_tag_ = TAG_FIRST_AVAILABLE - 1; + } + + /** + * Returns a new tag for use. + * + * All machines have to use NewTag in an exactly identical way. + */ + int NewTag() { + return ++last_tag_; + } + + void Register(RawRemoteObjectBackend *channel) { + if (channel->channel() >= channels_.size()) { + channels_.Resize(channel->channel() + 1); + } + channels_[channel->channel()] = channel_; + } + + void Loop(int n_workers_total) { + ArrayList data_recv; + ArrayList data_send; + int n_workers_born = 0; + int n_workers_done = 0; + + data_send.Init(); + data_recv.Init(); + + while (n_workers_done != n_workers_total) { + MPI_Status status; + + MPI_Probe(MPI_ANY_SOURCE, MPI_ANY_TAG, MPI_COMM_WORLD, &status); + data_recv.Resize(status.MPI_LENGTH); + MPI_Recv(data_recv.begin(), data_recv.size(), MPI_CHAR, + MPI_ANY_SOURCE, MPI_ANY_TAG, MPI_COMM_WORLD, &status); + + if (status.MPI_TAG == TAG_BORN) { + n_workers_born++; + } else if (status.MPI_TAG == TAG_DONE) { + n_workers_done++; + } else { + channels_[status.MPI_SOURCE]->HandleRawRequest( + &data_recv, &data_send); + MPI_Send(data_send.begin(), data_send.size(), MPI_CHAR, + status.MPI_SOURCE, status.MPI_TAG, MPI_COMM_WORLD); + } + } + } +}; + +/** + * Protocol request for networked block devices. + */ +struct BlockRequest { + BlockDevice::block_t blockid; + BlockDevice::offset_t begin; + BlockDevice::offset_t end; + enum { READ, WRITE, ALLOC } operation; + ArrayList payload; + + OT_DEF(BlockRequest) { + OT_MY_OBJECT(remember to make sure it compiles); + OT_MY_OBJECT(blockid); + OT_MY_OBJECT(begin); + OT_MY_OBJECT(end); + OT_MY_OBJECT(operation); + OT_MY_OBJECT(payload); + } +}; + +/** + * Protocol response for networked block devices. + */ +struct BlockResponse { + BlockDevice::blockid_t blockid; + ArrayList payload; + + OT_DEF(BlockResponse) { + OT_MY_OBJECT(remember to make sure it compiles); + OT_MY_OBJECT(blockid); + OT_MY_OBJECT(payload); + } +}; + +class BlockDeviceRemoteObjectBackend + : public RemoteObjectBackend { + private: + BlockDevice *blockdev_; + + public: + void HandleRequest(const BlockRequest& request, BlockResponse *response) { + + if (request.operation == BlockRequest::WRITE) { + response->payload.Init(); + blockdev_->Write(request.blockid, request.begin, request.end, + request.payload.begin()); + response->blockid = request.blockid; + } else if (request.operation == BlockRequest::READ) { + response->payload.Init(request.end - request.begin); + blockdev_->Read(request.blockid, request.begin, request.end, + response->payload.begin()); + response->blockid = request.blockid; + } else if (request.operation == BlockRequest::ALLOC) { + response->payload.Init(); + response->blockid = blockdev_->AllocBlock(); + } else { + FATAL("Only valid block operations are READ/WRITE."); + } + } +}; + +/** + * A block device sitting on another computer. + * + * Individual instances of this object are not + * thread safe? + */ +class BlockDeviceRemote + : public BlockDevice { + private: + RemoteObjectStub stub_; + + public: + void Init(int channel_in, int destination_in) { + stub_.Init(channel_in, destination_in); + } + + virtual void Read(blockid_t blockid, + offset_t begin, offset_t end, char *data) { + BlockRequest request; + + request.blockid = blockid; + request.begin = begin; + request.end = end; + request.operation = BlockRequest::READ; + request.payload.Init(); + + stub_.Lock(); + const BlockResponse *response = stub_.Request(request); + memcpy(data, response->payload.begin(), response->payload.size()); + DEBUG_SAME_INT(response->payload.size() == end - begin); + stub_.Unlock(); + } + + virtual void Write(blockid_t blockid, + offset_t begin, offset_t end, const char *data) { + BlockRequest request; + + request.blockid = blockid; + request.begin = begin; + request.end = end; + request.operation = BlockRequest::WRITE; + request.payload.Copy(data, end - begin); + + stub_.Lock(); + const BlockResponse *response = stub_.Request(request); + DEBUG_ASSERT(response->payload.size(), 0); + stub_.Unlock(); + } + + virtual blockid_t AllocBlock() { + BlockRequest request; + BlockDevice::blockid_t blockid; + + request.blockid = 0; + request.begin = 0; + request.end = 0; + request.operation = BlockRequest::ALLOC; + request.payload.Init(); + + stub_.Lock(); + const BlockResponse *response = stub_.Request(request); + DEBUG_ASSERT(response->payload.size(), 0); + blockid = response->blockid; + stub_.Unlock(); + + return blockid; + } + + virtual void Close() {} +}; + +#endif diff --git a/fastlib/u/garryb/nbr/tkde.cc b/fastlib/u/garryb/nbr/tkde.cc index 1b901035a3..aaa9dd5796 100644 --- a/fastlib/u/garryb/nbr/tkde.cc +++ b/fastlib/u/garryb/nbr/tkde.cc @@ -85,7 +85,7 @@ class Tkde { thresh = other.thresh; dim = other.dim; } - + /** * Initialize parameters from a data node (Req NBR). */