全局池与 Registry

pub(super) struct Registry {
    thread_infos: Vec<ThreadInfo>,
    sleep: Sleep,
    injected_jobs: Injector<JobRef>,
    broadcasts: Mutex<Vec<Worker<JobRef>>>,
    panic_handler: Option<Box<PanicHandler>>,
    start_handler: Option<Box<StartHandler>>,
    exit_handler: Option<Box<ExitHandler>>,

    // When this latch reaches 0, it means that all work on this
    // registry must be complete. This is ensured in the following ways:
    // ...
    terminate_count: AtomicUsize,
}

        let (workers, stealers): (Vec<_>, Vec<_>) = (0..n_threads)
            .map(|_| {
                let worker = if breadth_first {
                    Worker::new_fifo()
                } else {
                    Worker::new_lifo()
                };

                let stealer = worker.stealer();
                (worker, stealer)
            })
            .unzip();

        let (broadcasts, broadcast_stealers): (Vec<_>, Vec<_>) = (0..n_threads)
            .map(|_| {
                let worker = Worker::new_fifo();
                let stealer = worker.stealer();
                (worker, stealer)
            })
            .unzip();

        // If we return early or panic, make sure to terminate existing threads.
        let t1000 = Terminator(&registry);

        for (index, (worker, stealer)) in workers.into_iter().zip(broadcast_stealers).enumerate() {
            let thread = ThreadBuilder {
                name: builder.get_thread_name(index),
                stack_size: builder.get_stack_size(),
                registry: Arc::clone(&registry),
                worker,
                stealer,
                index,
            };

            if index == 0 && builder.use_current_thread {
                // ...（见 ThreadPoolBuilder 一页）
                continue;
            }

            if let Err(e) = builder.get_spawn_handler().spawn(thread) {
                return Err(ThreadPoolBuildError::new(ErrorKind::IOError(e)));
            }
        }

        // Returning normally now, without termination.
        mem::forget(t1000);

static mut THE_REGISTRY: Option<Arc<Registry>> = None;
static THE_REGISTRY_SET: Once = Once::new();

/// Starts the worker threads (if that has not already happened). If
/// initialization has not already occurred, use the default
/// configuration.
pub(super) fn global_registry() -> &'static Arc<Registry> {
    set_global_registry(default_global_registry)
        .or_else(|err| {
            // SAFETY: we only create a shared reference to `THE_REGISTRY` after the `call_once`
            // that initializes it, and there will be no more mutable accesses at all.
            debug_assert!(THE_REGISTRY_SET.is_completed());
            let the_registry = unsafe { &*ptr::addr_of!(THE_REGISTRY) };
            the_registry.as_ref().ok_or(err)
        })
        .expect("The global thread pool has not been initialized.")
}

fn set_global_registry<F>(registry: F) -> Result<&'static Arc<Registry>, ThreadPoolBuildError>
where
    F: FnOnce() -> Result<Arc<Registry>, ThreadPoolBuildError>,
{
    let mut result = Err(ThreadPoolBuildError::new(
        ErrorKind::GlobalPoolAlreadyInitialized,
    ));

    THE_REGISTRY_SET.call_once(|| {
        result = registry().map(|registry: Arc<Registry>| {
            // SAFETY: this is the only mutable access to `THE_REGISTRY`, thanks to `Once`, and
            // `global_registry()` only takes a shared reference **after** this `call_once`.
            unsafe {
                ptr::addr_of_mut!(THE_REGISTRY).write(Some(registry));
                (*ptr::addr_of!(THE_REGISTRY)).as_ref().unwrap_unchecked()
            }
        })
    });

    result
}

join -> (1, 2), threads=12
build_global after join -> Err(ThreadPoolBuildError { kind: GlobalPoolAlreadyInitialized })
Display: The global thread pool has already been initialized.
threads still=12

first build_global -> Ok(()), threads=3
second build_global -> Err(ThreadPoolBuildError { kind: GlobalPoolAlreadyInitialized }), threads=3

fn default_global_registry() -> Result<Arc<Registry>, ThreadPoolBuildError> {
    let result = Registry::new(ThreadPoolBuilder::new());

    // If we're running in an environment that doesn't support threads at all, we can fall back to
    // using the current thread alone. This is crude, and probably won't work for non-blocking
    // calls like `spawn` or `broadcast_spawn`, but a lot of stuff does work fine.
    //
    // Notably, this allows current WebAssembly targets to work even though their threading support
    // is stubbed out, and we won't have to change anything if they do add real threading.
    let unsupported = matches!(&result, Err(e) if e.is_unsupported());
    if unsupported && WorkerThread::current().is_null() {
        let builder = ThreadPoolBuilder::new().num_threads(1).use_current_thread();
        let fallback_result = Registry::new(builder);
        if fallback_result.is_ok() {
            return fallback_result;
        }
    }

    result
}

    pub(super) fn increment_terminate_count(&self) {
        let previous = self.terminate_count.fetch_add(1, Ordering::AcqRel);
        debug_assert!(previous != 0, "registry ref count incremented from zero");
        assert!(previous != usize::MAX, "overflow in registry ref count");
    }

    pub(super) fn terminate(&self) {
        if self.terminate_count.fetch_sub(1, Ordering::AcqRel) == 1 {
            for (i, thread_info) in self.thread_infos.iter().enumerate() {
                unsafe { OnceLatch::set_and_tickle_one(&thread_info.terminate, self, i) };
            }
        }
    }

impl Drop for ThreadPool {
    fn drop(&mut self) {
        self.registry.terminate();
    }
}