From 616a655c227bbb8988a3e5abdad960284cd52349 Mon Sep 17 00:00:00 2001
From: Sage Weil <sage@newdream.net>
Date: Mon, 7 Dec 2009 14:28:45 -0800
Subject: [PATCH] filestore: limit size of flusher queue

Limit the size of the flusher queue to avoid hitting max open files
limit.  If we hit the max, just don't queue async flush, so we close the
fd immediately.
---
 src/config.cc       |  1 +
 src/config.h        |  1 +
 src/os/FileStore.cc | 38 +++++++++++++++++++++++++++-----------
 src/os/FileStore.h  |  5 +++--
 4 files changed, 32 insertions(+), 13 deletions(-)

diff --git a/src/config.cc b/src/config.cc
index 6bca011f74d1a..4f6bd40e9262e 100644
--- a/src/config.cc
+++ b/src/config.cc
@@ -522,6 +522,7 @@ static struct config_option config_optionsp[] = {
 	OPTION(filestore_dev, 0, OPT_STR, 0),
 	OPTION(filestore_btrfs_trans, 0, OPT_BOOL, true),
 	OPTION(filestore_flusher, 0, OPT_BOOL, true),
+	OPTION(filestore_flusher_max_fds, 0, OPT_INT, 512),
 	OPTION(filestore_sync_flush, 0, OPT_BOOL, false),
 	OPTION(ebofs, 0, OPT_BOOL, false),
 	OPTION(ebofs_cloneable, 0, OPT_BOOL, true),
diff --git a/src/config.h b/src/config.h
index ad21e6ae8932e..f7f9f7e87e432 100644
--- a/src/config.h
+++ b/src/config.h
@@ -334,6 +334,7 @@ struct md_config_t {
   const char  *filestore_dev;
   bool filestore_btrfs_trans;
   bool filestore_flusher;
+  int filestore_flusher_max_fds;
   bool filestore_sync_flush;
   
   // ebofs
diff --git a/src/os/FileStore.cc b/src/os/FileStore.cc
index dbdd11ce051d3..7b0c78261fd3e 100644
--- a/src/os/FileStore.cc
+++ b/src/os/FileStore.cc
@@ -1450,9 +1450,8 @@ int FileStore::_write(coll_t cid, const sobject_t& oid,
       derr(0) << "couldn't write to " << fn << " len " << len << " off " << offset << " errno " << errno << " " << strerror_r(errno, buf, sizeof(buf)) << dendl;
     }
 
-    if (g_conf.filestore_flusher)
-      queue_flusher(fd, offset, len);
-    else {
+    if (!g_conf.filestore_flusher ||
+	!queue_flusher(fd, offset, len)) {
       if (g_conf.filestore_sync_flush)
 	::sync_file_range(fd, offset, len, SYNC_FILE_RANGE_WRITE);
       ::close(fd);
@@ -1585,16 +1584,30 @@ int FileStore::_clone_range(coll_t cid, const sobject_t& oldoid, const sobject_t
 }
 
 
-void FileStore::queue_flusher(int fd, __u64 off, __u64 len)
+bool FileStore::queue_flusher(int fd, __u64 off, __u64 len)
 {
+  bool queued;
   lock.Lock();
-  dout(10) << "queue_flusher ep " << sync_epoch << " fd " << fd << " " << off << "~" << len << dendl;
-  flusher_queue.push_back(sync_epoch);
-  flusher_queue.push_back(fd);
-  flusher_queue.push_back(off);
-  flusher_queue.push_back(len);
-  flusher_cond.Signal();
+  if (flusher_queue_len < g_conf.filestore_flusher_max_fds) {
+    flusher_queue.push_back(sync_epoch);
+    flusher_queue.push_back(fd);
+    flusher_queue.push_back(off);
+    flusher_queue.push_back(len);
+    flusher_queue_len++;
+    flusher_cond.Signal();
+    dout(10) << "queue_flusher ep " << sync_epoch << " fd " << fd << " " << off << "~" << len
+	     << " qlen " << flusher_queue_len
+	     << dendl;
+    queued = true;
+  } else {
+    dout(10) << "queue_flusher ep " << sync_epoch << " fd " << fd << " " << off << "~" << len
+	     << " qlen " << flusher_queue_len 
+	     << " hit flusher_max_fds " << g_conf.filestore_flusher_max_fds
+	     << ", skipping async flush" << dendl;
+    queued = false;
+  }
   lock.Unlock();
+  return queued;
 }
 
 void FileStore::flusher_entry()
@@ -1605,7 +1618,9 @@ void FileStore::flusher_entry()
     if (!flusher_queue.empty()) {
       list<__u64> q;
       q.swap(flusher_queue);
-      
+
+      int num = flusher_queue_len;  // see how many we're taking, here
+
       lock.Unlock();
       while (!q.empty()) {
 	__u64 ep = q.front();
@@ -1625,6 +1640,7 @@ void FileStore::flusher_entry()
 	::close(fd);
       }
       lock.Lock();
+      flusher_queue_len -= num;   // they're definitely closed, forget
     } else {
       if (stop)
 	break;
diff --git a/src/os/FileStore.h b/src/os/FileStore.h
index 53a82b411ed04..a74469598e3bd 100644
--- a/src/os/FileStore.h
+++ b/src/os/FileStore.h
@@ -84,6 +84,7 @@ class FileStore : public JournalingObjectStore {
   // flusher thread
   Cond flusher_cond;
   list<__u64> flusher_queue;
+  int flusher_queue_len;
   void flusher_entry();
   struct FlusherThread : public Thread {
     FileStore *fs;
@@ -93,7 +94,7 @@ class FileStore : public JournalingObjectStore {
       return 0;
     }
   } flusher_thread;
-  void queue_flusher(int fd, __u64 off, __u64 len);
+  bool queue_flusher(int fd, __u64 off, __u64 len);
   int open_journal();
 
  public:
@@ -104,7 +105,7 @@ class FileStore : public JournalingObjectStore {
     attrs(this), fake_attrs(false), 
     collections(this), fake_collections(false),
     lock("FileStore::lock"),
-    sync_epoch(0), stop(false), sync_thread(this), flusher_thread(this) { }
+    sync_epoch(0), stop(false), sync_thread(this), flusher_queue_len(0), flusher_thread(this) { }
 
   int mount();
   int umount();
-- 
2.39.5