Fix places where tokio is blocked

[0xdeafbeef]

tycho/collator/src/manager/mod.rs

Lines 465 to 466 in 136466b

	self.mpool_adapter
	.clear_anchors_cache(top_processed_to_anchor)?;

tycho/collator/src/collator/do_collate/mod.rs

Lines 998 to 1013 in 136466b

	// spawn update PrevData and working state
	self.prepare_working_state_update(
	block_id,
	finalized.new_observable_state,
	finalized.new_state_root,
	finalized.state_update,
	store_new_state_task,
	new_queue_diff_hash,
	new_mc_data,
	collation_config,
	has_unprocessed_messages,
	reader_state,
	ref_mc_state_handle,
	resume_collation_elapsed,
	)?;

tycho/collator/src/state_node.rs

Lines 295 to 365 in 136466b

	async fn handle_state(&self, state: &ShardStateStuff) -> Result<()> {
	let _histogram = HistogramGuard::begin("tycho_collator_state_adapter_handle_state_time");
	let sync_context = *self.sync_context_tx.borrow();
	tracing::debug!(target: tracing_targets::STATE_NODE_ADAPTER, "handle_state: block {}", state.block_id());
	let block_id = *state.block_id();
	let mut to_split = Vec::new();
	let shard = block_id.shard;
	let seqno = block_id.seqno;
	{
	let has_block = if let Some(shard_blocks) = self.blocks.get(&shard) {
	let has_block = shard_blocks.contains_key(&seqno);
	if shard.is_masterchain() {
	let prev_mc_block = shard_blocks
	.range(..=seqno)
	.rev()
	.find_map(|(&key, value)| if key < seqno { Some(value) } else { None });
	if let Some(prev_mc_block) = prev_mc_block {
	for id in &prev_mc_block.top_shard_blocks_ids {
	to_split.push((id.shard, id.seqno + 1));
	}
	to_split.push((shard, prev_mc_block.block_stuff_aug.id().seqno + 1));
	}
	}
	has_block
	} else {
	false
	};
	self.delayed_state_notifier
	.send_or_delay(
	self.listener.clone(),
	state.clone(),
	!has_block,
	sync_context,
	|sync_ctx| {
	let check = sync_ctx == CollatorSyncContext::Recent;
	if !check {
	tracing::debug!(
	target: tracing_targets::STATE_NODE_ADAPTER,
	block_id = %state.block_id().as_short_id(),
	sync_ctx = ?sync_context,
	"handle_state: will delay state",
	);
	}
	check
	},
	)
	.await?;
	}
	let mut to_drop = Vec::new();
	for (shard, seqno) in &to_split {
	if let Some(mut shard_blocks) = self.blocks.get_mut(shard) {
	let retained_blocks = shard_blocks.split_off(seqno);
	to_drop.push(std::mem::replace(&mut *shard_blocks, retained_blocks));
	}
	}
	// Don't wait for drop inside a tokio context.
	Reclaimer::instance().drop(to_drop);
	Ok(())
	}