hi

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<char>(), buffer, end - begin);
+  mem::CopyBytes(request->data_as<char>(), buffer, end - begin);
   handler->BlockFreeze(blockid, begin, end, buffer, request->data_as<char>());
   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<Request*>(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();

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;

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> config_backend;
+    DataGetterBackend<typename GNP::Param> 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<GNP, Solver>::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<typename GNP::QPoint>::InitDistributedCacheMaster(
       DATA_POINTS_CHANNEL, n_block_points,
@@ -425,7 +425,7 @@ void RpcMonochromaticDualTreeRunner<GNP, Solver>::ReadData_() {
 
   // also, set up the results array
   typename GNP::QResult default_result;
-  default_result.Init(config_.param);
+  default_result.Init(param_);
   CacheArray<typename GNP::QResult>::InitDistributedCacheMaster(
       Q_RESULTS_CHANNEL, n_block_points,
       config_.q_results_mb * MEGABYTE, default_result,
@@ -442,14 +442,14 @@ void RpcMonochromaticDualTreeRunner<GNP, Solver>::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<typename GNP::QNode>::InitDistributedCacheMaster(
       DATA_NODES_CHANNEL, n_block_nodes,
       config_.data_nodes_mb * MEGABYTE, data_example_node,
       &data_nodes_);
   KdTreeHybridBuilder
       <typename GNP::QPoint, typename GNP::QNode, typename GNP::Param>
-      ::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<GNP, Solver>::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<typename GNP, typename Solver>
 void RpcMonochromaticDualTreeRunner<GNP, Solver>::SetupMaster_() {
-  master_ = new Master();
-  master_->config_backend.Init(&config_);
-  rpc::Register(CONFIG_CHANNEL, &master_->config_backend);
+  master_->param_backend.Init(&param_);
+  rpc::Register(PARAM_CHANNEL, &master_->param_backend);
+
 
   // Make a static work queue
   CentroidWorkQueue<typename GNP::QNode> *actual_work_queue =
@@ -522,12 +526,14 @@ void RpcMonochromaticDualTreeRunner<GNP, Solver>::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<typename GNP::QPoint>);
@@ -538,7 +544,7 @@ void RpcMonochromaticDualTreeRunner<GNP, Solver>::Doit(
         config_.q_results_mb * MEGABYTE,
         new CacheArrayBlockHandler<typename GNP::QResult>);
 
-    rpc::GetRemoteData(CONFIG_CHANNEL, MASTER_RANK, &config_);
+    rpc::GetRemoteData(PARAM_CHANNEL, MASTER_RANK, &param_);
 
     RemoteWorkQueue *remote_work_queue = new RemoteWorkQueue();
     remote_work_queue->Init(WORK_CHANNEL, MASTER_RANK);
@@ -563,7 +569,7 @@ void RpcMonochromaticDualTreeRunner<GNP, Solver>::Doit(
   ThreadedDualTreeSolver<GNP, Solver> 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<GNP, Solver>::Doit(
   //rpc::Reduce(IOSTATS_RESULTS_CHANNEL, IoStatsReductor(), &q_results_.stats());
 
   GlobalResultReductor<GNP> global_result_reductor;
-  global_result_reductor.Init(&config_.param);
+  global_result_reductor.Init(&param_);
   rpc::Reduce(GLOBAL_RESULT_CHANNEL,
       global_result_reductor, &solver.global_result());
 
@@ -585,7 +591,7 @@ void RpcMonochromaticDualTreeRunner<GNP, Solver>::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();

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();