trie prefetcher: cleanup blocking get

[jakmeier]

This busy loop was a quick hack to write simpler code under time pressure. Let's replace it with a proper design, such as a Condvar..

nearcore/core/store/src/trie/prefetching_trie_storage.rs

Lines 330 to 339 in aad3bf2

	pub(crate) fn blocking_get(&self, key: CryptoHash) -> Option<Arc<[u8]>> {
	loop {
	match self.0.lock().expect(POISONED_LOCK_ERR).slots.get(&key) {
	Some(PrefetchSlot::Done(value)) => return Some(value.clone()),
	Some(_) => (),
	None => return None,
	}
	std::thread::sleep(std::time::Duration::from_micros(1));
	}
	}

Also, when calling this from a main thread, we should check the shard cache when blocking_get returns None. Due to forks, there can be multiple main threads and in that case the fetched value will be in the shard cache. Only if that also fails should we go and fetch the data from DB.

nearcore/core/store/src/trie/trie_storage.rs

Lines 528 to 538 in aad3bf2

	match prefetcher.prefetching.blocking_get(hash.clone()) {
	Some(value) => value,
	// Only main thread (this one) removes values from staging area,
	// therefore blocking read will usually not return empty unless there
	// was a storage error. Or in the case of forks and parallel chunk
	// processing where one chunk cleans up prefetched data from the other.
	// In any case, we can try again from the main thread.
	None => {
	self.metrics.prefetch_retry.inc();
	self.read_from_db(hash)?
	}

The only other call-site is here:

nearcore/core/store/src/trie/prefetching_trie_storage.rs

Lines 240 to 264 in aad3bf2

	PrefetcherResult::Pending => {
	// yield once before calling `block_get` that will check for data to be present again.
	std::thread::yield_now();
	self.prefetching
	.blocking_get(hash.clone())
	.or_else(|| {
	// `blocking_get` will return None if the prefetch slot has been removed
	// by the main thread and the value inserted into the shard cache.
	let mut guard = self.shard_cache.0.lock().expect(POISONED_LOCK_ERR);
	guard.get(hash)
	})
	.ok_or_else(|| {
	// This could only happen if this thread started prefetching a value
	// while also another thread was already prefetching it. When the
	// other thread finishes, the main thread takes it out, and moves it to
	// the shard cache. And then this current thread gets delayed for long
	// enough that the value gets evicted from the shard cache again before
	// this thread has a chance to read it.
	// In this rare occasion, we shall abort the current prefetch request and
	// move on to the next.
	StorageError::StorageInconsistentState(format!(
	"Prefetcher failed on hash {hash}"
	))
	})
	}

[jakmeier]

to clarify a bit more, there are actually two issues in here:

1. remove the busy loop: refactor: clean up busy waiting in prefetcher blocking get #8215
2. avoid an unnecessary storage lookup from the main thread in the scenario of forks: feat: try reading shard_cache if prefetcher blocking_get returns None #8287