Add mapreduce example

Author: GregorKonzett

benchmarking/queuebenchmark.py

@@ -5,16 +5,16 @@
 producerAmounts = 4
 vals = 100
 
-print("Mutex")
+print("Junctions:")
 for i in range(producerAmounts):
     prodConAmount = 10 ** i
     print("Producer/Consumer Amount: " + str(prodConAmount))
     for j in range(amount):
-        subprocess.call(["./go_build_queuebenchmarking_go.exe","mutex",str(prodConAmount), str(prodConAmount), str(vals)])
+        subprocess.call(["./go_build_queuebenchmarking_go.exe","junction",str(prodConAmount), str(prodConAmount), str(vals)])
 
-print("Junctions:")
+print("Mutex")
 for i in range(producerAmounts):
     prodConAmount = 10 ** i
     print("Producer/Consumer Amount: " + str(prodConAmount))
     for j in range(amount):
-        subprocess.call(["./go_build_queuebenchmarking_go.exe","junction",str(prodConAmount), str(prodConAmount), str(vals)])
+        subprocess.call(["./go_build_queuebenchmarking_go.exe","mutex",str(prodConAmount), str(prodConAmount), str(vals)])

examples/mapreduce.go

@@ -0,0 +1,58 @@
+package main
+
+import (
+	"../gojo/junction"
+	"../gojo/types"
+	"constraints"
+	"fmt"
+)
+
+func createMapReduce[T constraints.Integer](vals []T) func(types.Unit) ([]T, error) {
+	j := junction.NewJunction()
+
+	mapPort, mapSignal := junction.NewAsyncSignal[T](j)
+	reducePort, reduceSignal := junction.NewAsyncSignal[T](j)
+	listPort, listSignal := junction.NewAsyncSignal[[]T](j)
+	finalPort, finalSignal := junction.NewAsyncSignal[[]T](j)
+	getPort, getSignal := junction.NewSyncSignal[types.Unit, []T](j)
+
+	// map
+	junction.NewUnaryAsyncJoinPattern[T](mapPort).Action(func(val T) {
+		fmt.Println("Mapping: ", val)
+		reduceSignal(val * 2)
+	})
+
+	// reduce
+	junction.NewBinaryAsyncJoinPattern[[]T, T](listPort, reducePort).Action(func(list []T, val T) {
+		fmt.Printf("Reducing: %v %d \n", list, val)
+		list = append(list, val)
+
+		if len(list) == len(vals) {
+			finalSignal(list)
+		} else {
+			listSignal(list)
+		}
+	})
+
+	// final
+	junction.NewBinarySyncJoinPattern[[]T, types.Unit, []T](finalPort, getPort).Action(func(list []T, a types.Unit) []T {
+		return list
+	})
+
+	listSignal(make([]T, 0))
+
+	for _, val := range vals {
+		go mapSignal(val)
+	}
+
+	return getSignal
+}
+
+func main() {
+	arr := []int{1, 2, 3, 5}
+	get := createMapReduce[int](arr)
+
+	ret, _ := get(types.Unit{})
+
+	fmt.Printf("Result: %v", ret)
+}

examples/queue/benchmarking/queuebenchmarking.go

@@ -19,31 +19,38 @@ func test(producerCount int, consumerCount int, vals int, enqueue func(int), deq
 		enqueue(i)
 	}
 
-	func() {
+	wg.Add(1)
+
+	go func() {
+		defer wg.Done()
+
 		for i := 1; i <= producerCount; i++ {
 			wg.Add(1)
 			go func(num int) {
 				defer wg.Done()
 
 				for j := 1; j <= vals; j++ {
 					enqueue(j * num)
-					time.Sleep(time.Duration(j * 10))
 				}
 
 			}(i)
 		}
 	}()
 
+	wg.Add(1)
+
 	// Consumer
-	func() {
+	go func() {
+		defer wg.Done()
+
 		for i := 0; i < consumerCount; i++ {
 			wg.Add(1)
+
 			go func() {
 				defer wg.Done()
 
 				for j := 0; j < vals; j++ {
 					dequeue(types.Unit{})
-					time.Sleep(time.Duration(j * 10))
 				}
 
 			}()

examples/queue/queue.go

@@ -25,11 +25,12 @@ Problem when reading and writing async:
 
 func NewQueue[T any]() (func(T), func(types.Unit) (T, error)) {
 	j := junction.NewJunction()
+	j1 := junction.NewJunction()
 
 	firstPort, firstSignal := junction.NewAsyncSignal[QueueElement[T]](j)
-	lastPort, lastSignal := junction.NewAsyncSignal[QueueElement[T]](j)
+	lastPort, lastSignal := junction.NewAsyncSignal[QueueElement[T]](j1)
 
-	enqueuePort, enqueueSignal := junction.NewAsyncSignal[T](j)
+	enqueuePort, enqueueSignal := junction.NewAsyncSignal[T](j1)
 	dequeuePort, dequeueSignal := junction.NewSyncSignal[types.Unit, T](j)
 
 	junction.NewBinaryAsyncJoinPattern[QueueElement[T], T](lastPort, enqueuePort).Action(func(last QueueElement[T], value T) {
@@ -43,10 +44,10 @@ func NewQueue[T any]() (func(T), func(types.Unit) (T, error)) {
 
 	junction.NewBinarySyncJoinPattern[QueueElement[T], types.Unit, T](firstPort, dequeuePort).Action(func(first QueueElement[T], a types.Unit) T {
 		nextSignal, _ := first.getNextSignal(types.Unit{})
-		val, _ := first.getValueSignal(types.Unit{})
-
 		firstSignal(nextSignal)
 
+		val, _ := first.getValueSignal(types.Unit{})
+
 		return val
 	})
 
@@ -73,6 +74,8 @@ func newQueueElement[T any]() QueueElement[T] {
 
 	junction.NewBinarySyncJoinPattern[T, types.Unit, T](setValuePort, getValuePort).
 		Action(func(val T, a types.Unit) T {
+			// Added for benchmarking
+			time.Sleep(time.Duration(10))
 			return val
 		})
 

examples/queue/queueMutex.go

@@ -3,6 +3,7 @@ package queue
 import (
 	"../../gojo/types"
 	"sync"
+	"time"
 )
 
 type QueueElementMutex[T any] struct {
@@ -48,7 +49,7 @@ func NewQueueMutex[T any]() (func(T), func(types.Unit) (T, error)) {
 
 		firstElem := queue.head
 		queue.head = queue.head.next
-
+		time.Sleep(time.Duration(10))
 		c.L.Unlock()
 
 		return (*firstElem).val, nil

gojo/junction/controller/controllerdata.go

@@ -1,6 +1,8 @@
 package controller
 
-import "../../types"
+import (
+	"../../types"
+)
 
 type JoinPatterns struct {
 	portIds             int