EREW MERGE ALGORITHM

erew_merge.go

package main

import (
	"fmt"
	"math"
	"sort"
	"sync"
)

type quad struct {
	startA, endA, startB, endB int
}

type pair struct {
	first, second int
	arr           rune
}

// Shared Memory Elements
var A []int = make([]int, 0, 30)       // Sequence A of length r
var B []int = make([]int, 0, 30)       // Sequence B of length s
var C []int = make([]int, 0, 60)       // Final merged sequence C
var Quads []quad = make([]quad, 0, 60) // Shared memory to store quadruples
var N int                              // Number of processors
var r, s int                           // Size of sequences A and B

// helper function to calculate ceil
func ceil(a, b int) int {
	return int(math.Ceil(float64(a) / float64(b)))
}

// helper function to calculate floor
func floor(a, b int) int {
	return int(math.Floor(float64(a) / float64(b)))
}

// helper function to calculate min
func min(a, b int) int {
	return int(math.Min(float64(a), float64(b)))
}

// helper function to calculate max
func max(a, b int) int {
	return int(math.Max(float64(a), float64(b)))
}

// Function to perform sequential merge of two sequences
func SequentialMerge(A []int, B []int) []int {
	r, s := len(A), len(B)
	C := make([]int, r+s)
	i, j, k := 0, 0, 0

	for i < r && j < s {
		if A[i] <= B[j] {
			C[k] = A[i]
			i += 1
		} else {
			C[k] = B[j]
			j += 1
		}
		k += 1
	}

	for i < r {
		C[k] = A[i]
		i, k = i+1, k+1
	}

	for j < s {
		C[k] = B[j]
		j, k = j+1, k+1
	}

	return C
}

func isValidPair(A, B []int, x, y, r, s int, arr *rune) bool {
	var m int = r + s
	// fmt.Println("A =", A)
	// fmt.Println("B =", B)
	// fmt.Println("indices", x, y)
	var a, b int = A[x], B[y]
	var medianIndex int = ceil(m, 2)

	// check if a is median
	bCount := sort.Search(s, func(k int) bool { return a < B[k] })
	lessCountA := x + bCount - 2
	if lessCountA == medianIndex-1 {
		index1 := sort.Search(s, func(k int) bool { return B[k] >= a })
		index2 := sort.Search(s, func(k int) bool { return B[k] > a }) - 1

		// condition 2 for median pair
		if B[index1] == b || B[index2] == b {
			*arr = 'A'
			return true
		}
	}

	// check if b is median
	aCount := sort.Search(r, func(k int) bool { return b < A[k] })
	lessCountB := y + aCount - 2
	if lessCountB == medianIndex-1 {
		index1 := sort.Search(r, func(k int) bool { return A[k] >= b })
		index2 := sort.Search(r, func(k int) bool { return A[k] > b }) - 1

		// condition 2 for median pair
		if A[index1] == a || A[index2] == a {
			*arr = 'B'
			return true
		}
	}

	return false
}

// function to find the median of two non-decreasing sequences
func TwoSequenceMedian(A []int, B []int) pair {
	var r, s, lowA, lowB, highA, highB, nA, nB int
	var u, v, w int
	r, s = len(A), len(B)
	A = append([]int{0}, A...)
	B = append([]int{0}, B...)
	lowA, lowB = 1, 1
	highA, highB = r, s
	nA, nB = r, s

	for nA > 1 && nB > 1 {
		u = lowA + ceil(highA-lowA-1, 2)
		v = lowB + ceil(highB-lowB-1, 2)
		w = min(floor(nA, 2), floor(nB, 2))

		nA, nB = nA-w, nB-w

		if A[u] >= B[v] {
			highA, lowB = highA-w, lowB+w
		} else {
			lowA, highB = lowA+w, highB-w
		}
	}

	var i, j, min int
	var ans pair
	var arr rune
	min = 1000000

	// find median pairs among a set of candidate pairs
	for i = max(1, u-1); i <= u+1; i++ {
		for j = max(1, v-1); j <= v+1; j++ {
			if isValidPair(A, B, i, j, r, s, &arr) && i+j < min {
				min = i + j
				ans = pair{i, j, arr}
			}
		}
	}

	return ans
}

// function to execute Step 1.2 of EREW MERGE
func Step1Point2(i int, wg *sync.WaitGroup) {
	defer wg.Done()

	var p1, q1, p2, q2 int

	// Step 1.2.1
	medianPair := TwoSequenceMedian(A[Quads[i].startA:Quads[i].endA+1], B[Quads[i].startB:Quads[i].endB+1])
	medianPair.first += Quads[i].startA - 1
	medianPair.second += Quads[i].startB - 1
	fmt.Println(i, medianPair)

	// Step 1.2.2
	if medianPair.arr == 'A' {
		p1, q1 = medianPair.first, medianPair.first+1

		if B[medianPair.second] <= A[medianPair.first] {
			p2, q2 = medianPair.second, medianPair.second+1
		} else {
			p2, q2 = medianPair.second-1, medianPair.second
		}

	} else {
		p2, q2 = medianPair.second, medianPair.second+1

		if A[medianPair.first] <= B[medianPair.second] {
			p1, q1 = medianPair.first, medianPair.first+1
		} else {
			p1, q1 = medianPair.first-1, medianPair.first
		}
	}

	quadI := Quads[i]

	// Step 1.2.3
	Quads[2*i-1] = quad{quadI.startA, p1, quadI.startB, p2}
	fmt.Println(Quads[2*i-1])

	// Step 1.2.4
	Quads[2*i] = quad{q1, quadI.endA, q2, quadI.endB}
	fmt.Println(Quads[2*i])
}

// function to execute Step 2 of EREW MERGE
func Step2(i int, wg *sync.WaitGroup) {
	defer wg.Done()

	// w := 1 + ((i-1)*(r+s))/N
	w := Quads[i].startA + Quads[i].startB - 1
	_ = min(i*(r+s)/N, r+s)

	fmt.Printf("Merging A[%d : %d] and B[%d : %d]\n", Quads[i].startA, Quads[i].endA, Quads[i].startB, Quads[i].endB)
	subC := SequentialMerge(A[Quads[i].startA:Quads[i].endA+1], B[Quads[i].startB:Quads[i].endB+1])
	fmt.Println(subC)

	for j := 0; j < len(subC); j++ {
		C[w+j] = subC[j]
	}
}

func main() {
	var i, j int

	fmt.Println("Enter length of first sequence")
	fmt.Scanf("%d", &r)

	A = A[0 : r+1]
	for i = 1; i <= r; i++ {
		fmt.Scanf("%d", &A[i])
	}

	fmt.Println("Enter length of second sequence")
	fmt.Scanf("%d", &s)

	B = B[0 : s+1]
	for i = 1; i <= s; i++ {
		fmt.Scanf("%d", &B[i])
	}

	C = C[0 : r+s+1]

	fmt.Println("Enter number of processors : N = 2^q and 1 <= N <= r+s")
	fmt.Scanf("%d", &N)

	lgN := int(math.Log2(float64(N)))

	Quads = Quads[0 : N+1]

	var wg sync.WaitGroup

	/*********************************************** Start of EREW MERGE ALGORITHM ********************************************/

	// Step 1
	// Step 1.1
	Quads[1] = quad{1, r, 1, s}

	// Step 1.2
	for j = 1; j <= lgN; j++ {
		limit := int(math.Pow(float64(2), float64(j-1)))
		for i = 1; i <= limit; i++ {
			wg.Add(1)
			go Step1Point2(i, &wg) // Call the functions in different threads
		}
		wg.Wait() // Wait for Step 1.2 to finish
	}

	// Step 2
	for i = 1; i <= N; i++ {
		wg.Add(1)
		go Step2(i, &wg)
	}
	wg.Wait() // Wait for Step 2 to finish

	/*********************************************** End of EREW MERGE ALGORITHM ********************************************/

	fmt.Println(C[1:])
}