Consolidate concurrency limits (#3493)

## Summary

This PR consolidates the concurrency limits used throughout `uv` and
exposes two limits, `UV_CONCURRENT_DOWNLOADS` and
`UV_CONCURRENT_BUILDS`, as environment variables.

Currently, `uv` has a number of concurrent streams that it buffers using
relatively arbitrary limits for backpressure. However, many of these
limits are conflated. We run a relatively small number of tasks overall
and should start most things as soon as possible. What we really want to
limit are three separate operations:
- File I/O. This is managed by tokio's blocking pool and we should not
really have to worry about it.
- Network I/O.
- Python build processes.

Because the current limits span a broad range of tasks, it's possible
that a limit meant for network I/O is occupied by tasks performing
builds, reading from the file system, or even waiting on a `OnceMap`. We
also don't limit build processes that end up being required to perform a
download. While this may not pose a performance problem because our
limits are relatively high, it does mean that the limits do not do what
we want, making it tricky to expose them to users
(https://github.com/astral-sh/uv/issues/1205,
https://github.com/astral-sh/uv/issues/3311).

After this change, the limits on network I/O and build processes are
centralized and managed by semaphores. All other tasks are unbuffered
(note that these tasks are still bounded, so backpressure should not be
a problem).

crates/uv-dispatch/src/lib.rs

@@ -16,7 +16,9 @@ use distribution_types::{IndexLocations, Name, Requirement, Resolution, SourceDi
 use uv_build::{SourceBuild, SourceBuildContext};
 use uv_cache::Cache;
 use uv_client::RegistryClient;
+use uv_configuration::Concurrency;
 use uv_configuration::{BuildKind, ConfigSettings, NoBinary, NoBuild, Reinstall, SetupPyStrategy};
+use uv_distribution::DistributionDatabase;
 use uv_installer::{Downloader, Installer, Plan, Planner, SitePackages};
 use uv_interpreter::{Interpreter, PythonEnvironment};
 use uv_resolver::{FlatIndex, InMemoryIndex, Manifest, Options, PythonRequirement, Resolver};
@@ -41,6 +43,7 @@ pub struct BuildDispatch<'a> {
     source_build_context: SourceBuildContext,
     options: Options,
     build_extra_env_vars: FxHashMap<OsString, OsString>,
+    concurrency: Concurrency,
 }
 
 impl<'a> BuildDispatch<'a> {
@@ -59,6 +62,7 @@ impl<'a> BuildDispatch<'a> {
         link_mode: install_wheel_rs::linker::LinkMode,
         no_build: &'a NoBuild,
         no_binary: &'a NoBinary,
+        concurrency: Concurrency,
     ) -> Self {
         Self {
             client,
@@ -74,6 +78,7 @@ impl<'a> BuildDispatch<'a> {
             link_mode,
             no_build,
             no_binary,
+            concurrency,
             source_build_context: SourceBuildContext::default(),
             options: Options::default(),
             build_extra_env_vars: FxHashMap::default(),
@@ -144,12 +149,12 @@ impl<'a> BuildContext for BuildDispatch<'a> {
             &python_requirement,
             Some(markers),
             tags,
-            self.client,
             self.flat_index,
             self.index,
             &HashStrategy::None,
             self,
             &EmptyInstalledPackages,
+            DistributionDatabase::new(self.client, self, self.concurrency.downloads),
         )?;
         let graph = resolver.resolve().await.with_context(|| {
             format!(
@@ -226,8 +231,13 @@ impl<'a> BuildContext for BuildDispatch<'a> {
             vec![]
         } else {
             // TODO(konstin): Check that there is no endless recursion.
-            let downloader =
-                Downloader::new(self.cache, tags, &HashStrategy::None, self.client, self);
+            let downloader = Downloader::new(
+                self.cache,
+                tags,
+                &HashStrategy::None,
+                DistributionDatabase::new(self.client, self, self.concurrency.downloads),
+            );
+
             debug!(
                 "Downloading and building requirement{} for build: {}",
                 if remote.len() == 1 { "" } else { "s" },
@@ -315,6 +325,7 @@ impl<'a> BuildContext for BuildDispatch<'a> {
             self.build_isolation,
             build_kind,
             self.build_extra_env_vars.clone(),
+            self.concurrency.builds,
         )
         .boxed_local()
         .await?;