apaam/mlpack
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Fixing again.

Browse Source
This commit is contained in:
Dongryeol Lee
2010-10-09 14:42:43 +00:00
parent a910415844
commit 37d413883f
3 changed files with 47 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files
fastlib/trunk/contrib/dongryel/thesis_research/core/table/distributed_table.h
+21 -19
View File
@@ -33,11 +33,11 @@ class DistributedTable: public boost::noncopyable {
boost::mpi::communicator *comm_;
core::table::PointInbox point_inbox_;
//core::table::PointInbox point_inbox_;
core::table::PointRequestMessageInbox point_request_message_inbox_;
core::table::PointRequestMessageOutbox point_request_message_outbox_;
//core::table::PointRequestMessageOutbox point_request_message_outbox_;
public:
@@ -63,20 +63,22 @@ class DistributedTable: public boost::noncopyable {
comm_->barrier();
// Terminate the point inbox.
comm_->isend(
comm_->rank(),
core::table::DistributedTableMessage::TERMINATE_POINT_INBOX, 0);
//comm_->isend(
//comm_->rank(),
//core::table::DistributedTableMessage::TERMINATE_POINT_INBOX, 0);
// Terminate the point request message inbox.
comm_->isend(
comm_->rank(),
core::table::DistributedTableMessage::TERMINATE_POINT_REQUEST_MESSAGE_INBOX, 0);
boost::unique_lock<boost::mutex> lock(point_request_message_inbox_.mutex());
point_request_message_inbox_.point_request_message_inbox_quitting().wait(lock);
// Wait until the point inbox is terminated.
{
boost::unique_lock<boost::mutex> lock(point_inbox_.termination_mutex());
point_inbox_.termination_cond().wait(lock);
}
//{
//boost::unique_lock<boost::mutex> lock(point_inbox_.termination_mutex());
//point_inbox_.termination_cond().wait(lock);
//}
// Put a barrier so that all processes are ready to destroy each
// of their own tables and trees.
@@ -155,15 +157,15 @@ class DistributedTable: public boost::noncopyable {
*comm_, owned_table_->n_entries(), local_n_entries_);
// Initialize the mail boxes.
point_inbox_.Init(comm_);
//point_inbox_.Init(comm_);
point_request_message_inbox_.Init(comm_);
point_request_message_outbox_.Init(
comm_, owned_table_, &point_request_message_inbox_);
//point_request_message_outbox_.Init(
//comm_, owned_table_, &point_request_message_inbox_);
// Detach the server threads for each distributed process.
point_inbox_.Detach();
//point_inbox_.Detach();
point_request_message_inbox_.Detach();
point_request_message_outbox_.Detach();
//point_request_message_outbox_.Detach();
// Put a barrier to ensure that every process has started up the
// mailboxes.
@@ -205,15 +207,15 @@ class DistributedTable: public boost::noncopyable {
point_request_message);
// Do a conditional wait until the point is ready.
boost::unique_lock<boost::mutex> lock(
point_inbox_.point_received_mutex());
point_inbox_.wait(lock);
//boost::unique_lock<boost::mutex> lock(
//point_inbox_.point_received_mutex());
//point_inbox_.wait(lock);
// If we are here, then the point is ready. Copy the point.
entry->Init(point_inbox_.point());
//entry->Init(point_inbox_.point());
// Signal that we are done copying out the point.
point_inbox_.invalidate_point();
//point_inbox_.invalidate_point();
}
}
fastlib/trunk/contrib/dongryel/thesis_research/core/table/distributed_table.test.cc
+3 -3
View File
@@ -70,9 +70,9 @@ int main(int argc, char *argv[]) {
world.rank(), target_point_id, target_rank);
distributed_table.get(target_rank, target_point_id, &point);
printf("Process %d received point %d of length %d from Process %d.\n",
world.rank(), target_point_id, point.reference().n_elem, target_rank);
point.reference().print();
//printf("Process %d received point %d of length %d from Process %d.\n",
// world.rank(), target_point_id, point.reference().n_elem, target_rank);
//point.reference().print();
}
printf("Process %d is all done!\n", world.rank());
fastlib/trunk/contrib/dongryel/thesis_research/core/table/mailbox.h
+23 -25
View File
@@ -13,6 +13,7 @@ namespace table {
class Table;
/*
class PointInbox {
private:
@@ -86,7 +87,7 @@ class PointInbox {
this)));
}
bool has_outstanding_point_messages() {
boost::optional<boost::mpi::status> has_outstanding_point_messages() {
return comm_->iprobe(
boost::mpi::any_source,
core::table::DistributedTableMessage::RECEIVE_POINT);
@@ -100,8 +101,11 @@ class PointInbox {
}
bool time_to_quit() {
printf("Condition: %d %d %d\n", (! point_handle_is_valid_),
( ! has_outstanding_point_messages() ),
termination_signal_arrived());
return point_handle_is_valid_ == false &&
has_outstanding_point_messages() == false &&
(! has_outstanding_point_messages() ) &&
termination_signal_arrived();
}
@@ -142,11 +146,12 @@ class PointInbox {
termination_cond_.notify_one();
}
};
*/
class PointRequestMessageInbox {
private:
boost::condition_variable point_request_message_received_cond_;
boost::condition_variable point_request_message_inbox_quitting_;
boost::mpi::communicator *comm_;
@@ -154,8 +159,6 @@ class PointRequestMessageInbox {
bool point_request_message_is_valid_;
bool do_test_;
boost::mpi::request point_request_message_handle_;
boost::shared_ptr<boost::thread> point_request_message_inbox_thread_;
@@ -164,8 +167,12 @@ class PointRequestMessageInbox {
public:
void wait(boost::unique_lock<boost::mutex> &lock_in) {
point_request_message_received_cond_.wait(lock_in);
boost::mutex &mutex() {
return mutex_;
}
boost::condition_variable &point_request_message_inbox_quitting() {
return point_request_message_inbox_quitting_;
}
void Detach() {
@@ -193,13 +200,11 @@ class PointRequestMessageInbox {
PointRequestMessageInbox() {
comm_ = NULL;
do_test_ = true;
point_request_message_is_valid_ = false;
}
void invalidate_point_request_message() {
point_request_message_is_valid_ = false;
do_test_ = true;
}
bool termination_signal_arrived() {
@@ -214,7 +219,7 @@ class PointRequestMessageInbox {
// It is time to quit when there are no valid messages to
// handle, and the terminate signal is here.
return point_request_message_is_valid_ == false &&
has_outstanding_point_request_messages() == false &&
(! has_outstanding_point_request_messages()) &&
termination_signal_arrived();
}
@@ -223,7 +228,7 @@ class PointRequestMessageInbox {
point_request_message_handle_.test();
}
bool has_outstanding_point_request_messages() {
boost::optional<boost::mpi::status> has_outstanding_point_request_messages() {
return comm_->iprobe(
boost::mpi::any_source,
core::table::DistributedTableMessage::REQUEST_POINT);
@@ -248,29 +253,21 @@ class PointRequestMessageInbox {
comm_->rank());
}
// Check whether the request is done.
if(do_test_ && this->point_request_message_received()) {
// Wake up the thread waiting on the request. This thread
// turns off the validity flag after grabbing whch process
// wants a point.
printf("Waking up the point request outbox.\n");
do_test_ = false;
point_request_message_received_cond_.notify_one();
// Check whether the request is done. If so, send a MPI
// message to self for the outbox.
if(this->point_request_message_received()) {
point_request_message_is_valid_ = false;
}
} // end of the infinite server loop.
// Kill the outbox handling the point request messages.
comm_->isend(
comm_->rank(),
core::table::DistributedTableMessage::TERMINATE_POINT_REQUEST_MESSAGE_OUTBOX,
0);
printf("Point request message inbox for Process %d is quitting.\n",
comm_->rank());
point_request_message_inbox_quitting_.notify_one();
}
};
/*
class PointRequestMessageOutbox {
private:
@@ -385,6 +382,7 @@ class PointRequestMessageOutbox {
comm_->rank());
}
};
*/
};
};