diff --git a/fastlib/trunk/contrib/dongryel/thesis_research/core/table/distributed_table.h b/fastlib/trunk/contrib/dongryel/thesis_research/core/table/distributed_table.h index 13f7578df5..b08caffb2d 100644 --- a/fastlib/trunk/contrib/dongryel/thesis_research/core/table/distributed_table.h +++ b/fastlib/trunk/contrib/dongryel/thesis_research/core/table/distributed_table.h @@ -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 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 lock(point_inbox_.termination_mutex()); - point_inbox_.termination_cond().wait(lock); - } + //{ + //boost::unique_lock 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 lock( - point_inbox_.point_received_mutex()); - point_inbox_.wait(lock); + //boost::unique_lock 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(); } } diff --git a/fastlib/trunk/contrib/dongryel/thesis_research/core/table/distributed_table.test.cc b/fastlib/trunk/contrib/dongryel/thesis_research/core/table/distributed_table.test.cc index 6f3ea0fb62..dbc0b1dcc0 100644 --- a/fastlib/trunk/contrib/dongryel/thesis_research/core/table/distributed_table.test.cc +++ b/fastlib/trunk/contrib/dongryel/thesis_research/core/table/distributed_table.test.cc @@ -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()); diff --git a/fastlib/trunk/contrib/dongryel/thesis_research/core/table/mailbox.h b/fastlib/trunk/contrib/dongryel/thesis_research/core/table/mailbox.h index 1d1eef25ac..da75ce1cdf 100644 --- a/fastlib/trunk/contrib/dongryel/thesis_research/core/table/mailbox.h +++ b/fastlib/trunk/contrib/dongryel/thesis_research/core/table/mailbox.h @@ -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 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 point_request_message_inbox_thread_; @@ -164,8 +167,12 @@ class PointRequestMessageInbox { public: - void wait(boost::unique_lock &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 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()); } }; +*/ }; };