shutdown limiters on ticker shutdown

Author: linkdata

operation.go

@@ -95,6 +95,8 @@ func (op *Operation) run(ch chan<- int64) {
 				select {
 				case <-stopCh:
 					return
+				case <-op.Ticker.doneCh:
+					return
 				case limitCh <- batch:
 					todo -= batch
 					todo += op.avail.Swap(0)

ticker_test.go

@@ -2,6 +2,7 @@ package bwlimit
 
 import (
 	"bytes"
+	"io"
 	"testing"
 	"testing/synctest"
 	"time"
@@ -27,3 +28,34 @@ func TestTicker_NewTicker_NewLimiter(t *testing.T) {
 		}
 	})
 }
+
+func TestTicker_Stop_unblocksLimitedOperation(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		ticker := NewTicker()
+		l := ticker.NewLimiter(1000)
+		defer l.Stop()
+
+		ticker.Stop()
+
+		done := make(chan struct{})
+		var n int
+		var err error
+		go func() {
+			n, err = l.Reads.io(bytes.NewReader(make([]byte, 1)).Read, make([]byte, 1))
+			close(done)
+		}()
+
+		select {
+		case <-done:
+		case <-time.After(10 * time.Second):
+			t.Fatal("read stalled after ticker stop")
+		}
+
+		if n != 0 {
+			t.Fatalf("got n=%d, want 0", n)
+		}
+		if err != io.EOF {
+			t.Fatalf("got err=%v, want %v", err, io.EOF)
+		}
+	})
+}