src/balance/pool/mod.rs - platform/external/rust/crates/tower - Git at Google

 //! This module defines a load-balanced pool of services that adds new services when load is high.
 //!
 //! The pool uses `poll_ready` as a signal indicating whether additional services should be spawned
 //! to handle the current level of load. Specifically, every time `poll_ready` on the inner service
 //! returns `Ready`, [`Pool`] consider that a 0, and every time it returns `Pending`, [`Pool`]
 //! considers it a 1. [`Pool`] then maintains an [exponential moving
 //! average](https://en.wikipedia.org/wiki/Moving_average#Exponential_moving_average) over those
 //! samples, which gives an estimate of how often the underlying service has been ready when it was
 //! needed "recently" (see [`Builder::urgency`]). If the service is loaded (see
 //! [`Builder::loaded_above`]), a new service is created and added to the underlying [`Balance`].
 //! If the service is underutilized (see [`Builder::underutilized_below`]) and there are two or
 //! more services, then the latest added service is removed. In either case, the load estimate is
 //! reset to its initial value (see [`Builder::initial`] to prevent services from being rapidly
 //! added or removed.
 #![deny(missing_docs)]

 use super::p2c::Balance;
 use crate::discover::Change;
 use crate::load::Load;
 use crate::make::MakeService;
 use futures_core::{ready, Stream};
 use pin_project_lite::pin_project;
 use slab::Slab;
 use std::{
     fmt,
     future::Future,
     pin::Pin,
     task::{Context, Poll},
 };
 use tower_service::Service;

 #[cfg(test)]
 mod test;

 #[derive(Debug, Clone, Copy, Eq, PartialEq)]
 enum Level {
     /// Load is low -- remove a service instance.
     Low,
     /// Load is normal -- keep the service set as it is.
     Normal,
     /// Load is high -- add another service instance.
     High,
 }

 pin_project! {
     /// A wrapper around `MakeService` that discovers a new service when load is high, and removes a
     /// service when load is low. See [`Pool`].
     pub struct PoolDiscoverer<MS, Target, Request>
     where
         MS: MakeService<Target, Request>,
     {
         maker: MS,
         #[pin]
         making: Option<MS::Future>,
         target: Target,
         load: Level,
         services: Slab<()>,
         died_tx: tokio::sync::mpsc::UnboundedSender<usize>,
         #[pin]
         died_rx: tokio::sync::mpsc::UnboundedReceiver<usize>,
         limit: Option<usize>,
     }
 }

 impl<MS, Target, Request> fmt::Debug for PoolDiscoverer<MS, Target, Request>
 where
     MS: MakeService<Target, Request> + fmt::Debug,
     Target: fmt::Debug,
 {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.debug_struct("PoolDiscoverer")
             .field("maker", &self.maker)
             .field("making", &self.making.is_some())
             .field("target", &self.target)
             .field("load", &self.load)
             .field("services", &self.services)
             .field("limit", &self.limit)
             .finish()
     }
 }

 impl<MS, Target, Request> Stream for PoolDiscoverer<MS, Target, Request>
 where
     MS: MakeService<Target, Request>,
     MS::MakeError: Into<crate::BoxError>,
     MS::Error: Into<crate::BoxError>,
     Target: Clone,
 {
     type Item = Result<Change<usize, DropNotifyService<MS::Service>>, MS::MakeError>;

     fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         let mut this = self.project();

         while let Poll::Ready(Some(sid)) = this.died_rx.as_mut().poll_recv(cx) {
             this.services.remove(sid);
             tracing::trace!(
                 pool.services = this.services.len(),
                 message = "removing dropped service"
             );
         }

         if this.services.is_empty() && this.making.is_none() {
             let _ = ready!(this.maker.poll_ready(cx))?;
             tracing::trace!("construct initial pool connection");
             this.making
                 .set(Some(this.maker.make_service(this.target.clone())));
         }

         if let Level::High = this.load {
             if this.making.is_none() {
                 if this
                     .limit
                     .map(|limit| this.services.len() >= limit)
                     .unwrap_or(false)
                 {
                     return Poll::Pending;
                 }

                 tracing::trace!(
                     pool.services = this.services.len(),
                     message = "decided to add service to loaded pool"
                 );
                 ready!(this.maker.poll_ready(cx))?;
                 tracing::trace!("making new service");
                 // TODO: it'd be great if we could avoid the clone here and use, say, &Target
                 this.making
                     .set(Some(this.maker.make_service(this.target.clone())));
             }
         }

         if let Some(fut) = this.making.as_mut().as_pin_mut() {
             let svc = ready!(fut.poll(cx))?;
             this.making.set(None);

             let id = this.services.insert(());
             let svc = DropNotifyService {
                 svc,
                 id,
                 notify: this.died_tx.clone(),
             };
             tracing::trace!(
                 pool.services = this.services.len(),
                 message = "finished creating new service"
             );
             *this.load = Level::Normal;
             return Poll::Ready(Some(Ok(Change::Insert(id, svc))));
         }

         match this.load {
             Level::High => {
                 unreachable!("found high load but no Service being made");
             }
             Level::Normal => Poll::Pending,
             Level::Low if this.services.len() == 1 => Poll::Pending,
             Level::Low => {
                 *this.load = Level::Normal;
                 // NOTE: this is a little sad -- we'd prefer to kill short-living services
                 let rm = this.services.iter().next().unwrap().0;
                 // note that we _don't_ remove from self.services here
                 // that'll happen automatically on drop
                 tracing::trace!(
                     pool.services = this.services.len(),
                     message = "removing service for over-provisioned pool"
                 );
                 Poll::Ready(Some(Ok(Change::Remove(rm))))
             }
         }
     }
 }

 /// A [builder] that lets you configure how a [`Pool`] determines whether the underlying service is
 /// loaded or not. See the [module-level documentation](self) and the builder's methods for
 /// details.
 ///
 ///  [builder]: https://rust-lang-nursery.github.io/api-guidelines/type-safety.html#builders-enable-construction-of-complex-values-c-builder
 #[derive(Copy, Clone, Debug)]
 pub struct Builder {
     low: f64,
     high: f64,
     init: f64,
     alpha: f64,
     limit: Option<usize>,
 }

 impl Default for Builder {
     fn default() -> Self {
         Builder {
             init: 0.1,
             low: 0.00001,
             high: 0.2,
             alpha: 0.03,
             limit: None,
         }
     }
 }

 impl Builder {
     /// Create a new builder with default values for all load settings.
     ///
     /// If you just want to use the defaults, you can just use [`Pool::new`].
     pub fn new() -> Self {
         Self::default()
     }

     /// When the estimated load (see the [module-level docs](self)) drops below this
     /// threshold, and there are at least two services active, a service is removed.
     ///
     /// The default value is 0.01. That is, when one in every 100 `poll_ready` calls return
     /// `Pending`, then the underlying service is considered underutilized.
     pub fn underutilized_below(&mut self, low: f64) -> &mut Self {
         self.low = low;
         self
     }

     /// When the estimated load (see the [module-level docs](self)) exceeds this
     /// threshold, and no service is currently in the process of being added, a new service is
     /// scheduled to be added to the underlying [`Balance`].
     ///
     /// The default value is 0.5. That is, when every other call to `poll_ready` returns
     /// `Pending`, then the underlying service is considered highly loaded.
     pub fn loaded_above(&mut self, high: f64) -> &mut Self {
         self.high = high;
         self
     }

     /// The initial estimated load average.
     ///
     /// This is also the value that the estimated load will be reset to whenever a service is added
     /// or removed.
     ///
     /// The default value is 0.1.
     pub fn initial(&mut self, init: f64) -> &mut Self {
         self.init = init;
         self
     }

     /// How aggressively the estimated load average is updated.
     ///
     /// This is the α parameter of the formula for the [exponential moving
     /// average](https://en.wikipedia.org/wiki/Moving_average#Exponential_moving_average), and
     /// dictates how quickly new samples of the current load affect the estimated load. If the
     /// value is closer to 1, newer samples affect the load average a lot (when α is 1, the load
     /// average is immediately set to the current load). If the value is closer to 0, newer samples
     /// affect the load average very little at a time.
     ///
     /// The given value is clamped to `[0,1]`.
     ///
     /// The default value is 0.05, meaning, in very approximate terms, that each new load sample
     /// affects the estimated load by 5%.
     pub fn urgency(&mut self, alpha: f64) -> &mut Self {
         self.alpha = alpha.max(0.0).min(1.0);
         self
     }

     /// The maximum number of backing `Service` instances to maintain.
     ///
     /// When the limit is reached, the load estimate is clamped to the high load threshhold, and no
     /// new service is spawned.
     ///
     /// No maximum limit is imposed by default.
     pub fn max_services(&mut self, limit: Option<usize>) -> &mut Self {
         self.limit = limit;
         self
     }

     /// See [`Pool::new`].
     pub fn build<MS, Target, Request>(
         &self,
         make_service: MS,
         target: Target,
     ) -> Pool<MS, Target, Request>
     where
         MS: MakeService<Target, Request>,
         MS::Service: Load,
         <MS::Service as Load>::Metric: std::fmt::Debug,
         MS::MakeError: Into<crate::BoxError>,
         MS::Error: Into<crate::BoxError>,
         Target: Clone,
     {
         let (died_tx, died_rx) = tokio::sync::mpsc::unbounded_channel();
         let d = PoolDiscoverer {
             maker: make_service,
             making: None,
             target,
             load: Level::Normal,
             services: Slab::new(),
             died_tx,
             died_rx,
             limit: self.limit,
         };

         Pool {
             balance: Balance::new(Box::pin(d)),
             options: *self,
             ewma: self.init,
         }
     }
 }

 /// A dynamically sized, load-balanced pool of `Service` instances.
 pub struct Pool<MS, Target, Request>
 where
     MS: MakeService<Target, Request>,
     MS::MakeError: Into<crate::BoxError>,
     MS::Error: Into<crate::BoxError>,
     Target: Clone,
 {
     // the Pin<Box<_>> here is needed since Balance requires the Service to be Unpin
     balance: Balance<Pin<Box<PoolDiscoverer<MS, Target, Request>>>, Request>,
     options: Builder,
     ewma: f64,
 }

 impl<MS, Target, Request> fmt::Debug for Pool<MS, Target, Request>
 where
     MS: MakeService<Target, Request> + fmt::Debug,
     MS::MakeError: Into<crate::BoxError>,
     MS::Error: Into<crate::BoxError>,
     Target: Clone + fmt::Debug,
     MS::Service: fmt::Debug,
 {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.debug_struct("Pool")
             .field("balance", &self.balance)
             .field("options", &self.options)
             .field("ewma", &self.ewma)
             .finish()
     }
 }

 impl<MS, Target, Request> Pool<MS, Target, Request>
 where
     MS: MakeService<Target, Request>,
     MS::Service: Load,
     <MS::Service as Load>::Metric: std::fmt::Debug,
     MS::MakeError: Into<crate::BoxError>,
     MS::Error: Into<crate::BoxError>,
     Target: Clone,
 {
     /// Construct a new dynamically sized `Pool`.
     ///
     /// If many calls to `poll_ready` return `Pending`, `new_service` is used to
     /// construct another `Service` that is then added to the load-balanced pool.
     /// If many calls to `poll_ready` succeed, the most recently added `Service`
     /// is dropped from the pool.
     pub fn new(make_service: MS, target: Target) -> Self {
         Builder::new().build(make_service, target)
     }
 }

 type PinBalance<S, Request> = Balance<Pin<Box<S>>, Request>;

 impl<MS, Target, Req> Service<Req> for Pool<MS, Target, Req>
 where
     MS: MakeService<Target, Req>,
     MS::Service: Load,
     <MS::Service as Load>::Metric: std::fmt::Debug,
     MS::MakeError: Into<crate::BoxError>,
     MS::Error: Into<crate::BoxError>,
     Target: Clone,
 {
     type Response = <PinBalance<PoolDiscoverer<MS, Target, Req>, Req> as Service<Req>>::Response;
     type Error = <PinBalance<PoolDiscoverer<MS, Target, Req>, Req> as Service<Req>>::Error;
     type Future = <PinBalance<PoolDiscoverer<MS, Target, Req>, Req> as Service<Req>>::Future;

     fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
         if let Poll::Ready(()) = self.balance.poll_ready(cx)? {
             // services was ready -- there are enough services
             // update ewma with a 0 sample
             self.ewma *= 1.0 - self.options.alpha;

             let discover = self.balance.discover_mut().as_mut().project();
             if self.ewma < self.options.low {
                 if *discover.load != Level::Low {
                     tracing::trace!({ ewma = %self.ewma }, "pool is over-provisioned");
                 }
                 *discover.load = Level::Low;

                 if discover.services.len() > 1 {
                     // reset EWMA so we don't immediately try to remove another service
                     self.ewma = self.options.init;
                 }
             } else {
                 if *discover.load != Level::Normal {
                     tracing::trace!({ ewma = %self.ewma }, "pool is appropriately provisioned");
                 }
                 *discover.load = Level::Normal;
             }

             return Poll::Ready(Ok(()));
         }

         let discover = self.balance.discover_mut().as_mut().project();
         if discover.making.is_none() {
             // no services are ready -- we're overloaded
             // update ewma with a 1 sample
             self.ewma = self.options.alpha + (1.0 - self.options.alpha) * self.ewma;

             if self.ewma > self.options.high {
                 if *discover.load != Level::High {
                     tracing::trace!({ ewma = %self.ewma }, "pool is under-provisioned");
                 }
                 *discover.load = Level::High;

                 // don't reset the EWMA -- in theory, poll_ready should now start returning
                 // `Ready`, so we won't try to launch another service immediately.
                 // we clamp it to high though in case the # of services is limited.
                 self.ewma = self.options.high;

                 // we need to call balance again for PoolDiscover to realize
                 // it can make a new service
                 return self.balance.poll_ready(cx);
             } else {
                 *discover.load = Level::Normal;
             }
         }

         Poll::Pending
     }

     fn call(&mut self, req: Req) -> Self::Future {
         self.balance.call(req)
     }
 }

 #[doc(hidden)]
 #[derive(Debug)]
 pub struct DropNotifyService<Svc> {
     svc: Svc,
     id: usize,
     notify: tokio::sync::mpsc::UnboundedSender<usize>,
 }

 impl<Svc> Drop for DropNotifyService<Svc> {
     fn drop(&mut self) {
         let _ = self.notify.send(self.id).is_ok();
     }
 }

 impl<Svc: Load> Load for DropNotifyService<Svc> {
     type Metric = Svc::Metric;
     fn load(&self) -> Self::Metric {
         self.svc.load()
     }
 }

 impl<Request, Svc: Service<Request>> Service<Request> for DropNotifyService<Svc> {
     type Response = Svc::Response;
     type Future = Svc::Future;
     type Error = Svc::Error;

     fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
         self.svc.poll_ready(cx)
     }

     fn call(&mut self, req: Request) -> Self::Future {
         self.svc.call(req)
     }
 }
	//! This module defines a load-balanced pool of services that adds new services when load is high.
	//!
	//! The pool uses `poll_ready` as a signal indicating whether additional services should be spawned
	//! to handle the current level of load. Specifically, every time `poll_ready` on the inner service
	//! returns `Ready`, [`Pool`] consider that a 0, and every time it returns `Pending`, [`Pool`]
	//! considers it a 1. [`Pool`] then maintains an [exponential moving
	//! average](https://en.wikipedia.org/wiki/Moving_average#Exponential_moving_average) over those
	//! samples, which gives an estimate of how often the underlying service has been ready when it was
	//! needed "recently" (see [`Builder::urgency`]). If the service is loaded (see
	//! [`Builder::loaded_above`]), a new service is created and added to the underlying [`Balance`].
	//! If the service is underutilized (see [`Builder::underutilized_below`]) and there are two or
	//! more services, then the latest added service is removed. In either case, the load estimate is
	//! reset to its initial value (see [`Builder::initial`] to prevent services from being rapidly
	//! added or removed.
	#![deny(missing_docs)]

	use super::p2c::Balance;
	use crate::discover::Change;
	use crate::load::Load;
	use crate::make::MakeService;
	use futures_core::{ready, Stream};
	use pin_project_lite::pin_project;
	use slab::Slab;
	use std::{
	fmt,
	future::Future,
	pin::Pin,
	task::{Context, Poll},
	};
	use tower_service::Service;

	#[cfg(test)]
	mod test;

	#[derive(Debug, Clone, Copy, Eq, PartialEq)]
	enum Level {
	/// Load is low -- remove a service instance.
	Low,
	/// Load is normal -- keep the service set as it is.
	Normal,
	/// Load is high -- add another service instance.
	High,
	}

	pin_project! {
	/// A wrapper around `MakeService` that discovers a new service when load is high, and removes a
	/// service when load is low. See [`Pool`].
	pub struct PoolDiscoverer<MS, Target, Request>
	where
	MS: MakeService<Target, Request>,
	{
	maker: MS,
	#[pin]
	making: Option<MS::Future>,
	target: Target,
	load: Level,
	services: Slab<()>,
	died_tx: tokio::sync::mpsc::UnboundedSender<usize>,
	#[pin]
	died_rx: tokio::sync::mpsc::UnboundedReceiver<usize>,
	limit: Option<usize>,
	}
	}

	impl<MS, Target, Request> fmt::Debug for PoolDiscoverer<MS, Target, Request>
	where
	MS: MakeService<Target, Request> + fmt::Debug,
	Target: fmt::Debug,
	{
	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
	f.debug_struct("PoolDiscoverer")
	.field("maker", &self.maker)
	.field("making", &self.making.is_some())
	.field("target", &self.target)
	.field("load", &self.load)
	.field("services", &self.services)
	.field("limit", &self.limit)
	.finish()
	}
	}

	impl<MS, Target, Request> Stream for PoolDiscoverer<MS, Target, Request>
	where
	MS: MakeService<Target, Request>,
	MS::MakeError: Into<crate::BoxError>,
	MS::Error: Into<crate::BoxError>,
	Target: Clone,
	{
	type Item = Result<Change<usize, DropNotifyService<MS::Service>>, MS::MakeError>;

	fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
	let mut this = self.project();

	while let Poll::Ready(Some(sid)) = this.died_rx.as_mut().poll_recv(cx) {
	this.services.remove(sid);
	tracing::trace!(
	pool.services = this.services.len(),
	message = "removing dropped service"
	);
	}

	if this.services.is_empty() && this.making.is_none() {
	let _ = ready!(this.maker.poll_ready(cx))?;
	tracing::trace!("construct initial pool connection");
	this.making
	.set(Some(this.maker.make_service(this.target.clone())));
	}

	if let Level::High = this.load {
	if this.making.is_none() {
	if this
	.limit
	.map(\|limit\| this.services.len() >= limit)
	.unwrap_or(false)
	{
	return Poll::Pending;
	}

	tracing::trace!(
	pool.services = this.services.len(),
	message = "decided to add service to loaded pool"
	);
	ready!(this.maker.poll_ready(cx))?;
	tracing::trace!("making new service");
	// TODO: it'd be great if we could avoid the clone here and use, say, &Target
	this.making
	.set(Some(this.maker.make_service(this.target.clone())));
	}
	}

	if let Some(fut) = this.making.as_mut().as_pin_mut() {
	let svc = ready!(fut.poll(cx))?;
	this.making.set(None);

	let id = this.services.insert(());
	let svc = DropNotifyService {
	svc,
	id,
	notify: this.died_tx.clone(),
	};
	tracing::trace!(
	pool.services = this.services.len(),
	message = "finished creating new service"
	);
	*this.load = Level::Normal;
	return Poll::Ready(Some(Ok(Change::Insert(id, svc))));
	}

	match this.load {
	Level::High => {
	unreachable!("found high load but no Service being made");
	}
	Level::Normal => Poll::Pending,
	Level::Low if this.services.len() == 1 => Poll::Pending,
	Level::Low => {
	*this.load = Level::Normal;
	// NOTE: this is a little sad -- we'd prefer to kill short-living services
	let rm = this.services.iter().next().unwrap().0;
	// note that we _don't_ remove from self.services here
	// that'll happen automatically on drop
	tracing::trace!(
	pool.services = this.services.len(),
	message = "removing service for over-provisioned pool"
	);
	Poll::Ready(Some(Ok(Change::Remove(rm))))
	}
	}
	}
	}

	/// A [builder] that lets you configure how a [`Pool`] determines whether the underlying service is
	/// loaded or not. See the [module-level documentation](self) and the builder's methods for
	/// details.
	///
	/// [builder]: https://rust-lang-nursery.github.io/api-guidelines/type-safety.html#builders-enable-construction-of-complex-values-c-builder
	#[derive(Copy, Clone, Debug)]
	pub struct Builder {
	low: f64,
	high: f64,
	init: f64,
	alpha: f64,
	limit: Option<usize>,
	}

	impl Default for Builder {
	fn default() -> Self {
	Builder {
	init: 0.1,
	low: 0.00001,
	high: 0.2,
	alpha: 0.03,
	limit: None,
	}
	}
	}

	impl Builder {
	/// Create a new builder with default values for all load settings.
	///
	/// If you just want to use the defaults, you can just use [`Pool::new`].
	pub fn new() -> Self {
	Self::default()
	}

	/// When the estimated load (see the [module-level docs](self)) drops below this
	/// threshold, and there are at least two services active, a service is removed.
	///
	/// The default value is 0.01. That is, when one in every 100 `poll_ready` calls return
	/// `Pending`, then the underlying service is considered underutilized.
	pub fn underutilized_below(&mut self, low: f64) -> &mut Self {
	self.low = low;
	self
	}

	/// When the estimated load (see the [module-level docs](self)) exceeds this
	/// threshold, and no service is currently in the process of being added, a new service is
	/// scheduled to be added to the underlying [`Balance`].
	///
	/// The default value is 0.5. That is, when every other call to `poll_ready` returns
	/// `Pending`, then the underlying service is considered highly loaded.
	pub fn loaded_above(&mut self, high: f64) -> &mut Self {
	self.high = high;
	self
	}

	/// The initial estimated load average.
	///
	/// This is also the value that the estimated load will be reset to whenever a service is added
	/// or removed.
	///
	/// The default value is 0.1.
	pub fn initial(&mut self, init: f64) -> &mut Self {
	self.init = init;
	self
	}

	/// How aggressively the estimated load average is updated.
	///
	/// This is the α parameter of the formula for the [exponential moving
	/// average](https://en.wikipedia.org/wiki/Moving_average#Exponential_moving_average), and
	/// dictates how quickly new samples of the current load affect the estimated load. If the
	/// value is closer to 1, newer samples affect the load average a lot (when α is 1, the load
	/// average is immediately set to the current load). If the value is closer to 0, newer samples
	/// affect the load average very little at a time.
	///
	/// The given value is clamped to `[0,1]`.
	///
	/// The default value is 0.05, meaning, in very approximate terms, that each new load sample
	/// affects the estimated load by 5%.
	pub fn urgency(&mut self, alpha: f64) -> &mut Self {
	self.alpha = alpha.max(0.0).min(1.0);
	self
	}

	/// The maximum number of backing `Service` instances to maintain.
	///
	/// When the limit is reached, the load estimate is clamped to the high load threshhold, and no
	/// new service is spawned.
	///
	/// No maximum limit is imposed by default.
	pub fn max_services(&mut self, limit: Option<usize>) -> &mut Self {
	self.limit = limit;
	self
	}

	/// See [`Pool::new`].
	pub fn build<MS, Target, Request>(
	&self,
	make_service: MS,
	target: Target,
	) -> Pool<MS, Target, Request>
	where
	MS: MakeService<Target, Request>,
	MS::Service: Load,
	<MS::Service as Load>::Metric: std::fmt::Debug,
	MS::MakeError: Into<crate::BoxError>,
	MS::Error: Into<crate::BoxError>,
	Target: Clone,
	{
	let (died_tx, died_rx) = tokio::sync::mpsc::unbounded_channel();
	let d = PoolDiscoverer {
	maker: make_service,
	making: None,
	target,
	load: Level::Normal,
	services: Slab::new(),
	died_tx,
	died_rx,
	limit: self.limit,
	};

	Pool {
	balance: Balance::new(Box::pin(d)),
	options: *self,
	ewma: self.init,
	}
	}
	}

	/// A dynamically sized, load-balanced pool of `Service` instances.
	pub struct Pool<MS, Target, Request>
	where
	MS: MakeService<Target, Request>,
	MS::MakeError: Into<crate::BoxError>,
	MS::Error: Into<crate::BoxError>,
	Target: Clone,
	{
	// the Pin<Box<_>> here is needed since Balance requires the Service to be Unpin
	balance: Balance<Pin<Box<PoolDiscoverer<MS, Target, Request>>>, Request>,
	options: Builder,
	ewma: f64,
	}

	impl<MS, Target, Request> fmt::Debug for Pool<MS, Target, Request>
	where
	MS: MakeService<Target, Request> + fmt::Debug,
	MS::MakeError: Into<crate::BoxError>,
	MS::Error: Into<crate::BoxError>,
	Target: Clone + fmt::Debug,
	MS::Service: fmt::Debug,
	{
	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
	f.debug_struct("Pool")
	.field("balance", &self.balance)
	.field("options", &self.options)
	.field("ewma", &self.ewma)
	.finish()
	}
	}

	impl<MS, Target, Request> Pool<MS, Target, Request>
	where
	MS: MakeService<Target, Request>,
	MS::Service: Load,
	<MS::Service as Load>::Metric: std::fmt::Debug,
	MS::MakeError: Into<crate::BoxError>,
	MS::Error: Into<crate::BoxError>,
	Target: Clone,
	{
	/// Construct a new dynamically sized `Pool`.
	///
	/// If many calls to `poll_ready` return `Pending`, `new_service` is used to
	/// construct another `Service` that is then added to the load-balanced pool.
	/// If many calls to `poll_ready` succeed, the most recently added `Service`
	/// is dropped from the pool.
	pub fn new(make_service: MS, target: Target) -> Self {
	Builder::new().build(make_service, target)
	}
	}

	type PinBalance<S, Request> = Balance<Pin<Box<S>>, Request>;

	impl<MS, Target, Req> Service<Req> for Pool<MS, Target, Req>
	where
	MS: MakeService<Target, Req>,
	MS::Service: Load,
	<MS::Service as Load>::Metric: std::fmt::Debug,
	MS::MakeError: Into<crate::BoxError>,
	MS::Error: Into<crate::BoxError>,
	Target: Clone,
	{
	type Response = <PinBalance<PoolDiscoverer<MS, Target, Req>, Req> as Service<Req>>::Response;
	type Error = <PinBalance<PoolDiscoverer<MS, Target, Req>, Req> as Service<Req>>::Error;
	type Future = <PinBalance<PoolDiscoverer<MS, Target, Req>, Req> as Service<Req>>::Future;

	fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
	if let Poll::Ready(()) = self.balance.poll_ready(cx)? {
	// services was ready -- there are enough services
	// update ewma with a 0 sample
	self.ewma *= 1.0 - self.options.alpha;

	let discover = self.balance.discover_mut().as_mut().project();
	if self.ewma < self.options.low {
	if *discover.load != Level::Low {
	tracing::trace!({ ewma = %self.ewma }, "pool is over-provisioned");
	}
	*discover.load = Level::Low;

	if discover.services.len() > 1 {
	// reset EWMA so we don't immediately try to remove another service
	self.ewma = self.options.init;
	}
	} else {
	if *discover.load != Level::Normal {
	tracing::trace!({ ewma = %self.ewma }, "pool is appropriately provisioned");
	}
	*discover.load = Level::Normal;
	}

	return Poll::Ready(Ok(()));
	}

	let discover = self.balance.discover_mut().as_mut().project();
	if discover.making.is_none() {
	// no services are ready -- we're overloaded
	// update ewma with a 1 sample
	self.ewma = self.options.alpha + (1.0 - self.options.alpha) * self.ewma;

	if self.ewma > self.options.high {
	if *discover.load != Level::High {
	tracing::trace!({ ewma = %self.ewma }, "pool is under-provisioned");
	}
	*discover.load = Level::High;

	// don't reset the EWMA -- in theory, poll_ready should now start returning
	// `Ready`, so we won't try to launch another service immediately.
	// we clamp it to high though in case the # of services is limited.
	self.ewma = self.options.high;

	// we need to call balance again for PoolDiscover to realize
	// it can make a new service
	return self.balance.poll_ready(cx);
	} else {
	*discover.load = Level::Normal;
	}
	}

	Poll::Pending
	}

	fn call(&mut self, req: Req) -> Self::Future {
	self.balance.call(req)
	}
	}

	#[doc(hidden)]
	#[derive(Debug)]
	pub struct DropNotifyService<Svc> {
	svc: Svc,
	id: usize,
	notify: tokio::sync::mpsc::UnboundedSender<usize>,
	}

	impl<Svc> Drop for DropNotifyService<Svc> {
	fn drop(&mut self) {
	let _ = self.notify.send(self.id).is_ok();
	}
	}

	impl<Svc: Load> Load for DropNotifyService<Svc> {
	type Metric = Svc::Metric;
	fn load(&self) -> Self::Metric {
	self.svc.load()
	}
	}

	impl<Request, Svc: Service<Request>> Service<Request> for DropNotifyService<Svc> {
	type Response = Svc::Response;
	type Future = Svc::Future;
	type Error = Svc::Error;

	fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
	self.svc.poll_ready(cx)
	}

	fn call(&mut self, req: Request) -> Self::Future {
	self.svc.call(req)
	}
	}