|
| 1 | +//! A [`Load`] implementation which implements weighting on top of an inner [`Load`]. |
| 2 | +//! |
| 3 | +//! This can be useful in such cases as canary deployments, where it is desirable for a |
| 4 | +//! particular service to receive less than its fair share of load than other services. |
| 5 | +
|
| 6 | +#[cfg(feature = "discover")] |
| 7 | +use crate::discover::{Change, Discover}; |
| 8 | +#[cfg(feature = "discover")] |
| 9 | +use futures_core::Stream; |
| 10 | +#[cfg(feature = "discover")] |
| 11 | +use pin_project_lite::pin_project; |
| 12 | +#[cfg(feature = "discover")] |
| 13 | +use std::pin::Pin; |
| 14 | + |
| 15 | +use futures_core::ready; |
| 16 | +use std::ops; |
| 17 | +use std::task::{Context, Poll}; |
| 18 | +use tower_service::Service; |
| 19 | + |
| 20 | +use super::Load; |
| 21 | + |
| 22 | +/// A weight on [0.0, ∞]. |
| 23 | +/// |
| 24 | +/// Lesser-weighted nodes receive less traffic than heavier-weighted nodes. |
| 25 | +/// |
| 26 | +/// This is represented internally as an integer, rather than a float, so that it can implement |
| 27 | +/// `Hash` and `Eq`. |
| 28 | +#[derive(Copy, Clone, Debug, PartialEq, PartialOrd, Eq, Ord, Hash)] |
| 29 | +pub struct Weight(u32); |
| 30 | + |
| 31 | +impl Weight { |
| 32 | + /// Minimum Weight |
| 33 | + pub const MIN: Weight = Weight(0); |
| 34 | + /// Unit of Weight - what 1.0_f64 corresponds to |
| 35 | + pub const UNIT: Weight = Weight(10_000); |
| 36 | + /// Maximum Weight |
| 37 | + pub const MAX: Weight = Weight(u32::MAX); |
| 38 | +} |
| 39 | + |
| 40 | +impl Default for Weight { |
| 41 | + fn default() -> Self { |
| 42 | + Weight::UNIT |
| 43 | + } |
| 44 | +} |
| 45 | + |
| 46 | +impl From<f64> for Weight { |
| 47 | + fn from(w: f64) -> Self { |
| 48 | + if w < 0.0 || w == f64::NAN { |
| 49 | + Self::MIN |
| 50 | + } else if w == f64::INFINITY { |
| 51 | + Self::MAX |
| 52 | + } else { |
| 53 | + Weight((w * (Weight::UNIT.0 as f64)).round() as u32) |
| 54 | + } |
| 55 | + } |
| 56 | +} |
| 57 | + |
| 58 | +impl Into<f64> for Weight { |
| 59 | + fn into(self) -> f64 { |
| 60 | + (self.0 as f64) / (Weight::UNIT.0 as f64) |
| 61 | + } |
| 62 | +} |
| 63 | + |
| 64 | +impl ops::Div<Weight> for f64 { |
| 65 | + type Output = f64; |
| 66 | + |
| 67 | + fn div(self, w: Weight) -> f64 { |
| 68 | + if w == Weight::MIN { |
| 69 | + f64::INFINITY |
| 70 | + } else { |
| 71 | + let w: f64 = w.into(); |
| 72 | + self / w |
| 73 | + } |
| 74 | + } |
| 75 | +} |
| 76 | + |
| 77 | +impl ops::Div<Weight> for usize { |
| 78 | + type Output = f64; |
| 79 | + |
| 80 | + fn div(self, w: Weight) -> f64 { |
| 81 | + (self as f64) / w |
| 82 | + } |
| 83 | +} |
| 84 | + |
| 85 | +/// Measures the load of the underlying service by weighting that service's load by a constant |
| 86 | +/// weighting factor. |
| 87 | +#[derive(Clone, Debug, PartialEq, PartialOrd)] |
| 88 | +pub struct Weighted<S> { |
| 89 | + inner: S, |
| 90 | + weight: Weight, |
| 91 | +} |
| 92 | + |
| 93 | +impl<S> Weighted<S> { |
| 94 | + /// Wraps an `S`-typed service so that its load is weighted by the given weight. |
| 95 | + pub fn new<W: Into<Weight>>(inner: S, w: W) -> Self { |
| 96 | + let weight = w.into(); |
| 97 | + Self { inner, weight } |
| 98 | + } |
| 99 | +} |
| 100 | + |
| 101 | +impl<S> Load for Weighted<S> |
| 102 | +where |
| 103 | + S: Load, |
| 104 | + S::Metric: ops::Div<Weight>, |
| 105 | + <S::Metric as ops::Div<Weight>>::Output: PartialOrd, |
| 106 | +{ |
| 107 | + type Metric = <S::Metric as ops::Div<Weight>>::Output; |
| 108 | + |
| 109 | + fn load(&self) -> Self::Metric { |
| 110 | + self.inner.load() / self.weight |
| 111 | + } |
| 112 | +} |
| 113 | + |
| 114 | +impl<R, S: Service<R>> Service<R> for Weighted<S> { |
| 115 | + type Response = S::Response; |
| 116 | + type Error = S::Error; |
| 117 | + type Future = S::Future; |
| 118 | + |
| 119 | + fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> { |
| 120 | + self.inner.poll_ready(cx) |
| 121 | + } |
| 122 | + |
| 123 | + fn call(&mut self, req: R) -> Self::Future { |
| 124 | + self.inner.call(req) |
| 125 | + } |
| 126 | +} |
| 127 | + |
| 128 | +#[cfg(feature = "discover")] |
| 129 | +pin_project! { |
| 130 | + /// Wraps a `D`-typed stream of discovered services with [`Weighted`]. |
| 131 | + #[cfg_attr(docsrs, doc(cfg(feature = "discover")))] |
| 132 | + #[derive(Debug)] |
| 133 | + pub struct WeightedDiscover<D>{ |
| 134 | + #[pin] |
| 135 | + discover: D, |
| 136 | + } |
| 137 | +} |
| 138 | + |
| 139 | +impl<D> WeightedDiscover<D> { |
| 140 | + /// Wraps a [`Discover`], wrapping all of its services with [`Weighted`]. |
| 141 | + pub fn new(discover: D) -> Self { |
| 142 | + Self { discover } |
| 143 | + } |
| 144 | +} |
| 145 | + |
| 146 | +/// Allows [`tower::Discover::Key`] to expose a weight, so that they can be included in a discover |
| 147 | +/// stream |
| 148 | +pub trait HasWeight { |
| 149 | + /// Returns the [`Weight`] |
| 150 | + fn weight(&self) -> Weight; |
| 151 | +} |
| 152 | + |
| 153 | +impl<T: HasWeight> From<T> for Weighted<T> { |
| 154 | + fn from(inner: T) -> Self { |
| 155 | + let weight = inner.weight(); |
| 156 | + Self { inner, weight } |
| 157 | + } |
| 158 | +} |
| 159 | + |
| 160 | +impl<T> HasWeight for Weighted<T> { |
| 161 | + fn weight(&self) -> Weight { |
| 162 | + self.weight |
| 163 | + } |
| 164 | +} |
| 165 | + |
| 166 | +#[cfg(feature = "discover")] |
| 167 | +impl<D> Stream for WeightedDiscover<D> |
| 168 | +where |
| 169 | + D: Discover, |
| 170 | + D::Key: HasWeight, |
| 171 | +{ |
| 172 | + type Item = Result<Change<D::Key, Weighted<D::Service>>, D::Error>; |
| 173 | + |
| 174 | + fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> { |
| 175 | + use self::Change::*; |
| 176 | + |
| 177 | + let this = self.project(); |
| 178 | + let change = match ready!(this.discover.poll_discover(cx)).transpose()? { |
| 179 | + None => return Poll::Ready(None), |
| 180 | + Some(Insert(k, svc)) => { |
| 181 | + let w = k.weight(); |
| 182 | + Insert(k, Weighted::new(svc, w)) |
| 183 | + } |
| 184 | + Some(Remove(k)) => Remove(k), |
| 185 | + }; |
| 186 | + |
| 187 | + Poll::Ready(Some(Ok(change))) |
| 188 | + } |
| 189 | +} |
| 190 | + |
| 191 | +#[test] |
| 192 | +fn div_min() { |
| 193 | + assert_eq!(10.0 / Weight::MIN, f64::INFINITY); |
| 194 | + assert_eq!(10 / Weight::MIN, f64::INFINITY); |
| 195 | + assert_eq!(0 / Weight::MIN, f64::INFINITY); |
| 196 | +} |
0 commit comments