dley unmount until sync writes are flushed

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

ceph/client/Buffercache.cc

@@ -93,19 +93,19 @@ void Bufferhead::alloc_buffers(off_t size)
   assert(size > 0);
   while (size > 0) {
     if (size <= (unsigned)g_conf.client_bcache_alloc_maxsize) {
-          off_t k = g_conf.client_bcache_alloc_minsize;
-          off_t asize = size - size % k + (size % k > 0) * k;
+	  off_t k = g_conf.client_bcache_alloc_minsize;
+	  off_t asize = size - size % k + (size % k > 0) * k;
 	  buffer *b = new buffer(asize);
 	  b->set_length(size);
 	  bl.push_back(b);
 	  bc->increase_size(size);
-          dout(10) << "bc: new buffer(" << asize << "), total: " << bl.length() << endl;
+	  dout(10) << "bc: new buffer(" << asize << "), total: " << bl.length() << endl;
 	  break;
 	}
-        buffer *b = new buffer(g_conf.client_bcache_alloc_maxsize);
+	buffer *b = new buffer(g_conf.client_bcache_alloc_maxsize);
 	b->set_length(g_conf.client_bcache_alloc_maxsize);
 	bl.push_back(b);
-        dout(10) << "bc: new buffer(" << g_conf.client_bcache_alloc_maxsize << "), total: " << bl.length() << endl;
+	dout(10) << "bc: new buffer(" << g_conf.client_bcache_alloc_maxsize << "), total: " << bl.length() << endl;
 	size -= g_conf.client_bcache_alloc_maxsize;
 	bc->increase_size(g_conf.client_bcache_alloc_maxsize);
   }

ceph/client/Client.cc

@@ -67,6 +67,8 @@ Client::Client(Messenger *m)
   mounted = false;
   unmounting = false;
 
+  unsafe_sync_write = 0;
+
   // 
   root = 0;
 
@@ -976,8 +978,12 @@ int Client::unmount()
   lru.lru_set_max(0);
   trim_cache();
 
-  while (lru.lru_get_size() > 0 || !inode_map.empty()) {
-	dout(3) << "cache still has " << lru.lru_get_size() << "+" << inode_map.size() << " items, waiting (presumably for caps to be released?)" << endl;
+  while (lru.lru_get_size() > 0 || 
+		 !inode_map.empty() ||
+		 unsafe_sync_write > 0) {
+	dout(3) << "cache still has " << lru.lru_get_size() 
+			<< "+" << inode_map.size() << " items + " 
+			<< unsafe_sync_write << " unsafe_sync_writes, waiting (presumably for caps to be released?)" << endl;
 	unmount_cond.Wait(client_lock);
   }
   assert(lru.lru_get_size() == 0);
@@ -1882,6 +1888,32 @@ public:
 };
 
 
+/*
+ * hack -- 
+ *  until we properly implement synchronous writes wrt buffer cache,
+ *  make sure we delay shutdown until they're all safe on disk!
+ */
+class C_Client_HackUnsafe : public Context {
+  Client *cl;
+public:
+  C_Client_HackUnsafe(Client *c) : cl(c) {}
+  void finish(int) {
+	cl->hack_sync_write_safe();
+  }
+};
+
+void Client::hack_sync_write_safe()
+{
+  client_lock.Lock();
+  assert(unsafe_sync_write > 0);
+  unsafe_sync_write--;
+  if (unmounting) {
+	cerr << "hack_sync_write_safe -- no more unsafe writes, unmount can proceed" << endl;
+	unmount_cond.Signal();
+  }
+  client_lock.Unlock();
+}
+
 int Client::write(fh_t fh, const char *buf, off_t size, off_t offset) 
 {
   client_lock.Lock();
@@ -1969,10 +2001,14 @@ int Client::write(fh_t fh, const char *buf, off_t size, off_t offset)
 	  
 	  bool done = false;
 	  C_Client_Cond *onfinish = new C_Client_Cond(&done, &cond, &client_lock, &rvalue);
+	  C_Client_HackUnsafe *onsafe = new C_Client_HackUnsafe(this);
+	  unsafe_sync_write++;
+
 	  filer->write(in->inode, size, offset, blist, 0, 
 				   //NULL,NULL);  // no wait hack
-				   onfinish, NULL);   // applied
-				   //NULL, onfinish); // safe on disk
+				   onfinish, onsafe);
+	  
+
 	  while (!done)
 		cond.Wait(client_lock);
 	}

ceph/client/Client.h

@@ -243,6 +243,11 @@ class Client : public Dispatcher {
   bool   mounted;
   bool   unmounting;
   Cond   unmount_cond;  
+
+  int    unsafe_sync_write;
+public:
+  void hack_sync_write_safe();
+protected:
   
   Filer                 *filer;  // (non-blocking) osd interface