change *Action to Action, add a new func named Map

Author: SYM01

README.md

@@ -12,7 +12,7 @@ Features:
 - [x] `ipfilter`: a high performance IP filter based on AVL tree
 - [x] `concurrent`: some well-tested concurrent features, such as Goroutine pool
 
-More common algorithms and data structures will be implemented here.
+More and more common algorithms and data structures will be implemented here.
 
 
 ## Install

concurrent/pool.go

@@ -29,24 +29,31 @@ type Action struct {
 	// Runner is a func need to be executed. It may return a list of new
 	// actions. These actions will be executed after the current action had
 	// finished
-	Runner func() []*Action
+	Runner func() []Action
 
 	// Dependencies for the current action. The current action will only be
 	// executed after all dependencies had been finished
-	Dependencies []*Action
+	Dependencies []Action
 }
 
 // ActionPool is an action pool based on goroutines
 type ActionPool interface {
 	// Do an action. It's panic-free if using it correctly.
-	// It will panic if you do a nil action, or do an action in a closed pool.
-	Do(*Action)
+	// It will panic if you do an action in a closed pool.
+	Do(Action)
+
+	// Map returns a slice of result that applies `f` to every item of `input`.
+	// The `ret` is in-order.
+	Map(input []interface{}, f func(interface{}) interface{}) (ret []interface{})
+
+	// Wait until all the actions had been finished.
+	Wait()
 
 	// Wait until all the actions had been finished, and close the pool.
 	WaitAndClose()
 
 	// Set a logger to output some err msg when panic. os.Stderr will be used
-	// by default
+	// by default.
 	SetLogger(io.Writer)
 }
 
@@ -59,11 +66,7 @@ type actionPool struct {
 }
 
 // Do implements ActionPool's method.
-func (p *actionPool) Do(a *Action) {
-	if a == nil {
-		panic("ActionPool: the action is nil")
-	}
-
+func (p *actionPool) Do(a Action) {
 	if atomic.LoadUint32(&p.closed) > 0 {
 		panic("ActionPool: the current pool had already been closed.")
 	}
@@ -76,7 +79,7 @@ func (p *actionPool) Do(a *Action) {
 }
 
 // do an action and free a slot when finished executing.
-func (p *actionPool) do(a *Action, parentWg *sync.WaitGroup) {
+func (p *actionPool) do(a Action, parentWg *sync.WaitGroup) {
 	// clean up
 	defer func() {
 		// parent waitgroup
@@ -122,6 +125,11 @@ func (p *actionPool) panicCatcher() {
 	}
 }
 
+// Wait implements ActionPool's method.
+func (p *actionPool) Wait() {
+	p.wg.Wait()
+}
+
 // WaitAndClose implements ActionPool's method.
 func (p *actionPool) WaitAndClose() {
 	atomic.StoreUint32(&p.closed, 1)
@@ -132,3 +140,28 @@ func (p *actionPool) WaitAndClose() {
 func (p *actionPool) SetLogger(l io.Writer) {
 	p.logger = l
 }
+
+// Map implements ActionPool's method.
+func (p *actionPool) Map(input []interface{}, f func(interface{}) interface{}) (ret []interface{}) {
+	if len(input) == 0 {
+		return
+	}
+
+	ret = make([]interface{}, len(input))
+	mu := sync.Mutex{}
+	for tmpIdx := range input {
+		idx := tmpIdx // just copy the value
+		p.Do(Action{
+			Runner: func() []Action {
+				r := f(input[idx])
+				mu.Lock()
+				ret[idx] = r
+				mu.Unlock()
+				return nil
+			},
+		})
+	}
+
+	p.Wait()
+	return
+}

concurrent/pool_test.go

@@ -2,6 +2,7 @@ package concurrent_test
 
 import (
 	"bytes"
+	"fmt"
 	"strconv"
 	"strings"
 	"sync/atomic"
@@ -19,8 +20,8 @@ func TestPoolWithOneSlot(t *testing.T) {
 	for i := 0; i < 100; i++ {
 		c := strconv.Itoa(i) + "."
 		expected.WriteString(c)
-		pool.Do(&concurrent.Action{
-			Runner: func() []*concurrent.Action {
+		pool.Do(concurrent.Action{
+			Runner: func() []concurrent.Action {
 				buf.WriteString(c)
 				return nil
 			},
@@ -45,22 +46,22 @@ func TestPoolWithOneSlotAndDeps(t *testing.T) {
 		expected.WriteString("2@")
 		expected.WriteString(c)
 
-		pool.Do(&concurrent.Action{
-			Runner: func() []*concurrent.Action {
+		pool.Do(concurrent.Action{
+			Runner: func() []concurrent.Action {
 				buf.WriteString(strconv.Itoa(int(atomic.LoadInt32(p))))
 				buf.WriteString("@")
 				buf.WriteString(c)
 				return nil
 			},
-			Dependencies: []*concurrent.Action{
+			Dependencies: []concurrent.Action{
 				{
-					Runner: func() []*concurrent.Action {
+					Runner: func() []concurrent.Action {
 						atomic.AddInt32(p, 1)
 						return nil
 					},
 				},
 				{
-					Runner: func() []*concurrent.Action {
+					Runner: func() []concurrent.Action {
 						atomic.AddInt32(p, 1)
 						return nil
 					},
@@ -81,27 +82,27 @@ func TestPoolConcurrency(t *testing.T) {
 	tStart := time.Now()
 
 	for i := 0; i < 5; i++ {
-		pool.Do(&concurrent.Action{
-			Runner: func() []*concurrent.Action {
+		pool.Do(concurrent.Action{
+			Runner: func() []concurrent.Action {
 				// 2s will elapse
 				time.Sleep(time2sleep)
 
 				// 2s will elapse
-				return []*concurrent.Action{
+				return []concurrent.Action{
 					{
-						Runner: func() []*concurrent.Action {
+						Runner: func() []concurrent.Action {
 							time.Sleep(time2sleep)
 							return nil
 						},
 					},
 					{
-						Runner: func() []*concurrent.Action {
+						Runner: func() []concurrent.Action {
 							time.Sleep(time2sleep)
 							return nil
 						},
 					},
 					{
-						Runner: func() []*concurrent.Action {
+						Runner: func() []concurrent.Action {
 							time.Sleep(time2sleep)
 							return nil
 						},
@@ -119,25 +120,22 @@ func TestPoolConcurrency(t *testing.T) {
 
 func TestPanic(t *testing.T) {
 	pool := concurrent.NewActionPool(0)
-	expectPanic(func() {
-		pool.Do(nil)
-	}, t)
-
-	pool.Do(&concurrent.Action{})
+	pool.Map(nil, nil)
+	pool.Do(concurrent.Action{})
 	pool.WaitAndClose()
 	expectPanic(func() {
-		pool.Do(&concurrent.Action{})
+		pool.Do(concurrent.Action{})
 	}, t)
 
 	pool = concurrent.NewActionPool(0)
 	buf := new(bytes.Buffer)
 	pool.SetLogger(buf)
-	pool.Do(&concurrent.Action{
-		Runner: func() []*concurrent.Action {
+	pool.Do(concurrent.Action{
+		Runner: func() []concurrent.Action {
 			panic("panic test")
 		},
 	})
-	pool.WaitAndClose()
+	pool.Wait()
 	if !strings.HasPrefix(buf.String(), "panic test") {
 		t.Fatalf("unexpected result, expect logger with panic msg")
 	}
@@ -152,3 +150,31 @@ func expectPanic(f func(), t *testing.T) {
 
 	f()
 }
+
+func ExampleActionPool_Map() {
+	pool := concurrent.NewActionPool(10)
+	input := []interface{}{
+		1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11,
+	}
+	transFunc := func(i interface{}) interface{} {
+		return 12 - i.(int)
+	}
+
+	ret := pool.Map(input, transFunc)
+	for _, item := range ret {
+		fmt.Println(item)
+	}
+
+	// Output:
+	// 11
+	// 10
+	// 9
+	// 8
+	// 7
+	// 6
+	// 5
+	// 4
+	// 3
+	// 2
+	// 1
+}