use std::cell::{Cell, Ref, RefCell, RefMut};
use std::cmp::Ordering;
use std::collections::{HashMap, HashSet};
use std::fmt;
use std::hash;
use std::mem;
use std::path::{Path, PathBuf};
use std::time::{Duration, Instant};

use bytesize::ByteSize;
use curl::easy::{Easy, HttpVersion};
use curl::multi::{EasyHandle, Multi};
use failure::ResultExt;
use lazycell::LazyCell;
use log::{debug, warn};
use semver::Version;
use serde::ser;
use serde::Serialize;

use crate::core::interning::InternedString;
use crate::core::source::MaybePackage;
use crate::core::{Dependency, Manifest, PackageId, SourceId, Target};
use crate::core::{FeatureMap, SourceMap, Summary};
use crate::ops;
use crate::util::config::PackageCacheLock;
use crate::util::errors::{CargoResult, CargoResultExt, HttpNot200};
use crate::util::network::Retry;
use crate::util::{self, internal, Config, Progress, ProgressStyle};

/// Information about a package that is available somewhere in the file system.
///
/// A package is a `Cargo.toml` file plus all the files that are part of it.
//
// TODO: is `manifest_path` a relic?
#[derive(Clone)]
pub struct Package {
    /// The package's manifest.
    manifest: Manifest,
    /// The root of the package.
    manifest_path: PathBuf,
}

impl Ord for Package {
    fn cmp(&self, other: &Package) -> Ordering {
        self.package_id().cmp(&other.package_id())
    }
}

impl PartialOrd for Package {
    fn partial_cmp(&self, other: &Package) -> Option<Ordering> {
        Some(self.cmp(other))
    }
}

/// A Package in a form where `Serialize` can be derived.
#[derive(Serialize)]
struct SerializedPackage<'a> {
    name: &'a str,
    version: &'a Version,
    id: PackageId,
    license: Option<&'a str>,
    license_file: Option<&'a str>,
    description: Option<&'a str>,
    source: SourceId,
    dependencies: &'a [Dependency],
    targets: Vec<&'a Target>,
    features: &'a FeatureMap,
    manifest_path: &'a Path,
    metadata: Option<&'a toml::Value>,
    publish: Option<&'a Vec<String>>,
    authors: &'a [String],
    categories: &'a [String],
    keywords: &'a [String],
    readme: Option<&'a str>,
    repository: Option<&'a str>,
    edition: &'a str,
    links: Option<&'a str>,
    #[serde(skip_serializing_if = "Option::is_none")]
    metabuild: Option<&'a Vec<String>>,
}

impl ser::Serialize for Package {
    fn serialize<S>(&self, s: S) -> Result<S::Ok, S::Error>
    where
        S: ser::Serializer,
    {
        let summary = self.manifest.summary();
        let package_id = summary.package_id();
        let manmeta = self.manifest.metadata();
        let license = manmeta.license.as_ref().map(String::as_ref);
        let license_file = manmeta.license_file.as_ref().map(String::as_ref);
        let description = manmeta.description.as_ref().map(String::as_ref);
        let authors = manmeta.authors.as_ref();
        let categories = manmeta.categories.as_ref();
        let keywords = manmeta.keywords.as_ref();
        let readme = manmeta.readme.as_ref().map(String::as_ref);
        let repository = manmeta.repository.as_ref().map(String::as_ref);
        // Filter out metabuild targets. They are an internal implementation
        // detail that is probably not relevant externally. There's also not a
        // real path to show in `src_path`, and this avoids changing the format.
        let targets: Vec<&Target> = self
            .manifest
            .targets()
            .iter()
            .filter(|t| t.src_path().is_path())
            .collect();

        SerializedPackage {
            name: &*package_id.name(),
            version: package_id.version(),
            id: package_id,
            license,
            license_file,
            description,
            source: summary.source_id(),
            dependencies: summary.dependencies(),
            targets,
            features: summary.features(),
            manifest_path: &self.manifest_path,
            metadata: self.manifest.custom_metadata(),
            authors,
            categories,
            keywords,
            readme,
            repository,
            edition: &self.manifest.edition().to_string(),
            links: self.manifest.links(),
            metabuild: self.manifest.metabuild(),
            publish: self.publish().as_ref(),
        }
        .serialize(s)
    }
}

impl Package {
    /// Creates a package from a manifest and its location.
    pub fn new(manifest: Manifest, manifest_path: &Path) -> Package {
        Package {
            manifest,
            manifest_path: manifest_path.to_path_buf(),
        }
    }

    /// Gets the manifest dependencies.
    pub fn dependencies(&self) -> &[Dependency] {
        self.manifest.dependencies()
    }
    /// Gets the manifest.
    pub fn manifest(&self) -> &Manifest {
        &self.manifest
    }
    /// Gets the manifest.
    pub fn manifest_mut(&mut self) -> &mut Manifest {
        &mut self.manifest
    }
    /// Gets the path to the manifest.
    pub fn manifest_path(&self) -> &Path {
        &self.manifest_path
    }
    /// Gets the name of the package.
    pub fn name(&self) -> InternedString {
        self.package_id().name()
    }
    /// Gets the `PackageId` object for the package (fully defines a package).
    pub fn package_id(&self) -> PackageId {
        self.manifest.package_id()
    }
    /// Gets the root folder of the package.
    pub fn root(&self) -> &Path {
        self.manifest_path.parent().unwrap()
    }
    /// Gets the summary for the package.
    pub fn summary(&self) -> &Summary {
        self.manifest.summary()
    }
    /// Gets the targets specified in the manifest.
    pub fn targets(&self) -> &[Target] {
        self.manifest.targets()
    }
    /// Gets the current package version.
    pub fn version(&self) -> &Version {
        self.package_id().version()
    }
    /// Gets the package authors.
    pub fn authors(&self) -> &Vec<String> {
        &self.manifest.metadata().authors
    }
    /// Returns `true` if the package is set to publish.
    pub fn publish(&self) -> &Option<Vec<String>> {
        self.manifest.publish()
    }

    /// Returns `true` if the package uses a custom build script for any target.
    pub fn has_custom_build(&self) -> bool {
        self.targets().iter().any(|t| t.is_custom_build())
    }

    pub fn map_source(self, to_replace: SourceId, replace_with: SourceId) -> Package {
        Package {
            manifest: self.manifest.map_source(to_replace, replace_with),
            manifest_path: self.manifest_path,
        }
    }

    pub fn to_registry_toml(&self, config: &Config) -> CargoResult<String> {
        let manifest = self.manifest().original().prepare_for_publish(config)?;
        let toml = toml::to_string(&manifest)?;
        Ok(format!(
            "# THIS FILE IS AUTOMATICALLY GENERATED BY CARGO\n\
             #\n\
             # When uploading crates to the registry Cargo will automatically\n\
             # \"normalize\" Cargo.toml files for maximal compatibility\n\
             # with all versions of Cargo and also rewrite `path` dependencies\n\
             # to registry (e.g., crates.io) dependencies\n\
             #\n\
             # If you believe there's an error in this file please file an\n\
             # issue against the rust-lang/cargo repository. If you're\n\
             # editing this file be aware that the upstream Cargo.toml\n\
             # will likely look very different (and much more reasonable)\n\
             \n\
             {}\
             ",
            toml
        ))
    }

    /// Returns if package should include `Cargo.lock`.
    pub fn include_lockfile(&self) -> bool {
        self.targets().iter().any(|t| t.is_example() || t.is_bin())
    }
}

impl fmt::Display for Package {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "{}", self.summary().package_id())
    }
}

impl fmt::Debug for Package {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("Package")
            .field("id", &self.summary().package_id())
            .field("..", &"..")
            .finish()
    }
}

impl PartialEq for Package {
    fn eq(&self, other: &Package) -> bool {
        self.package_id() == other.package_id()
    }
}

impl Eq for Package {}

impl hash::Hash for Package {
    fn hash<H: hash::Hasher>(&self, into: &mut H) {
        self.package_id().hash(into)
    }
}

/// A set of packages, with the intent to download.
///
/// This is primarily used to convert a set of `PackageId`s to `Package`s. It
/// will download as needed, or used the cached download if available.
pub struct PackageSet<'cfg> {
    packages: HashMap<PackageId, LazyCell<Package>>,
    sources: RefCell<SourceMap<'cfg>>,
    config: &'cfg Config,
    multi: Multi,
    /// Used to prevent reusing the PackageSet to download twice.
    downloading: Cell<bool>,
    /// Whether or not to use curl HTTP/2 multiplexing.
    multiplexing: bool,
}

/// Helper for downloading crates.
pub struct Downloads<'a, 'cfg> {
    set: &'a PackageSet<'cfg>,
    /// When a download is started, it is added to this map. The key is a
    /// "token" (see `Download::token`). It is removed once the download is
    /// finished.
    pending: HashMap<usize, (Download<'cfg>, EasyHandle)>,
    /// Set of packages currently being downloaded. This should stay in sync
    /// with `pending`.
    pending_ids: HashSet<PackageId>,
    /// The final result of each download. A pair `(token, result)`. This is a
    /// temporary holding area, needed because curl can report multiple
    /// downloads at once, but the main loop (`wait`) is written to only
    /// handle one at a time.
    results: Vec<(usize, Result<(), curl::Error>)>,
    /// The next ID to use for creating a token (see `Download::token`).
    next: usize,
    /// Progress bar.
    progress: RefCell<Option<Progress<'cfg>>>,
    /// Number of downloads that have successfully finished.
    downloads_finished: usize,
    /// Total bytes for all successfully downloaded packages.
    downloaded_bytes: u64,
    /// Size (in bytes) and package name of the largest downloaded package.
    largest: (u64, String),
    /// Time when downloading started.
    start: Instant,
    /// Indicates *all* downloads were successful.
    success: bool,

    /// Timeout management, both of timeout thresholds as well as whether or not
    /// our connection has timed out (and accompanying message if it has).
    ///
    /// Note that timeout management is done manually here instead of in libcurl
    /// because we want to apply timeouts to an entire batch of operations, not
    /// any one particular single operation.
    timeout: ops::HttpTimeout,
    /// Last time bytes were received.
    updated_at: Cell<Instant>,
    /// This is a slow-speed check. It is reset to `now + timeout_duration`
    /// every time at least `threshold` bytes are received. If the current
    /// time ever exceeds `next_speed_check`, then give up and report a
    /// timeout error.
    next_speed_check: Cell<Instant>,
    /// This is the slow-speed threshold byte count. It starts at the
    /// configured threshold value (default 10), and is decremented by the
    /// number of bytes received in each chunk. If it is <= zero, the
    /// threshold has been met and data is being received fast enough not to
    /// trigger a timeout; reset `next_speed_check` and set this back to the
    /// configured threshold.
    next_speed_check_bytes_threshold: Cell<u64>,
    /// Global filesystem lock to ensure only one Cargo is downloading at a
    /// time.
    _lock: PackageCacheLock<'cfg>,
}

struct Download<'cfg> {
    /// The token for this download, used as the key of the `Downloads::pending` map
    /// and stored in `EasyHandle` as well.
    token: usize,

    /// The package that we're downloading.
    id: PackageId,

    /// Actual downloaded data, updated throughout the lifetime of this download.
    data: RefCell<Vec<u8>>,

    /// The URL that we're downloading from, cached here for error messages and
    /// reenqueuing.
    url: String,

    /// A descriptive string to print when we've finished downloading this crate.
    descriptor: String,

    /// Statistics updated from the progress callback in libcurl.
    total: Cell<u64>,
    current: Cell<u64>,

    /// The moment we started this transfer at.
    start: Instant,
    timed_out: Cell<Option<String>>,

    /// Logic used to track retrying this download if it's a spurious failure.
    retry: Retry<'cfg>,
}

impl<'cfg> PackageSet<'cfg> {
    pub fn new(
        package_ids: &[PackageId],
        sources: SourceMap<'cfg>,
        config: &'cfg Config,
    ) -> CargoResult<PackageSet<'cfg>> {
        // We've enabled the `http2` feature of `curl` in Cargo, so treat
        // failures here as fatal as it would indicate a build-time problem.
        //
        // Note that the multiplexing support is pretty new so we're having it
        // off-by-default temporarily.
        //
        // Also note that pipelining is disabled as curl authors have indicated
        // that it's buggy, and we've empirically seen that it's buggy with HTTP
        // proxies.
        let mut multi = Multi::new();
        let multiplexing = config.http_config()?.multiplexing.unwrap_or(true);
        multi
            .pipelining(false, multiplexing)
            .chain_err(|| "failed to enable multiplexing/pipelining in curl")?;

        // let's not flood crates.io with connections
        multi.set_max_host_connections(2)?;

        Ok(PackageSet {
            packages: package_ids
                .iter()
                .map(|&id| (id, LazyCell::new()))
                .collect(),
            sources: RefCell::new(sources),
            config,
            multi,
            downloading: Cell::new(false),
            multiplexing,
        })
    }

    pub fn package_ids<'a>(&'a self) -> impl Iterator<Item = PackageId> + 'a {
        self.packages.keys().cloned()
    }

    pub fn enable_download<'a>(&'a self) -> CargoResult<Downloads<'a, 'cfg>> {
        assert!(!self.downloading.replace(true));
        let timeout = ops::HttpTimeout::new(self.config)?;
        Ok(Downloads {
            start: Instant::now(),
            set: self,
            next: 0,
            pending: HashMap::new(),
            pending_ids: HashSet::new(),
            results: Vec::new(),
            progress: RefCell::new(Some(Progress::with_style(
                "Downloading",
                ProgressStyle::Ratio,
                self.config,
            ))),
            downloads_finished: 0,
            downloaded_bytes: 0,
            largest: (0, String::new()),
            success: false,
            updated_at: Cell::new(Instant::now()),
            timeout,
            next_speed_check: Cell::new(Instant::now()),
            next_speed_check_bytes_threshold: Cell::new(0),
            _lock: self.config.acquire_package_cache_lock()?,
        })
    }

    pub fn get_one(&self, id: PackageId) -> CargoResult<&Package> {
        Ok(self.get_many(Some(id))?.remove(0))
    }

    pub fn get_many(&self, ids: impl IntoIterator<Item = PackageId>) -> CargoResult<Vec<&Package>> {
        let mut pkgs = Vec::new();
        let mut downloads = self.enable_download()?;
        for id in ids {
            pkgs.extend(downloads.start(id)?);
        }
        while downloads.remaining() > 0 {
            pkgs.push(downloads.wait()?);
        }
        downloads.success = true;
        Ok(pkgs)
    }

    pub fn sources(&self) -> Ref<'_, SourceMap<'cfg>> {
        self.sources.borrow()
    }

    pub fn sources_mut(&self) -> RefMut<'_, SourceMap<'cfg>> {
        self.sources.borrow_mut()
    }

    /// Merge the given set into self.
    pub fn add_set(&mut self, set: PackageSet<'cfg>) {
        assert!(!self.downloading.get());
        assert!(!set.downloading.get());
        for (pkg_id, p_cell) in set.packages {
            self.packages.entry(pkg_id).or_insert(p_cell);
        }
        let mut sources = self.sources.borrow_mut();
        let other_sources = set.sources.into_inner();
        sources.add_source_map(other_sources);
    }

    /// Get mutable access to an already downloaded package, if it's already
    /// downoaded and it's part of this set. Does not actually attempt to
    /// download anything if it's not already downloaded.
    pub fn lookup_mut(&mut self, id: PackageId) -> Option<&mut Package> {
        self.packages
            .get_mut(&id)
            .and_then(|cell| cell.borrow_mut())
    }
}

// When dynamically linked against libcurl, we want to ignore some failures
// when using old versions that don't support certain features.
macro_rules! try_old_curl {
    ($e:expr, $msg:expr) => {
        let result = $e;
        if cfg!(target_os = "macos") {
            if let Err(e) = result {
                warn!("ignoring libcurl {} error: {}", $msg, e);
            }
        } else {
            result.with_context(|_| {
                failure::format_err!("failed to enable {}, is curl not built right?", $msg)
            })?;
        }
    };
}

impl<'a, 'cfg> Downloads<'a, 'cfg> {
    /// Starts to download the package for the `id` specified.
    ///
    /// Returns `None` if the package is queued up for download and will
    /// eventually be returned from `wait_for_download`. Returns `Some(pkg)` if
    /// the package is ready and doesn't need to be downloaded.
    pub fn start(&mut self, id: PackageId) -> CargoResult<Option<&'a Package>> {
        Ok(self
            .start_inner(id)
            .chain_err(|| format!("failed to download `{}`", id))?)
    }

    fn start_inner(&mut self, id: PackageId) -> CargoResult<Option<&'a Package>> {
        // First up see if we've already cached this package, in which case
        // there's nothing to do.
        let slot = self
            .set
            .packages
            .get(&id)
            .ok_or_else(|| internal(format!("couldn't find `{}` in package set", id)))?;
        if let Some(pkg) = slot.borrow() {
            return Ok(Some(pkg));
        }

        // Ask the original source fo this `PackageId` for the corresponding
        // package. That may immediately come back and tell us that the package
        // is ready, or it could tell us that it needs to be downloaded.
        let mut sources = self.set.sources.borrow_mut();
        let source = sources
            .get_mut(id.source_id())
            .ok_or_else(|| internal(format!("couldn't find source for `{}`", id)))?;
        let pkg = source
            .download(id)
            .chain_err(|| failure::format_err!("unable to get packages from source"))?;
        let (url, descriptor) = match pkg {
            MaybePackage::Ready(pkg) => {
                debug!("{} doesn't need a download", id);
                assert!(slot.fill(pkg).is_ok());
                return Ok(Some(slot.borrow().unwrap()));
            }
            MaybePackage::Download { url, descriptor } => (url, descriptor),
        };

        // Ok we're going to download this crate, so let's set up all our
        // internal state and hand off an `Easy` handle to our libcurl `Multi`
        // handle. This won't actually start the transfer, but later it'll
        // happen during `wait_for_download`
        let token = self.next;
        self.next += 1;
        debug!("downloading {} as {}", id, token);
        assert!(self.pending_ids.insert(id));

        let (mut handle, _timeout) = ops::http_handle_and_timeout(self.set.config)?;
        handle.get(true)?;
        handle.url(&url)?;
        handle.follow_location(true)?; // follow redirects

        // Enable HTTP/2 to be used as it'll allow true multiplexing which makes
        // downloads much faster.
        //
        // Currently Cargo requests the `http2` feature of the `curl` crate
        // which means it should always be built in. On OSX, however, we ship
        // cargo still linked against the system libcurl. Building curl with
        // ALPN support for HTTP/2 requires newer versions of OSX (the
        // SecureTransport API) than we want to ship Cargo for. By linking Cargo
        // against the system libcurl then older curl installations won't use
        // HTTP/2 but newer ones will. All that to basically say we ignore
        // errors here on OSX, but consider this a fatal error to not activate
        // HTTP/2 on all other platforms.
        if self.set.multiplexing {
            try_old_curl!(handle.http_version(HttpVersion::V2), "HTTP2");
        } else {
            handle.http_version(HttpVersion::V11)?;
        }

        // This is an option to `libcurl` which indicates that if there's a
        // bunch of parallel requests to the same host they all wait until the
        // pipelining status of the host is known. This means that we won't
        // initiate dozens of connections to crates.io, but rather only one.
        // Once the main one is opened we realized that pipelining is possible
        // and multiplexing is possible with static.crates.io. All in all this
        // reduces the number of connections done to a more manageable state.
        try_old_curl!(handle.pipewait(true), "pipewait");

        handle.write_function(move |buf| {
            debug!("{} - {} bytes of data", token, buf.len());
            tls::with(|downloads| {
                if let Some(downloads) = downloads {
                    downloads.pending[&token]
                        .0
                        .data
                        .borrow_mut()
                        .extend_from_slice(buf);
                }
            });
            Ok(buf.len())
        })?;

        handle.progress(true)?;
        handle.progress_function(move |dl_total, dl_cur, _, _| {
            tls::with(|downloads| match downloads {
                Some(d) => d.progress(token, dl_total as u64, dl_cur as u64),
                None => false,
            })
        })?;

        // If the progress bar isn't enabled then it may be awhile before the
        // first crate finishes downloading so we inform immediately that we're
        // downloading crates here.
        if self.downloads_finished == 0
            && self.pending.is_empty()
            && !self.progress.borrow().as_ref().unwrap().is_enabled()
        {
            self.set
                .config
                .shell()
                .status("Downloading", "crates ...")?;
        }

        let dl = Download {
            token,
            data: RefCell::new(Vec::new()),
            id,
            url,
            descriptor,
            total: Cell::new(0),
            current: Cell::new(0),
            start: Instant::now(),
            timed_out: Cell::new(None),
            retry: Retry::new(self.set.config)?,
        };
        self.enqueue(dl, handle)?;
        self.tick(WhyTick::DownloadStarted)?;

        Ok(None)
    }

    /// Returns the number of crates that are still downloading.
    pub fn remaining(&self) -> usize {
        self.pending.len()
    }

    /// Blocks the current thread waiting for a package to finish downloading.
    ///
    /// This method will wait for a previously enqueued package to finish
    /// downloading and return a reference to it after it's done downloading.
    ///
    /// # Panics
    ///
    /// This function will panic if there are no remaining downloads.
    pub fn wait(&mut self) -> CargoResult<&'a Package> {
        let (dl, data) = loop {
            assert_eq!(self.pending.len(), self.pending_ids.len());
            let (token, result) = self.wait_for_curl()?;
            debug!("{} finished with {:?}", token, result);

            let (mut dl, handle) = self
                .pending
                .remove(&token)
                .expect("got a token for a non-in-progress transfer");
            let data = mem::replace(&mut *dl.data.borrow_mut(), Vec::new());
            let mut handle = self.set.multi.remove(handle)?;
            self.pending_ids.remove(&dl.id);

            // Check if this was a spurious error. If it was a spurious error
            // then we want to re-enqueue our request for another attempt and
            // then we wait for another request to finish.
            let ret = {
                let timed_out = &dl.timed_out;
                let url = &dl.url;
                dl.retry
                    .r#try(|| {
                        if let Err(e) = result {
                            // If this error is "aborted by callback" then that's
                            // probably because our progress callback aborted due to
                            // a timeout. We'll find out by looking at the
                            // `timed_out` field, looking for a descriptive message.
                            // If one is found we switch the error code (to ensure
                            // it's flagged as spurious) and then attach our extra
                            // information to the error.
                            if !e.is_aborted_by_callback() {
                                return Err(e.into());
                            }

                            return Err(match timed_out.replace(None) {
                                Some(msg) => {
                                    let code = curl_sys::CURLE_OPERATION_TIMEDOUT;
                                    let mut err = curl::Error::new(code);
                                    err.set_extra(msg);
                                    err
                                }
                                None => e,
                            }
                            .into());
                        }

                        let code = handle.response_code()?;
                        if code != 200 && code != 0 {
                            let url = handle.effective_url()?.unwrap_or(url);
                            return Err(HttpNot200 {
                                code,
                                url: url.to_string(),
                            }
                            .into());
                        }
                        Ok(())
                    })
                    .chain_err(|| format!("failed to download from `{}`", dl.url))?
            };
            match ret {
                Some(()) => break (dl, data),
                None => {
                    self.pending_ids.insert(dl.id);
                    self.enqueue(dl, handle)?
                }
            }
        };

        // If the progress bar isn't enabled then we still want to provide some
        // semblance of progress of how we're downloading crates, and if the
        // progress bar is enabled this provides a good log of what's happening.
        self.progress.borrow_mut().as_mut().unwrap().clear();
        self.set
            .config
            .shell()
            .status("Downloaded", &dl.descriptor)?;

        self.downloads_finished += 1;
        self.downloaded_bytes += dl.total.get();
        if dl.total.get() > self.largest.0 {
            self.largest = (dl.total.get(), dl.id.name().to_string());
        }

        // We're about to synchronously extract the crate below. While we're
        // doing that our download progress won't actually be updated, nor do we
        // have a great view into the progress of the extraction. Let's prepare
        // the user for this CPU-heavy step if it looks like it'll take some
        // time to do so.
        if dl.total.get() < ByteSize::kb(400).0 {
            self.tick(WhyTick::DownloadFinished)?;
        } else {
            self.tick(WhyTick::Extracting(&dl.id.name()))?;
        }

        // Inform the original source that the download is finished which
        // should allow us to actually get the package and fill it in now.
        let mut sources = self.set.sources.borrow_mut();
        let source = sources
            .get_mut(dl.id.source_id())
            .ok_or_else(|| internal(format!("couldn't find source for `{}`", dl.id)))?;
        let start = Instant::now();
        let pkg = source.finish_download(dl.id, data)?;

        // Assume that no time has passed while we were calling
        // `finish_download`, update all speed checks and timeout limits of all
        // active downloads to make sure they don't fire because of a slowly
        // extracted tarball.
        let finish_dur = start.elapsed();
        self.updated_at.set(self.updated_at.get() + finish_dur);
        self.next_speed_check
            .set(self.next_speed_check.get() + finish_dur);

        let slot = &self.set.packages[&dl.id];
        assert!(slot.fill(pkg).is_ok());
        Ok(slot.borrow().unwrap())
    }

    fn enqueue(&mut self, dl: Download<'cfg>, handle: Easy) -> CargoResult<()> {
        let mut handle = self.set.multi.add(handle)?;
        let now = Instant::now();
        handle.set_token(dl.token)?;
        self.updated_at.set(now);
        self.next_speed_check.set(now + self.timeout.dur);
        self.next_speed_check_bytes_threshold
            .set(u64::from(self.timeout.low_speed_limit));
        dl.timed_out.set(None);
        dl.current.set(0);
        dl.total.set(0);
        self.pending.insert(dl.token, (dl, handle));
        Ok(())
    }

    /// Block, waiting for curl. Returns a token and a `Result` for that token
    /// (`Ok` means the download successfully finished).
    fn wait_for_curl(&mut self) -> CargoResult<(usize, Result<(), curl::Error>)> {
        // This is the main workhorse loop. We use libcurl's portable `wait`
        // method to actually perform blocking. This isn't necessarily too
        // efficient in terms of fd management, but we should only be juggling
        // a few anyway.
        //
        // Here we start off by asking the `multi` handle to do some work via
        // the `perform` method. This will actually do I/O work (non-blocking)
        // and attempt to make progress. Afterwards we ask about the `messages`
        // contained in the handle which will inform us if anything has finished
        // transferring.
        //
        // If we've got a finished transfer after all that work we break out
        // and process the finished transfer at the end. Otherwise we need to
        // actually block waiting for I/O to happen, which we achieve with the
        // `wait` method on `multi`.
        loop {
            let n = tls::set(self, || {
                self.set
                    .multi
                    .perform()
                    .chain_err(|| "failed to perform http requests")
            })?;
            debug!("handles remaining: {}", n);
            let results = &mut self.results;
            let pending = &self.pending;
            self.set.multi.messages(|msg| {
                let token = msg.token().expect("failed to read token");
                let handle = &pending[&token].1;
                if let Some(result) = msg.result_for(handle) {
                    results.push((token, result));
                } else {
                    debug!("message without a result (?)");
                }
            });

            if let Some(pair) = results.pop() {
                break Ok(pair);
            }
            assert!(!self.pending.is_empty());
            let timeout = self
                .set
                .multi
                .get_timeout()?
                .unwrap_or_else(|| Duration::new(5, 0));
            self.set
                .multi
                .wait(&mut [], timeout)
                .chain_err(|| "failed to wait on curl `Multi`")?;
        }
    }

    fn progress(&self, token: usize, total: u64, cur: u64) -> bool {
        let dl = &self.pending[&token].0;
        dl.total.set(total);
        let now = Instant::now();
        if cur != dl.current.get() {
            let delta = cur - dl.current.get();
            let threshold = self.next_speed_check_bytes_threshold.get();

            dl.current.set(cur);
            self.updated_at.set(now);

            if delta >= threshold {
                self.next_speed_check.set(now + self.timeout.dur);
                self.next_speed_check_bytes_threshold
                    .set(u64::from(self.timeout.low_speed_limit));
            } else {
                self.next_speed_check_bytes_threshold.set(threshold - delta);
            }
        }
        if self.tick(WhyTick::DownloadUpdate).is_err() {
            return false;
        }

        // If we've spent too long not actually receiving any data we time out.
        if now - self.updated_at.get() > self.timeout.dur {
            self.updated_at.set(now);
            let msg = format!(
                "failed to download any data for `{}` within {}s",
                dl.id,
                self.timeout.dur.as_secs()
            );
            dl.timed_out.set(Some(msg));
            return false;
        }

        // If we reached the point in time that we need to check our speed
        // limit, see if we've transferred enough data during this threshold. If
        // it fails this check then we fail because the download is going too
        // slowly.
        if now >= self.next_speed_check.get() {
            self.next_speed_check.set(now + self.timeout.dur);
            assert!(self.next_speed_check_bytes_threshold.get() > 0);
            let msg = format!(
                "download of `{}` failed to transfer more \
                 than {} bytes in {}s",
                dl.id,
                self.timeout.low_speed_limit,
                self.timeout.dur.as_secs()
            );
            dl.timed_out.set(Some(msg));
            return false;
        }

        true
    }

    fn tick(&self, why: WhyTick<'_>) -> CargoResult<()> {
        let mut progress = self.progress.borrow_mut();
        let progress = progress.as_mut().unwrap();

        if let WhyTick::DownloadUpdate = why {
            if !progress.update_allowed() {
                return Ok(());
            }
        }
        let pending = self.pending.len();
        let mut msg = if pending == 1 {
            format!("{} crate", pending)
        } else {
            format!("{} crates", pending)
        };
        match why {
            WhyTick::Extracting(krate) => {
                msg.push_str(&format!(", extracting {} ...", krate));
            }
            _ => {
                let mut dur = Duration::new(0, 0);
                let mut remaining = 0;
                for (dl, _) in self.pending.values() {
                    dur += dl.start.elapsed();
                    // If the total/current look weird just throw out the data
                    // point, sounds like curl has more to learn before we have
                    // the true information.
                    if dl.total.get() >= dl.current.get() {
                        remaining += dl.total.get() - dl.current.get();
                    }
                }
                if remaining > 0 && dur > Duration::from_millis(500) {
                    msg.push_str(&format!(", remaining bytes: {}", ByteSize(remaining)));
                }
            }
        }
        progress.print_now(&msg)
    }
}

#[derive(Copy, Clone)]
enum WhyTick<'a> {
    DownloadStarted,
    DownloadUpdate,
    DownloadFinished,
    Extracting(&'a str),
}

impl<'a, 'cfg> Drop for Downloads<'a, 'cfg> {
    fn drop(&mut self) {
        self.set.downloading.set(false);
        let progress = self.progress.get_mut().take().unwrap();
        // Don't print a download summary if we're not using a progress bar,
        // we've already printed lots of `Downloading...` items.
        if !progress.is_enabled() {
            return;
        }
        // If we didn't download anything, no need for a summary.
        if self.downloads_finished == 0 {
            return;
        }
        // If an error happened, let's not clutter up the output.
        if !self.success {
            return;
        }
        // pick the correct plural of crate(s)
        let crate_string = if self.downloads_finished == 1 {
            "crate"
        } else {
            "crates"
        };
        let mut status = format!(
            "{} {} ({}) in {}",
            self.downloads_finished,
            crate_string,
            ByteSize(self.downloaded_bytes),
            util::elapsed(self.start.elapsed())
        );
        // print the size of largest crate if it was >1mb
        // however don't print if only a single crate was downloaded
        // because it is obvious that it will be the largest then
        if self.largest.0 > ByteSize::mb(1).0 && self.downloads_finished > 1 {
            status.push_str(&format!(
                " (largest was `{}` at {})",
                self.largest.1,
                ByteSize(self.largest.0),
            ));
        }
        // Clear progress before displaying final summary.
        drop(progress);
        drop(self.set.config.shell().status("Downloaded", status));
    }
}

mod tls {
    use std::cell::Cell;

    use super::Downloads;

    thread_local!(static PTR: Cell<usize> = Cell::new(0));

    pub(crate) fn with<R>(f: impl FnOnce(Option<&Downloads<'_, '_>>) -> R) -> R {
        let ptr = PTR.with(|p| p.get());
        if ptr == 0 {
            f(None)
        } else {
            unsafe { f(Some(&*(ptr as *const Downloads<'_, '_>))) }
        }
    }

    pub(crate) fn set<R>(dl: &Downloads<'_, '_>, f: impl FnOnce() -> R) -> R {
        struct Reset<'a, T: Copy>(&'a Cell<T>, T);

        impl<'a, T: Copy> Drop for Reset<'a, T> {
            fn drop(&mut self) {
                self.0.set(self.1);
            }
        }

        PTR.with(|p| {
            let _reset = Reset(p, p.get());
            p.set(dl as *const Downloads<'_, '_> as usize);
            f()
        })
    }
}