Use an interval map for tracking in-progress writes

2 files changed, 40 insertions, 25 deletions

diff --git a/netfs/fuse/fuseApp.h b/netfs/fuse/fuseApp.h
index 3f4d40d..9486e41 100644
--- a/netfs/fuse/fuseApp.h
+++ b/netfs/fuse/fuseApp.h
@@ -13,6 +13,7 @@
 #include "fuseConfig.h"
 #include "cache.h"
 #include <visibility.h>
+#include <boost/icl/interval_map.hpp>
 
 namespace NetFS {
 	class DLL_PUBLIC FuseApp : public FuseAppBase {
@@ -33,13 +34,14 @@ namespace NetFS {
 					OpenFile(FilePrx remote, const std::string & path, int flags);
 
 					void flush();
-					void nonBlockingFlush();
 					void wait() const;
 
 					FilePrx remote;
 					const std::string path;
 					const int flags;
-					std::list<Ice::AsyncResultPtr> bg;
+					typedef boost::icl::interval_map<Ice::Long, Ice::AsyncResultPtr> BGs;
+					static BGs::interval_type range(off_t, size_t);
+					BGs bg;
 					mutable boost::shared_mutex _lock;
 			};
 			typedef std::shared_ptr<OpenFile> OpenFilePtr;
diff --git a/netfs/fuse/fuseFiles.cpp b/netfs/fuse/fuseFiles.cpp
index 4c93d56..eb80531 100644
--- a/netfs/fuse/fuseFiles.cpp
+++ b/netfs/fuse/fuseFiles.cpp
@@ -23,7 +23,7 @@ FuseApp::OpenFile::wait() const
 {
 	SharedLock(_lock);
 	for (const auto & w : bg) {
-		w->waitForCompleted();
+		w.second->waitForCompleted();
 	}
 }
 
@@ -32,29 +32,18 @@ FuseApp::OpenFile::flush()
 {
 	Lock(_lock);
 	while (!bg.empty()) {
-		auto w = bg.front();
-		bg.pop_front();
+		auto wi = bg.begin();
+		auto w = wi->second;
+		bg.erase(w);
 		// bg operations are void, so no need to actually get the return value
 		w->throwLocalException();
 	}
 }
 
-void
-FuseApp::OpenFile::nonBlockingFlush()
+FuseApp::OpenFile::BGs::interval_type
+FuseApp::OpenFile::range(off_t o, size_t s)
 {
-	boost::unique_lock<boost::shared_mutex> lock(_lock, boost::try_to_lock);
-	if (lock.owns_lock()) {
-		for (auto wi = bg.begin(); wi != bg.end(); ) {
-			auto w = *wi;
-			if (w->isCompleted()) {
-				wi = bg.erase(wi);
-				w->throwLocalException();
-			}
-			else {
-				wi++;
-			}
-		}
-	}
+	return OpenFile::BGs::interval_type::right_open(o, o + s);
 }
 
 int
@@ -138,15 +127,39 @@ FuseApp::write(const char *, const char * buf, size_t s, off_t o, struct fuse_fi
 	try {
 		auto of = getProxy<OpenFilePtr>(fi->fh);
 		auto remote = of->remote;
-		auto r = remote->begin_write(o, s, Buffer(buf, buf + s));
 		if (fcr->Async) {
-			ScopeLock(of->_lock) {
-				of->bg.push_back(r);
+			const auto key = OpenFile::range(s, o);
+			while (true) {
+				ScopeLock(of->_lock) {
+					// Acquire operations to wait for
+					std::vector<Ice::AsyncResultPtr> overlap;
+					auto R = of->bg.equal_range(key);
+					if (R.first == R.second) {
+						// Begin write and store operation
+						auto r = remote->begin_write(o, s, Buffer(buf, buf + s), [of,key](const Ice::AsyncResultPtr &) {
+							ScopeLock(of->_lock) {
+								of->bg.erase(key);
+							}
+						});
+						of->bg.insert({key, r});
+						return s;
+					}
+					else {
+						// Wait for them whilst unlocked
+						of->_lock.unlock();
+						overlap.reserve(std::distance(R.first, R.second));
+						for (auto i = R.first; i != R.second; i++) {
+							overlap.push_back(i->second);
+						}
+						for (const auto & r : overlap) {
+							r->waitForCompleted();
+						}
+					}
+				}
 			}
-			of->nonBlockingFlush();
 		}
 		else {
-			r->throwLocalException();
+			remote->write(o, s, Buffer(buf, buf + s));
 		}
 		return s;
 	}