Batch closer can wait merely on the closer goroutines (treeverse#1392)

arielshaqed · web-flow · commit ebebde11bcb3 · 2021-02-01T21:42:17.000+02:00
Simplifies using a waitgroup, which is now a regular blocker usage (Add only called at
startup) rather than dynamically increasing and decreasing.  This way does not have to
synchronize as closely across the waitgroup.
diff --git a/graveler/committed/batch.go b/graveler/committed/batch.go
@@ -16,7 +16,6 @@ type BatchCloser struct {
 	err     error
 
 	wg sync.WaitGroup
-
 	ch chan ResultCloser
 }
 
@@ -27,6 +26,7 @@ func NewBatchCloser(numClosers int) *BatchCloser {
 		ch: make(chan ResultCloser),
 	}
 
+	ret.wg.Add(numClosers)
 	for i := 0; i < numClosers; i++ {
 		go ret.handleClose()
 	}
@@ -51,7 +51,6 @@ func (bc *BatchCloser) CloseWriterAsync(w ResultCloser) error {
 		return bc.err
 	}
 
-	bc.wg.Add(1)
 	bc.mu.Unlock()
 	bc.ch <- w
 
@@ -62,10 +61,10 @@ func (bc *BatchCloser) handleClose() {
 	for w := range bc.ch {
 		bc.closeWriter(w)
 	}
+	bc.wg.Done()
 }
 
 func (bc *BatchCloser) closeWriter(w ResultCloser) {
-	defer bc.wg.Done()
 	res, err := w.Close()
 
 	// long operation is over, we can lock to have synchronized access to err and results

Original file line number	Diff line number	Diff line change
`@@ -16,7 +16,6 @@ type BatchCloser struct {`
`16`	`16`	`err error`
`17`	`17`
`18`	`18`	`wg sync.WaitGroup`
`19`		`-`
`20`	`19`	`ch chan ResultCloser`
`21`	`20`	`}`
`22`	`21`
`@@ -27,6 +26,7 @@ func NewBatchCloser(numClosers int) *BatchCloser {`
`27`	`26`	`ch: make(chan ResultCloser),`
`28`	`27`	`}`
`29`	`28`
	`29`	`+ ret.wg.Add(numClosers)`
`30`	`30`	`for i := 0; i < numClosers; i++ {`
`31`	`31`	`go ret.handleClose()`
`32`	`32`	`}`
`@@ -51,7 +51,6 @@ func (bc *BatchCloser) CloseWriterAsync(w ResultCloser) error {`
`51`	`51`	`return bc.err`
`52`	`52`	`}`
`53`	`53`
`54`		`- bc.wg.Add(1)`
`55`	`54`	`bc.mu.Unlock()`
`56`	`55`	`bc.ch <- w`
`57`	`56`
`@@ -62,10 +61,10 @@ func (bc *BatchCloser) handleClose() {`
`62`	`61`	`for w := range bc.ch {`
`63`	`62`	`bc.closeWriter(w)`
`64`	`63`	`}`
	`64`	`+ bc.wg.Done()`
`65`	`65`	`}`
`66`	`66`
`67`	`67`	`func (bc *BatchCloser) closeWriter(w ResultCloser) {`
`68`		`- defer bc.wg.Done()`
`69`	`68`	`res, err := w.Close()`
`70`	`69`
`71`	`70`	`// long operation is over, we can lock to have synchronized access to err and results`