avoid race conditions and add a method to prevent channel leakage

.github/workflows/test.yml

@@ -38,7 +38,7 @@ jobs:
       - name: Checkout code
         uses: actions/checkout@v2
       - name: Run tests
-        run: go test -v -covermode=count
+        run: go test -v -covermode=count && go test -race -v
 
   coverage:
     runs-on: ubuntu-latest

README.md

@@ -47,6 +47,9 @@ func main() {
     // This function has to call to ensure all goroutines have finished
     // after close the main program.
     c.WaitAllDone()
+
+    // This function will release all the channels associated with goccm
+    c.Release()
 }
 ```
 
@@ -76,5 +79,8 @@ func main() {
 
     // Returns the number of goroutines which are running
     c.RunningCount()
+
+    // Closes all the channels associated with goccm
+    c.Release()
 }
 ```

goccm.go

@@ -1,6 +1,8 @@
 package goccm
 
-import "sync/atomic"
+import (
+	"sync/atomic"
+)
 
 type (
 	// ConcurrencyManager Interface
@@ -19,6 +21,9 @@ type (
 
 		// RunningCount Returns the number of goroutines which are running
 		RunningCount() int32
+
+		// Release Closes all the channels associated with goccm
+		Release()
 	}
 
 	concurrencyManager struct {
@@ -35,7 +40,7 @@ type (
 		allDoneCh chan bool
 
 		// The close flag allows we know when we can close the manager
-		closed bool
+		closed int32
 
 		// The running count allows we know the number of goroutines are running
 		runningCount int32
@@ -75,7 +80,7 @@ func (c *concurrencyManager) controller() {
 
 		// When the closed flag is set,
 		// we need to close the manager if it doesn't have any running goroutine
-		if c.closed && c.runningCount == 0 {
+		if c.IsClosed() && c.RunningCount() == 0 {
 			break
 		}
 	}
@@ -84,6 +89,10 @@ func (c *concurrencyManager) controller() {
 	c.allDoneCh <- true
 }
 
+func (c *concurrencyManager) IsClosed() bool {
+	return atomic.LoadInt32(&c.closed) > 0
+}
+
 // Wait until a slot is available for the new goroutine.
 // A goroutine have to start after this function.
 func (c *concurrencyManager) Wait() {
@@ -106,7 +115,7 @@ func (c *concurrencyManager) Done() {
 
 // Close the manager manually
 func (c *concurrencyManager) Close() {
-	c.closed = true
+	atomic.AddInt32(&c.closed, 1)
 }
 
 // WaitAllDone Wait for all goroutines are done
@@ -120,5 +129,12 @@ func (c *concurrencyManager) WaitAllDone() {
 
 // RunningCount Returns the number of goroutines which are running
 func (c *concurrencyManager) RunningCount() int32 {
-	return c.runningCount
+	return atomic.LoadInt32(&c.runningCount)
+}
+
+// Release Closes all the channels associated with goccm
+func (c *concurrencyManager) Release() {
+	close(c.managerCh)
+	close(c.doneCh)
+	close(c.allDoneCh)
 }

goccm_test.go

@@ -2,6 +2,7 @@ package goccm
 
 import (
 	"fmt"
+	"sync/atomic"
 	"testing"
 	"time"
 )
@@ -20,13 +21,13 @@ func TestExample(t *testing.T) {
 }
 
 func TestManuallyClose(t *testing.T) {
-	executedJobs := 0
+	executedJobs := int32(0)
 	c := New(3)
 	for i := 1; i <= 1000; i++ {
 		c.Wait()
 		go func() {
-			executedJobs++
-			fmt.Printf("Executed jobs %d\n", executedJobs)
+			atomic.AddInt32(&executedJobs, 1)
+			fmt.Printf("Executed jobs %d\n", atomic.LoadInt32(&executedJobs))
 			time.Sleep(2 * time.Second)
 			c.Done()
 		}()
@@ -46,8 +47,8 @@ func TestConcurrency(t *testing.T) {
 		c.Wait()
 		go func(i int) {
 			fmt.Printf("Current running jobs %d\n", c.RunningCount())
-			if c.RunningCount() > testMaxRunningJobs {
-				testMaxRunningJobs = c.RunningCount()
+			if c.RunningCount() > atomic.LoadInt32(&testMaxRunningJobs) {
+				atomic.StoreInt32(&testMaxRunningJobs, c.RunningCount())
 			}
 			time.Sleep(2 * time.Second)
 			c.Done()