From fbc265f85d3c35b372370ae66e41361d98c06152 Mon Sep 17 00:00:00 2001
From: Alain Spineux <alain@baculasystems.com>
Date: Wed, 14 Jun 2017 14:01:12 +0200
Subject: [PATCH] core: bwlimit measure bandwidth

- measure BW on the 10 last seconds (or more)
---
 bacula/src/lib/bwlimit.c | 74 ++++++++++++++++++++++++++++++++++++++--
 bacula/src/lib/bwlimit.h | 21 ++++++++++--
 2 files changed, 90 insertions(+), 5 deletions(-)

diff --git a/bacula/src/lib/bwlimit.c b/bacula/src/lib/bwlimit.c
index edabfd6cd1..8d24f95883 100644
--- a/bacula/src/lib/bwlimit.c
+++ b/bacula/src/lib/bwlimit.c
@@ -22,6 +22,68 @@
 
 #define ONE_SEC 1000000L /* number of microseconds in a second */
 
+void bwlimit::reset_sample()
+{
+   memset(samples_time, '\0', sizeof(samples_time));
+   memset(samples_byte, '\0', sizeof(samples_byte));
+   memset(samples_sleep, '\0', sizeof(samples_sleep));
+   total_time=total_byte=total_sleep=0;
+   current_sample=0;
+   current_byte=0;
+}
+
+void bwlimit::push_sample(int64_t t, int64_t bytes, int64_t sleep)
+{
+   // accumulate data in current sample
+   current_time+=t;
+   current_byte+=bytes;
+   current_sleep+=sleep;
+   if (current_time>ONE_SEC) {
+      // we have accumulated enough data for this sample go to the next one
+      // First "pop" the data from the total
+      total_time-=samples_time[current_sample];
+      total_byte-=samples_byte[current_sample];
+      total_sleep-=samples_sleep[current_sample];
+      // Push the new one
+      total_time+=current_time;
+      total_byte+=current_byte;
+      total_sleep+=current_sleep;
+      // record the data in the table
+      samples_time[current_sample]=current_time;
+      samples_byte[current_sample]=current_byte;
+      samples_sleep[current_sample]=current_sleep;
+      // be ready for next sample
+      current_time=0;
+      current_byte=0;
+      current_sleep=0;
+      current_sample=(current_sample+1)%sample_capacity;
+   }
+}
+
+void bwlimit::get_total(int64_t *t, int64_t *bytes, int64_t *sleep)
+{
+   pthread_mutex_lock(&m_bw_mutex);
+   *t=total_time+current_time;
+   *bytes=total_byte+current_byte;
+   *sleep=total_sleep+current_sleep;
+   pthread_mutex_unlock(&m_bw_mutex);
+}
+
+int64_t bwlimit::get_bw()
+{
+   int64_t bw = 0;
+   btime_t temp = get_current_btime() - m_last_tick;
+   if (temp < 0) {
+      temp = 0;
+   }
+   pthread_mutex_lock(&m_bw_mutex);
+   if (total_time+current_time>0) {
+      bw=(total_byte+current_byte)*ONE_SEC/(total_time+current_time+temp);
+   }
+   pthread_mutex_unlock(&m_bw_mutex);
+   return bw;
+}
+
 void bwlimit::control_bwlimit(int bytes)
 {
    btime_t now, temp;
@@ -36,14 +98,16 @@ void bwlimit::control_bwlimit(int bytes)
    if (temp < 0 || temp > m_backlog_limit) { /* Take care of clock problems (>10s) or back in time */
       m_nb_bytes = bytes;
       m_last_tick = now;
+      reset_sample();
       return;
    }
 
    /* remove what as been consumed */
-   m_nb_bytes -= bytes;;
+   m_nb_bytes -= bytes;
 
    /* Less than 0.1ms since the last call, see the next time */
    if (temp < 100) {
+      push_sample(temp, bytes, 0);
       return;
    }
 
@@ -52,14 +116,18 @@ void bwlimit::control_bwlimit(int bytes)
    m_last_tick = now;
 
    /* limit the backlog */
-   if (m_nb_bytes > m_backlog_limit) {
-      m_nb_bytes = m_backlog_limit;
+   if (m_nb_bytes > m_backlog_limit*m_bwlimit) {
+      m_nb_bytes = m_backlog_limit*m_bwlimit;
+      push_sample(temp, bytes, 0);
    } else if (m_nb_bytes < 0) {
       /* What exceed should be converted in sleep time */
       int64_t usec_sleep = (int64_t)(-m_nb_bytes /((double)m_bwlimit / ONE_SEC));
       if (usec_sleep > 100) {
+         pthread_mutex_unlock(&m_bw_mutex);
          bmicrosleep(usec_sleep / ONE_SEC, usec_sleep % ONE_SEC);
+         pthread_mutex_lock(&m_bw_mutex);
       }
+      push_sample(temp, bytes, usec_sleep>100?usec_sleep:0);
       /* m_nb_bytes & m_last_tick will be updated at next iteration */
    }
 }
diff --git a/bacula/src/lib/bwlimit.h b/bacula/src/lib/bwlimit.h
index a26bbe41d0..3b4312a10d 100644
--- a/bacula/src/lib/bwlimit.h
+++ b/bacula/src/lib/bwlimit.h
@@ -22,18 +22,31 @@
 
 class bwlimit: public SMARTALLOC
 {
+   static const int sample_capacity=10;
+
 private:
    int64_t m_bwlimit;           /* set to limit bandwidth */
    int64_t m_nb_bytes;          /* bytes sent/recv since the last tick */
    btime_t m_last_tick;         /* last tick used by bwlimit */
    btime_t m_backlog_limit;     /* don't handle more backlog thna this us */
-   int64_t m_provision;
    pthread_mutex_t m_bw_mutex;
 
+   int64_t samples_time[sample_capacity];
+   int64_t samples_byte[sample_capacity];
+   int64_t samples_sleep[sample_capacity];
+   int64_t total_time, total_byte, total_sleep;
+   int64_t current_time, current_byte, current_sleep;
+   int64_t current_sample;
+
+   void push_sample(int64_t t, int64_t bytes, int64_t sleep);
+
 public:
    bwlimit(int64_t speed=0): m_bwlimit(speed), m_nb_bytes(0), m_last_tick(0),
-         m_backlog_limit(10*1000*1000) {
+         m_backlog_limit(10*1000*1000),
+         total_time(0), total_byte(0), total_sleep(0), current_sample(0)
+   {
       pthread_mutex_init(&m_bw_mutex, NULL);
+      reset_sample();
    };
    ~bwlimit() {
       pthread_mutex_destroy(&m_bw_mutex);
@@ -41,6 +54,10 @@ public:
 
    void control_bwlimit(int bytes);
    void set_bwlimit(int64_t maxspeed) { m_bwlimit = maxspeed; };
+   int64_t get_bwlimit() { return m_bwlimit; };
    bool use_bwlimit() { return m_bwlimit > 0;};
+   int64_t get_bw();
+   void get_total(int64_t *t, int64_t *bytes, int64_t *sleep);
+   void reset_sample();
 };
 #endif
-- 
2.39.2