Commit d1f4e9f3 authored by Ian Jackson's avatar Ian Jackson 💬
Browse files

Merge branch 'pending_error' into 'main' 

Handle panics from circuit construction

Closes #347

See merge request !336
parents b4c0bd6e 3ff9b187

crates/tor-circmgr/src/err.rs

+1 −2
Original line number Diff line number Diff line
@@ -19,7 +19,6 @@ pub enum Error { 
    GuardNotUsable,



    /// We were waiting on a pending circuit, but it failed to report

    /// success _or_ failure.

    #[error("Pending circuit(s) failed without reporting status")]

    PendingCanceled,
@@ -128,7 +127,7 @@ impl HasKind for Error { 
            E::Bug(e) => e.kind(),

            E::NoPath(_) => EK::NoPath,

            E::NoExit(_) => EK::NoExit,

            E::PendingCanceled => EK::Canceled,

            E::PendingCanceled => EK::ReactorShuttingDown,

            E::CircTimeout => EK::TorNetworkTimeout,

            E::GuardNotUsable => EK::TransientFailure,

            E::RequestTimeout => EK::TorNetworkTimeout,

crates/tor-circmgr/src/mgr.rs

+64 −45
Original line number Diff line number Diff line
@@ -27,6 +27,7 @@ use crate::{DirInfo, Error, Result}; 


use retry_error::RetryError;

use tor_config::MutCfg;

use tor_error::internal;

use tor_rtcompat::{Runtime, SleepProviderExt};



use async_trait::async_trait;
@@ -38,6 +39,7 @@ use std::collections::HashMap; 
use std::convert::TryInto;

use std::fmt::Debug;

use std::hash::Hash;

use std::panic::AssertUnwindSafe;

use std::sync::{self, Arc, Weak};

use std::time::{Duration, Instant};

use tracing::{debug, info, warn};
@@ -806,7 +808,7 @@ impl<B: AbstractCircBuilder + 'static, R: Runtime> AbstractCircMgr<B, R> { 
        if let Action::Build(plans) = action {

            for plan in plans {

                let self_clone = Arc::clone(self);

                let _ignore_receiver = self_clone.launch(usage, plan);

                let _ignore_receiver = self_clone.spawn_launch(usage, plan);

            }

        }
@@ -890,7 +892,7 @@ impl<B: AbstractCircBuilder + 'static, R: Runtime> AbstractCircMgr<B, R> { 
                for plan in plans {

                    let self_clone = Arc::clone(&self);

                    // (This is where we actually launch circuits.)

                    futures.push(self_clone.launch(usage, plan));

                    futures.push(self_clone.spawn_launch(usage, plan));

                }

                futures

            }
@@ -1025,14 +1027,14 @@ impl<B: AbstractCircBuilder + 'static, R: Runtime> AbstractCircMgr<B, R> { 
            .expect("Poisoned lock for circuit list")

            .add_pending_circ(pending);



        Ok(Arc::clone(self).launch(usage, plan))

        Ok(Arc::clone(self).spawn_launch(usage, plan))

    }



    /// Actually launch a circuit in a background task.

    /// Spawn a background task to launch a circuit, and report its status.

    ///

    /// The `usage` argument is the usage from the original request that made

    /// us build this circuit.

    fn launch(

    fn spawn_launch(

        self: Arc<Self>,

        usage: &<B::Spec as AbstractSpec>::Usage,

        plan: CircBuildPlan<B>,
@@ -1059,21 +1061,67 @@ impl<B: AbstractCircBuilder + 'static, R: Runtime> AbstractCircMgr<B, R> { 


        runtime

            .spawn(async move {

                let self_clone = Arc::clone(&self);

                let future = AssertUnwindSafe(self_clone.do_launch(plan, pending)).catch_unwind();

                let (new_spec, reply) = match future.await {

                    Ok(x) => x, // Success or regular failure

                    Err(e) => {

                        // Okay, this is a panic.  We have to tell the calling

                        // thread about it, then exit this circuit builder task.

                        let _ = sender.send(Err(internal!("circuit build task panicked").into()));

                        std::panic::panic_any(e);

                    }

                };



                // Tell anybody who was listening about it that this

                // circuit is now usable or failed.

                //

                // (We ignore any errors from `send`: That just means that nobody

                // was waiting for this circuit.)

                let _ = sender.send(reply.clone());



                if let Some(new_spec) = new_spec {

                    // Wait briefly before we notify opportunistically.  This

                    // delay will give the circuits that were originally

                    // specifically intended for a request a little more time

                    // to finish, before we offer it this circuit instead.

                    let sl = runtime_copy.sleep(request_loyalty);

                    runtime_copy.allow_one_advance(request_loyalty);

                    sl.await;



                    let pending = {

                        let list = self.circs.lock().expect("poisoned lock");

                        list.find_pending_requests(&new_spec)

                    };

                    for pending_request in pending {

                        let _ = pending_request.notify.clone().try_send(reply.clone());

                    }

                }

                runtime_copy.release_advance(format!("circuit builder task {}", tid));

            })

            .expect("Couldn't spawn circuit-building task");



        wait_on_future

    }



    /// Run in the background to launch a circuit. Return a 2-tuple of the new

    /// circuit spec and the outcome that should be sent to the initiator.

    async fn do_launch(

        self: Arc<Self>,

        plan: <B as AbstractCircBuilder>::Plan,

        pending: Arc<PendingEntry<B>>,

    ) -> (Option<<B as AbstractCircBuilder>::Spec>, PendResult<B>) {

        let outcome = self.builder.build_circuit(plan).await;



                let (new_spec, reply) = match outcome {

        match outcome {

            Err(e) => (None, Err(e)),

            Ok((new_spec, circ)) => {

                let id = circ.id();



                let use_duration = self.pick_use_duration();

                let exp_inst = self.runtime.now() + use_duration;

                        spawn_expiration_task(

                            &runtime_copy,

                            Arc::downgrade(&self),

                            circ.id(),

                            exp_inst,

                        );

                let runtime_copy = self.runtime.clone();

                spawn_expiration_task(&runtime_copy, Arc::downgrade(&self), circ.id(), exp_inst);

                // I used to call restrict_mut here, but now I'm not so

                // sure. Doing restrict_mut makes sure that this

                // circuit will be suitable for the request that asked
@@ -1100,36 +1148,7 @@ impl<B: AbstractCircBuilder + 'static, R: Runtime> AbstractCircMgr<B, R> { 
                    }

                }

            }

                };

                // Tell anybody who was listening about it that this

                // circuit is now usable or failed.

                //

                // (We ignore any errors from `send`: That just means that nobody

                // was waiting for this circuit.)

                let _ = sender.send(reply.clone());



                if let Some(new_spec) = new_spec {

                    // Wait briefly before we notify opportunistically.  This

                    // delay will give the circuits that were originally

                    // specifically intended for a request a little more time

                    // to finish, before we offer it this circuit instead.

                    let sl = runtime_copy.sleep(request_loyalty);

                    runtime_copy.allow_one_advance(request_loyalty);

                    sl.await;



                    let pending = {

                        let list = self.circs.lock().expect("poisoned lock");

                        list.find_pending_requests(&new_spec)

                    };

                    for pending_request in pending {

                        let _ = pending_request.notify.clone().try_send(reply.clone());

                    }

        }

                runtime_copy.release_advance(format!("circuit builder task {}", tid));

            })

            .expect("Couldn't spawn circuit-building task");



        wait_on_future

    }



    /// Remove the circuit with a given `id` from this manager.