progress: rename "ctime" -> "tm_left"

"ctime", an abbreviation of "cycle time", collides with "ctime", an
abbreviation of "st_ctime", a well-known filesystem term.

"tm_left" fits in the column, so use that.

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

docs/options.md

@@ -36,6 +36,34 @@
 
  Has no effect unless `--loadavg-target` is used to specify a target load.
 
+* `--throttle-factor FACTOR`
+
+ In order to avoid saturating btrfs deferred work queues, bees tracks
+ the time that operations with delayed effect (dedupe and tmpfile copy)
+ and operations with long run times (`LOGICAL_INO`) run.  If an operation
+ finishes before the average run time for that operation, bees will
+ sleep for the remainder of the average run time, so that operations
+ are submitted to btrfs at a rate similar to the rate that btrfs can
+ complete them.
+
+ The `FACTOR` is multiplied by the average run time for each operation
+ to calculate the target delay time.
+
+ `FACTOR` 0 is the default, which adds no delays.  bees will attempt
+ to saturate btrfs delayed work queues as quickly as possible, which
+ may impact other processes on the same filesystem, or even slow down
+ bees itself.
+
+ `FACTOR` 1.0 will attempt to keep btrfs delayed work queues filled at
+ a steady average rate.
+
+ `FACTOR` more than 1.0 will add delays longer than the average
+ run time (e.g. 10.0 will delay all operations that take less than 10x
+ the average run time).  High values of `FACTOR` may be desirable when
+ using bees with other applications on the same filesystem.
+
+ The maximum delay per operation is 60 seconds.
+
 ## Filesystem tree traversal options
 
 * `--scan-mode MODE` or `-m`

docs/running.md

@@ -75,9 +75,8 @@ in the shell script that launches `bees`:
         schedtool -D -n20 $$
         ionice -c3 -p $$
 
-You can also use the [`--loadavg-target` and `--thread-min`
-options](options.md) to further control the impact of bees on the rest
-of the system.
+You can also use the [load management options](options.md) to further
+control the impact of bees on the rest of the system.
 
 Let the bees fly:
 

include/crucible/bytevector.h

@@ -55,7 +55,6 @@ namespace crucible {
 		Pointer m_ptr;
 		size_t m_size = 0;
 		mutable mutex m_mutex;
-	friend ostream & operator<<(ostream &os, const ByteVector &bv);
 	};
 
 	template <class T>
@@ -74,6 +73,8 @@ namespace crucible {
 		THROW_CHECK2(out_of_range, size(), sizeof(T), size() >= sizeof(T));
 		return reinterpret_cast<T*>(data());
 	}
+
+	ostream& operator<<(ostream &os, const ByteVector &bv);
 }
 
 #endif // _CRUCIBLE_BYTEVECTOR_H_

include/crucible/fs.h

@@ -243,8 +243,6 @@ namespace crucible {
 		unsigned long available() const;
 	};
 
-	template<class V> ostream &hexdump(ostream &os, const V &v);
-
 	struct BtrfsIoctlFsInfoArgs : public btrfs_ioctl_fs_info_args_v3 {
 		BtrfsIoctlFsInfoArgs();
 		void do_ioctl(int fd);

include/crucible/hexdump.h

@@ -12,12 +12,14 @@ namespace crucible {
 	ostream &
 	hexdump(ostream &os, const V &v)
 	{
-		os << "V { size = " << v.size() << ", data:\n";
-		for (size_t i = 0; i < v.size(); i += 8) {
+		const auto v_size = v.size();
+		const uint8_t* const v_data = reinterpret_cast<uint8_t*>(v.data());
+		os << "V { size = " << v_size << ", data:\n";
+		for (size_t i = 0; i < v_size; i += 8) {
 			string hex, ascii;
 			for (size_t j = i; j < i + 8; ++j) {
-				if (j < v.size()) {
-					uint8_t c = v[j];
+				if (j < v_size) {
+					const uint8_t c = v_data[j];
 					char buf[8];
 					sprintf(buf, "%02x ", c);
 					hex += buf;

include/crucible/time.h

@@ -34,7 +34,7 @@ namespace crucible {
 		double	m_rate;
 		double	m_burst;
 		double  m_tokens = 0.0;
-		mutex	m_mutex;
+		mutable mutex m_mutex;
 
 		void update_tokens();
 		RateLimiter() = delete;
@@ -45,6 +45,8 @@ namespace crucible {
 		double sleep_time(double cost = 1.0);
 		bool is_ready();
 		void borrow(double cost = 1.0);
+		void rate(double new_rate);
+		double rate() const;
 	};
 
 	class RateEstimator {
@@ -88,6 +90,9 @@ namespace crucible {
 		// Read count
 		uint64_t count() const;
 
+		/// Increment count (like update(count() + more), but atomic)
+		void increment(uint64_t more = 1);
+
 		// Convert counts to chrono types
 		chrono::high_resolution_clock::time_point time_point(uint64_t absolute_count) const;
 		chrono::duration<double> duration(uint64_t relative_count) const;

lib/bytevector.cc

@@ -44,10 +44,10 @@ namespace crucible {
 	}
 
 	ByteVector::value_type&
-	ByteVector::operator[](size_t size) const
+	ByteVector::operator[](size_t index) const
 	{
 		unique_lock<mutex> lock(m_mutex);
-		return m_ptr.get()[size];
+		return m_ptr.get()[index];
 	}
 
 	ByteVector::ByteVector(const ByteVector &that)
@@ -183,7 +183,6 @@ namespace crucible {
 
 	ostream&
 	operator<<(ostream &os, const ByteVector &bv) {
-		unique_lock<mutex> lock(bv.m_mutex);
 		hexdump(os, bv);
 		return os;
 	}

lib/fs.cc

@@ -781,7 +781,7 @@ namespace crucible {
 			++s_calls;
 			if (rv != 0 && errno == ENOENT) {
 				// If we are searching a tree that is deleted or no longer exists, just return an empty list
-				nr_items = 0;
+				ioctl_ptr->key.nr_items = 0;
 				break;
 			}
 			if (rv != 0 && errno != EOVERFLOW) {

lib/task.cc

@@ -530,11 +530,6 @@ namespace crucible {
 	size_t
 	TaskMasterState::calculate_thread_count_nolock()
 	{
-		if (m_paused) {
-			// No threads running while paused or cancelled
-			return 0;
-		}
-
 		if (m_load_target == 0) {
 			// No limits, no stats, use configured thread count
 			return m_configured_thread_max;
@@ -645,6 +640,9 @@ namespace crucible {
 		unique_lock<mutex> lock(m_mutex);
 		m_paused = paused;
 		m_condvar.notify_all();
+		if (!m_paused) {
+			start_threads_nolock();
+		}
 		lock.unlock();
 	}
 

lib/time.cc

@@ -98,12 +98,16 @@ namespace crucible {
 		m_rate(rate),
 		m_burst(burst)
 	{
+		THROW_CHECK1(invalid_argument, m_rate, m_rate > 0);
+		THROW_CHECK1(invalid_argument, m_burst, m_burst >= 0);
 	}
 
 	RateLimiter::RateLimiter(double rate) :
 		m_rate(rate),
 		m_burst(rate)
 	{
+		THROW_CHECK1(invalid_argument, m_rate, m_rate > 0);
+		THROW_CHECK1(invalid_argument, m_burst, m_burst >= 0);
 	}
 
 	void
@@ -119,6 +123,7 @@ namespace crucible {
 	double
 	RateLimiter::sleep_time(double cost)
 	{
+		THROW_CHECK1(invalid_argument, m_rate, m_rate > 0);
 		borrow(cost);
 		unique_lock<mutex> lock(m_mutex);
 		update_tokens();
@@ -154,6 +159,21 @@ namespace crucible {
 		m_tokens -= cost;
 	}
 
+	void
+	RateLimiter::rate(double const new_rate)
+	{
+		THROW_CHECK1(invalid_argument, new_rate, new_rate > 0);
+		unique_lock<mutex> lock(m_mutex);
+		m_rate = new_rate;
+	}
+
+	double
+	RateLimiter::rate() const
+	{
+		unique_lock<mutex> lock(m_mutex);
+		return m_rate;
+	}
+
 	RateEstimator::RateEstimator(double min_delay, double max_delay) :
 		m_min_delay(min_delay),
 		m_max_delay(max_delay)
@@ -202,6 +222,13 @@ namespace crucible {
 		}
 	}
 
+	void
+	RateEstimator::increment(const uint64_t more)
+	{
+		unique_lock<mutex> lock(m_mutex);
+		return update_unlocked(m_last_count + more);
+	}
+
 	uint64_t
 	RateEstimator::count() const
 	{

src/bees-context.cc

@@ -20,7 +20,6 @@
 using namespace crucible;
 using namespace std;
 
-
 BeesFdCache::BeesFdCache(shared_ptr<BeesContext> ctx) :
 	m_ctx(ctx)
 {
@@ -98,6 +97,7 @@ BeesContext::dump_status()
 		TaskMaster::print_queue(ofs);
 #endif
 
+		ofs << "PROGRESS:\n";
 		ofs << get_progress();
 
 		ofs.close();
@@ -125,6 +125,9 @@ string
 BeesContext::get_progress()
 {
 	unique_lock<mutex> lock(m_progress_mtx);
+	if (m_progress_str.empty()) {
+		return "[No progress estimate available]\n";
+	}
 	return m_progress_str;
 }
 
@@ -210,6 +213,7 @@ BeesContext::dedup(const BeesRangePair &brp_in)
 {
 	// TOOLONG and NOTE can retroactively fill in the filename details, but LOG can't
 	BEESNOTE("dedup " << brp_in);
+	BEESTRACE("dedup " << brp_in);
 
 	if (is_root_ro(brp_in.second.fid().root())) {
 		// BEESLOGDEBUG("WORKAROUND: dst root " << (brp_in.second.fid().root()) << " is read-only);
@@ -237,27 +241,40 @@ BeesContext::dedup(const BeesRangePair &brp_in)
 	BEESCOUNT(dedup_try);
 
 	BEESNOTE("waiting to dedup " << brp);
-	const auto lock = MultiLocker::get_lock("dedupe");
-
-	Timer dedup_timer;
+	auto lock = MultiLocker::get_lock("dedupe");
 
 	BEESLOGINFO("dedup: src " << pretty(brp.first.size())  << " [" << to_hex(brp.first.begin())  << ".." << to_hex(brp.first.end())  << "] {" << first_addr  << "} " << name_fd(brp.first.fd()) << "\n"
 		 << "       dst " << pretty(brp.second.size()) << " [" << to_hex(brp.second.begin()) << ".." << to_hex(brp.second.end()) << "] {" << second_addr << "} " << name_fd(brp.second.fd()));
 	BEESNOTE("dedup: src " << pretty(brp.first.size())  << " [" << to_hex(brp.first.begin())  << ".." << to_hex(brp.first.end())  << "] {" << first_addr  << "} " << name_fd(brp.first.fd()) << "\n"
 		 << "       dst " << pretty(brp.second.size()) << " [" << to_hex(brp.second.begin()) << ".." << to_hex(brp.second.end()) << "] {" << second_addr << "} " << name_fd(brp.second.fd()));
 
-	const bool rv = btrfs_extent_same(brp.first.fd(), brp.first.begin(), brp.first.size(), brp.second.fd(), brp.second.begin());
-	BEESCOUNTADD(dedup_ms, dedup_timer.age() * 1000);
+	while (true) {
+		try {
+			Timer dedup_timer;
+			const bool rv = btrfs_extent_same(brp.first.fd(), brp.first.begin(), brp.first.size(), brp.second.fd(), brp.second.begin());
+			BEESCOUNTADD(dedup_ms, dedup_timer.age() * 1000);
 
-	if (rv) {
-		BEESCOUNT(dedup_hit);
-		BEESCOUNTADD(dedup_bytes, brp.first.size());
-	} else {
-		BEESCOUNT(dedup_miss);
-		BEESLOGWARN("NO Dedup! " << brp);
+			if (rv) {
+				BEESCOUNT(dedup_hit);
+				BEESCOUNTADD(dedup_bytes, brp.first.size());
+			} else {
+				BEESCOUNT(dedup_miss);
+				BEESLOGWARN("NO Dedup! " << brp);
+			}
+
+			lock.reset();
+			bees_throttle(dedup_timer.age(), "dedup");
+			return rv;
+		} catch (const std::system_error &e) {
+			if (e.code().value() == EAGAIN) {
+				BEESNOTE("dedup waiting for btrfs send on " << brp.second);
+				BEESLOGDEBUG("dedup waiting for btrfs send on " << brp.second);
+				roots()->wait_for_transid(1);
+			} else {
+				throw;
+			}
+		}
 	}
-
-	return rv;
 }
 
 BeesRangePair
@@ -341,6 +358,8 @@ BeesContext::scan_one_extent(const BeesFileRange &bfr, const Extent &e)
 	BEESTRACE("scan bfr " << bfr);
 	BEESCOUNT(scan_extent);
 
+	Timer one_timer;
+
 	// We keep moving this method around
 	auto m_ctx = shared_from_this();
 
@@ -837,7 +856,8 @@ BeesContext::scan_one_extent(const BeesFileRange &bfr, const Extent &e)
 			<< pretty(e.size()) << " "
 			<< dedupe_list.size() << "d" << copy_list.size() << "c"
 			<< ((bytes_zeroed + BLOCK_SIZE_SUMS - 1) / BLOCK_SIZE_SUMS) << "p"
-			<< (extent_compressed ? "z {" : " {")
+			<< (extent_compressed ? "z " : " ")
+			<< one_timer << "s {"
 			<< to_hex(e.bytenr()) << "+" << to_hex(e.offset()) << "} "
 			<< to_hex(e.begin()) << " [" << bar << "] " << to_hex(e.end())
 			<< ' ' << name_fd(bfr.fd())
@@ -974,9 +994,10 @@ BeesContext::resolve_addr_uncached(BeesAddress addr)
 	Timer resolve_timer;
 
 	struct rusage usage_before;
+	struct rusage usage_after;
 	{
 		BEESNOTE("waiting to resolve addr " << addr << " with LOGICAL_INO");
-		const auto lock = MultiLocker::get_lock("logical_ino");
+		auto lock = MultiLocker::get_lock("logical_ino");
 
 		// Get this thread's system CPU usage
 		DIE_IF_MINUS_ONE(getrusage(RUSAGE_THREAD, &usage_before));
@@ -990,13 +1011,13 @@ BeesContext::resolve_addr_uncached(BeesAddress addr)
 		} else {
 			BEESCOUNT(resolve_fail);
 		}
-		BEESCOUNTADD(resolve_ms, resolve_timer.age() * 1000);
+		DIE_IF_MINUS_ONE(getrusage(RUSAGE_THREAD, &usage_after));
+		const auto resolve_timer_age = resolve_timer.age();
+		BEESCOUNTADD(resolve_ms, resolve_timer_age * 1000);
+		lock.reset();
+		bees_throttle(resolve_timer_age, "resolve_addr");
 	}
 
-	// Again!
-	struct rusage usage_after;
-	DIE_IF_MINUS_ONE(getrusage(RUSAGE_THREAD, &usage_after));
-
 	const double sys_usage_delta =
 		(usage_after.ru_stime.tv_sec + usage_after.ru_stime.tv_usec / 1000000.0) -
 		(usage_before.ru_stime.tv_sec + usage_before.ru_stime.tv_usec / 1000000.0);

src/bees-roots.cc

@@ -130,7 +130,7 @@ BeesScanMode::start_scan()
 			st->scan();
 		});
 	}
-	m_scan_task.idle();
+	m_scan_task.run();
 }
 
 bool
@@ -592,7 +592,7 @@ BeesScanModeExtent::create_extent_map(const uint64_t bytenr, const ProgressTrack
 
 	{
 		BEESNOTE("waiting to create extent map for " << to_hex(bytenr) << " with LOGICAL_INO");
-		const auto lock = MultiLocker::get_lock("logical_ino");
+		auto lock = MultiLocker::get_lock("logical_ino");
 
 		BEESNOTE("Resolving bytenr " << to_hex(bytenr) << " refs " << log_ino.m_iors.size());
 		BEESTOOLONG("Resolving bytenr " << to_hex(bytenr) << " refs " << log_ino.m_iors.size());
@@ -605,8 +605,11 @@ BeesScanModeExtent::create_extent_map(const uint64_t bytenr, const ProgressTrack
 		} else {
 			BEESCOUNT(extent_fail);
 		}
+		const auto resolve_age = resolve_timer.age();
 
-		BEESCOUNTADD(extent_ms, resolve_timer.age() * 1000);
+		BEESCOUNTADD(extent_ms, resolve_age * 1000);
+		lock.reset();
+		bees_throttle(resolve_age, "extent_map");
 	}
 
         const size_t rv_count = log_ino.m_iors.size();
@@ -645,7 +648,7 @@ BeesScanModeExtent::create_extent_map(const uint64_t bytenr, const ProgressTrack
 			bedf.objectid(i.m_inum);
 			const auto bti = bedf.at(i.m_offset);
 			if (!bti) {
-				BEESLOGDEBUG("No ref for extent " << to_hex(bytenr) << " at root " << i.m_root << " ino " << i.m_inum << " offset " << to_hex(i.m_offset));
+				// BEESLOGDEBUG("No ref for extent " << to_hex(bytenr) << " at root " << i.m_root << " ino " << i.m_inum << " offset " << to_hex(i.m_offset));
 				BEESCOUNT(extent_ref_missing);
 				return;
 			}
@@ -765,7 +768,7 @@ BeesScanModeExtent::scan()
 
 	// Good to go, start everything running
 	for (const auto &i : task_map_copy) {
-		i.second.idle();
+		i.second.run();
 	}
 }
 
@@ -898,7 +901,7 @@ BeesScanModeExtent::map_next_extent(uint64_t const subvol)
 				<< " time " << crawl_time << " subvol " << subvol);
 		}
 
-		// We did something!  Get in line to run again
+		// We did something!  Get in line to run again (but don't preempt work already queued)
 		Task::current_task().idle();
 		return;
 	}
@@ -907,6 +910,18 @@ BeesScanModeExtent::map_next_extent(uint64_t const subvol)
 	BEESCOUNT(crawl_done);
 }
 
+static
+string
+strf_localtime(const time_t &when)
+{
+	struct tm ltm = { 0 };
+	DIE_IF_ZERO(localtime_r(&when, &ltm));
+
+	char buf[100] = { 0 };
+	DIE_IF_ZERO(strftime(buf, sizeof(buf), "%Y-%m-%d %H:%M", &ltm));
+	return buf;
+}
+
 void
 BeesScanModeExtent::next_transid(const CrawlMap &crawl_map_unused)
 {
@@ -963,13 +978,15 @@ BeesScanModeExtent::next_transid(const CrawlMap &crawl_map_unused)
 		THROW_CHECK0(runtime_error, offset > 0);
 		THROW_CHECK0(runtime_error, chunk_length > 0);
 		last_bgaddr = offset + chunk_length;
+		// Mixed-bg filesystems have block groups that are data _and_ metadata.
+		// Block groups that are _only_ metadata should be filtered out.
+		if (0 == (bti.chunk_type() & BTRFS_BLOCK_GROUP_DATA)) {
+			continue;
+		}
 		bg_info_map[last_bgaddr] = (bg_info) {
 			.first_bytenr = offset,
 			.first_total = fs_size,
 		};
-		if (bti.chunk_type() & (BTRFS_BLOCK_GROUP_METADATA | BTRFS_BLOCK_GROUP_SYSTEM)) {
-			continue;
-		}
 		fs_size += chunk_length;
 	}
 
@@ -995,6 +1012,18 @@ BeesScanModeExtent::next_transid(const CrawlMap &crawl_map_unused)
 
 	// Report on progress using extent bytenr map
 	Table::Table eta;
+	eta.insert_row(0, vector<Table::Content> {
+		Table::Text("extsz"),
+		Table::Text("datasz"),
+		Table::Text("point"),
+		Table::Text("gen_min"),
+		Table::Text("gen_max"),
+		Table::Text("this cycle start"),
+		Table::Text("tm_left"),
+		Table::Text("next cycle ETA"),
+	});
+	const auto dash_fill = Table::Fill('-');
+	eta.insert_row(1, vector<Table::Content>(eta.cols().size(), dash_fill));
 	for (const auto &i : s_magic_crawl_map) {
 		const auto &subvol = i.first;
 		const auto &magic = i.second;
@@ -1033,55 +1062,48 @@ BeesScanModeExtent::next_transid(const CrawlMap &crawl_map_unused)
 			BEESCOUNT(progress_out_of_bg);
 		}
 		const auto bytenr_offset = min(bi_last_bytenr, max(bytenr, bi.first_bytenr)) - bi.first_bytenr + bi.first_total;
-		const auto bytenr_percent = bytenr_offset / (0.01 * fs_size);
+		const auto bytenr_norm = bytenr_offset / double(fs_size);
 		const auto now = time(NULL);
 		const auto time_so_far = now - min(now, this_state.m_started);
+		const string start_stamp = strf_localtime(this_state.m_started);
 		string eta_stamp = "-";
 		string eta_pretty = "-";
 		const auto &deferred_finished = deferred_map.at(subvol);
-		const bool deferred = deferred_finished.first;
 		const bool finished = deferred_finished.second;
-		if (time_so_far > 1 && bytenr_percent > 0 && !finished) {
-			const time_t eta_duration = time_so_far / (bytenr_percent / 100);
+		if (finished) {
+			// eta_stamp = "idle";
+		} else if (time_so_far > 1 && bytenr_norm > 0.01) {
+			const time_t eta_duration = time_so_far / bytenr_norm;
 			const time_t eta_time = eta_duration + now;
-			struct tm ltm = { 0 };
-			DIE_IF_ZERO(localtime_r(&eta_time, &ltm));
-
-			char buf[1024] = { 0 };
-			DIE_IF_ZERO(strftime(buf, sizeof(buf), "%Y-%m-%d %H:%M", &ltm));
-			eta_stamp = string(buf);
+			eta_stamp = strf_localtime(eta_time);
 			eta_pretty = pretty_seconds(eta_duration);
 		}
 		const auto &mma = mes.m_map.at(subvol);
 		const auto mma_ratio = mes_sample_size_ok ? (mma.m_bytes / double(mes.m_total)) : 1.0;
-		const auto pos_scaled_text = mes_sample_size_ok ? pretty(bytenr_offset * mma_ratio) : "-";
-		const auto pos_text = Table::Text(deferred ? "deferred" : pos_scaled_text);
-		const auto pct_text = Table::Text(finished ? "finished" : astringprintf("%.4f%%", bytenr_percent));
+		const auto posn_text = Table::Text(finished ? "idle" : astringprintf("%06d", int(floor(bytenr_norm * 1000000))));
 		const auto size_text = Table::Text( mes_sample_size_ok ? pretty(fs_size * mma_ratio) : "-");
 		eta.insert_row(Table::endpos, vector<Table::Content> {
-			pos_text,
-			size_text,
-			pct_text,
 			Table::Text(magic.m_max_size == numeric_limits<uint64_t>::max() ? "max" : pretty(magic.m_max_size)),
+			size_text,
+			posn_text,
 			Table::Number(this_state.m_min_transid),
 			Table::Number(this_state.m_max_transid),
+			Table::Text(start_stamp),
 			Table::Text(eta_pretty),
 			Table::Text(eta_stamp),
 		});
 		BEESCOUNT(progress_ok);
 	}
-	eta.insert_row(0, vector<Table::Content> {
-		Table::Text("done"),
+	eta.insert_row(Table::endpos, vector<Table::Content> {
+		Table::Text("total"),
 		Table::Text(pretty(fs_size)),
-		Table::Text("%done"),
-		Table::Text("size"),
-		Table::Text("transid"),
+		Table::Text(""),
+		Table::Text("gen_now"),
 		Table::Number(m_roots->transid_max()),
-		Table::Text("todo"),
-		Table::Text("ETA"),
+		Table::Text(""),
+		Table::Text(""),
+		Table::Text(""),
 	});
-	const auto dash_fill = Table::Fill('-');
-	eta.insert_row(1, vector<Table::Content>(eta.cols().size(), dash_fill));
 	eta.left("");
 	eta.mid(" ");
 	eta.right("");
@@ -1279,6 +1301,9 @@ BeesRoots::transid_max_nocache()
 	THROW_CHECK1(runtime_error, rv, rv > 0);
 	// transid must be less than max, or we did something very wrong
 	THROW_CHECK1(runtime_error, rv, rv < numeric_limits<uint64_t>::max());
+
+	// Update the rate estimator
+	m_transid_re.update(rv);
 	return rv;
 }
 
@@ -1303,8 +1328,6 @@ struct BeesFileCrawl {
 	BeesCrawlState					m_state;
 	/// Currently processed offset in file
 	off_t						m_offset;
-	/// Btrfs file fetcher
-	BtrfsExtentDataFetcher				m_bedf;
 
 	/// Method that does one unit of work for the Task
 	bool crawl_one_extent();
@@ -1316,10 +1339,15 @@ BeesFileCrawl::crawl_one_extent()
 	BEESNOTE("crawl_one_extent m_offset " << to_hex(m_offset) << " state " << m_state);
 	BEESTRACE("crawl_one_extent m_offset " << to_hex(m_offset) << " state " << m_state);
 
+	BtrfsExtentDataFetcher bedf(m_ctx->root_fd());
+	bedf.tree(m_state.m_root);
+	bedf.objectid(m_state.m_objectid);
+	bedf.transid(m_state.m_min_transid);
+
 	// Only one thread can dedupe a file.  btrfs will lock others out.
 	// Inodes are usually full of shared extents, especially in the case of snapshots,
 	// so when we lock an inode, we'll lock the same inode number in all subvols at once.
-	auto inode_mutex = m_ctx->get_inode_mutex(m_bedf.objectid());
+	auto inode_mutex = m_ctx->get_inode_mutex(bedf.objectid());
 	auto inode_lock = inode_mutex->try_lock(Task::current_task());
 	if (!inode_lock) {
 		BEESCOUNT(crawl_deferred_inode);
@@ -1331,12 +1359,12 @@ BeesFileCrawl::crawl_one_extent()
 	// It will mean the file or subvol was deleted or there's metadata corruption,
 	// and we should stop trying to scan the inode in that case.
 	// The calling Task will be aborted.
-	const auto bti = m_bedf.lower_bound(m_offset);
+	const auto bti = bedf.lower_bound(m_offset);
 	if (!bti) {
 		return false;
 	}
 	// Make sure we advance
-	m_offset = max(bti.offset() + m_bedf.block_size(), bti.offset());
+	m_offset = max(bti.offset() + bedf.block_size(), bti.offset());
 	// Check extent item generation is in range
 	const auto gen = bti.file_extent_generation();
 	if (gen < m_state.m_min_transid) {
@@ -1446,11 +1474,7 @@ BeesRoots::crawl_batch(shared_ptr<BeesCrawl> this_crawl)
 		.m_hold = this_crawl->hold_state(this_state),
 		.m_state = this_state,
 		.m_offset = this_range.begin(),
-		.m_bedf = BtrfsExtentDataFetcher(m_ctx->root_fd()),
 	});
-	bfc->m_bedf.tree(subvol);
-	bfc->m_bedf.objectid(inode);
-	bfc->m_bedf.transid(this_state.m_min_transid);
 	BEESNOTE("Starting task " << this_range);
 	Task(task_title, [bfc]() {
 		BEESNOTE("crawl_batch " << bfc->m_hold->get());
@@ -1476,6 +1500,15 @@ BeesRoots::clear_caches()
 	m_ctx->resolve_cache_clear();
 }
 
+void
+BeesRoots::wait_for_transid(const uint64_t count)
+{
+	const auto now_transid = transid_max_nocache();
+	const auto target_transid = now_transid + count;
+	BEESLOGDEBUG("Waiting for transid " << target_transid << ", current transid is " << now_transid);
+	m_transid_re.wait_until(target_transid);
+}
+
 void
 BeesRoots::crawl_thread()
 {
@@ -1497,7 +1530,8 @@ BeesRoots::crawl_thread()
 		BEESTRACE("Measure current transid");
 		catch_all([&]() {
 			BEESTRACE("calling transid_max_nocache");
-			m_transid_re.update(transid_max_nocache());
+			// Will update m_transid_re as side effect
+			transid_max_nocache();
 		});
 
 		const auto new_transid = m_transid_re.count();
@@ -1614,7 +1648,7 @@ BeesRoots::insert_new_crawl()
 	lock.unlock();
 
 	// Nothing to crawl?  Seems suspicious...
-	if (m_root_crawl_map.empty()) {
+	if (crawl_map_copy.empty()) {
 		BEESLOGINFO("crawl map is empty!");
 	}
 
@@ -1682,7 +1716,7 @@ BeesRoots::start()
 	m_crawl_thread.exec([&]() {
 		// Measure current transid before creating any crawlers
 		catch_all([&]() {
-			m_transid_re.update(transid_max_nocache());
+			transid_max_nocache();
 		});
 
 		// Make sure we have a full complement of crawlers
@@ -2026,12 +2060,6 @@ BeesRoots::open_root_ino(uint64_t root, uint64_t ino)
 	return m_ctx->fd_cache()->open_root_ino(root, ino);
 }
 
-RateEstimator &
-BeesRoots::transid_re()
-{
-	return m_transid_re;
-}
-
 void
 BeesRoots::insert_tmpfile(Fd fd)
 {

src/bees-usage.txt

@@ -12,6 +12,7 @@ Load management options:
     -C, --thread-factor   Worker thread factor (default 1)
     -G, --thread-min      Minimum worker thread count (default 0)
     -g, --loadavg-target  Target load average for worker threads (default none)
+        --throttle-factor Idle time between operations (default 1.0)
 
 Filesystem tree traversal options:
     -m, --scan-mode       Scanning mode (0..4, default 4)

src/bees.cc

@@ -220,7 +220,7 @@ bees_readahead_check(int const fd, off_t const offset, size_t const size)
 {
 	// FIXME: the rest of the code calls this function more often than necessary,
 	// usually back-to-back calls on the same range in a loop.
-	// Simply discard requests that are identical to recent requests from the same thread.
+	// Simply discard requests that are identical to recent requests.
 	const Stat stat_rv(fd);
 	auto tup = make_tuple(offset, size, stat_rv.st_dev, stat_rv.st_ino);
 	static mutex s_recent_mutex;
@@ -242,7 +242,7 @@ static
 void
 bees_readahead_nolock(int const fd, const off_t offset, const size_t size)
 {
-	if (!bees_readahead_check(fd, size, offset)) return;
+	if (!bees_readahead_check(fd, offset, size)) return;
 	Timer readahead_timer;
 	BEESNOTE("readahead " << name_fd(fd) << " offset " << to_hex(offset) << " len " << pretty(size));
 	BEESTOOLONG("readahead " << name_fd(fd) << " offset " << to_hex(offset) << " len " << pretty(size));
@@ -278,7 +278,7 @@ static mutex s_only_one;
 void
 bees_readahead_pair(int fd, off_t offset, size_t size, int fd2, off_t offset2, size_t size2)
 {
-	if (!bees_readahead_check(fd, size, offset) && !bees_readahead_check(fd2, offset2, size2)) return;
+	if (!bees_readahead_check(fd, offset, size) && !bees_readahead_check(fd2, offset2, size2)) return;
 	BEESNOTE("waiting to readahead " << name_fd(fd) << " offset " << to_hex(offset) << " len " << pretty(size) << ","
 		<< "\n\t" << name_fd(fd2) << " offset " << to_hex(offset2) << " len " << pretty(size2));
 	unique_lock<mutex> m_lock(s_only_one);
@@ -289,7 +289,7 @@ bees_readahead_pair(int fd, off_t offset, size_t size, int fd2, off_t offset2, s
 void
 bees_readahead(int const fd, const off_t offset, const size_t size)
 {
-	if (!bees_readahead_check(fd, size, offset)) return;
+	if (!bees_readahead_check(fd, offset, size)) return;
 	BEESNOTE("waiting to readahead " << name_fd(fd) << " offset " << to_hex(offset) << " len " << pretty(size));
 	unique_lock<mutex> m_lock(s_only_one);
 	bees_readahead_nolock(fd, offset, size);
@@ -305,6 +305,48 @@ bees_unreadahead(int const fd, off_t offset, size_t size)
 	BEESCOUNTADD(readahead_unread_ms, unreadahead_timer.age() * 1000);
 }
 
+static double bees_throttle_factor = 0.0;
+
+void
+bees_throttle(const double time_used, const char *const context)
+{
+	static mutex s_mutex;
+	unique_lock<mutex> throttle_lock(s_mutex);
+	struct time_pair {
+		double time_used = 0;
+		double time_count = 0;
+		double longest_sleep_time = 0;
+	};
+	static map<string, time_pair> s_time_map;
+	auto &this_time = s_time_map[context];
+	auto &this_time_used = this_time.time_used;
+	auto &this_time_count = this_time.time_count;
+	auto &longest_sleep_time = this_time.longest_sleep_time;
+	this_time_used += time_used;
+	++this_time_count;
+	// Keep the timing data fresh
+	static Timer s_fresh_timer;
+	if (s_fresh_timer.age() > 60) {
+		s_fresh_timer.reset();
+		this_time_count *= 0.9;
+		this_time_used *= 0.9;
+	}
+	// Wait for enough data to calculate rates
+	if (this_time_used < 1.0 || this_time_count < 1.0) return;
+	const auto avg_time = this_time_used / this_time_count;
+	const auto sleep_time = min(60.0, bees_throttle_factor * avg_time - time_used);
+	if (sleep_time <= 0) {
+		return;
+	}
+	if (sleep_time > longest_sleep_time) {
+		BEESLOGDEBUG(context << ": throttle delay " << sleep_time << " s, time used " << time_used << " s, avg time " << avg_time << " s");
+		longest_sleep_time = sleep_time;
+	}
+	throttle_lock.unlock();
+	BEESNOTE(context << ": throttle delay " << sleep_time << " s, time used " << time_used << " s, avg time " << avg_time << " s");
+	nanosleep(sleep_time);
+}
+
 thread_local random_device bees_random_device;
 thread_local uniform_int_distribution<default_random_engine::result_type> bees_random_seed_dist(
 	numeric_limits<default_random_engine::result_type>::min(),
@@ -401,6 +443,8 @@ BeesTempFile::resize(off_t offset)
 
 	// Count time spent here
 	BEESCOUNTADD(tmp_resize_ms, resize_timer.age() * 1000);
+
+	bees_throttle(resize_timer.age(), "tmpfile_resize");
 }
 
 void
@@ -536,6 +580,8 @@ BeesTempFile::make_copy(const BeesFileRange &src)
 	}
 	BEESCOUNTADD(tmp_copy_ms, copy_timer.age() * 1000);
 
+	bees_throttle(copy_timer.age(), "tmpfile_copy");
+
 	BEESCOUNT(tmp_copy);
 	return rv;
 }
@@ -574,19 +620,23 @@ operator<<(ostream &os, const siginfo_t &si)
 
 static sigset_t new_sigset, old_sigset;
 
+static
 void
-block_term_signal()
+block_signals()
 {
 	BEESLOGDEBUG("Masking signals");
 
 	DIE_IF_NON_ZERO(sigemptyset(&new_sigset));
 	DIE_IF_NON_ZERO(sigaddset(&new_sigset, SIGTERM));
 	DIE_IF_NON_ZERO(sigaddset(&new_sigset, SIGINT));
+	DIE_IF_NON_ZERO(sigaddset(&new_sigset, SIGUSR1));
+	DIE_IF_NON_ZERO(sigaddset(&new_sigset, SIGUSR2));
 	DIE_IF_NON_ZERO(sigprocmask(SIG_BLOCK, &new_sigset, &old_sigset));
 }
 
+static
 void
-wait_for_term_signal()
+wait_for_signals()
 {
 	BEESNOTE("waiting for signals");
 	BEESLOGDEBUG("Waiting for signals...");
@@ -603,14 +653,28 @@ wait_for_term_signal()
 			THROW_ERRNO("sigwaitinfo errno = " << errno);
 		} else {
 			BEESLOGNOTICE("Received signal " << rv << " info " << info);
-			// Unblock so we die immediately if signalled again
-			DIE_IF_NON_ZERO(sigprocmask(SIG_BLOCK, &old_sigset, &new_sigset));
-			break;
+			// If SIGTERM or SIGINT, unblock so we die immediately if signalled again
+			switch (info.si_signo) {
+				case SIGUSR1:
+					BEESLOGNOTICE("Received SIGUSR1 - pausing workers");
+					TaskMaster::pause(true);
+					break;
+				case SIGUSR2:
+					BEESLOGNOTICE("Received SIGUSR2 - unpausing workers");
+					TaskMaster::pause(false);
+					break;
+				case SIGTERM:
+				case SIGINT:
+				default:
+					DIE_IF_NON_ZERO(sigprocmask(SIG_BLOCK, &old_sigset, &new_sigset));
+					BEESLOGDEBUG("Signal catcher exiting");
+					return;
+			}
 		}
 	}
-	BEESLOGDEBUG("Signal catcher exiting");
 }
 
+static
 int
 bees_main(int argc, char *argv[])
 {
@@ -634,7 +698,7 @@ bees_main(int argc, char *argv[])
 
 	// Have to block signals now before we create a bunch of threads
 	// so the threads will also have the signals blocked.
-	block_term_signal();
+	block_signals();
 
 	// Create a context so we can apply configuration to it
 	shared_ptr<BeesContext> bc = make_shared<BeesContext>();
@@ -652,29 +716,34 @@ bees_main(int argc, char *argv[])
 	BeesRoots::ScanMode root_scan_mode = BeesRoots::SCAN_MODE_EXTENT;
 
 	// Configure getopt_long
+	// Options with no short form
+	enum {
+		BEES_OPT_THROTTLE_FACTOR = 256,
+	};
 	static const struct option long_options[] = {
-		{ "thread-factor",         required_argument, NULL, 'C' },
-		{ "thread-min",            required_argument, NULL, 'G' },
-		{ "strip-paths",           no_argument,       NULL, 'P' },
-		{ "no-timestamps",         no_argument,       NULL, 'T' },
-		{ "workaround-btrfs-send", no_argument,       NULL, 'a' },
-		{ "thread-count",          required_argument, NULL, 'c' },
-		{ "loadavg-target",        required_argument, NULL, 'g' },
-		{ "help",                  no_argument,       NULL, 'h' },
-		{ "scan-mode",             required_argument, NULL, 'm' },
-		{ "absolute-paths",        no_argument,       NULL, 'p' },
-		{ "timestamps",            no_argument,       NULL, 't' },
-		{ "verbose",               required_argument, NULL, 'v' },
-		{ 0, 0, 0, 0 },
+		{ .name = "thread-factor",         .has_arg = required_argument, .val = 'C' },
+		{ .name = "throttle-factor",       .has_arg = required_argument, .val = BEES_OPT_THROTTLE_FACTOR },
+		{ .name = "thread-min",            .has_arg = required_argument, .val = 'G' },
+		{ .name = "strip-paths",           .has_arg = no_argument,       .val = 'P' },
+		{ .name = "no-timestamps",         .has_arg = no_argument,       .val = 'T' },
+		{ .name = "workaround-btrfs-send", .has_arg = no_argument,       .val = 'a' },
+		{ .name = "thread-count",          .has_arg = required_argument, .val = 'c' },
+		{ .name = "loadavg-target",        .has_arg = required_argument, .val = 'g' },
+		{ .name = "help",                  .has_arg = no_argument,       .val = 'h' },
+		{ .name = "scan-mode",             .has_arg = required_argument, .val = 'm' },
+		{ .name = "absolute-paths",        .has_arg = no_argument,       .val = 'p' },
+		{ .name = "timestamps",            .has_arg = no_argument,       .val = 't' },
+		{ .name = "verbose",               .has_arg = required_argument, .val = 'v' },
+		{ 0 },
 	};
 
 	// Build getopt_long's short option list from the long_options table.
 	// While we're at it, make sure we didn't duplicate any options.
 	string getopt_list;
-	set<decltype(option::val)> option_vals;
+	map<decltype(option::val), string> option_vals;
 	for (const struct option *op = long_options; op->val; ++op) {
-		THROW_CHECK1(runtime_error, op->val, !option_vals.count(op->val));
-		option_vals.insert(op->val);
+		const auto ins_rv = option_vals.insert(make_pair(op->val, op->name));
+		THROW_CHECK1(runtime_error, op->val, ins_rv.second);
 		if ((op->val & 0xff) != op->val) {
 			continue;
 		}
@@ -685,22 +754,26 @@ bees_main(int argc, char *argv[])
 	}
 
 	// Parse options
-	int c;
 	while (true) {
 		int option_index = 0;
 
-		c = getopt_long(argc, argv, getopt_list.c_str(), long_options, &option_index);
+		const auto c = getopt_long(argc, argv, getopt_list.c_str(), long_options, &option_index);
 		if (-1 == c) {
 			break;
 		}
 
-		BEESLOGDEBUG("Parsing option '" << static_cast<char>(c) << "'");
+		// getopt_long should have weeded out any invalid options,
+		// so we can go ahead and throw here
+		BEESLOGDEBUG("Parsing option '" << option_vals.at(c) << "'");
 
 		switch (c) {
 
 			case 'C':
 				thread_factor = stod(optarg);
 				break;
+			case BEES_OPT_THROTTLE_FACTOR:
+				bees_throttle_factor = stod(optarg);
+				break;
 			case 'G':
 				thread_min = stoul(optarg);
 				break;
@@ -741,12 +814,12 @@ bees_main(int argc, char *argv[])
 			case 'h':
 			default:
 				do_cmd_help(argv);
-				return EXIT_FAILURE;
+				return EXIT_SUCCESS;
 		}
 	}
 
 	if (optind + 1 != argc) {
-		BEESLOGERR("Only one filesystem path per bees process");
+		BEESLOGERR("Exactly one filesystem path required");
 		return EXIT_FAILURE;
 	}
 
@@ -786,6 +859,8 @@ bees_main(int argc, char *argv[])
 	BEESLOGNOTICE("setting worker thread pool maximum size to " << thread_count);
 	TaskMaster::set_thread_count(thread_count);
 
+	BEESLOGNOTICE("setting throttle factor to " << bees_throttle_factor);
+
 	// Set root path
 	string root_path = argv[optind++];
 	BEESLOGNOTICE("setting root path to '" << root_path << "'");
@@ -808,7 +883,7 @@ bees_main(int argc, char *argv[])
 	bc->start();
 
 	// Now we just wait forever
-	wait_for_term_signal();
+	wait_for_signals();
 
 	// Shut it down
 	bc->stop();

src/bees.h

@@ -78,10 +78,10 @@ const int BEES_PROGRESS_INTERVAL = BEES_STATS_INTERVAL;
 const int BEES_STATUS_INTERVAL = 1;
 
 // Number of file FDs to cache when not in active use
-const size_t BEES_FILE_FD_CACHE_SIZE = 32768;
+const size_t BEES_FILE_FD_CACHE_SIZE = 524288;
 
 // Number of root FDs to cache when not in active use
-const size_t BEES_ROOT_FD_CACHE_SIZE = 4096;
+const size_t BEES_ROOT_FD_CACHE_SIZE = 65536;
 
 // Number of FDs to open (rlimit)
 const size_t BEES_OPEN_FILE_LIMIT = BEES_FILE_FD_CACHE_SIZE + BEES_ROOT_FD_CACHE_SIZE + 100;
@@ -576,7 +576,6 @@ class BeesRoots : public enable_shared_from_this<BeesRoots> {
 	void writeback_thread();
 	uint64_t next_root(uint64_t root = 0);
 	void current_state_set(const BeesCrawlState &bcs);
-	RateEstimator& transid_re();
 	bool crawl_batch(shared_ptr<BeesCrawl> crawl);
 	void clear_caches();
 
@@ -615,6 +614,8 @@ public:
 
 	uint64_t transid_min();
 	uint64_t transid_max();
+
+	void wait_for_transid(const uint64_t count);
 };
 
 struct BeesHash {
@@ -887,6 +888,7 @@ string pretty(double d);
 void bees_readahead(int fd, off_t offset, size_t size);
 void bees_readahead_pair(int fd, off_t offset, size_t size, int fd2, off_t offset2, size_t size2);
 void bees_unreadahead(int fd, off_t offset, size_t size);
+void bees_throttle(double time_used, const char *context);
 string format_time(time_t t);
 
 #endif