WIP: try to fix halt channel closed after starting UDP tracker

[josecelano]

Hi @da2ce7 @WarmBeer I'm been debugging this error some hours but I can't find the problem.

It seems that the "halt" channel is immediately closed after starting the UDP tracker. So the app panics with:

thread 'tokio-runtime-worker' panicked at src/servers/signals.rs:52:25:
Failed to install stop signal: channel closed

The UDP is started but it's not possible to shut it down gracefully.

I've added a lot of debug lines in this PR:

$ cargo run
   Compiling torrust-tracker v3.0.0-alpha.12-develop (/home/josecelano/Documents/git/committer/me/github/torrust/torrust-tracker)
    Finished dev [optimized + debuginfo] target(s) in 7.05s
     Running `target/debug/torrust-tracker`
Loading default configuration file: `./share/default/config/tracker.development.sqlite3.toml` ...
2024-01-10T17:46:40.348930411+00:00 [torrust_tracker::bootstrap::logging][INFO] logging initialized.
2024-01-10T17:46:40.349651636+00:00 [UDP Tracker][DEBUG] Launcher starting ...
2024-01-10T17:46:40.349677556+00:00 [UDP Tracker][INFO] Starting on: udp://0.0.0.0:6969
2024-01-10T17:46:40.349682046+00:00 [UDP Tracker][INFO] Started on: udp://0.0.0.0:6969
2024-01-10T17:46:40.349684556+00:00 [UDP Tracker][DEBUG] Launcher finished ...
2024-01-10T17:46:40.349687856+00:00 [UDP Tracker][DEBUG] Waiting for packets ...
2024-01-10T17:46:40.349687736+00:00 [torrust_tracker::bootstrap::jobs::http_tracker][INFO] Note: Not loading Http Tracker Service, Not Enabled in Configuration.
2024-01-10T17:46:40.349691306+00:00 [UDP Tracker][DEBUG] Wait for launcher (UDP service) to finish ...
2024-01-10T17:46:40.349689536+00:00 [UDP Tracker][DEBUG] Server on socket address: udp://0.0.0.0:6969 waiting for halt signal ...
2024-01-10T17:46:40.349700506+00:00 [torrust_tracker::bootstrap::jobs::health_check_api][INFO] Starting Health Check API server: http://127.0.0.1:1313
thread 'tokio-runtime-worker' panicked at src/servers/signals.rs:52:25:
Failed to install stop signal: channel closed
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace
2024-01-10T17:46:40.349793665+00:00 [torrust_tracker::bootstrap::jobs::health_check_api][INFO] Torrust Health Check API server started on: http://127.0.0.1:1313
2024-01-10T17:46:40.349793225+00:00 [UDP Tracker][DEBUG] Halt signal spawned task stopped on address: udp://0.0.0.0:6969

The function panicking is:

pub async fn shutdown_signal(rx_halt: tokio::sync::oneshot::Receiver<Halted>) {
    let halt = async {
        match rx_halt.await {
            Ok(signal) => signal,
            Err(err) => panic!("Failed to install stop signal: {err}"),
        }
    };

    tokio::select! {
        signal = halt => { info!("Halt signal processed: {}", signal) },
        () = global_shutdown_signal() => { info!("Global shutdown signal processed") }
    }
}

When the thread starts waiting for the signal it receives the error "channel closed".

I fixed a similar problem for the Tracker API and the HTTP Tracker:

• Fix halt channel closed after starting HTTP tracker #593
• Fix broken grateful shutdown for tracker API #589

Apparently the problem was the sender in the channel was dropped (I don't know why, it looks like a compiler optimization)

Are we using this feature (sending a halt signal to the services to stop them). I think we are not using it yet and we do not have tests. Maybe I should just remove it. It's an interesting topic but it might take long to solve it.

Could you take a look? Maybe I'm missing something obvious.

The only problem with this bug it's the UDP is forced to close when we close the app and we don't see anything in the logs. But maybe it does not make sense to use the signal if we do nothing with it.

[codecov]

Codecov Report

Attention: 14 lines in your changes are missing coverage. Please review.

Comparison is base (49c961c) 80.64% compared to head (75c07ff) 80.59%.

Files	Patch %	Lines
src/servers/udp/server.rs	70.27%	11 Missing ⚠️
src/bootstrap/jobs/udp_tracker.rs	0.00%	3 Missing ⚠️

Additional details and impacted files

@@             Coverage Diff             @@
##           develop     #594      +/-   ##
===========================================
- Coverage    80.64%   80.59%   -0.05%     
===========================================
  Files          117      117              
  Lines         7831     7853      +22     
===========================================
+ Hits          6315     6329      +14     
- Misses        1516     1524       +8     

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

[da2ce7]

@josecelano

My branch uses the following udp service management code: (for reference)

impl UdpServer<Stopped>

pub async fn start(self, tracker: Arc<Tracker>, form: RegistrationForm) -> Result<UdpServer<Running>, Error> {
    let (tx_start, rx_start) = tokio::sync::oneshot::channel::<Started>();
    let (tx_halt, rx_halt) = tokio::sync::oneshot::channel::<Halted>();

    let task = self.state.launcher.start(tracker, tx_start, rx_halt);

    let binding = rx_start.await.expect("it should be able to start the service").address;

    form.send(Listing::new(binding, super::check))
        .expect("it should be able to send service registration");

    let running_udp_server: UdpServer<Running> = UdpServer {
        state: Running {
            binding,
            halt_task: tx_halt,
            task,
        },
    };

    trace!("Running UDP Tracker on Socket: {}", running_udp_server.state.binding);

    Ok(running_udp_server)
}

impl Launcher

pub fn start(
    &self,
    tracker: Arc<Tracker>,
    tx_start: oneshot::Sender<Started>,
    rx_halt: oneshot::Receiver<Halted>,
) -> JoinHandle<Launcher> {
    let launcher = Launcher::new(self.bind_to);
    tokio::spawn(async move {
        Udp::run_with_graceful_shutdown(tracker, launcher.bind_to, tx_start, rx_halt).await;
        launcher
    })
}

struct ActiveRequests

struct ActiveRequests {
    rb: LocalRb<Static<AbortHandle, 50>>, // the number of requests we handle at the same time.
}

impl Drop for ActiveRequests {
    fn drop(&mut self) {
        for h in self.rb.pop_iter() {
            if !h.is_finished() {
                h.abort();
            }
        }
    }
}

impl Udp

async fn run_with_graceful_shutdown(
    tracker: Arc<Tracker>,
    bind_to: SocketAddr,
    tx_start: oneshot::Sender<Started>,
    rx_halt: oneshot::Receiver<Halted>,
) {
    let socket = Arc::new(UdpSocket::bind(bind_to).await.expect("Could not bind to {self.socket}."));
    let address = socket.local_addr().expect("Could not get local_addr from {binding}.");
    let halt = shutdown_signal_with_message(rx_halt, format!("Halting Http Service Bound to Socket: {address}"));

    let running = tokio::task::spawn(async move {
        let tracker = tracker.clone();
        let socket = socket.clone();

        let reqs = &mut ActiveRequests::default();

        loop {
            let old = reqs.rb.push_overwrite(
                Self::do_request(Self::receive_request(socket.clone()).await, tracker.clone(), socket.clone()).abort_handle(),
            );

            if let Some(h) = old {
                if !h.is_finished() {
                    h.abort();
                }
            }
        }
    });

    tx_start
        .send(Started { address })
        .expect("the UDP Tracker service should not be dropped");

    let stop = running.abort_handle();

    select! {
        _ = running => {},
        () = halt => {}
    }
    stop.abort();

    task::yield_now().await; // lets allow the other threads to complete.
}

This version seems to work well...

[josecelano]

@josecelano

My branch uses the following udp service management code: (for reference)

impl UdpServer<Stopped>

pub async fn start(self, tracker: Arc<Tracker>, form: RegistrationForm) -> Result<UdpServer<Running>, Error> {
    let (tx_start, rx_start) = tokio::sync::oneshot::channel::<Started>();
    let (tx_halt, rx_halt) = tokio::sync::oneshot::channel::<Halted>();

    let task = self.state.launcher.start(tracker, tx_start, rx_halt);

    let binding = rx_start.await.expect("it should be able to start the service").address;

    form.send(Listing::new(binding, super::check))
        .expect("it should be able to send service registration");

    let running_udp_server: UdpServer<Running> = UdpServer {
        state: Running {
            binding,
            halt_task: tx_halt,
            task,
        },
    };

    trace!("Running UDP Tracker on Socket: {}", running_udp_server.state.binding);

    Ok(running_udp_server)
}

impl Launcher

pub fn start(
    &self,
    tracker: Arc<Tracker>,
    tx_start: oneshot::Sender<Started>,
    rx_halt: oneshot::Receiver<Halted>,
) -> JoinHandle<Launcher> {
    let launcher = Launcher::new(self.bind_to);
    tokio::spawn(async move {
        Udp::run_with_graceful_shutdown(tracker, launcher.bind_to, tx_start, rx_halt).await;
        launcher
    })
}

struct ActiveRequests

struct ActiveRequests {
    rb: LocalRb<Static<AbortHandle, 50>>, // the number of requests we handle at the same time.
}

impl Drop for ActiveRequests {
    fn drop(&mut self) {
        for h in self.rb.pop_iter() {
            if !h.is_finished() {
                h.abort();
            }
        }
    }
}

impl Udp

async fn run_with_graceful_shutdown(
    tracker: Arc<Tracker>,
    bind_to: SocketAddr,
    tx_start: oneshot::Sender<Started>,
    rx_halt: oneshot::Receiver<Halted>,
) {
    let socket = Arc::new(UdpSocket::bind(bind_to).await.expect("Could not bind to {self.socket}."));
    let address = socket.local_addr().expect("Could not get local_addr from {binding}.");
    let halt = shutdown_signal_with_message(rx_halt, format!("Halting Http Service Bound to Socket: {address}"));

    let running = tokio::task::spawn(async move {
        let tracker = tracker.clone();
        let socket = socket.clone();

        let reqs = &mut ActiveRequests::default();

        loop {
            let old = reqs.rb.push_overwrite(
                Self::do_request(Self::receive_request(socket.clone()).await, tracker.clone(), socket.clone()).abort_handle(),
            );

            if let Some(h) = old {
                if !h.is_finished() {
                    h.abort();
                }
            }
        }
    });

    tx_start
        .send(Started { address })
        .expect("the UDP Tracker service should not be dropped");

    let stop = running.abort_handle();

    select! {
        _ = running => {},
        () = halt => {}
    }
    stop.abort();

    task::yield_now().await; // lets allow the other threads to complete.
}

This version seems to work well...

Hi @da2ce7 OK. Then, it does not make sense I continue working on the issue.

[josecelano]

Hi @da2ce7 could you tell me what branch is that?

[josecelano]

Hi @da2ce7 in the end I decided to re-open this PR to fix the problem even if you are working on a better solution in the long-term. I want to fix problems in the live demo asap. I did some changes similar to want you are doing alhougth the graceful shutdown is not implemented yet.