hi

2007-07-27 05:24:37 +00:00
parent 9b593b345c
commit edfae28a8b
5 changed files with 167 additions and 123 deletions
@@ -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"])
@@ -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);

@@ -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);
@@ -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;
@@ -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);
  /**