use crate::{Multiaddr, core::{Transport, transport::{ListenerEvent, TransportError}}};
use futures::{prelude::*, try_ready};
use lazy_static::lazy_static;
use parking_lot::Mutex;
use smallvec::{smallvec, SmallVec};
use std::{cmp, io, io::Read, io::Write, sync::Arc, time::Duration};
use wasm_timer::Instant;
#[derive(Clone)]
pub struct BandwidthLogging<TInner> {
inner: TInner,
sinks: Arc<BandwidthSinks>,
}
impl<TInner> BandwidthLogging<TInner> {
#[inline]
pub fn new(inner: TInner, period: Duration) -> (Self, Arc<BandwidthSinks>) {
let mut period_seconds = cmp::min(period.as_secs(), 86400) as u32;
if period.subsec_nanos() > 0 {
period_seconds += 1;
}
let sink = Arc::new(BandwidthSinks {
download: Mutex::new(BandwidthSink::new(period_seconds)),
upload: Mutex::new(BandwidthSink::new(period_seconds)),
});
let trans = BandwidthLogging {
inner,
sinks: sink.clone(),
};
(trans, sink)
}
}
impl<TInner> Transport for BandwidthLogging<TInner>
where
TInner: Transport,
{
type Output = BandwidthConnecLogging<TInner::Output>;
type Error = TInner::Error;
type Listener = BandwidthListener<TInner::Listener>;
type ListenerUpgrade = BandwidthFuture<TInner::ListenerUpgrade>;
type Dial = BandwidthFuture<TInner::Dial>;
fn listen_on(self, addr: Multiaddr) -> Result<Self::Listener, TransportError<Self::Error>> {
let sinks = self.sinks;
self.inner
.listen_on(addr)
.map(move |inner| BandwidthListener { inner, sinks })
}
fn dial(self, addr: Multiaddr) -> Result<Self::Dial, TransportError<Self::Error>> {
let sinks = self.sinks;
self.inner
.dial(addr)
.map(move |fut| BandwidthFuture { inner: fut, sinks })
}
}
pub struct BandwidthListener<TInner> {
inner: TInner,
sinks: Arc<BandwidthSinks>,
}
impl<TInner, TConn> Stream for BandwidthListener<TInner>
where
TInner: Stream<Item = ListenerEvent<TConn>>,
{
type Item = ListenerEvent<BandwidthFuture<TConn>>;
type Error = TInner::Error;
fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
let event = match try_ready!(self.inner.poll()) {
Some(v) => v,
None => return Ok(Async::Ready(None))
};
let event = event.map(|inner| {
BandwidthFuture { inner, sinks: self.sinks.clone() }
});
Ok(Async::Ready(Some(event)))
}
}
pub struct BandwidthFuture<TInner> {
inner: TInner,
sinks: Arc<BandwidthSinks>,
}
impl<TInner> Future for BandwidthFuture<TInner>
where TInner: Future,
{
type Item = BandwidthConnecLogging<TInner::Item>;
type Error = TInner::Error;
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
let inner = try_ready!(self.inner.poll());
Ok(Async::Ready(BandwidthConnecLogging {
inner,
sinks: self.sinks.clone(),
}))
}
}
pub struct BandwidthSinks {
download: Mutex<BandwidthSink>,
upload: Mutex<BandwidthSink>,
}
impl BandwidthSinks {
#[inline]
pub fn average_download_per_sec(&self) -> u64 {
self.download.lock().get()
}
#[inline]
pub fn average_upload_per_sec(&self) -> u64 {
self.upload.lock().get()
}
}
pub struct BandwidthConnecLogging<TInner> {
inner: TInner,
sinks: Arc<BandwidthSinks>,
}
impl<TInner> Read for BandwidthConnecLogging<TInner>
where TInner: Read
{
#[inline]
fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
let num_bytes = self.inner.read(buf)?;
self.sinks.download.lock().inject(num_bytes);
Ok(num_bytes)
}
}
impl<TInner> tokio_io::AsyncRead for BandwidthConnecLogging<TInner>
where TInner: tokio_io::AsyncRead
{
unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [u8]) -> bool {
self.inner.prepare_uninitialized_buffer(buf)
}
fn read_buf<B: bytes::BufMut>(&mut self, buf: &mut B) -> Poll<usize, io::Error> {
self.inner.read_buf(buf)
}
}
impl<TInner> Write for BandwidthConnecLogging<TInner>
where TInner: Write
{
#[inline]
fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
let num_bytes = self.inner.write(buf)?;
self.sinks.upload.lock().inject(num_bytes);
Ok(num_bytes)
}
#[inline]
fn flush(&mut self) -> io::Result<()> {
self.inner.flush()
}
}
impl<TInner> tokio_io::AsyncWrite for BandwidthConnecLogging<TInner>
where TInner: tokio_io::AsyncWrite
{
#[inline]
fn shutdown(&mut self) -> Poll<(), io::Error> {
self.inner.shutdown()
}
}
#[inline]
fn current_second() -> u32 {
lazy_static! {
static ref EPOCH: Instant = Instant::now();
}
EPOCH.elapsed().as_secs() as u32
}
struct BandwidthSink {
bytes: SmallVec<[u64; 8]>,
latest_update: u32,
}
impl BandwidthSink {
fn new(seconds: u32) -> Self {
BandwidthSink {
bytes: smallvec![0; seconds as usize + 1],
latest_update: current_second(),
}
}
fn get(&mut self) -> u64 {
self.update();
let seconds = self.bytes.len() - 1;
self.bytes.iter()
.take(seconds)
.fold(0u64, |a, &b| a.saturating_add(b)) / seconds as u64
}
fn inject(&mut self, bytes: usize) {
self.update();
if let Some(last) = self.bytes.last_mut() {
*last = last.saturating_add(bytes as u64);
}
}
fn update(&mut self) {
let current_second = current_second();
debug_assert!(current_second >= self.latest_update);
let num_iter = cmp::min(current_second - self.latest_update, self.bytes.len() as u32);
for _ in 0..num_iter {
self.bytes.remove(0);
self.bytes.push(0);
}
self.latest_update = current_second;
}
}
#[cfg(test)]
mod tests {
use std::{thread, time::Duration};
use super::*;
#[test]
fn sink_works() {
let mut sink = BandwidthSink::new(5);
sink.inject(100);
thread::sleep(Duration::from_millis(1000));
assert_eq!(sink.get(), 20);
sink.inject(100);
thread::sleep(Duration::from_millis(1000));
assert_eq!(sink.get(), 40);
sink.inject(100);
thread::sleep(Duration::from_millis(1000));
assert_eq!(sink.get(), 60);
sink.inject(100);
thread::sleep(Duration::from_millis(1000));
assert_eq!(sink.get(), 80);
sink.inject(100);
thread::sleep(Duration::from_millis(1000));
assert_eq!(sink.get(), 100);
thread::sleep(Duration::from_millis(1000));
assert_eq!(sink.get(), 80);
}
}