Modified Files:

 	client/Buffercache.cc client/Buffercache.h

Support for implementation of inode_flush and flush. UNTESTED

git-svn-id: https://ceph.svn.sf.net/svnroot/ceph@360 29311d96-e01e-0410-9327-a35deaab8ce9

diff --git a/ceph/client/Buffercache.cc b/ceph/client/Buffercache.cc
index 21386874b94c0996a4365476e9819e3ac3f8e033..ed6572add66ef3edd4c18d0990269e36cf1b3f81 100644 (file)
--- a/ceph/client/Buffercache.cc
+++ b/ceph/client/Buffercache.cc
@@ -46,12 +46,10 @@ void Filecache::map_existing(size_t len,
   if (next_off < off + len) {
     holes[next_off] = (size_t) (off + len - next_off);
   }
-  // FIXME: consolidate holes
 }
 
-list<Bufferhead*> Filecache::simplify()
+void Filecache::simplify(list<Bufferhead*>& removed)
 {
-  list<Bufferhead*> removed;
   map<off_t, Bufferhead*>::iterator start, next;
   start = buffer_map.begin();
   while (start != buffer_map.end()) {
@@ -61,7 +59,7 @@ list<Bufferhead*> Filecache::simplify()
                  (*start)->second->state == (*next)->second->state &&
                  (*start)->second->offset + (*start)->second->len == (*next)->second->offset) {
                (*start)->second->claim_append((*next)->second);
-               buffer_map.erase((*next)->first);
+               buffer_map.erase((*next)->first); (*next)->second->put();
                removed.push_back((*next)->second);
                next++;
          } else {
@@ -70,7 +68,6 @@ list<Bufferhead*> Filecache::simplify()
        }
        start = next;
   }
-  return removed;
 }
 
 void Filecache::copy_out(size_t size, off_t offset, char *dst) 
@@ -111,8 +108,8 @@ void Buffercache::insert(Bufferhead *bh) {
     bcache_map[bh->ino] = fc;
   }
   if (fc->buffermap.count(bh->offset)) assert(0); // fail loudly if offset already exists!
-  fc->buffer_map[bh->offset] = bh;
-  lru.lru_insert_top(bh);
+  fc->buffer_map[bh->offset] = bh; bh->get();
+  lru.lru_insert_top(bh); bh->get();
   clean_size += bh->len;
 }
 
@@ -165,12 +162,20 @@ size_t Buffercache::touch_continuous(map<off_t, Bufferhead*>& hits, size_t size,
 void Buffercache::simplify(inodeno_t ino)
 {
   Filecache *fc = bcache_map[ino];
-  list<Bufferhead*> removed = fc->simplify();
+  list<Bufferhead*> removed;
+  fc->simplify(&removed);
   for (list<Bufferhead*>::iterator it = removed.begin();
           it != removed.end();
           it++) {
-       lru.lru_remove(*it);
-       delete *it;
+       lru.lru_remove(*it); (*it)->put();
+    if (dirty_buffers.count(*it)) {
+      dirty_buffers.erase(*it);
+      (*it)->put();
+    }
+    if (bcache_map[ino]->dirty_buffers.count(*it)) {
+      bcache_map[ino]->dirty_buffers.erase(*it)
+      (*it)->put();
+    }
   }
 }
 
@@ -191,8 +196,8 @@ Bufferhead *Buffercache::alloc_buffers(ino, offset, size)
 
 
 void Buffercache::map_or_alloc(inodeno_t ino, size_t len, off_t off, 
-                               map<off_t, Bufferhead*> *buffers, 
-                               map<off_t, Bufferhead*> *inflight)
+                               map<off_t, Bufferhead*>& buffers, 
+                               map<off_t, Bufferhead*>& inflight)
 {
   Filecache *fc = bcache_map[ino];
   map<off_t, size_t> holes;
@@ -214,10 +219,10 @@ void Buffercache::free_buffers(Bufferhead *bh)
 {
   assert(bh->state == BUFH_STATE_CLEAN);
   assert(bh->lru_is_expirable());
-  bcache_map[bh->ino]->buffer_map.erase(bh->offset);
-  lru.lru_remove(bh);    
   clean_size -= bh->len;
-  delete bh;
+  lru.lru_remove(bh); bh->put();
+  assert(bh->ref == 1); // next put is going to delete it
+  bcache_map[bh->ino]->buffer_map.erase(bh->offset); bh->put();
 }
 
 void Buffercache::release_file(inodeno_t ino) 

diff --git a/ceph/client/Buffercache.h b/ceph/client/Buffercache.h
index fa30708a822ec8ce62e48e7c84d1116b83c06cfb..5988adf28b47705f666f503e5b1ad238d40459bc 100644 (file)
--- a/ceph/client/Buffercache.h
+++ b/ceph/client/Buffercache.h
@@ -16,17 +16,6 @@
 class Buffercache;
 
 class Bufferhead : public LRUObject {
-  // reference counter
-  int ref;  
-  void get() {
-       if (ref == 0) lru_pin();
-       ++ref;
-  }
-  void put() {
-       --ref;
-       if (ref == 0) lru_unpin();
-  }
-
  public: // FIXME: make more private and write some accessors
   off_t offset;
   size_t len;
@@ -38,10 +27,10 @@ class Bufferhead : public LRUObject {
   // write_waiters: threads waiting for writes into the buffer
   list<Cond*> read_waiters, write_waiters;
   Buffercache *bc;
+  ref = 0;
   
   // cons/destructors
   Bufferhead(inodeno_t ino, off_t off, size_t len, Buffercache *bc, int state=BUFHD_STATE_CLEAN) {
-       this->ref = 0;
     this->ino = ino;
     this->offset = off;
        this->len = len;
@@ -52,32 +41,41 @@ class Bufferhead : public LRUObject {
   }
   
   ~Bufferhead() {
-       // no need to delete bufferlist bufferptr's explicitly; ~list() does that (since it's list<bufferptr>, not list<bufferptr*>)
+    list<bufferptr> bl = bh->bl.buffers();
+    for (list<bufferptr>::iterator it == bl.begin();
+         it != bl.end();
+         it++) {
+      delete *it;
+    }
   }
   
   //Bufferhead(inodeno_t ino, off_t off, size_t len, int state);
-  
   // ~Bufferhead(); FIXME: need to mesh with allocator scheme
-
-
-  // -- wait for read, write: these will block
-  // i think this will work okay?  and reference coutning in the waiter makes sure the wakeup fn doesn't 
-  // inadvertantly unpin the bufferhead before the waiters get to go
-  void wait_for_read(Mutex& lock) {
-       Cond cond;  // on local stack
-       get();
-       read_waiters.push_back(&cond);
-       cond.Wait(lock);
-       put();
-  }
-  void wait_for_write(Mutex& lock) {
-       Cond cond;  // on local stack
-       get();
-       write_waiters.push_back(&cond);
-       cond.Wait(lock);
-       put();  
+  
+  void get() {
+    ref++;
+    assert(ref > 0);
+  }
+  
+  void put() {
+    assert (ref > 0);
+    ref--;
+    if (ref == 0) {
+      assert(!lru_pinned);
+      delete this;
+    }
   }
 
+  void add_read_waiter(Cond *cond) {
+    read_waiters->push_back(cond); 
+       lru_pin(); 
+  }
+  
+  void add_write_waiter(Cond *cond) { 
+    write_waiters->push_back(cond); 
+       lru_pin(); 
+  }
+  
   void wakeup_read_waiters() { 
     for (list<Cond*>::iterator it = read_waiters.begin();
                 it != read_waiters.end();
@@ -85,6 +83,7 @@ class Bufferhead : public LRUObject {
          (*it)->Signal();
        }
     read_waiters.clear(); 
+       if (write_waiters.empty()) lru_unpin(); 
   }
   
   void wakeup_write_waiters() {
@@ -93,6 +92,8 @@ class Bufferhead : public LRUObject {
                 it++) {
          (*it)->Signal();
        }
+    write_waiters.clear(); 
+       if (read_waiters.empty()) lru_unpin(); 
   }
   
   void miss_start() {
@@ -112,7 +113,14 @@ class Bufferhead : public LRUObject {
       state = BUFHD_STATE_DIRTY;
       bc->dirty_size += bh->len;
       bc->clean_size -= bh->len;
-      bc->dirty_map[last_written] = this;
+      if (bc->dirty_buffers.count(offset)) {
+        bc->dirty_buffers.insert(this); 
+        get();
+      }
+      if (bc->bcache_map[ino]->dirty_buffers.count(offset)) {
+        bc->bcache_map[ino]->dirty_buffers.insert(this); 
+        get();
+      }
     }    
   }
   
@@ -126,8 +134,11 @@ class Bufferhead : public LRUObject {
   void flush_finish() {
        assert(state == BUFHD_STATE_INFLIGHT);
        state = BUFHD_STATE_CLEAN;
+    last_written = time();
     bc->flush_size -= len;
     bc->clean_size += len;
+    bc->dirty_buffers.erase(this); put();
+    bc->bcache_map[ino]->dirty_buffers.erase(this); put();
        wakeup_write_waiters(); // readers never wait on flushes
   }
   
@@ -143,6 +154,8 @@ class Bufferhead : public LRUObject {
 class Filecache {
  public: 
   map<off_t, Bufferhead*> buffer_map;
+  set<Bufferhead*> dirty_buffers;
+  list<Cond*> waitfor_flushed;
 
   size_t length() {
     size_t len = 0;
@@ -159,7 +172,7 @@ class Filecache {
   void map_existing(size_t len, off_t start_off, 
                     map<off_t, Bufferhead*>& hits, inflight,
                     map<off_t, size_t>& holes);
-  void simplify();
+  void simplify(list<Bufferhead*>& removed);
 
 };
 
@@ -168,8 +181,9 @@ class Buffercache {
   map<inodeno_t, Filecache*> bcache_map;
   LRU lru;
   size_t dirty_size = 0, flushing_size = 0, clean_size = 0;
-  map<time_t, Bufferhead*> dirty_map;
-
+  set<Bufferhead*> dirty_buffers;
+  list<Cond*> waitfor_flushed;
+  
   // FIXME: constructor & destructor need to mesh with allocator scheme
   ~Buffercache() {
     // FIXME: make sure all buffers are cleaned  and then free them