From 249d701dcccf07b83084aae9bac8ea17b1645840 Mon Sep 17 00:00:00 2001 From: Garry Boyer Date: Sat, 28 Jul 2007 05:32:24 +0000 Subject: [PATCH] hi --- fastlib/u/garryb/nbr/distribcache.cc | 96 +++++++++++++++++----------- fastlib/u/garryb/nbr/distribcache.h | 7 ++ fastlib/u/garryb/nbr/nbr_utils.h | 38 ++++++----- fastlib/u/garryb/nbr/rpc_sock.cc | 2 +- 4 files changed, 89 insertions(+), 54 deletions(-) diff --git a/fastlib/u/garryb/nbr/distribcache.cc b/fastlib/u/garryb/nbr/distribcache.cc index 7512c4080f..36d18d2085 100644 --- a/fastlib/u/garryb/nbr/distribcache.cc +++ b/fastlib/u/garryb/nbr/distribcache.cc @@ -141,8 +141,8 @@ void DistributedCache::HandleStatusInformation_( // performing the sync barrier and couldn't have written to the block // myself, and if some other machine had written it, I'd actually // be the owner. - DEBUG_ASSERT_MSG(!block->is_owner(), - "Lost ownership unexpectedly"); + DEBUG_ASSERT_MSG(status->owner != my_rank_, + "Received ownership unexpectedly"); DEBUG_ASSERT_MSG(!block->is_dirty(), "Remote blocks shouldn't be dirty during a sync."); block->value = ~status->owner; @@ -163,7 +163,7 @@ void DistributedCache::ComputeStatusInformation_( status->is_new = block->is_new(); } else { status->owner = -1; - status->is_new = NOT_DIRTY_NEW; + status->is_new = false; } } mutex_.Unlock(); @@ -209,15 +209,28 @@ void DistributedCache::StartSync() { blockid_t blockid = slot->blockid; if (blockid >= 0) { BlockMetadata *block = &blocks[blockid]; - slot++; - if (unlikely(!block->is_owner())) { + if (!block->is_owner()) { slot->blockid = -1; DEBUG_ASSERT_MSG(block->locks == 0, "Why is a locked block in LRU?"); Purge_(blockid); + DEBUG_ASSERT_MSG(!block->is_dirty(), + "We purged a block and it's still marked as dirty?"); } } + slot++; } while (i != 0); +#ifdef DEBUG + for (index_t i = 0; i < n_blocks_; i++) { + BlockMetadata *block = &blocks[i]; + DEBUG_ASSERT_MSG(!block->is_busy(), "A block is busy during sync."); + if (block->is_dirty()) { + DEBUG_ASSERT(block->is_in_core()); + DEBUG_ASSERT(block->is_owner()); + } + } +#endif + // TODO: Make absolutely certain nobody is currently accessesing the cache write_ranges_.Reset(); mutex_.Unlock(); @@ -240,7 +253,7 @@ void DistributedCache::Read(blockid_t blockid, block->locks = 0; } offset_t n_bytes = end - begin; - mem::Copy(buf, block->data + begin, n_bytes); + mem::CopyBytes(buf, block->data + begin, n_bytes); handler_->BlockFreeze(blockid, begin, n_bytes, block->data + begin, buf); if (unlikely(block->locks == 0)) { EncacheBlock_(blockid); @@ -263,11 +276,12 @@ void DistributedCache::RemoteRead(blockid_t blockid, void DistributedCache::Write(blockid_t blockid, offset_t begin, offset_t end, const char *buf) { - // hmm, i have to be the owner of the block - char *dest = StartWrite(blockid, true); + // i have to be the owner of the block + char *dest = StartWrite(blockid, true) + begin; size_t n_bytes = end - begin; - mem::CopyBytes(dest + begin, buf, n_bytes); + mem::CopyBytes(dest, buf, n_bytes); + handler_->BlockThaw(blockid, begin, n_bytes, dest); StopWrite(blockid); } @@ -284,6 +298,7 @@ void DistributedCache::RemoteWrite(blockid_t blockid, if (!block->is_owner()) { // when we receive a remote write, we are always the owner block->value = SELF_OWNER_UNALLOCATED; // mark as owner + block->status = NOT_DIRTY_NEW; } mutex_.Unlock(); Write(blockid, begin, end, buf); @@ -452,6 +467,7 @@ void DistributedCache::HandleMiss_(BlockDevice::blockid_t blockid) { DEBUG_ASSERT(block->data != NULL); if (block->is_new()) { + fprintf(stderr, "%d: New block %d\n", rpc::rank(), blockid); DEBUG_ASSERT_MSG(block->status == NOT_DIRTY_NEW, "Block should be NOT_DIRTY_NEW, because that's what is_new() means"); handler_->BlockInitFrozen(blockid, 0, n_block_bytes_, block->data); @@ -470,7 +486,8 @@ void DistributedCache::HandleLocalMiss_(BlockDevice::blockid_t blockid) { blockid_t local_blockid = block->local_blockid(); DEBUG_ASSERT(block->is_owner()); - fprintf(stderr, "DISK: reading %d from %d\n", blockid, local_blockid); + fprintf(stderr, "DISK: reading %d from %d (%d bytes)\n", + blockid, local_blockid, n_block_bytes_); overflow_device_->Read(local_blockid, 0, n_block_bytes_, block->data); } @@ -480,7 +497,7 @@ void DistributedCache::HandleRemoteMiss_(BlockDevice::blockid_t blockid) { DEBUG_ASSERT(!block->is_owner()); ReadTransaction read_transaction; - read_transaction.Doit(channel_num_, block->value, + read_transaction.Doit(channel_num_, block->owner(), blockid, 0, n_block_bytes_, block->data); } @@ -510,7 +527,7 @@ void DistributedCache::WritebackDirtyLocalFreeze_( DEBUG_ASSERT(block->is_owner()); - BlockDevice::blockid_t local_blockid = block->value; + BlockDevice::blockid_t local_blockid = block->local_blockid(); if (local_blockid == SELF_OWNER_UNALLOCATED) { local_blockid = overflow_free_; @@ -524,7 +541,8 @@ void DistributedCache::WritebackDirtyLocalFreeze_( } handler_->BlockFreeze(blockid, 0, n_block_bytes_, block->data, block->data); - fprintf(stderr, "DISK: writing %d to %d\n", blockid, local_blockid); + fprintf(stderr, "DISK: writing %d to %d (%d bytes)\n", + blockid, local_blockid, n_block_bytes_); overflow_device_->Write(local_blockid, 0, n_block_bytes_, block->data); block->status = NOT_DIRTY_OLD; } @@ -538,7 +556,7 @@ void DistributedCache::WritebackDirtyRemote_(BlockDevice::blockid_t blockid) { if (block->status == FULLY_DIRTY) { // The entire block is dirty WriteTransaction write_transaction; - write_transaction.Doit(channel_num_, block->value, blockid, handler_, + write_transaction.Doit(channel_num_, block->owner(), blockid, handler_, 0, n_block_bytes_, block->data); } else { DEBUG_ASSERT(block->status == PARTIALLY_DIRTY); @@ -561,8 +579,9 @@ void DistributedCache::WritebackDirtyRemote_(BlockDevice::blockid_t blockid) { end_offset = end.offset; } WriteTransaction write_transaction; - write_transaction.Doit(channel_num_, block->value, blockid, handler_, - begin_offset, end_offset, block->data); + write_transaction.Doit(channel_num_, block->owner(), blockid, + handler_, + begin_offset, end_offset, block->data + begin_offset); DEBUG_ONLY(anything_done = true); } } @@ -603,7 +622,7 @@ void DistributedCache::ReadTransaction::Doit( response = NULL; Send(message); cond.Wait(); - mem::Copy(buffer, response->data(), end - begin); + mem::CopyBytes(buffer, response->data(), end - begin); delete response; } @@ -654,7 +673,7 @@ void DistributedCache::WriteTransaction::Doit( request->begin = begin; request->end = end; request->rank = 0; - mem::Copy(request->data_as(), buffer, end - begin); + mem::CopyBytes(request->data_as(), buffer, end - begin); handler->BlockFreeze(blockid, begin, end, buffer, request->data_as()); Send(message); Done(); @@ -693,6 +712,7 @@ void DistributedCache::ConfigTransaction::HandleMessage(Message *message) { void DistributedCache::ResponseTransaction::Init( DistributedCache *cache_in) { cache_ = cache_in; + Transaction::Init(cache_->channel_num()); } void DistributedCache::ResponseTransaction::HandleMessage( @@ -751,8 +771,9 @@ void DistributedCache::SyncInfo::Init(const DistributedCache& cache) { void DistributedCache::SyncInfo::MergeWith(const SyncInfo& other) { index_t old_size = statuses.size(); + index_t min_size = min(statuses.size(), other.statuses.size()); - for (index_t i = 0; i < old_size; i++) { + for (index_t i = 0; i < min_size; i++) { BlockStatus *orig = &statuses[i]; const BlockStatus *in = &other.statuses[i]; if (in->owner >= 0) { @@ -801,9 +822,10 @@ void DistributedCache::SyncTransaction::HandleMessage(Message *message) { default: FATAL("Unknown state"); } + bool is_done = (state_ == DONE); mutex_.Unlock(); delete message; - if (state_ == DONE) { + if (is_done) { delete this; } } @@ -811,8 +833,9 @@ void DistributedCache::SyncTransaction::HandleMessage(Message *message) { void DistributedCache::SyncTransaction::StartSyncFlushDone() { mutex_.Lock(); ChildFlushed_(); + bool is_done = (state_ == DONE); mutex_.Unlock(); - if (state_ == DONE) { + if (is_done) { delete this; } } @@ -904,14 +927,7 @@ void DistributedCache::CacheChannel::Init(DistributedCache *cache_in) { } void DistributedCache::CacheChannel::StartSyncFlushDone() { - SyncTransaction *t; - mutex_.Lock(); - if (sync_transaction_ == NULL) { - sync_transaction_ = new SyncTransaction(); - sync_transaction_->Init(cache_); - } - t = sync_transaction_; - mutex_.Unlock(); + SyncTransaction *t = GetSyncTransaction_();; t->StartSyncFlushDone(); } @@ -927,18 +943,24 @@ void DistributedCache::CacheChannel::SyncDone() { sync_done_.Done(); } +DistributedCache::SyncTransaction * +DistributedCache::CacheChannel::GetSyncTransaction_() { + SyncTransaction *t; + mutex_.Lock(); + if (sync_transaction_ == NULL) { + sync_transaction_ = new SyncTransaction(); + sync_transaction_->Init(cache_); + } + t = sync_transaction_; + mutex_.Unlock(); + return t; +} + Transaction *DistributedCache::CacheChannel::GetTransaction( Message *message) { Request *request = reinterpret_cast(message->data()); if (unlikely(request->type == Request::SYNC)) { - Transaction *t; - mutex_.Lock(); - if (sync_transaction_ == NULL) { - sync_transaction_ = new SyncTransaction(); - sync_transaction_->Init(cache_); - } - t = sync_transaction_; - mutex_.Unlock(); + Transaction *t = GetSyncTransaction_(); return t; } else { ResponseTransaction *t = new ResponseTransaction(); diff --git a/fastlib/u/garryb/nbr/distribcache.h b/fastlib/u/garryb/nbr/distribcache.h index 06ead1ab18..ad535659ce 100644 --- a/fastlib/u/garryb/nbr/distribcache.h +++ b/fastlib/u/garryb/nbr/distribcache.h @@ -224,6 +224,9 @@ class DistributedCache : public BlockDevice { Mutex mutex_; DoneCondition sync_done_; + private: + SyncTransaction *GetSyncTransaction_(); + public: void Init(DistributedCache *cache_in); Transaction *GetTransaction(Message *message); @@ -315,6 +318,10 @@ class DistributedCache : public BlockDevice { int owner(const struct DistributedCache *cache) const { return unlikely(value >= 0) ? cache->my_rank_ : (~value); } + int owner() const { + DEBUG_ASSERT_MSG(!is_owner(), "owner() doesn't work if i'm the owner"); + return ~value; + } /** Determines whether I am the owner. */ bool is_owner() const { return value >= 0; diff --git a/fastlib/u/garryb/nbr/nbr_utils.h b/fastlib/u/garryb/nbr/nbr_utils.h index b161d1f48f..1f0684bb86 100644 --- a/fastlib/u/garryb/nbr/nbr_utils.h +++ b/fastlib/u/garryb/nbr/nbr_utils.h @@ -296,7 +296,6 @@ class RpcMonochromaticDualTreeRunner { * The configuration of the problem. */ struct Config { - typename GNP::Param param; int n_threads; int data_points_mb; int data_nodes_mb; @@ -307,7 +306,6 @@ class RpcMonochromaticDualTreeRunner { } OT_DEF(Config) { - OT_MY_OBJECT(param); OT_MY_OBJECT(n_threads); OT_MY_OBJECT(data_points_mb); OT_MY_OBJECT(data_nodes_mb); @@ -317,6 +315,7 @@ class RpcMonochromaticDualTreeRunner { struct Master { DataGetterBackend config_backend; + DataGetterBackend param_backend; RemoteWorkQueueBackend work_backend; }; @@ -344,6 +343,7 @@ class RpcMonochromaticDualTreeRunner { datanode *data_module_; const char *gnp_name_; + typename GNP::Param param_; Config config_; WorkQueueInterface *work_queue_; index_t n_points_; @@ -407,7 +407,7 @@ void RpcMonochromaticDualTreeRunner::ReadData_() { fx_timer_start(module_, "copy"); typename GNP::QPoint default_point; default_point.vec().Init(dim_); - config_.param.BootstrapMonochromatic(&default_point, n_points_); + param_.BootstrapMonochromatic(&default_point, n_points_); CacheArray::InitDistributedCacheMaster( DATA_POINTS_CHANNEL, n_block_points, @@ -425,7 +425,7 @@ void RpcMonochromaticDualTreeRunner::ReadData_() { // also, set up the results array typename GNP::QResult default_result; - default_result.Init(config_.param); + default_result.Init(param_); CacheArray::InitDistributedCacheMaster( Q_RESULTS_CHANNEL, n_block_points, config_.q_results_mb * MEGABYTE, default_result, @@ -442,14 +442,14 @@ void RpcMonochromaticDualTreeRunner::MakeTree_() { fprintf(stderr, "master: Building tree\n"); fx_timer_start(module_, "tree"); typename GNP::QNode data_example_node; - data_example_node.Init(dim_, config_.param); + data_example_node.Init(dim_, param_); CacheArray::InitDistributedCacheMaster( DATA_NODES_CHANNEL, n_block_nodes, config_.data_nodes_mb * MEGABYTE, data_example_node, &data_nodes_); KdTreeHybridBuilder - ::Build(data_module_, config_.param, 0, n_points_, + ::Build(data_module_, param_, 0, n_points_, &data_points_, &data_nodes_); fx_timer_stop(module_, "tree"); } @@ -459,15 +459,19 @@ void RpcMonochromaticDualTreeRunner::SetupConfig_() { // Set up and export the config object config_.n_threads = fx_param_int(module_, "n_threads", 1); config_.data_points_mb = fx_param_int(module_, "data_points_mb", 400); - config_.data_nodes_mb = fx_param_int(module_, "data_nodes_mb", 100); - config_.q_results_mb = fx_param_int(module_, "q_results_mb", 100); + config_.data_nodes_mb = fx_param_int(module_, "data_nodes_mb", 150); + config_.q_results_mb = fx_param_int(module_, "q_results_mb", 200); + + master_ = new Master(); + master_->config_backend.Init(&config_); + rpc::Register(CONFIG_CHANNEL, &master_->config_backend); } template void RpcMonochromaticDualTreeRunner::SetupMaster_() { - master_ = new Master(); - master_->config_backend.Init(&config_); - rpc::Register(CONFIG_CHANNEL, &master_->config_backend); + master_->param_backend.Init(¶m_); + rpc::Register(PARAM_CHANNEL, &master_->param_backend); + // Make a static work queue CentroidWorkQueue *actual_work_queue = @@ -522,12 +526,14 @@ void RpcMonochromaticDualTreeRunner::Doit( SetupConfig_(); fprintf(stderr, "nbr_utils(%d): reading in data, making tree, and distributing data\n", rpc::rank()); - config_.param.Init(fx_submodule(module_, gnp_name_, gnp_name_)); + param_.Init(fx_submodule(module_, gnp_name_, gnp_name_)); ReadData_(); MakeTree_(); SetupMaster_(); } else { + rpc::GetRemoteData(CONFIG_CHANNEL, MASTER_RANK, &config_); + data_points_.InitWorker(DATA_POINTS_CHANNEL, config_.data_points_mb * MEGABYTE, new CacheArrayBlockHandler); @@ -538,7 +544,7 @@ void RpcMonochromaticDualTreeRunner::Doit( config_.q_results_mb * MEGABYTE, new CacheArrayBlockHandler); - rpc::GetRemoteData(CONFIG_CHANNEL, MASTER_RANK, &config_); + rpc::GetRemoteData(PARAM_CHANNEL, MASTER_RANK, ¶m_); RemoteWorkQueue *remote_work_queue = new RemoteWorkQueue(); remote_work_queue->Init(WORK_CHANNEL, MASTER_RANK); @@ -563,7 +569,7 @@ void RpcMonochromaticDualTreeRunner::Doit( ThreadedDualTreeSolver solver; solver.Doit( fx_submodule(module_, "solver", "local"), - config_.n_threads, rpc::rank(), work_queue_, config_.param, + config_.n_threads, rpc::rank(), work_queue_, param_, &data_points_, &data_nodes_, &data_points_, &data_nodes_, &q_results_); @@ -576,7 +582,7 @@ void RpcMonochromaticDualTreeRunner::Doit( //rpc::Reduce(IOSTATS_RESULTS_CHANNEL, IoStatsReductor(), &q_results_.stats()); GlobalResultReductor global_result_reductor; - global_result_reductor.Init(&config_.param); + global_result_reductor.Init(¶m_); rpc::Reduce(GLOBAL_RESULT_CHANNEL, global_result_reductor, &solver.global_result()); @@ -585,7 +591,7 @@ void RpcMonochromaticDualTreeRunner::Doit( //q_results_.ReportStats(true, fx_submodule(module_, NULL, "gnp_results")); work_queue_->Report(fx_submodule(module_, NULL, "work_queue")); - solver.global_result().Report(config_.param, + solver.global_result().Report(param_, fx_submodule(module_, NULL, "global_result")); rpc::Done(); diff --git a/fastlib/u/garryb/nbr/rpc_sock.cc b/fastlib/u/garryb/nbr/rpc_sock.cc index 691aa686db..de06e16158 100644 --- a/fastlib/u/garryb/nbr/rpc_sock.cc +++ b/fastlib/u/garryb/nbr/rpc_sock.cc @@ -400,7 +400,7 @@ void RpcSockImpl::PollingLoop_() { Peer *peer = &peers_[peer_num]; peer->mutex.Lock(); if (!peer->connection.TryRead() && !errors_ok) { - FATAL("Unexpected end of file for peer %d", i); + FATAL("Unexpected end of file for peer %d", peer_num); } GatherReadyMessages_(peer, &work_items); peer->mutex.Unlock();