diff --git a/fastlib/u/garryb/nbr/build.py b/fastlib/u/garryb/nbr/build.py
index a71c90a302..50b2a55cff 100644
--- a/fastlib/u/garryb/nbr/build.py
+++ b/fastlib/u/garryb/nbr/build.py
@@ -10,6 +10,15 @@ librule(name = "nbr",
               "cache.h"],
    deplibs = ["fastlib:fastlib_int"])
 
+librule(name = "dc",
+   sources = ["distribcache.cc"],
+   headers = ["blockdev.h", "cachearray.h",
+              "dfs.h", "gnp.h", "kdtree.h", "nbr_utils.h",
+              "spbounds.h", "spnode.h", "work.h",
+              "rpc.h", "rpc_sock.h", "distribcache.h",
+              "cache.h"],
+   deplibs = ["fastlib:fastlib_int"])
+
 binrule(name = "rpc_sock_test",
    sources = ["rpc_sock_test.cc"],
    deplibs = [":nbr"])
diff --git a/fastlib/u/garryb/nbr/cachearray.h b/fastlib/u/garryb/nbr/cachearray.h
index 84af76fd1e..2d6c4a05c5 100644
--- a/fastlib/u/garryb/nbr/cachearray.h
+++ b/fastlib/u/garryb/nbr/cachearray.h
@@ -35,7 +35,7 @@ class CacheArrayBlockHandler : public BlockHandler {
   }
   
   void Deserialize(const ArrayList<char>& data) {
-    default_elem_->Copy(*data);
+    default_elem_.Copy(data);
   }
 
 //  void WriteHeader(BlockDevice *inner_device) {
@@ -147,6 +147,7 @@ class CacheArray {
   ArrayList<Metadata> metadatas_;
 
   BlockDevice::blockid_t *fifo_;
+  int fifo_index_;
 
   unsigned int n_elem_bytes_;
   index_t begin_;
@@ -156,12 +157,12 @@ class CacheArray {
   BlockDevice::blockid_t skip_blocks_;
   BlockDevice::mode_t mode_;
 
-  DistribtedCache *cache_;
+  DistributedCache *cache_;
 
  public:
   CacheArray() {}
   ~CacheArray() {
-    if (BlockDevice::need_write(mode_) {
+    if (BlockDevice::need_write(mode_)) {
       Flush();
     }
     mem::Free(fifo_);
@@ -178,19 +179,16 @@ class CacheArray {
     Init(other->cache_, mode_in, begin_index_in, end_index_in);
   }
 
-  /** Opens an existing SmallCache, a sub-range only (static use-case). */
-  void Init(SmallCache *cache_in, BlockDevice::mode_t mode_in,
+  /** Opens an existing DistributedCache, a sub-range only (static use-case). */
+  void Init(DistributedCache *cache_in, BlockDevice::mode_t mode_in,
       index_t begin_index_in, index_t end_index_in);
 
-#error need the static domain decomposition code
-#error need to register with the cache
-
   /**
-   * Opens an existing SmallCache, a sub-range only.
+   * Opens an existing DistributedCache, a sub-range only.
    *
    * Behavior is inferred via the mode.
    */
-  void Init(SmallCache *cache_in, BlockDevice::mode_t mode_in) {
+  void Init(DistributedCache *cache_in, BlockDevice::mode_t mode_in) {
     Init(cache_in, mode_in, 0, 0);
     Grow();
   }
@@ -234,7 +232,7 @@ class CacheArray {
     return n_block_elems_mask_;
   }
 
-  SmallCache *cache() const {
+  DistributedCache *cache() const {
     return cache_;
   }
 
@@ -251,13 +249,13 @@ class CacheArray {
 
   void StopRead(index_t element_id) {
     DEBUG_ONLY(BoundsCheck_(element_id));
-    ReleaseElement_(element_id);
+    ReleaseElement(element_id);
   }
 
   void StopWrite(index_t element_id) {
     DEBUG_ONLY(BoundsCheck_(element_id));
     DEBUG_ASSERT(BlockDevice::can_write(mode_));
-    ReleaseElement_(element_id);
+    ReleaseElement(element_id);
   }
 
   void Swap(index_t index_a, index_t index_b) {
@@ -269,8 +267,8 @@ class CacheArray {
     mem::Swap(a, b, n_elem_bytes_);
     ot::PointerRelocate<Element>(a, b);
     ot::PointerRelocate<Element>(b, a);
-    ReleaseElement_(index_a);
-    ReleaseElement_(index_b);
+    ReleaseElement(index_a);
+    ReleaseElement(index_b);
   }
 
   void Copy(index_t index_src, index_t index_dest) {
@@ -281,8 +279,8 @@ class CacheArray {
     char *dest = reinterpret_cast<char*>(StartWrite(index_dest));
     mem::Copy(dest, src, n_elem_bytes_);
     ot::PointerRelocate<Element>(src, dest);
-    ReleaseElement_(index_src);
-    ReleaseElement_(index_dest);
+    ReleaseElement(index_src);
+    ReleaseElement(index_dest);
   }
 
   index_t AllocD(int owner, index_t count) {
@@ -292,7 +290,7 @@ class CacheArray {
       BlockDevice::blockid_t blocks_to_alloc =
           (count + n_block_elems_mask()) >> n_block_elems_log();
       BlockDevice::blockid_t blockid = cache_->AllocBlocks(
-          blocks_to_alloc, peer);
+          blocks_to_alloc, owner);
 
       metadatas_.Resize(blockid + blocks_to_alloc - skip_blocks_);
       adjusted_metadatas_ = metadatas_.begin() - skip_blocks_;
@@ -372,7 +370,7 @@ class CacheArray {
   /* these are public so various classes can use them efficiently */
 
   void ReleaseBlock(BlockDevice::blockid_t blockid) {
-    DEBUG_ONLY(--adjusted_metadatas_[fakeid].lock_count);
+    DEBUG_ONLY(--adjusted_metadatas_[blockid].lock_count);
   }
 
   index_t BlockElement(BlockDevice::blockid_t blockid) {
@@ -383,19 +381,19 @@ class CacheArray {
     return element_id >> n_block_elems_log();
   }
   
-  BlockDevice::offset_t Offset(index element_id) {
+  BlockDevice::offset_t Offset(index_t element_id) {
     return (element_id & n_block_elems_mask()) * n_elem_bytes_;
   }
 
   void ReleaseElement(index_t element_id) {
     DEBUG_ONLY(BoundsCheck_(element_id));
-    ReleaseBlock_(Blockid_(element_id));
+    ReleaseBlock_(Blockid(element_id));
   }
 };
 
 template<typename TElement>
 void CacheArray<TElement>::Init(
-    SmallCache *cache_in, BlockDevice::mode_t mode_in,
+    DistributedCache *cache_in, BlockDevice::mode_t mode_in,
     index_t begin_index_in, index_t end_index_in) {
   CacheArrayBlockHandler<TElement>* handler =
       static_cast<CacheArrayBlockHandler<TElement>*>(
@@ -408,6 +406,7 @@ void CacheArray<TElement>::Init(
   mode_ = mode_in;
   n_elem_bytes_ = handler->n_elem_bytes();
   fifo_ = mem::Alloc<BlockDevice::blockid_t>(FIFO_SIZE);
+  fifo_index_ = 0;
 
   unsigned n_block_elems_calc = cache_->n_block_bytes() / n_elem_bytes_;
   // Cache size must be a power of 2.
@@ -418,8 +417,9 @@ void CacheArray<TElement>::Init(
       "Block size must be a multiple of element size.");
 
   if (!BlockDevice::is_dynamic(mode_)) {
-    FOO
-    cache_->AddPartialDirtyRange();
+    cache_->AddPartialDirtyRange(
+        Blockid(begin_), Offset(end_),
+        Blockid(end_), Offset(end_));
   }
 
   metadatas_.Init(((end_ + n_block_elems_mask()) >> n_block_elems_log())
@@ -459,7 +459,7 @@ typename CacheArray<TElement>::Element* CacheArray<TElement>::HandleCacheMiss_(
     if (unlikely(victim < 0)) {
       break;
     }
-    victim_metadata = adjusted_metadats_ + victim;
+    victim_metadata = adjusted_metadatas_ + victim;
     if (unlikely(victim_metadata->lock_count != 0)) {
       continue;
     }
@@ -473,7 +473,7 @@ typename CacheArray<TElement>::Element* CacheArray<TElement>::HandleCacheMiss_(
     break;
   }
 
-  BlockDevice::blockid_t blockid = Blockid_(element_id);
+  BlockDevice::blockid_t blockid = Blockid(element_id);
   Metadata *metadata = adjusted_metadatas_ + blockid;
 
   if (BlockDevice::can_write(mode_)) {
@@ -673,13 +673,14 @@ class CacheWriteIter
 
 //------------------------------------------------------------------------
 
+#error what *is* a TempCache now?
 /**
  * Specialed cache-array to simplify the creation/cleanup process.
  */
 template<typename TElement>
 class TempCacheArray : public CacheArray<TElement> {
  private:
-  SmallCache underlying_cache_;
+  DistributedCache underlying_cache_;
   NullBlockDevice null_device_;
 
  public:
@@ -697,7 +698,7 @@ class TempCacheArray : public CacheArray<TElement> {
     handler->Init(default_obj);
 
     null_device_.Init(0, n_block_elems_in * handler->n_elem_bytes());
-    underlying_cache_.Init(&null_device_, handler, BlockDevice::M_TEMP);
+    underlying_cache_.InitMaster(&null_device_, handler, BlockDevice::M_TEMP);
 
     CacheArray<TElement>::Init(&underlying_cache_, BlockDevice::M_TEMP, 0, 0);
 
diff --git a/fastlib/u/garryb/nbr/dfs.h b/fastlib/u/garryb/nbr/dfs.h
index 37344b9f5b..603c7345c4 100644
--- a/fastlib/u/garryb/nbr/dfs.h
+++ b/fastlib/u/garryb/nbr/dfs.h
@@ -57,11 +57,11 @@ class DualTreeDepthFirst {
       datanode *datanode_in,
       const typename GNP::Param& param_in,
       index_t q_root_index,
-      SmallCache *q_points,
-      SmallCache *q_nodes,
-      SmallCache *r_points,
-      SmallCache *r_nodes,
-      SmallCache *q_results);
+      DistributedCache *q_points,
+      DistributedCache *q_nodes,
+      DistributedCache *r_points,
+      DistributedCache *r_nodes,
+      DistributedCache *q_results);
 
   /**
    * Gets the global result after computation.
@@ -102,11 +102,11 @@ void DualTreeDepthFirst<GNP>::Doit(
     struct datanode *datanode_in,
     const typename GNP::Param& param_in,
     index_t q_root_index,
-    SmallCache *q_points,
-    SmallCache *q_nodes,
-    SmallCache *r_points,
-    SmallCache *r_nodes,
-    SmallCache *q_results) {
+    DistributedCache *q_points,
+    DistributedCache *q_nodes,
+    DistributedCache *r_points,
+    DistributedCache *r_nodes,
+    DistributedCache *q_results) {
   param_.Copy(param_in);
 
   q_nodes_.Init(q_nodes, BlockDevice::M_READ);
diff --git a/fastlib/u/garryb/nbr/distribcache.cc b/fastlib/u/garryb/nbr/distribcache.cc
index 2f9f3a919c..3b7b0a8f27 100644
--- a/fastlib/u/garryb/nbr/distribcache.cc
+++ b/fastlib/u/garryb/nbr/distribcache.cc
@@ -66,8 +66,10 @@ void DistributedCache::InitFile_(const char *filename) {
   } else {
     filename_str.Copy(filename);
   }
-  overflow_device_ = new DiskBlockDevice(filename_str.c_str(),
+  DiskBlockDevice *db = new DiskBlockDevice();
+  db->Init(filename_str.c_str(),
       BlockDevice::M_TEMP, n_block_bytes_);
+  overflow_device_ = db;
 }
 
 void DistributedCache::InitChannel_(int channel_num_in) {
@@ -75,8 +77,7 @@ void DistributedCache::InitChannel_(int channel_num_in) {
   channel_.Init(this);
 }
 
-void DistributedCache::InitCommon_(
-    BlockDevice::offset_t n_block_bytes) {
+void DistributedCache::InitCommon_() {
   blocks_.Init();
   handler_ = NULL;
 
@@ -85,12 +86,12 @@ void DistributedCache::InitCommon_(
   overflow_next_.default_value() = BIG_BAD_NUMBER;
   overflow_device_ = NULL;
 
-  my_rank_ = rpc::Rank();
+  write_ranges_.Init();
+
+  my_rank_ = rpc::rank();
 }
 
 void DistributedCache::InitCache_(size_t total_ram) {
-  ASSOC = 8;
-  LOG_ASSOC = 3;
   // give enough cache sets, but rounded up
   n_sets_ = (total_ram + ASSOC*n_block_bytes_ - 1) / (ASSOC*n_block_bytes_);
   slots_.Init(n_sets_ << LOG_ASSOC);
@@ -107,7 +108,7 @@ void DistributedCache::HandleStatusInformation_(
 
   DEBUG_ASSERT(statuses.size() >= n_blocks_);
 
-  if (n_blocks_ != statuses.size()) [
+  if (n_blocks_ != statuses.size()) {
     n_blocks_ = statuses.size();
     blocks_.Resize(n_blocks_);
   }
@@ -140,31 +141,30 @@ void DistributedCache::HandleStatusInformation_(
   mutex_.Unlock();
 }
 
-#error need GiveOwnership method that marks entire page FULLY_DIRTY
-
 void DistributedCache::ComputeStatusInformation_(
-    ArrayList<BlockStatus> *statuses)
+    ArrayList<BlockStatus> *statuses) {
   mutex_.Lock();
-  statuses->Init(cache_->n_blocks());
+  statuses->Init(n_blocks());
   for (index_t i = 0; i < statuses->size(); i++) {
-    BlockStatus *status = &statuses[i];
-    BlockMetadata *owner = &cache_->blocks_[i];
+    BlockStatus *status = &(*statuses)[i];
+    const BlockMetadata *block = &blocks_[i];
     if (block->is_owner()) {
-      status->owner = rank_;
-    } else if (self_only) {
+      status->owner = my_rank_;
+      status->is_new = block->is_new();
+    } else {
       status->owner = -1;
+      status->is_new = NOT_DIRTY_NEW;
     }
-    status->is_new = block->is_new();
   }
   mutex_.Unlock();
 }
 
-void DistributedCache::BestEffortFlush(double portion) {
+void DistributedCache::BestEffortWriteback(double portion) {
   mutex_.Lock();
-  Slot *slot = slots_.ptr();
+  Slot *slot = slots_.begin();
   index_t i = slots_.size();
   int start_col = int(nearbyint(ASSOC * portion));
-  BlockMetadata *blocks = blocks_.ptr();
+  BlockMetadata *blocks = blocks_.begin();
 
   // Might want to software-pipeline this loop, because of the really nasty
   // indirect load going on.
@@ -174,8 +174,8 @@ void DistributedCache::BestEffortFlush(double portion) {
       blockid_t blockid = slot[j].blockid;
       if (blockid >= 0) {
         BlockMetadata *block = &blocks[blockid];
+        DEBUG_ASSERT_MSG(!block->is_busy(), "Why is a busy block in LRU?");
         if (unlikely(block->is_dirty()) && unlikely(!block->is_owner())) {
-          DEBUG_ASSERT_MSG(block->locks == 0, "Why is a locked block in LRU?");
           WritebackDirtyRemote_(blockid);
         }
       }
@@ -188,9 +188,9 @@ void DistributedCache::BestEffortFlush(double portion) {
 void DistributedCache::StartSync() {
   // We'll assume everything we have locally is no longer valid.
   mutex_.Lock();
-  Slot *slot = slots_.ptr();
+  Slot *slot = slots_.begin();
   index_t i = slots_.size();
-  BlockMetadata *blocks = blocks_.ptr();
+  BlockMetadata *blocks = blocks_.begin();
 
   // Might want to software-pipeline this loop, because of the really nasty
   // indirect load going on.
@@ -208,7 +208,7 @@ void DistributedCache::StartSync() {
   mutex_.Unlock();
 
   // TODO: Make absolutely certain nobody is currently accessesing the cache
-  write_ranges_.Clear();
+  write_ranges_.Reset();
 
   channel_.StartSyncFlushDone();
 }
@@ -226,18 +226,19 @@ void DistributedCache::Read(blockid_t blockid,
 
   // TODO: consider read-through
   mem::Copy(buf, src + begin, n_bytes);
-  block_handler_->BlockFreeze(blockid, begin, n_bytes, src + begin, buf);
+  handler_->BlockFreeze(blockid, begin, n_bytes, src + begin, buf);
 
   StopRead(blockid);
 }
 
 void DistributedCache::Write(blockid_t blockid,
     offset_t begin, offset_t end, const char *buf) {
-  char *dest = StartWrite(blockid);
+  // hmm, i have to be the owner of the block
+  char *dest = StartWrite(blockid, true);
   size_t n_bytes = end - begin;
 
   mem::CopyBytes(dest + begin, buf, n_bytes);
-  block_handler_->BlockThaw(blockid, begin, n_bytes, dest + begin);
+  handler_->BlockThaw(blockid, begin, n_bytes, dest + begin);
 
   StopWrite(blockid);
 }
@@ -279,8 +280,8 @@ BlockDevice::blockid_t DistributedCache::AllocBlocks(
 
     mutex_.Unlock();
   } else {
-    AllocTransacation t;
-    blockid = t.Doit(channel_, MASTER_RANK, n_blocks_to_alloc, owner);
+    AllocTransaction t;
+    blockid = t.Doit(channel_num_, MASTER_RANK, n_blocks_to_alloc, owner);
     mutex_.Lock();
     n_blocks_ = blockid + n_blocks_to_alloc;
     blocks_.GrowTo(n_blocks_);
@@ -295,8 +296,9 @@ void DistributedCache::GiveOwnership(blockid_t my_blockid, int new_owner) {
     // mark whole block as dirty and change its owner.
     StartWrite(my_blockid, false);
     mutex_.Lock();
-    BlockMetadata *block = &blocks_[blockid];
-    DEBUG_ASSERT(block->is_owner());
+    BlockMetadata *block = &blocks_[my_blockid];
+    DEBUG_ASSERT_MSG(block->is_owner(),
+        "Can only give ownership if I'm the owner");
     if (block->local_blockid() != SELF_OWNER_UNALLOCATED) {
       // this block has a location on disk -- since it's not ours anymore,
       // recycle its allocated disk space.
@@ -311,8 +313,7 @@ void DistributedCache::GiveOwnership(blockid_t my_blockid, int new_owner) {
 
 //----
 
-char *DistributedCache::StartWrite(BlockDevice::blockid blockid,
-    bool partial) {
+char *DistributedCache::StartWrite(blockid_t blockid, bool is_partial) {
   mutex_.Lock();
   BlockMetadata *block = &blocks_[blockid];
   if (likely(block->locks)) {
@@ -320,7 +321,7 @@ char *DistributedCache::StartWrite(BlockDevice::blockid blockid,
   } else {
     DecacheBlock_(blockid);
   }
-  if (partial) {
+  if (is_partial) {
     block->status &= PARTIALLY_DIRTY;
   } else {
     block->status = FULLY_DIRTY;
@@ -329,7 +330,7 @@ char *DistributedCache::StartWrite(BlockDevice::blockid blockid,
   return block->data;
 }
 
-char *DistributedCache::StartRead(BlockDevice::blockid blockid) {
+char *DistributedCache::StartRead(BlockDevice::blockid_t blockid) {
   mutex_.Lock();
   BlockMetadata *block = &blocks_[blockid];
   if (likely(block->locks)) {
@@ -345,6 +346,7 @@ char *DistributedCache::StartRead(BlockDevice::blockid blockid) {
 
 void DistributedCache::StopRead(BlockDevice::blockid_t blockid) {
   mutex_.Lock();
+  BlockMetadata *block = &blocks_[blockid];
   if (unlikely(--block->locks == 0)) {
     EncacheBlock_(blockid);
   }
@@ -353,6 +355,7 @@ void DistributedCache::StopRead(BlockDevice::blockid_t blockid) {
 
 void DistributedCache::StopWrite(BlockDevice::blockid_t blockid) {
   mutex_.Lock();
+  BlockMetadata *block = &blocks_[blockid];
   if (unlikely(--block->locks == 0)) {
     EncacheBlock_(blockid);
   }
@@ -383,7 +386,6 @@ void DistributedCache::DecacheBlock_(BlockDevice::blockid_t blockid) {
 }
 
 void DistributedCache::EncacheBlock_(BlockDevice::blockid_t blockid) {
-  BlockMetadata *block = &blocks_[blockid];
   index_t slot = (unsigned(blockid) % unsigned(n_sets_)) << LOG_ASSOC;
   Slot *base_slot = &slots_[slot];
   int i;
@@ -418,8 +420,8 @@ void DistributedCache::HandleMiss_(BlockDevice::blockid_t blockid) {
   if (block->is_new()) {
     DEBUG_ASSERT_MSG(block->status == NOT_DIRTY_NEW,
         "Block should be NOT_DIRTY_NEW, because that's what is_new() means");
-    block_handler_->BlockInitFrozen(blockid, 0, n_block_bytes_, block->data);
-    block_handler_->BlockThaw(blockid, 0, n_block_bytes_, block->data);
+    handler_->BlockInitFrozen(blockid, 0, n_block_bytes_, block->data);
+    handler_->BlockThaw(blockid, 0, n_block_bytes_, block->data);
   } else if (block->is_owner()) {
     DEBUG_ASSERT(block->status == NOT_DIRTY_OLD);
     HandleLocalMiss_(blockid);
@@ -448,7 +450,7 @@ void DistributedCache::HandleRemoteMiss_(BlockDevice::blockid_t blockid) {
       blockid, 0, n_block_bytes_, block->data);
 }
 
-char *DistributedCache::Purge_(BlockDevice::blockid blockid) {
+void DistributedCache::Purge_(blockid_t blockid) {
   BlockMetadata *block = &blocks_[blockid];
 
   DEBUG_ASSERT(block->is_in_core());
@@ -458,8 +460,7 @@ char *DistributedCache::Purge_(BlockDevice::blockid blockid) {
 
   if (block->is_dirty()) {
     if (block->is_owner()) {
-      PurgeDirtyLocal_(blockid);
-      return;
+      WritebackDirtyLocal_(blockid);
     } else {
       WritebackDirtyRemote_(blockid);
     }
@@ -469,7 +470,7 @@ char *DistributedCache::Purge_(BlockDevice::blockid blockid) {
   block->data = NULL;
 }
 
-void DistributedCache::PurgeDirtyLocal_(BlockDevice::blockid_t blockid) {
+void DistributedCache::WritebackDirtyLocal_(BlockDevice::blockid_t blockid) {
   BlockMetadata *block = &blocks_[blockid];
 
   DEBUG_ASSERT(block->is_owner());
@@ -488,11 +489,8 @@ void DistributedCache::PurgeDirtyLocal_(BlockDevice::blockid_t blockid) {
   }
 
   fprintf(stderr, "DISK: Writing block %d to %d\n", blockid, local_blockid);
-  local_device_.Write(local_blockid, 0, n_block_bytes_, block->data);
+  overflow_device_->Write(local_blockid, 0, n_block_bytes_, block->data);
   block->status = NOT_DIRTY_OLD;
-
-  mem::Free(block->data);
-  block->data = NULL;
 }
 
 void DistributedCache::WritebackDirtyRemote_(BlockDevice::blockid_t blockid) {
@@ -504,7 +502,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,
+    write_transaction.Doit(channel_num_, block->value, blockid,
         0, n_block_bytes_, block->data);
   } else {
     DEBUG_ASSERT(block->status == PARTIALLY_DIRTY);
@@ -513,21 +511,22 @@ void DistributedCache::WritebackDirtyRemote_(BlockDevice::blockid_t blockid) {
     #ifdef DEBUG
     bool anything_done = false;
     #endif
-    for (index_t i = 0; i < ranges_.size(); i++) {
-      const Range *range = &ranges_[i];
-      if (block >= range->begin_block || block <= range->last_block) {
+    for (index_t i = 0; i < write_ranges_.size(); i++) {
+      Position begin = write_ranges_[i].begin;
+      Position end = write_ranges_[i].end;
+      if (blockid >= begin.block || blockid <= end.block) {
         // We found a partial range that overlaps.  Write it.
-        offset_t begin = 0;
-        offset_t end = n_block_bytes_;
-        if (block == range->begin_block) {
-          begin = range->begin;
+        offset_t begin_offset = 0;
+        offset_t end_offset = n_block_bytes_;
+        if (blockid == begin.block) {
+          begin_offset = begin.offset;
         }
-        if (block == range->last_block) {
-          end = range->end;
+        if (blockid == end.block) {
+          end_offset = end.offset;
         }
         WriteTransaction write_transaction;
-        write_transaction.Doit(channel_num, block->value, blockid,
-            begin, end, block->data);
+        write_transaction.Doit(channel_num_, block->value, blockid,
+            begin_offset, end_offset, block->data);
         DEBUG_ONLY(anything_done = true);
       }
     }
@@ -535,8 +534,10 @@ void DistributedCache::WritebackDirtyRemote_(BlockDevice::blockid_t blockid) {
         "A block marked partially dirty has no overlapping write ranges.");
   }
   block->status = NOT_DIRTY_OLD;
+}
 
-void AddPartialDirtyRange(blockid_t begin_block, offset_t begin_offset,
+void DistributedCache::AddPartialDirtyRange(
+    blockid_t begin_block, offset_t begin_offset,
     blockid_t last_block, offset_t end_offset) {
   Position begin;
   Position end;
@@ -544,7 +545,7 @@ void AddPartialDirtyRange(blockid_t begin_block, offset_t begin_offset,
   begin.offset = begin_offset;
   end.block = last_block;
   end.offset = end_offset;
-  ranges_.Union(begin, end);
+  write_ranges_.Union(begin, end);
 }
 
 //-------------------------------------------------------------------------
@@ -578,6 +579,32 @@ void DistributedCache::ReadTransaction::HandleMessage(Message *message) {
 
 //-------------------------------------------------------------------------
 
+BlockDevice::blockid_t DistributedCache::AllocTransaction::Doit(
+    int channel_num, int peer, blockid_t n_blocks_to_alloc, int owner) {
+  Transaction::Init(channel_num);
+  Message *message = CreateMessage(peer, sizeof(Request));
+  Request *request = message->data_as<Request>();
+  request->type = Request::ALLOC;
+  request->blockid = n_blocks_to_alloc;
+  request->begin = 0;
+  request->end = 0;
+  request->rank = owner;
+  response = NULL;
+  Send(message);
+  cond.Wait();
+  blockid_t retval = *message->data_as<blockid_t>();
+  delete response;
+  return retval;
+}
+
+void DistributedCache::AllocTransaction::HandleMessage(Message *message) {
+  response = message;
+  cond.Done();
+  Done();
+}
+
+//-------------------------------------------------------------------------
+
 void DistributedCache::WriteTransaction::Doit(
     int channel_num, int peer, BlockDevice::blockid_t blockid,
     BlockDevice::offset_t begin, BlockDevice::offset_t end,
@@ -637,7 +664,7 @@ void DistributedCache::ResponseTransaction::HandleMessage(
   switch (request->type) {
     case Request::CONFIG: {
       ConfigResponse config_response;
-      cache_->handler_->Serialize(&config_response.data);
+      cache_->handler_->Serialize(&config_response.block_handler_data);
       config_response.n_block_bytes = cache_->n_block_bytes_;
       Message *response = CreateMessage(message->peer(),
           ot::PointerFrozenSize(config_response));
@@ -661,8 +688,8 @@ void DistributedCache::ResponseTransaction::HandleMessage(
     case Request::ALLOC: {
       DEBUG_ASSERT(cache_->my_rank_ == MASTER_RANK);
       Message *response = CreateMessage(message->peer(), sizeof(blockid_t));
-      *reinterpret_cast<blockid_t>(message->data()) =
-          cache_->AllocBlocks(blockid, request->rank);
+      *message->data_as<blockid_t>() =
+          cache_->AllocBlocks(request->blockid, request->rank);
       Send(response);
     }
     break;
@@ -678,11 +705,10 @@ void DistributedCache::ResponseTransaction::HandleMessage(
 
 //-------------------------------------------------------------------------
 
-void DistributedCache::SyncTransaction::Init(Cache *cache) {
+void DistributedCache::SyncTransaction::Init(DistributedCache *cache_in) {
   cache_ = cache_in;
   state_ = CHILDREN_FLUSHING;
   n_ = 0;
-  sync_response_.Init();
 }
 
 void DistributedCache::SyncTransaction::HandleMessage(Message *message) {
@@ -699,9 +725,9 @@ void DistributedCache::SyncTransaction::HandleMessage(Message *message) {
       break;
     case OTHERS_ACCUMULATING: {
       DEBUG_ASSERT(message->peer() == rpc::parent());
+      char *data = message->data_as<Request>()->data_as<char>();
       ArrayList<BlockStatus> *in_statuses =
-          ot::PointerThaw< ArrayList<BlockStatus> >(
-              message->data_as<Request>()->data_as<char>());
+          ot::PointerThaw< ArrayList<BlockStatus> >(data);
       ParentAccumulated_(*in_statuses);
     }
     break;
diff --git a/fastlib/u/garryb/nbr/distribcache.h b/fastlib/u/garryb/nbr/distribcache.h
index 1e06b0972d..d3a4443477 100644
--- a/fastlib/u/garryb/nbr/distribcache.h
+++ b/fastlib/u/garryb/nbr/distribcache.h
@@ -71,7 +71,7 @@ class DistributedCache : public BlockDevice {
     }
 
     static size_t size(size_t data_size) {
-      return sizeof(Request) + data_size - sizeof(long_data);
+      return sizeof(Request) + data_size - sizeof(long);
     }
   };
 
@@ -98,19 +98,19 @@ class DistributedCache : public BlockDevice {
   };
 
   /** How to query information about the overall state */
-  struct ConfigTransaction {
+  struct ConfigTransaction : public Transaction {
    public:
     DoneCondition cond;
     Message *response;
    public:
     ~ConfigTransaction() { delete response; }
 
-    void Doit(int channel_num, int peer);
+    ConfigResponse *Doit(int channel_num, int peer);
     void HandleMessage(Message *message);
   };
 
   /** How to query information about the overall state */
-  struct AllocTransaction {
+  struct AllocTransaction : public Transaction {
    public:
     DoneCondition cond;
     Message *response;
@@ -123,7 +123,7 @@ class DistributedCache : public BlockDevice {
   };
 
   /** How to initiate a read messages */
-  struct ReadTransaction {
+  struct ReadTransaction : public Transaction {
    public:
     DoneCondition cond;
     Message *response;
@@ -137,7 +137,7 @@ class DistributedCache : public BlockDevice {
   };
 
   /** How to initiate a write message */
-  struct WriteTransaction {
+  struct WriteTransaction : public Transaction  {
    public:
     void Doit(int channel_num, int peer, BlockDevice::blockid_t blockid,
         BlockDevice::offset_t begin, BlockDevice::offset_t end,
@@ -160,7 +160,7 @@ class DistributedCache : public BlockDevice {
    * along with a reduction and scatter so each machine knows the updated
    * block owner information.
    */
-  struct SyncTransaction {
+  struct SyncTransaction : public Transaction  {
    private:
     enum State {
       /** My children and I are flushing data. */
@@ -332,14 +332,14 @@ class DistributedCache : public BlockDevice {
       *this = other;
     }
     
-    bool operator < (const Position& other) {
+    bool operator < (const Position& other) const {
       if (unlikely(block == other.block)) {
         return offset < other.offset;
       } else {
         return block < other.block;
       }
     }
-    bool operator == (const Position& other) {
+    bool operator == (const Position& other) const {
       return block == other.block && offset == other.offset;
     }
     DEFINE_ALL_COMPARATORS(Position);
@@ -351,7 +351,7 @@ class DistributedCache : public BlockDevice {
   BlockHandler *handler_;
 
   /* ranges that apply for partial writes */
-  ArrayList<Range> write_ranges_;
+  RangeSet<Position> write_ranges_;
 
   /* local device */
   BlockDevice *overflow_device_;
@@ -360,17 +360,20 @@ class DistributedCache : public BlockDevice {
 
   /* remote stuff */
   int channel_num_;
-  Channel channel_;
+  CacheChannel channel_;
 
   int my_rank_;
   
   /* cache stuff */
   ArrayList<Slot> slots_;
   unsigned n_sets_;
-  int assoc_;
-  int log_assoc_;
+
+  Mutex mutex_;
 
  public:
+  DistributedCache() {}
+  virtual ~DistributedCache();
+
   /**
    * Initializes me as the master.
    *
@@ -393,8 +396,12 @@ class DistributedCache : public BlockDevice {
    * It may be possible for a performance benefit from calling this
    * periodically -- maybe mostly due to the ability to avoid lots of
    * extra reads.
+   *
+   * @param portion the portion (out of 1.0) to attempt flushing, in case
+   *        a "soft flush" is desired -- i.e. a flush of 0.5 will flush out
+   *        only old blocks
    */
-  void BestEffortFlush();
+  void BestEffortWriteback(double portion = 1.0);
   /** Starts syncing. */
   void StartSync();
   /** Ensures that the sync point has passed before returning. */
@@ -429,10 +436,7 @@ class DistributedCache : public BlockDevice {
 
   /* our bread-and-butter cache methods, called by the FIFO */
   /** Start a write access to the whole page. */
-  char *StartWrite(blockid_t blockid);
-  /** Start a write access to part of a page. */
-  char *StartWrite(blockid_t blockid,
-      offset_t begin, offset_t end);
+  char *StartWrite(blockid_t blockid, bool is_partial);
   /** Start a read to a page. */
   char *StartRead(blockid_t blockid);
   /** End a read access. */
@@ -461,10 +465,14 @@ class DistributedCache : public BlockDevice {
   }
   /** Allocates blocks, but assign ownership to a specified machine. */
   blockid_t AllocBlocks(blockid_t n_blocks_to_alloc, int owner);
+  /** Gets the underlying block handler. */
+  BlockHandler *block_handler() const {
+    return handler_;
+  }
 
  private:
   void InitChannel_(int channel_num_in);
-  void InitCommon_(offset_t n_block_bytes_);
+  void InitCommon_();
   void InitCache_(size_t total_ram);
   void InitFile_(const char *fname);
   /**