feat: separate mapreduce from go-zero

Author: kevwan

atomicerror.go

@@ -0,0 +1,23 @@
+package mapreduce
+
+import "sync/atomic"
+
+// AtomicError defines an atomic error.
+type AtomicError struct {
+	err atomic.Value // error
+}
+
+// Set sets the error.
+func (ae *AtomicError) Set(err error) {
+	if err != nil {
+		ae.err.Store(err)
+	}
+}
+
+// Load returns the error.
+func (ae *AtomicError) Load() error {
+	if v := ae.err.Load(); v != nil {
+		return v.(error)
+	}
+	return nil
+}

atomicerror_test.go

@@ -0,0 +1,82 @@
+package mapreduce
+
+import (
+	"errors"
+	"sync"
+	"sync/atomic"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+var errAtomicDummy = errors.New("hello")
+
+func TestAtomicError(t *testing.T) {
+	var err AtomicError
+	err.Set(errAtomicDummy)
+	assert.Equal(t, errAtomicDummy, err.Load())
+}
+
+func TestAtomicErrorSetNil(t *testing.T) {
+	var (
+		errNil error
+		err    AtomicError
+	)
+	err.Set(errNil)
+	assert.Equal(t, errNil, err.Load())
+}
+
+func TestAtomicErrorNil(t *testing.T) {
+	var err AtomicError
+	assert.Nil(t, err.Load())
+}
+
+func BenchmarkAtomicError(b *testing.B) {
+	var aerr AtomicError
+	wg := sync.WaitGroup{}
+
+	b.Run("Load", func(b *testing.B) {
+		var done uint32
+		go func() {
+			for {
+				if atomic.LoadUint32(&done) != 0 {
+					break
+				}
+				wg.Add(1)
+				go func() {
+					aerr.Set(errAtomicDummy)
+					wg.Done()
+				}()
+			}
+		}()
+		b.ResetTimer()
+		for i := 0; i < b.N; i++ {
+			_ = aerr.Load()
+		}
+		b.StopTimer()
+		atomic.StoreUint32(&done, 1)
+		wg.Wait()
+	})
+	b.Run("Set", func(b *testing.B) {
+		var done uint32
+		go func() {
+			for {
+				if atomic.LoadUint32(&done) != 0 {
+					break
+				}
+				wg.Add(1)
+				go func() {
+					_ = aerr.Load()
+					wg.Done()
+				}()
+			}
+		}()
+		b.ResetTimer()
+		for i := 0; i < b.N; i++ {
+			aerr.Set(errAtomicDummy)
+		}
+		b.StopTimer()
+		atomic.StoreUint32(&done, 1)
+		wg.Wait()
+	})
+}

donechan.go

@@ -0,0 +1,28 @@
+package mapreduce
+
+import "sync"
+
+// A DoneChan is used as a channel that can be closed multiple times and wait for done.
+type DoneChan struct {
+	done chan struct{}
+	once sync.Once
+}
+
+// NewDoneChan returns a DoneChan.
+func NewDoneChan() *DoneChan {
+	return &DoneChan{
+		done: make(chan struct{}),
+	}
+}
+
+// Close closes dc, it's safe to close more than once.
+func (dc *DoneChan) Close() {
+	dc.once.Do(func() {
+		close(dc.done)
+	})
+}
+
+// Done returns a channel that can be notified on dc closed.
+func (dc *DoneChan) Done() chan struct{} {
+	return dc.done
+}

donechan_test.go

@@ -0,0 +1,31 @@
+package mapreduce
+
+import (
+	"sync"
+	"testing"
+)
+
+func TestDoneChanClose(t *testing.T) {
+	doneChan := NewDoneChan()
+
+	for i := 0; i < 5; i++ {
+		doneChan.Close()
+	}
+}
+
+func TestDoneChanDone(t *testing.T) {
+	var waitGroup sync.WaitGroup
+	doneChan := NewDoneChan()
+
+	waitGroup.Add(1)
+	go func() {
+		<-doneChan.Done()
+		waitGroup.Done()
+	}()
+
+	for i := 0; i < 5; i++ {
+		doneChan.Close()
+	}
+
+	waitGroup.Wait()
+}

go.mod

@@ -0,0 +1,12 @@
+module github.com/kevwan/mapreduce
+
+go 1.14
+
+require github.com/stretchr/testify v1.7.0
+
+require (
+	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/kr/text v0.2.0 // indirect
+	gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c // indirect
+	gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b // indirect
+)

go.sum

@@ -0,0 +1,21 @@
+github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/kr/pretty v0.2.1 h1:Fmg33tUaq4/8ym9TJN1x7sLJnHVwhP33CNkpYV/7rwI=
+github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
+github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
+github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
+github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
+github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.7.0 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5CcY=
+github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
+gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b h1:h8qDotaEPuJATrMmW04NCwg7v22aHH28wwpauUhK9Oo=
+gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=