From a62508af6fb0f6c686e99db15891eff6b22fbb8e Mon Sep 17 00:00:00 2001
From: tursom <tursom@foxmail.com>
Date: Thu, 20 Apr 2023 17:11:50 +0800
Subject: [PATCH] add MapReduce & Park

---
 concurrent/collections/Park.go      | 45 ++++++++++++++++++++
 concurrent/collections/Park_test.go | 38 +++++++++++++++++
 util/mr/MapReduce.go                | 66 +++++++++++++++++++++++++++++
 3 files changed, 149 insertions(+)
 create mode 100644 concurrent/collections/Park.go
 create mode 100644 concurrent/collections/Park_test.go
 create mode 100644 util/mr/MapReduce.go

diff --git a/concurrent/collections/Park.go b/concurrent/collections/Park.go
new file mode 100644
index 0000000..93b8193
--- /dev/null
+++ b/concurrent/collections/Park.go
@@ -0,0 +1,45 @@
+package collections
+
+import (
+	"sync"
+	"time"
+)
+
+type Park struct {
+	lock sync.Mutex
+	ch   chan struct{}
+}
+
+func (p *Park) getCh() chan struct{} {
+	p.lock.Lock()
+	defer p.lock.Unlock()
+
+	if p.ch == nil {
+		p.ch = make(chan struct{})
+	}
+
+	return p.ch
+}
+
+func (p *Park) Park() {
+	<-p.getCh()
+}
+
+func (p *Park) ParkT(timeout time.Duration) {
+	select {
+	case <-p.getCh():
+	case <-time.After(timeout):
+	}
+}
+
+func (p *Park) Unpark() {
+	p.lock.Lock()
+	defer p.lock.Unlock()
+
+	if p.ch == nil {
+		return
+	}
+
+	close(p.ch)
+	p.ch = nil
+}
diff --git a/concurrent/collections/Park_test.go b/concurrent/collections/Park_test.go
new file mode 100644
index 0000000..48347c1
--- /dev/null
+++ b/concurrent/collections/Park_test.go
@@ -0,0 +1,38 @@
+package collections
+
+import (
+	"testing"
+	"time"
+)
+
+func TestPark_Park(t *testing.T) {
+	var p Park
+	t1 := time.Now()
+	go func() {
+		<-time.After(time.Second)
+
+		p.Unpark()
+	}()
+
+	p.Park()
+	t2 := time.Now()
+
+	sub := t2.Sub(t1)
+	if sub > time.Duration(float64(time.Second)*1.01) ||
+		sub < time.Duration(float64(time.Second)*0.09) {
+		t.Fatal(sub)
+	}
+}
+
+func TestPark_ParkT(t *testing.T) {
+	var p Park
+	t1 := time.Now()
+	p.ParkT(time.Second)
+	t2 := time.Now()
+
+	sub := t2.Sub(t1)
+	if sub > time.Duration(float64(time.Second)*1.01) ||
+		sub < time.Duration(float64(time.Second)*0.09) {
+		t.Fatal(sub)
+	}
+}
diff --git a/util/mr/MapReduce.go b/util/mr/MapReduce.go
new file mode 100644
index 0000000..8de4ec5
--- /dev/null
+++ b/util/mr/MapReduce.go
@@ -0,0 +1,66 @@
+package mr
+
+import "github.com/tursom/GoCollections/lang/atomic"
+
+type (
+	MapReduce[V, R any] interface {
+		Map(value V) R
+		Reduce(results <-chan R) R
+	}
+)
+
+func LocalMap[V, R any](values <-chan V, m func(value V) R) <-chan R {
+	rc := make(chan R)
+
+	go func() {
+		for value := range values {
+			rc <- m(value)
+		}
+
+		close(rc)
+	}()
+
+	return rc
+}
+
+func LocalReduce[R any](values <-chan R, r func(results <-chan R) R) R {
+	return r(values)
+}
+
+func Local[V, R any](values <-chan V, mr MapReduce[V, R]) R {
+	return LocalReduce(LocalMap(values, mr.Map), mr.Reduce)
+}
+
+func MultiMap[V, R any](values <-chan V, m func(value V) R) <-chan R {
+	rc := make(chan R)
+
+	c := atomic.Int32(1)
+	for value0 := range values {
+		value := value0
+		go func() {
+			rc <- m(value)
+
+			if c.Add(-1) == 0 {
+				close(rc)
+			}
+		}()
+	}
+	if c.Add(-1) == 0 {
+		close(rc)
+	}
+
+	return rc
+}
+
+func MultiReduce[R any](values <-chan R, r func(results <-chan R) R) R {
+	rc := make(chan R)
+	go func() {
+		rc <- r(values)
+	}()
+
+	return <-rc
+}
+
+func Multi[V, R any](values <-chan V, mr MapReduce[V, R]) R {
+	return MultiReduce(MultiMap(values, mr.Map), mr.Reduce)
+}