9395: Add CollectionBuffers config to keep-balance.

diff --git a/services/keep-balance/balance.go b/services/keep-balance/balance.go
index 0ca6e713c40e739a585c2636e47a15b0bb656783..2d1a59e8909bc250b2ca995775496f1839adf9f9 100644 (file)
--- a/services/keep-balance/balance.go
+++ b/services/keep-balance/balance.go
@@ -80,7 +80,7 @@ func (bal *Balancer) Run(config Config, runOptions RunOptions) (err error) {
                        return
                }
        }
-       if err = bal.GetCurrentState(&config.Client, config.CollectionBatchSize); err != nil {
+       if err = bal.GetCurrentState(&config.Client, config.CollectionBatchSize, config.CollectionBuffers); err != nil {
                return
        }
        bal.ComputeChangeSets()
@@ -190,7 +190,7 @@ func (bal *Balancer) ClearTrashLists(c *arvados.Client) error {
 // collection manifests in the database (API server).
 //
 // It encodes the resulting information in BlockStateMap.
-func (bal *Balancer) GetCurrentState(c *arvados.Client, pageSize int) error {
+func (bal *Balancer) GetCurrentState(c *arvados.Client, pageSize, bufs int) error {
        defer timeMe(bal.Logger, "GetCurrentState")()
        bal.BlockStateMap = NewBlockStateMap()
 
@@ -225,7 +225,7 @@ func (bal *Balancer) GetCurrentState(c *arvados.Client, pageSize int) error {
        // collQ buffers incoming collections so we can start fetching
        // the next page without waiting for the current page to
        // finish processing.
-       collQ := make(chan arvados.Collection, 1000)
+       collQ := make(chan arvados.Collection, bufs)
 
        // Start a goroutine to process collections. (We could use a
        // worker pool here, but even with a single worker we already

diff --git a/services/keep-balance/main.go b/services/keep-balance/main.go
index 6e7d70d4e842a66323597cee51f75b4114117273..364bb3ffd3f6b437ec1d46edf52223539fb1161b 100644 (file)
--- a/services/keep-balance/main.go
+++ b/services/keep-balance/main.go
@@ -31,6 +31,11 @@ type Config struct {
 
        // Number of collections to request in each API call
        CollectionBatchSize int
+
+       // Max collections to buffer in memory (bigger values consume
+       // more memory, but can reduce store-and-forward latency when
+       // fetching pages)
+       CollectionBuffers int
 }
 
 // RunOptions controls runtime behavior. The flags/options that belong

diff --git a/services/keep-balance/usage.go b/services/keep-balance/usage.go
index c5363641cab6453289bfafc1553241fb8d8e3eca..2273f3a57fef08ec716320bcce660470e24ecb3a 100644 (file)
--- a/services/keep-balance/usage.go
+++ b/services/keep-balance/usage.go
@@ -17,7 +17,8 @@ var exampleConfigFile = []byte(`
            "disk"
        ],
        "RunPeriod": "600s",
-       "CollectionBatchSize": 100000
+       "CollectionBatchSize": 100000,
+       "CollectionBuffers": 1000
     }`)
 
 func usage() {