Replace trees computation tasks with a worker

[OlivierHecart]

No description provided.

[YuanYuYuan]

Hi @OlivierHecart, I summarize my comments as below.

Recap

1. TerminatableTask is designed not only for our ZRuntime but also for termination safety.
2. TerminatableTask will terminate itself once it was dropped.
3. If we want to spawn a background task, we can achieve this via ZRuntime::Net::spawn(...).
4. Background tasks spreaded everywhere will make us hard to monitor their status. That's why we adopt a worker pattern in this PR.
5. Repeatedly spawning a task causes some overhead compared to one single worker that is spawned at most once.

Review

1. Maybe we can explicitly use a Worker as below to clarify the content?
2. We can use spawn_abortable if we don't care about the cancellation token.
3. Regarding the TREES_COMPUTATION_DELAY_MS delay, I guess we needed this since we want to slow down the new tasks scheduled on the JoinHandle.
   However, with the new worker pattern, we make sure it would only take the next job (TablesLock) once it finished the computation.

struct HatTables {
    peer_subs: HashSet<Arc<Resource>>,
    peer_qabls: HashSet<Arc<Resource>>,
    peers_net: Option<Network>,
    worker: Worker,
}

struct Worker {
    // NOTE: Unless we need to have a pulic API, it's unecessary to implement Worker::terminate to terminate the task since this is
    // automatically called when Worker is dropped
    _task: TerminatableTask,
    tx: flume::Sender<Arc<TablesLock>>,
}

impl Worker {
    fn new() -> Self {
        let (tx, rx) = flume::bounded::<Arc<TablesLock>>(1);
        // NOTE: it seems that we don't care about the cancellation token, so we can use `spawn_abortable` here
        let task = TerminatableTask::spawn_abortable(zenoh_runtime::ZRuntime::Net, async move {
            loop {
                // NOTE: Why do we need this delay?
                tokio::time::sleep(std::time::Duration::from_millis(
                    *TREES_COMPUTATION_DELAY_MS,
                ))
                .await;

                if let Ok(tables_ref) = rx.recv_async().await {
                    // NOTE: Blocking codes omitted
                }
            }
        });
        Self { _task: task, tx }
    }
}

impl HatTables {
    fn new() -> Self {
        Self {
            peer_subs: HashSet::new(),
            peer_qabls: HashSet::new(),
            peers_net: None,
            worker: Worker::new(),
        }
    }

    fn schedule_compute_trees(&mut self, tables_ref: Arc<TablesLock>) {
        tracing::trace!("Schedule trees computation");
        // NOTE: What if this try_send fails?
        let _ = self.worker.tx.try_send(tables_ref);
    }
}

[OlivierHecart]

About the TREES_COMPUTATION_DELAY_MS:
Routers receive events (LinkState msgs) that modify the network graph and require a tree recomputation. As those computations are expensive we'd like to perform a single computation for multiple events. The TREES_COMPUTATION_DELAY_MS (both in old and new pattern) allows to give a chance for other events to arrive before performing the computation. (Note those events tend to arrive in groups as a change in the graph will induce a LinkState msg from all impacted nodes).

This is why we use a bounded queue of length 1 and try_send method. If a computation is already scheduled, no need to schedule again.

[YuanYuYuan]

LGTM