Merge pull request #579 from cmasone-attic/issue564

IterAllP for typed Maps

nomdl/codegen/map.tmpl

@@ -116,6 +116,11 @@ func (m {{.Name}}) IterAll(cb {{.Name}}IterAllCallback) {
 	})
 }
 
+func (m {{.Name}}) IterAllP(concurrency int, cb {{.Name}}IterAllCallback) {
+	m.m.IterAllP(concurrency, func(k, v {{$typesPackage}}Value) {
+		cb({{valueToUser "k" .KeyType}}, {{valueToUser "v" .ValueType}})
+	})
+}
 
 type {{.Name}}FilterCallback func(k {{userType .KeyType}}, v {{userType .ValueType}}) (keep bool)
 

nomdl/codegen/test/gen/map.noms.go

@@ -121,6 +121,12 @@ func (m MapOfBoolToString) IterAll(cb MapOfBoolToStringIterAllCallback) {
 	})
 }
 
+func (m MapOfBoolToString) IterAllP(concurrency int, cb MapOfBoolToStringIterAllCallback) {
+	m.m.IterAllP(concurrency, func(k, v types.Value) {
+		cb(bool(k.(types.Bool)), v.(types.String).String())
+	})
+}
+
 type MapOfBoolToStringFilterCallback func(k bool, v string) (keep bool)
 
 func (m MapOfBoolToString) Filter(cb MapOfBoolToStringFilterCallback) MapOfBoolToString {
@@ -247,6 +253,12 @@ func (m MapOfStringToValue) IterAll(cb MapOfStringToValueIterAllCallback) {
 	})
 }
 
+func (m MapOfStringToValue) IterAllP(concurrency int, cb MapOfStringToValueIterAllCallback) {
+	m.m.IterAllP(concurrency, func(k, v types.Value) {
+		cb(k.(types.String).String(), v)
+	})
+}
+
 type MapOfStringToValueFilterCallback func(k string, v types.Value) (keep bool)
 
 func (m MapOfStringToValue) Filter(cb MapOfStringToValueFilterCallback) MapOfStringToValue {

types/list.go

@@ -14,10 +14,6 @@ type List struct {
 	ref    *ref.Ref
 }
 
-type listIterFunc func(v Value, index uint64) (stop bool)
-type listIterAllFunc func(v Value, index uint64)
-type MapFunc func(v Value, index uint64) interface{}
-
 var listTypeRef = MakeCompoundTypeRef(ListKind, MakePrimitiveTypeRef(ValueKind))
 
 func NewList(v ...Value) List {
@@ -44,6 +40,8 @@ func (l List) Get(idx uint64) Value {
 	return l.values[idx]
 }
 
+type listIterFunc func(v Value, index uint64) (stop bool)
+
 func (l List) Iter(f listIterFunc) {
 	for i, v := range l.values {
 		if f(v, uint64(i)) {
@@ -52,6 +50,8 @@ func (l List) Iter(f listIterFunc) {
 	}
 }
 
+type listIterAllFunc func(v Value, index uint64)
+
 func (l List) IterAll(f listIterAllFunc) {
 	for i, v := range l.values {
 		f(v, uint64(i))
@@ -86,6 +86,8 @@ func (l List) iterInternal(sem chan int, lf listIterAllFunc, offset uint64) {
 	wg.Wait()
 }
 
+type MapFunc func(v Value, index uint64) interface{}
+
 func (l List) Map(mf MapFunc) []interface{} {
 	return l.MapP(1, mf)
 }

types/map.go

@@ -1,7 +1,9 @@
 package types
 
 import (
+	"runtime"
 	"sort"
+	"sync"
 
 	"github.com/attic-labs/noms/d"
 	"github.com/attic-labs/noms/ref"
@@ -101,7 +103,28 @@ func (m Map) IterAll(cb mapIterAllCallback) {
 	}
 }
 
-// TODO: Implement IterAllP() BUG 573
+func (m Map) IterAllP(concurrency int, f mapIterAllCallback) {
+	if concurrency == 0 {
+		concurrency = runtime.NumCPU()
+	}
+	sem := make(chan int, concurrency)
+
+	wg := sync.WaitGroup{}
+
+	for idx := range m.data {
+		wg.Add(1)
+
+		sem <- 1
+		go func(idx int) {
+			defer wg.Done()
+			md := m.data[idx]
+			f(md.key, md.value)
+			<-sem
+		}(idx)
+	}
+
+	wg.Wait()
+}
 
 type mapFilterCallback func(key, value Value) (keep bool)
 

types/map_test.go

@@ -2,6 +2,8 @@ package types
 
 import (
 	"bytes"
+	"runtime"
+	"sync"
 	"testing"
 
 	"github.com/attic-labs/noms/Godeps/_workspace/src/github.com/stretchr/testify/assert"
@@ -134,6 +136,61 @@ func TestMapIter(t *testing.T) {
 	assert.True(got(NewString("a"), Int32(0)) || got(NewString("b"), Int32(1)))
 }
 
+func TestMapIterAllP(t *testing.T) {
+	assert := assert.New(t)
+
+	testIter := func(concurrency, mapLen int) {
+		values := make([]Value, 2*mapLen)
+		for i := 0; i < mapLen; i++ {
+			values[2*i] = UInt64(i)
+			values[2*i+1] = UInt64(i)
+		}
+
+		m := NewMap(values...)
+
+		cur := 0
+		mu := sync.Mutex{}
+		getCur := func() int {
+			mu.Lock()
+			defer mu.Unlock()
+			return cur
+		}
+
+		expectConcurreny := concurrency
+		if concurrency == 0 {
+			expectConcurreny = runtime.NumCPU()
+		}
+		visited := make([]bool, mapLen)
+		f := func(k, v Value) {
+			mu.Lock()
+			cur++
+			mu.Unlock()
+
+			for getCur() < expectConcurreny {
+			}
+
+			visited[v.(UInt64)] = true
+		}
+
+		if concurrency == 1 {
+			m.IterAll(f)
+		} else {
+			m.IterAllP(concurrency, f)
+		}
+		numVisited := 0
+		for _, visit := range visited {
+			if visit {
+				numVisited++
+			}
+		}
+		assert.Equal(mapLen, numVisited, "IterAllP was not called with every map key")
+	}
+	testIter(0, 100)
+	testIter(10, 1000)
+	testIter(1, 100000)
+	testIter(64, 100000)
+}
+
 func TestMapFilter(t *testing.T) {
 	assert := assert.New(t)