accounting: cull the old time ranges when possible to save memory

fs/accounting/stats.go

@@ -37,8 +37,9 @@ type StatsInfo struct {
 	renameQueueSize   int64
 	deletes           int64
 	inProgress        *inProgress
-	startedTransfers  []*Transfer // currently active transfers
-	oldTimeRanges     timeRanges  // a merged list of time ranges for the transfers
+	startedTransfers  []*Transfer   // currently active transfers
+	oldTimeRanges     timeRanges    // a merged list of time ranges for the transfers
+	oldDuration       time.Duration // duration of transfers we have culled
 }
 
 // NewStats creates an initialised StatsInfo
@@ -155,6 +156,21 @@ func (trs *timeRanges) merge() {
 	*trs = newTrs
 }
 
+// cull remove any ranges whose start and end are before cutoff
+// returning their duration sum
+func (trs *timeRanges) cull(cutoff time.Time) (d time.Duration) {
+	var newTrs = (*trs)[:0]
+	for _, tr := range *trs {
+		if cutoff.Before(tr.start) || cutoff.Before(tr.end) {
+			newTrs = append(newTrs, tr)
+		} else {
+			d += tr.end.Sub(tr.start)
+		}
+	}
+	*trs = newTrs
+	return d
+}
+
 // total the time out of the time ranges
 func (trs timeRanges) total() (total time.Duration) {
 	for _, tr := range trs {
@@ -182,7 +198,7 @@ func (s *StatsInfo) totalDuration() time.Duration {
 	}
 
 	timeRanges.merge()
-	return timeRanges.total()
+	return s.oldDuration + timeRanges.total()
 }
 
 // eta returns the ETA of the current operation,
@@ -436,6 +452,7 @@ func (s *StatsInfo) ResetCounters() {
 	s.transfers = 0
 	s.deletes = 0
 	s.startedTransfers = nil
+	s.oldDuration = 0
 }
 
 // ResetErrors sets the errors count to 0 and resets lastError, fatalError and retryError
@@ -568,16 +585,30 @@ func (s *StatsInfo) AddTransfer(transfer *Transfer) {
 //
 // Must be called with the lock held
 func (s *StatsInfo) removeTransfer(transfer *Transfer, i int) {
+	now := time.Now()
+
 	// add finished transfer onto old time ranges
 	start, end := transfer.TimeRange()
 	if end.IsZero() {
-		end = time.Now()
+		end = now
 	}
 	s.oldTimeRanges = append(s.oldTimeRanges, timeRange{start, end})
 	s.oldTimeRanges.merge()
 
 	// remove the found entry
 	s.startedTransfers = append(s.startedTransfers[:i], s.startedTransfers[i+1:]...)
+
+	// Find youngest active transfer
+	oldestStart := now
+	for i := range s.startedTransfers {
+		start, _ := s.startedTransfers[i].TimeRange()
+		if start.Before(oldestStart) {
+			oldestStart = start
+		}
+	}
+
+	// remove old entries older than that
+	s.oldDuration += s.oldTimeRanges.cull(oldestStart)
 }
 
 // RemoveTransfer removes a reference to the started transfer.

fs/accounting/stats_test.go

@@ -254,6 +254,14 @@ func makeTimeRanges(t *testing.T, in []string) timeRanges {
 	return trs
 }
 
+func (trs timeRanges) toStrings() (out []string) {
+	out = []string{}
+	for _, tr := range trs {
+		out = append(out, fmt.Sprintf("%d-%d", tr.start.Unix(), tr.end.Unix()))
+	}
+	return out
+}
+
 func TestTimeRangeMerge(t *testing.T) {
 	for _, test := range []struct {
 		in   []string
@@ -293,15 +301,80 @@ func TestTimeRangeMerge(t *testing.T) {
 		in := makeTimeRanges(t, test.in)
 		in.merge()
 
-		got := []string{}
-		for _, tr := range in {
-			got = append(got, fmt.Sprintf("%d-%d", tr.start.Unix(), tr.end.Unix()))
-		}
-
+		got := in.toStrings()
 		assert.Equal(t, test.want, got)
 	}
 }
 
+func TestTimeRangeCull(t *testing.T) {
+	for _, test := range []struct {
+		in           []string
+		cutoff       int64
+		want         []string
+		wantDuration time.Duration
+	}{{
+		in:           []string{},
+		cutoff:       1,
+		want:         []string{},
+		wantDuration: 0 * time.Second,
+	}, {
+		in:           []string{"1-2"},
+		cutoff:       1,
+		want:         []string{"1-2"},
+		wantDuration: 0 * time.Second,
+	}, {
+		in:           []string{"2-5", "7-9"},
+		cutoff:       1,
+		want:         []string{"2-5", "7-9"},
+		wantDuration: 0 * time.Second,
+	}, {
+		in:           []string{"2-5", "7-9"},
+		cutoff:       4,
+		want:         []string{"2-5", "7-9"},
+		wantDuration: 0 * time.Second,
+	}, {
+		in:           []string{"2-5", "7-9"},
+		cutoff:       5,
+		want:         []string{"7-9"},
+		wantDuration: 3 * time.Second,
+	}, {
+		in:           []string{"2-5", "7-9", "2-5", "2-5"},
+		cutoff:       6,
+		want:         []string{"7-9"},
+		wantDuration: 9 * time.Second,
+	}, {
+		in:           []string{"7-9", "3-3", "2-5"},
+		cutoff:       7,
+		want:         []string{"7-9"},
+		wantDuration: 3 * time.Second,
+	}, {
+		in:           []string{"2-5", "7-9"},
+		cutoff:       8,
+		want:         []string{"7-9"},
+		wantDuration: 3 * time.Second,
+	}, {
+		in:           []string{"2-5", "7-9"},
+		cutoff:       9,
+		want:         []string{},
+		wantDuration: 5 * time.Second,
+	}, {
+		in:           []string{"2-5", "7-9"},
+		cutoff:       10,
+		want:         []string{},
+		wantDuration: 5 * time.Second,
+	}} {
+
+		in := makeTimeRanges(t, test.in)
+		cutoff := time.Unix(test.cutoff, 0)
+		gotDuration := in.cull(cutoff)
+
+		what := fmt.Sprintf("in=%q, cutoff=%d", test.in, test.cutoff)
+		got := in.toStrings()
+		assert.Equal(t, test.want, got, what)
+		assert.Equal(t, test.wantDuration, gotDuration, what)
+	}
+}
+
 func TestTimeRangeDuration(t *testing.T) {
 	assert.Equal(t, 0*time.Second, timeRanges{}.total())
 	assert.Equal(t, 1*time.Second, makeTimeRanges(t, []string{"1-2"}).total())