context: lock extents by bytenr instead of globally prohibiting tmpfiles

This prevents two threads from attempting to dispose of the same physical
extent at the same time.  This is a more precise exclusion than the
general lock on all tmpfiles.

Signed-off-by: Zygo Blaxell <bees@furryterror.org>

src/bees-context.cc

@@ -252,6 +252,7 @@ BeesContext::home_fd()
 BeesContext::BeesContext(shared_ptr<BeesContext> parent) :
 	m_parent_ctx(parent)
 {
+	m_extent_lock_set.max_size(bees_worker_thread_count());;
 	if (m_parent_ctx) {
 		m_fd_cache = m_parent_ctx->fd_cache();
 	}
@@ -440,9 +441,6 @@ BeesContext::scan_one_extent(const BeesFileRange &bfr, const Extent &e)
 	BEESTRACE(e << " block_count " << block_count);
 	string bar(block_count, '#');
 
-	// Only one thread may create tmpfiles at any given time
-	unique_lock<mutex> tmpfile_lock(bees_tmpfile_mutex, defer_lock);
-
 	for (off_t next_p = e.begin(); next_p < e.end(); ) {
 
 		// Guarantee forward progress
@@ -743,10 +741,6 @@ BeesContext::scan_one_extent(const BeesFileRange &bfr, const Extent &e)
 			// BEESLOG("noinsert_set.count(" << to_hex(p) << ") " << noinsert_set.count(p));
 			if (noinsert_set.count(p)) {
 				if (p - last_p > 0) {
-					if (!tmpfile_lock) {
-						BEESNOTE("waiting for tmpfile");
-						tmpfile_lock.lock();
-					}
 					rewrite_file_range(BeesFileRange(bfr.fd(), last_p, p));
 					blocks_rewritten = true;
 				}
@@ -758,10 +752,6 @@ BeesContext::scan_one_extent(const BeesFileRange &bfr, const Extent &e)
 		}
 		BEESTRACE("last");
 		if (next_p - last_p > 0) {
-			if (!tmpfile_lock) {
-				BEESNOTE("waiting for tmpfile");
-				tmpfile_lock.lock();
-			}
 			rewrite_file_range(BeesFileRange(bfr.fd(), last_p, next_p));
 			blocks_rewritten = true;
 		}
@@ -852,6 +842,9 @@ BeesContext::scan_forward(const BeesFileRange &bfr)
 			e = ew.current();
 
 			catch_all([&]() {
+				uint64_t extent_bytenr = e.bytenr();
+				BEESNOTE("waiting for extent bytenr " << to_hex(extent_bytenr));
+				decltype(m_extent_lock_set)::Lock extent_lock(m_extent_lock_set, extent_bytenr);
 				Timer one_extent_timer;
 				return_bfr = scan_one_extent(bfr, e);
 				BEESCOUNTADD(scanf_extent_ms, one_extent_timer.age() * 1000);
@@ -950,8 +943,9 @@ BeesContext::set_root_fd(Fd fd)
 	m_root_uuid = fsinfo.uuid();
 	BEESLOG("Filesystem UUID is " << m_root_uuid);
 
-	// 65536 is big enough for two max-sized extents
+	// 65536 is big enough for two max-sized extents.
-	m_resolve_cache.max_size(65536);
+	// Need enough total space in the cache for the maximum number of active threads.
+	m_resolve_cache.max_size(65536 * bees_worker_thread_count());
 	m_resolve_cache.func([&](BeesAddress addr) -> BeesResolveAddrResult {
 		return resolve_addr_uncached(addr);
 	});

src/bees-hash.cc

@@ -311,7 +311,7 @@ BeesHashTable::fetch_missing_extent(HashType hash)
 	BEESNOTE("waiting to fetch hash extent #" << extent_number << ", " << missing_buckets << " left to fetch");
 
 	// Acquire blocking lock on this extent only
-	LockSet<uint64_t>::Lock extent_lock(m_extent_lock_set, extent_number);
+	decltype(m_extent_lock_set)::Lock extent_lock(m_extent_lock_set, extent_number);
 
 	// Check missing again because someone else might have fetched this
 	// extent for us while we didn't hold any locks

src/bees-roots.cc

@@ -301,8 +301,7 @@ BeesRoots::BeesRoots(shared_ptr<BeesContext> ctx) :
 	m_crawl_state_file(ctx->home_fd(), crawl_state_filename()),
 	m_writeback_thread("crawl_writeback")
 {
-	unsigned max_crawlers = max(1U, thread::hardware_concurrency());
+	m_lock_set.max_size(bees_worker_thread_count());
-	m_lock_set.max_size(max_crawlers);
 
 	catch_all([&]() {
 		state_load();
@@ -555,20 +554,22 @@ void
 BeesCrawl::crawl_thread()
 {
 	Timer crawl_timer;
+	LockSet<uint64_t>::Lock crawl_lock(m_ctx->roots()->lock_set(), m_state.m_root, false);
 	while (!m_stopped) {
 		BEESNOTE("waiting for crawl thread limit " << m_state);
-		LockSet<uint64_t>::Lock crawl_lock(m_ctx->roots()->lock_set(), m_state.m_root);
+		crawl_lock.lock();
+		BEESNOTE("pop_front " << m_state);
 		auto this_range = pop_front();
+		crawl_lock.unlock();
 		if (this_range) {
 			catch_all([&]() {
-				// BEESINFO("scan_forward " << this_range);
+				BEESNOTE("scan_forward " << this_range);
 				m_ctx->scan_forward(this_range);
 			});
 			BEESCOUNT(crawl_scan);
 		} else {
 			auto crawl_time = crawl_timer.age();
 			BEESLOGNOTE("Crawl ran out of data after " << crawl_time << "s, waiting for more...");
-			crawl_lock.unlock();
 			unique_lock<mutex> lock(m_mutex);
 			if (m_stopped) {
 				break;

src/bees.cc

@@ -205,14 +205,6 @@ operator<<(ostream &os, const BeesStatTmpl<T> &bs)
  * effectively crash the kernel. */
 mutex bees_ioctl_mutex;
 
-/**
- * Don't allow two threads to create temporary copies of extent data at
- * the same time.  If two threads create temporary copies of the same
- * extent at the same time they will not be properly deduped.  This lock
- * goes into effect as the first temporary extent is created by a thread,
- * and is released after the source extent scan is finished. */
-mutex bees_tmpfile_mutex;
-
 template <class T>
 T&
 BeesStatTmpl<T>::at(string idx)
@@ -568,6 +560,12 @@ BeesTempFile::make_copy(const BeesFileRange &src)
 	return rv;
 }
 
+unsigned
+bees_worker_thread_count()
+{
+	return max(1U, thread::hardware_concurrency());
+}
+
 int
 bees_main(int argc, const char **argv)
 {

src/bees.h

@@ -703,6 +703,8 @@ class BeesContext : public enable_shared_from_this<BeesContext> {
 
 	Timer						m_total_timer;
 
+	LockSet<uint64_t>			m_extent_lock_set;
+
 	void set_root_fd(Fd fd);
 
 	BeesResolveAddrResult resolve_addr_uncached(BeesAddress addr);
@@ -740,6 +742,7 @@ public:
 	shared_ptr<BeesTempFile> tmpfile();
 
 	const Timer &total_timer() const { return m_total_timer; }
+	LockSet<uint64_t> &extent_lock_set() { return m_extent_lock_set; }
 
 	// TODO: move the rest of the FD cache methods here
 	void insert_root_ino(Fd fd);
@@ -828,6 +831,6 @@ extern RateLimiter bees_info_rate_limit;
 void bees_sync(int fd);
 string format_time(time_t t);
 extern mutex bees_ioctl_mutex;
-extern mutex bees_tmpfile_mutex;
+extern unsigned bees_worker_thread_count();
 
 #endif