Merge branch 'master' into subvol-threads

Somehow a delete failed to get merged.

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

include/crucible/cleanup.h

@@ -0,0 +1,18 @@
+#ifndef CRUCIBLE_CLEANUP_H
+#define CRUCIBLE_CLEANUP_H
+
+#include <functional>
+
+namespace crucible {
+	using namespace std;
+
+	class Cleanup {
+		function<void()> m_cleaner;
+	public:
+		Cleanup(function<void()> func);
+		~Cleanup();
+	};
+
+}
+
+#endif // CRUCIBLE_CLEANUP_H

include/crucible/lockset.h

@@ -1,6 +1,7 @@
 #ifndef CRUCIBLE_LOCKSET_H
 #define CRUCIBLE_LOCKSET_H
 
+#include <crucible/cleanup.h>
 #include <crucible/error.h>
 #include <crucible/process.h>
 
@@ -12,6 +13,8 @@
 #include <map>
 #include <memory>
 #include <mutex>
+#include <set>
+#include <thread>
 
 namespace crucible {
 	using namespace std;
@@ -29,8 +32,11 @@ namespace crucible {
 		mutex				m_mutex;
 		condition_variable		m_condvar;
 		size_t				m_max_size = numeric_limits<size_t>::max();
+		set<uint64_t>			m_priorities;
+		uint64_t			m_priority_counter;
 
 		bool full();
+		bool first_in_priority(uint64_t my_priority);
 		bool locked(const key_type &name);
 
 		class Lock {
@@ -61,7 +67,6 @@ namespace crucible {
 		size_t size();
 		bool empty();
 		set_type copy();
-		void wait_unlock(double interval);
 
 		void max_size(size_t max);
 
@@ -96,6 +101,26 @@ namespace crucible {
 		return m_set.size() >= m_max_size;
 	}
 
+	template <class T>
+	bool
+	LockSet<T>::first_in_priority(uint64_t my_priority)
+	{
+#if 1
+		auto counter = m_max_size;
+		for (auto i : m_priorities) {
+			if (i == my_priority) {
+				return true;
+			}
+			if (++counter > m_max_size) {
+				return false;
+			}
+		}
+		THROW_ERROR(runtime_error, "my_priority " << my_priority << " not in m_priorities (size " << m_priorities.size() << ")");
+#else
+		return *m_priorities.begin() == my_priority;
+#endif
+	}
+
 	template <class T>
 	bool
 	LockSet<T>::locked(const key_type &name)
@@ -105,9 +130,10 @@ namespace crucible {
 
 	template <class T>
 	void
-	LockSet<T>::max_size(size_t s)
+	LockSet<T>::max_size(size_t new_max_size)
 	{
-		m_max_size = s;
+		THROW_CHECK1(out_of_range, new_max_size, new_max_size > 0);
+		m_max_size = new_max_size;
 	}
 
 	template <class T>
@@ -115,11 +141,18 @@ namespace crucible {
 	LockSet<T>::lock(const key_type &name)
 	{
 		unique_lock<mutex> lock(m_mutex);
-		while (full() || locked(name)) {
+		auto my_priority = m_priority_counter++;
+		Cleanup cleanup([&]() {
+			m_priorities.erase(my_priority);
+		});
+		m_priorities.insert(my_priority);
+		while (full() || locked(name) || !first_in_priority(my_priority)) {
 			m_condvar.wait(lock);
 		}
 		auto rv = m_set.insert(make_pair(name, gettid()));
 		THROW_CHECK0(runtime_error, rv.second);
+		// We removed our priority slot so other threads have to check again
+		m_condvar.notify_all();
 	}
 
 	template <class T>
@@ -142,18 +175,11 @@ namespace crucible {
 		unique_lock<mutex> lock(m_mutex);
 		auto erase_count = m_set.erase(name);
 		m_condvar.notify_all();
+		lock.unlock();
+		this_thread::yield();
 		THROW_CHECK1(invalid_argument, erase_count, erase_count == 1);
 	}
 
-	template <class T>
-	void
-	LockSet<T>::wait_unlock(double interval)
-	{
-		unique_lock<mutex> lock(m_mutex);
-		if (m_set.empty()) return;
-		m_condvar.wait_for(lock, chrono::duration<double>(interval));
-	}
-
 	template <class T>
 	size_t
 	LockSet<T>::size()

include/crucible/resource.h

@@ -3,6 +3,7 @@
 
 #include "crucible/error.h"
 
+#include <cassert>
 #include <map>
 #include <memory>
 #include <mutex>
@@ -52,7 +53,6 @@ namespace crucible {
 
 		// A bunch of static variables and functions
 		static mutex s_map_mutex;
-		static mutex s_ptr_mutex;
 		static map_type s_map;
 		static resource_ptr_type insert(const key_type &key);
 		static resource_ptr_type insert(const resource_ptr_type &res);
@@ -83,14 +83,14 @@ namespace crucible {
 		ResourceHandle(const resource_ptr_type &res);
 		ResourceHandle& operator=(const resource_ptr_type &res);
 
-		// default constructor is public and mostly harmless
+		// default construct/assign/move is public and mostly harmless
 		ResourceHandle() = default;
+		ResourceHandle(const ResourceHandle &that) = default;
+		ResourceHandle(ResourceHandle &&that) = default;
+		ResourceHandle& operator=(const ResourceHandle &that) = default;
+		ResourceHandle& operator=(ResourceHandle &&that) = default;
 
-		// copy/assign/move/move-assign - with a mutex to help shared_ptr be atomic
-		ResourceHandle(const ResourceHandle &that);
-		ResourceHandle(ResourceHandle &&that);
-		ResourceHandle& operator=(const ResourceHandle &that);
-		ResourceHandle& operator=(ResourceHandle &&that);
+		// Nontrivial destructor
 		~ResourceHandle();
 
 		// forward anything else to the Resource constructor
@@ -239,7 +239,6 @@ namespace crucible {
 	template <class Key, class Resource>
 	ResourceHandle<Key, Resource>::ResourceHandle(const key_type &key)
 	{
-		unique_lock<mutex> lock(s_ptr_mutex);
 		m_ptr = insert(key);
 	}
 
@@ -247,7 +246,6 @@ namespace crucible {
 	ResourceHandle<Key, Resource>&
 	ResourceHandle<Key, Resource>::operator=(const key_type &key)
 	{
-		unique_lock<mutex> lock(s_ptr_mutex);
 		m_ptr = insert(key);
 		return *this;
 	}
@@ -255,7 +253,6 @@ namespace crucible {
 	template <class Key, class Resource>
 	ResourceHandle<Key, Resource>::ResourceHandle(const resource_ptr_type &res)
 	{
-		unique_lock<mutex> lock(s_ptr_mutex);
 		m_ptr = insert(res);
 	}
 
@@ -263,65 +260,32 @@ namespace crucible {
 	ResourceHandle<Key, Resource>&
 	ResourceHandle<Key, Resource>::operator=(const resource_ptr_type &res)
 	{
-		unique_lock<mutex> lock(s_ptr_mutex);
 		m_ptr = insert(res);
 		return *this;
 	}
 
-	template <class Key, class Resource>
-	ResourceHandle<Key, Resource>::ResourceHandle(const ResourceHandle &that)
-	{
-		unique_lock<mutex> lock(s_ptr_mutex);
-		m_ptr = that.m_ptr;
-	}
-
-	template <class Key, class Resource>
-	ResourceHandle<Key, Resource>::ResourceHandle(ResourceHandle &&that)
-	{
-		unique_lock<mutex> lock(s_ptr_mutex);
-		swap(m_ptr, that.m_ptr);
-	}
-
-	template <class Key, class Resource>
-	ResourceHandle<Key, Resource> &
-	ResourceHandle<Key, Resource>::operator=(ResourceHandle &&that)
-	{
-		unique_lock<mutex> lock(s_ptr_mutex);
-		m_ptr = that.m_ptr;
-		that.m_ptr.reset();
-		return *this;
-	}
-
-	template <class Key, class Resource>
-	ResourceHandle<Key, Resource> &
-	ResourceHandle<Key, Resource>::operator=(const ResourceHandle &that)
-	{
-		unique_lock<mutex> lock(s_ptr_mutex);
-		m_ptr = that.m_ptr;
-		return *this;
-	}
-
 	template <class Key, class Resource>
 	ResourceHandle<Key, Resource>::~ResourceHandle()
 	{
-		unique_lock<mutex> lock_ptr(s_ptr_mutex);
 		// No pointer, nothing to do
 		if (!m_ptr) {
 			return;
 		}
 		// Save key so we can clean the map
 		auto key = s_traits.get_key(*m_ptr);
-		// Save pointer so we can release lock before deleting
-		auto ptr_copy = m_ptr;
+		// Save a weak_ptr so we can tell if we need to clean the map
+		weak_ptr_type wp = m_ptr;
+		// Drop shared_ptr
 		m_ptr.reset();
-		// Release lock
-		lock_ptr.unlock();
-		// Delete our (possibly last) reference to pointer
-		ptr_copy.reset();
+		// If there are still other references to the shared_ptr, we can stop now
+		if (!wp.expired()) {
+			return;
+		}
 		// Remove weak_ptr from map if it has expired
 		// (and not been replaced in the meantime)
 		unique_lock<mutex> lock_map(s_map_mutex);
 		auto found = s_map.find(key);
+		// Map entry may have been replaced, so check for expiry again
 		if (found != s_map.end() && found->second.expired()) {
 			s_map.erase(key);
 		}
@@ -331,23 +295,17 @@ namespace crucible {
 	typename ResourceHandle<Key, Resource>::resource_ptr_type
 	ResourceHandle<Key, Resource>::get_resource_ptr() const
 	{
-		unique_lock<mutex> lock(s_ptr_mutex);
-		// Make isolated copy of pointer with lock held, and return the copy
-		auto rv = m_ptr;
-		return rv;
+		return m_ptr;
 	}
 
 	template <class Key, class Resource>
 	typename ResourceHandle<Key, Resource>::resource_ptr_type
 	ResourceHandle<Key, Resource>::operator->() const
 	{
-		unique_lock<mutex> lock(s_ptr_mutex);
 		if (!m_ptr) {
 			THROW_ERROR(out_of_range, __PRETTY_FUNCTION__ << " called on null Resource");
 		}
-		// Make isolated copy of pointer with lock held, and return the copy
-		auto rv = m_ptr;
-		return rv;
+		return m_ptr;
 	}
 
 	template <class Key, class Resource>
@@ -355,7 +313,6 @@ namespace crucible {
 	shared_ptr<T>
 	ResourceHandle<Key, Resource>::cast() const
 	{
-		unique_lock<mutex> lock(s_ptr_mutex);
 		shared_ptr<T> dp;
 		if (!m_ptr) {
 			return dp;
@@ -371,7 +328,6 @@ namespace crucible {
 	typename ResourceHandle<Key, Resource>::key_type
 	ResourceHandle<Key, Resource>::get_key() const
 	{
-		unique_lock<mutex> lock(s_ptr_mutex);
 		if (!m_ptr) {
 			return s_traits.get_null_key();
 		} else {
@@ -399,13 +355,9 @@ namespace crucible {
 	template <class Key, class Resource>
 	mutex ResourceHandle<Key, Resource>::s_map_mutex;
 
-	template <class Key, class Resource>
-	mutex ResourceHandle<Key, Resource>::s_ptr_mutex;
-
 	template <class Key, class Resource>
 	typename ResourceHandle<Key, Resource>::map_type ResourceHandle<Key, Resource>::s_map;
 
-
 }
 
 #endif // RESOURCE_H

lib/Makefile

@@ -3,8 +3,9 @@ TAG := $(shell git describe --always --dirty || echo UNKNOWN)
 default: libcrucible.so
 
 OBJS = \
-	crc64.o \
 	chatter.o \
+	cleanup.o \
+	crc64.o \
 	error.o \
 	extentwalker.o \
 	fd.o \

lib/cleanup.cc

@@ -0,0 +1,17 @@
+#include <crucible/cleanup.h>
+
+namespace crucible {
+
+	Cleanup::Cleanup(function<void()> func) :
+		m_cleaner(func)
+	{
+	}
+
+	Cleanup::~Cleanup()
+	{
+		if (m_cleaner) {
+			m_cleaner();
+		}
+	}
+
+}

src/bees-context.cc

@@ -1,6 +1,7 @@
 #include "bees.h"
 
 #include "crucible/limits.h"
+#include "crucible/process.h"
 #include "crucible/string.h"
 
 #include <fstream>
@@ -29,13 +30,14 @@ BeesFdCache::BeesFdCache()
 		BEESCOUNTADD(open_root_ms, open_timer.age() * 1000);
 		return rv;
 	});
+	m_root_cache.max_size(BEES_ROOT_FD_CACHE_SIZE);
 	m_file_cache.func([&](shared_ptr<BeesContext> ctx, uint64_t root, uint64_t ino) -> Fd {
 		Timer open_timer;
 		auto rv = ctx->roots()->open_root_ino_nocache(root, ino);
 		BEESCOUNTADD(open_ino_ms, open_timer.age() * 1000);
 		return rv;
 	});
-	m_file_cache.max_size(BEES_FD_CACHE_SIZE);
+	m_file_cache.max_size(BEES_FILE_FD_CACHE_SIZE);
 }
 
 Fd
@@ -158,6 +160,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();
 	}
@@ -166,19 +169,11 @@ BeesContext::BeesContext(shared_ptr<BeesContext> parent) :
 bool
 BeesContext::dedup(const BeesRangePair &brp)
 {
-	// TOOLONG and NOTE can retroactively fill in the filename details, but LOG can't
-	BEESNOTE("dedup " << brp);
-
+	// Open the files
 	brp.first.fd(shared_from_this());
 	brp.second.fd(shared_from_this());
 
-#if 0
-	// This avoids some sort of kernel race condition;
-	// however, it also doubles our dedup times.
-	// Is avoiding a crash every few weeks worth it?
-	bees_sync(brp.first.fd());
-#endif
-
+	BEESNOTE("dedup " << brp);
 	BEESTOOLONG("dedup " << brp);
 
 	BeesAddress first_addr(brp.first.fd(), brp.first.begin());
@@ -725,6 +720,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));
+				auto extent_lock = m_extent_lock_set.make_lock(extent_bytenr);
 				Timer one_extent_timer;
 				return_bfr = scan_one_extent(bfr, e);
 				BEESCOUNTADD(scanf_extent_ms, one_extent_timer.age() * 1000);
@@ -756,12 +754,19 @@ BeesContext::resolve_addr_uncached(BeesAddress addr)
 {
 	THROW_CHECK1(invalid_argument, addr, !addr.is_magic());
 	THROW_CHECK0(invalid_argument, !!root_fd());
+
+	// To avoid hammering all the cores with long-running ioctls,
+	// only do one resolve at any given time.
+	BEESNOTE("waiting to resolve addr " << addr);
+	auto lock = bees_ioctl_lock_set.make_lock(gettid());
+
 	Timer resolve_timer;
 
 	// There is no performance benefit if we restrict the buffer size.
         BtrfsIoctlLogicalInoArgs log_ino(addr.get_physical_or_zero());
 
 	{
+		BEESNOTE("resolving addr " << addr);
 		BEESTOOLONG("Resolving addr " << addr << " in " << root_path() << " refs " << log_ino.m_iors.size());
 		if (log_ino.do_ioctl_nothrow(root_fd())) {
 			BEESCOUNT(resolve_ok);
@@ -816,8 +821,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
-	m_resolve_cache.max_size(65536);
+	// 65536 is big enough for two max-sized extents.
+	// 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-roots.cc

@@ -1,11 +1,14 @@
 #include "bees.h"
 
 #include "crucible/cache.h"
+#include "crucible/process.h"
 #include "crucible/string.h"
 
 #include <fstream>
 #include <tuple>
 
+#include <inttypes.h>
+
 using namespace crucible;
 using namespace std;
 
@@ -150,9 +153,12 @@ BeesRoots::crawl_state_erase(const BeesCrawlState &bcs)
 		return;
 	}
 
-	if (m_root_crawl_map.count(bcs.m_root)) {
-		m_root_crawl_map.erase(bcs.m_root);
+	auto found = m_root_crawl_map.find(bcs.m_root);
+	if (found != m_root_crawl_map.end()) {
+		auto hold_this_until_unlocked = found->second;
+		m_root_crawl_map.erase(found);
 		m_crawl_dirty = true;
+		lock.unlock();
 	}
 }
 
@@ -191,97 +197,12 @@ BeesRoots::transid_max()
 	return rv;
 }
 
-void
-BeesRoots::crawl_roots()
-{
-	BEESNOTE("Crawling roots");
-
-	unique_lock<mutex> lock(m_mutex);
-	if (m_root_crawl_map.empty()) {
-		BEESNOTE("idle, crawl map is empty");
-		m_condvar.wait(lock);
-		// Don't count the time we were waiting as part of the crawl time
-		m_crawl_timer.reset();
-	}
-
-	// Work from a copy because BeesCrawl might change the world under us
-	auto crawl_map_copy = m_root_crawl_map;
-	lock.unlock();
-
-#if 0
-	// Scan the same inode/offset tuple in each subvol (good for snapshots)
-	BeesFileRange first_range;
-	shared_ptr<BeesCrawl> first_crawl;
-	for (auto i : crawl_map_copy) {
-		auto this_crawl = i.second;
-		auto this_range = this_crawl->peek_front();
-		if (this_range) {
-			if (!first_range || this_range < first_range) {
-				first_crawl = this_crawl;
-				first_range = this_range;
-			}
-		}
-	}
-
-	if (first_range) {
-		catch_all([&]() {
-			// BEESINFO("scan_forward " << first_range);
-			m_ctx->scan_forward(first_range);
-		});
-		BEESCOUNT(crawl_scan);
-		m_crawl_current = first_crawl->get_state();
-		auto first_range_popped = first_crawl->pop_front();
-		THROW_CHECK2(runtime_error, first_range, first_range_popped, first_range == first_range_popped);
-		return;
-	}
-#else
-	// Scan each subvol one extent at a time (good for continuous forward progress)
-	bool crawled = false;
-	for (auto i : crawl_map_copy) {
-		auto this_crawl = i.second;
-		auto this_range = this_crawl->peek_front();
-		if (this_range) {
-			catch_all([&]() {
-				// BEESINFO("scan_forward " << this_range);
-				m_ctx->scan_forward(this_range);
-			});
-			crawled = true;
-			BEESCOUNT(crawl_scan);
-			m_crawl_current = this_crawl->get_state();
-			auto this_range_popped = this_crawl->pop_front();
-			THROW_CHECK2(runtime_error, this_range, this_range_popped, this_range == this_range_popped);
-		}
-	}
-
-	if (crawled) return;
-#endif
-
-	BEESLOG("Crawl ran out of data after " << m_crawl_timer.lap() << "s, waiting for more...");
-	BEESCOUNT(crawl_done);
-	BEESNOTE("idle, waiting for more data");
-	lock.lock();
-	m_condvar.wait(lock);
-
-	// Don't count the time we were waiting as part of the crawl time
-	m_crawl_timer.reset();
-}
-
-void
-BeesRoots::crawl_thread()
-{
-	BEESNOTE("crawling");
-	while (1) {
-		catch_all([&]() {
-			crawl_roots();
-		});
-	}
-}
-
 void
 BeesRoots::writeback_thread()
 {
-	while (1) {
-		BEESNOTE(m_crawl_current << (m_crawl_dirty ? " (dirty)" : ""));
+	while (true) {
+		// BEESNOTE(m_crawl_current << (m_crawl_dirty ? " (dirty)" : ""));
+		BEESNOTE((m_crawl_dirty ? "dirty" : "clean") << ", interval " << BEES_WRITEBACK_INTERVAL << "s");
 
 		catch_all([&]() {
 			BEESNOTE("saving crawler state");
@@ -379,17 +300,16 @@ BeesRoots::state_load()
 BeesRoots::BeesRoots(shared_ptr<BeesContext> ctx) :
 	m_ctx(ctx),
 	m_crawl_state_file(ctx->home_fd(), crawl_state_filename()),
-	m_crawl_thread("crawl"),
 	m_writeback_thread("crawl_writeback")
 {
-	m_crawl_thread.exec([&]() {
-		catch_all([&]() {
-			state_load();
-		});
-		m_writeback_thread.exec([&]() {
-			writeback_thread();
-		});
-		crawl_thread();
+	// This is a sanity check to prevent us from running out of FDs
+	m_lock_set.max_size(BEES_WORKER_THREAD_LIMIT);
+
+	catch_all([&]() {
+		state_load();
+	});
+	m_writeback_thread.exec([&]() {
+		writeback_thread();
 	});
 }
 
@@ -397,6 +317,7 @@ Fd
 BeesRoots::open_root_nocache(uint64_t rootid)
 {
 	BEESTRACE("open_root_nocache " << rootid);
+	BEESNOTE("open_root_nocache " << rootid);
 
 	// Stop recursion at the root of the filesystem tree
 	if (rootid == BTRFS_FS_TREE_OBJECTID) {
@@ -469,7 +390,7 @@ BeesRoots::open_root_nocache(uint64_t rootid)
 				THROW_CHECK2(runtime_error, new_root_id, rootid, new_root_id == rootid);
 				Stat st(rv);
 				THROW_CHECK1(runtime_error, st.st_ino, st.st_ino == BTRFS_FIRST_FREE_OBJECTID);
-				BEESINFO("open_root_nocache " << rootid << ": " << name_fd(rv));
+				// BEESINFO("open_root_nocache " << rootid << ": " << name_fd(rv));
 				return rv;
 			}
 		}
@@ -632,8 +553,54 @@ BeesRoots::open_root_ino(uint64_t root, uint64_t ino)
 
 BeesCrawl::BeesCrawl(shared_ptr<BeesContext> ctx, BeesCrawlState initial_state) :
 	m_ctx(ctx),
-	m_state(initial_state)
+	m_state(initial_state),
+	m_thread(astringprintf("crawl_%" PRIu64, m_state.m_root))
 {
+	m_thread.exec([&]() {
+		crawl_thread();
+	});
+}
+
+BeesCrawl::~BeesCrawl()
+{
+	BEESLOGNOTE("Stopping crawl thread " << m_state);
+	unique_lock<mutex> lock(m_mutex);
+	m_stopped = true;
+	m_cond_stopped.notify_all();
+	lock.unlock();
+	BEESLOGNOTE("Joining crawl thread " << m_state);
+	m_thread.join();
+	BEESLOG("Stopped crawl thread " << m_state);
+}
+
+void
+BeesCrawl::crawl_thread()
+{
+	Timer crawl_timer;
+	while (!m_stopped) {
+		BEESNOTE("pop_front " << m_state);
+		auto this_range = pop_front();
+		if (this_range) {
+			catch_all([&]() {
+				BEESNOTE("waiting for scan thread limit " << m_state);
+				auto crawl_lock = m_ctx->roots()->lock_set().make_lock(m_state.m_root);
+
+				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...");
+			unique_lock<mutex> lock(m_mutex);
+			if (m_stopped) {
+				break;
+			}
+			m_cond_stopped.wait_for(lock, chrono::duration<double>(BEES_COMMIT_INTERVAL));
+			crawl_timer.reset();
+		}
+	}
+	BEESLOG("Crawl thread stopped");
 }
 
 bool
@@ -681,7 +648,7 @@ BeesCrawl::fetch_extents()
 	}
 
 	BEESNOTE("crawling " << get_state());
-	BEESLOG("Crawling " << get_state());
+	// BEESLOG("Crawling " << get_state());
 
 	Timer crawl_timer;
 
@@ -700,6 +667,11 @@ BeesCrawl::fetch_extents()
 	BEESTRACE("Searching crawl sk " << static_cast<btrfs_ioctl_search_key&>(sk));
 	bool ioctl_ok = false;
 	{
+#if 0
+		BEESNOTE("waiting to search crawl sk " << static_cast<btrfs_ioctl_search_key&>(sk));
+		auto lock = bees_ioctl_lock_set.make_lock(gettid());
+#endif
+
 		BEESNOTE("searching crawl sk " << static_cast<btrfs_ioctl_search_key&>(sk));
 		BEESTOOLONG("Searching crawl sk " << static_cast<btrfs_ioctl_search_key&>(sk));
 		Timer crawl_timer;
@@ -720,7 +692,7 @@ BeesCrawl::fetch_extents()
 		return next_transid();
 	}
 
-	BEESLOG("Crawling " << sk.m_result.size() << " results from " << get_state());
+	// BEESLOG("Crawling " << sk.m_result.size() << " results from " << get_state());
 	auto results_left = sk.m_result.size();
 	BEESNOTE("crawling " << results_left << " results from " << get_state());
 	size_t count_other = 0;
@@ -737,7 +709,6 @@ BeesCrawl::fetch_extents()
 
 		BEESTRACE("i = " << i);
 
-#if 1
 		// We need the "+ 1" and objectid rollover that next_min does.
 		auto new_state = get_state();
 		new_state.m_objectid = sk.min_objectid;
@@ -749,7 +720,6 @@ BeesCrawl::fetch_extents()
 		// is a lot of metadata we can't process.  Favor forward
 		// progress over losing search results.
 		set_state(new_state);
-#endif
 
 		// Ignore things that aren't EXTENT_DATA_KEY
 		if (i.type != BTRFS_EXTENT_DATA_KEY) {
@@ -762,13 +732,24 @@ BeesCrawl::fetch_extents()
 		if (gen < get_state().m_min_transid) {
 			BEESCOUNT(crawl_gen_low);
 			++count_low;
-			// We probably want (need?) to scan these anyway.
-			// continue;
+			// We want (need?) to scan these anyway?
+			// The header generation refers to the transid
+			// of the metadata page holding the current ref.
+			// This includes anything else in that page that
+			// happened to be modified, regardless of how
+			// old it is.
+			// The file_extent_generation refers to the
+			// transid of the extent item's page, which is
+			// a different approximation of what we want.
+			// Combine both of these filters to minimize
+			// the number of times we unnecessarily re-read
+			// an extent.
+			continue;
 		}
 		if (gen > get_state().m_max_transid) {
 			BEESCOUNT(crawl_gen_high);
 			++count_high;
-			// This shouldn't ever happen
+			// This shouldn't ever happen...and so far, doesn't.
 			// continue;
 		}
 
@@ -818,7 +799,7 @@ BeesCrawl::fetch_extents()
 			}
 		}
 	}
-	BEESLOG("Crawled inline " << count_inline << " data " << count_data << " other " << count_other << " unknown " << count_unknown << " gen_low " << count_low << " gen_high " << count_high << " " << get_state() << " in " << crawl_timer << "s");
+	// BEESLOG("Crawled inline " << count_inline << " data " << count_data << " other " << count_other << " unknown " << count_unknown << " gen_low " << count_low << " gen_high " << count_high << " " << get_state() << " in " << crawl_timer << "s");
 
 	return true;
 }
@@ -857,12 +838,6 @@ BeesCrawl::pop_front()
 	}
 	auto rv = *m_extents.begin();
 	m_extents.erase(m_extents.begin());
-#if 0
-	auto state = get_state();
-	state.m_objectid = rv.fid().ino();
-	state.m_offset = rv.begin();
-	set_state(state);
-#endif
 	return rv;
 }
 

src/bees.cc

@@ -19,6 +19,10 @@
 #include <linux/fs.h>
 #include <sys/ioctl.h>
 
+// setrlimit
+#include <sys/time.h>
+#include <sys/resource.h>
+
 #include <getopt.h>
 
 using namespace crucible;
@@ -213,6 +217,13 @@ operator<<(ostream &os, const BeesStatTmpl<T> &bs)
 
 // other ----------------------------------------
 
+/**
+ * Don't allow two threads to use some btrfs ioctls at the same time.
+ * Some of them consume egregious amounts of kernel CPU time and are
+ * not interruptible, so if we have more threads than cores we will
+ * effectively crash the kernel. */
+LockSet<pid_t> bees_ioctl_lock_set;
+
 template <class T>
 T&
 BeesStatTmpl<T>::at(string idx)
@@ -583,6 +594,13 @@ BeesTempFile::make_copy(const BeesFileRange &src)
 	return rv;
 }
 
+unsigned
+bees_worker_thread_count()
+{
+	// Maybe # of cores * (scalar from 0.25..4)?
+	return max(1U, thread::hardware_concurrency() * 4);
+}
+
 int
 bees_main(int argc, char *argv[])
 {
@@ -647,6 +665,21 @@ bees_main(int argc, char *argv[])
 		BEESLOG("using relative path " << relative_path() << "\n");
 	}
 
+	// There can be only one because we measure running time with it
+	// EXPERIMENT:  don't try this on kernels before v4.14
+	// bees_ioctl_lock_set.max_size(1);
+
+	BEESLOG("setting rlimit NOFILE to " << BEES_OPEN_FILE_LIMIT);
+
+	struct rlimit lim = {
+		.rlim_cur = BEES_OPEN_FILE_LIMIT,
+		.rlim_max = BEES_OPEN_FILE_LIMIT,
+	};
+	int rv = setrlimit(RLIMIT_NOFILE, &lim);
+	if (rv) {
+		BEESLOG("setrlimit(RLIMIT_NOFILE, { " << lim.rlim_cur << " }): " << strerror(errno));
+	};
+
 	// Create a context and start crawlers
 	bool did_subscription = false;
 	while (optind < argc) {

src/bees.h

@@ -75,14 +75,25 @@ const int BEES_PROGRESS_INTERVAL = BEES_STATS_INTERVAL;
 // Status is output every freakin second.  Use a ramdisk.
 const int BEES_STATUS_INTERVAL = 1;
 
-// Number of FDs to open (not counting 100 roots)
-const size_t BEES_FD_CACHE_SIZE = 384;
+// Number of file FDs to cache when not in active use
+const size_t BEES_FILE_FD_CACHE_SIZE = 4096;
+
+// Number of root FDs to cache when not in active use
+const size_t BEES_ROOT_FD_CACHE_SIZE = 1024;
+
+// Number of FDs to open (rlimit)
+const size_t BEES_OPEN_FILE_LIMIT = (BEES_FILE_FD_CACHE_SIZE + BEES_ROOT_FD_CACHE_SIZE) * 2 + 100;
+
+// Worker thread limit (more threads may be created, but only this number will be active concurrently)
+const size_t BEES_WORKER_THREAD_LIMIT = 128;
 
 // Log warnings when an operation takes too long
 const double BEES_TOO_LONG = 2.5;
 
 // Avoid any extent where LOGICAL_INO takes this long
-const double BEES_TOXIC_DURATION = 9.9;
+// const double BEES_TOXIC_DURATION = 9.9;
+// EXPERIMENT:  Kernel v4.14+ may let us ignore toxicity
+const double BEES_TOXIC_DURATION = BEES_COMMIT_INTERVAL;
 
 // How long between hash table histograms
 const double BEES_HASH_TABLE_ANALYZE_INTERVAL = BEES_STATS_INTERVAL;
@@ -95,7 +106,7 @@ const double BEES_INFO_BURST = 1.0;
 const size_t BEES_MAX_QUEUE_SIZE = 1024;
 
 // Read this many items at a time in SEARCHv2
-const size_t BEES_MAX_CRAWL_SIZE = 4096;
+const size_t BEES_MAX_CRAWL_SIZE = 1024;
 
 // If an extent has this many refs, pretend it does not exist
 // to avoid a crippling btrfs performance bug
@@ -495,30 +506,35 @@ class BeesCrawl {
 	mutex					m_state_mutex;
 	BeesCrawlState				m_state;
 
+	BeesThread				m_thread;
+	bool					m_stopped = false;
+	condition_variable			m_cond_stopped;
+
 	bool fetch_extents();
 	void fetch_extents_harder();
 	bool next_transid();
 
 public:
+	~BeesCrawl();
 	BeesCrawl(shared_ptr<BeesContext> ctx, BeesCrawlState initial_state);
 	BeesFileRange peek_front();
 	BeesFileRange pop_front();
 	BeesCrawlState get_state();
 	void set_state(const BeesCrawlState &bcs);
+	void crawl_thread();
 };
 
 class BeesRoots {
 	shared_ptr<BeesContext>			m_ctx;
 
 	BeesStringFile				m_crawl_state_file;
-	BeesCrawlState				m_crawl_current;
 	map<uint64_t, shared_ptr<BeesCrawl>>	m_root_crawl_map;
 	mutex					m_mutex;
 	condition_variable			m_condvar;
 	bool					m_crawl_dirty = false;
 	Timer					m_crawl_timer;
-	BeesThread				m_crawl_thread;
 	BeesThread				m_writeback_thread;
+	LockSet<uint64_t>			m_lock_set;
 
 	void insert_new_crawl();
 	void insert_root(const BeesCrawlState &bcs);
@@ -533,7 +549,6 @@ class BeesRoots {
 	BeesCrawlState crawl_state_get(uint64_t root);
 	void crawl_state_set_dirty();
 	void crawl_state_erase(const BeesCrawlState &bcs);
-	void crawl_thread();
 	void writeback_thread();
 	uint64_t next_root(uint64_t root = 0);
 	void current_state_set(const BeesCrawlState &bcs);
@@ -546,6 +561,7 @@ public:
 	Fd open_root(uint64_t root);
 	Fd open_root_ino(uint64_t root, uint64_t ino);
 	Fd open_root_ino(const BeesFileId &bfi) { return open_root_ino(bfi.root(), bfi.ino()); }
+	LockSet<uint64_t> &lock_set() { return m_lock_set; }
 };
 
 struct BeesHash {
@@ -668,6 +684,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);
@@ -705,6 +723,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);
@@ -792,5 +811,7 @@ string pretty(double d);
 extern RateLimiter bees_info_rate_limit;
 void bees_sync(int fd);
 string format_time(time_t t);
+extern LockSet<pid_t> bees_ioctl_lock_set;
+extern unsigned bees_worker_thread_count();
 
 #endif