E. 2 канала

package main import ( "runtime" "sync" ) type mutedSlice struct { mux sync.Mutex data []int done []bool } func mutedSliceConstructor(size int) *mutedSlice { this := new(mutedSlice) this.data = make([]int, size) this.done = make([]bool, size) return this } func reader(f func(int) int, in <-chan int, data *mutedSlice, n int) { for counter := 0; counter < n; counter++ { x := <- in go func(place int, val int) { defer data.mux.Unlock() res := f(val) data.mux.Lock() data.data[place] = res data.done[place] = true return } (counter, x) } } func checker(out chan<- int, data1 *mutedSlice, data2 *mutedSlice, n int) { go func() { for counter := 0; counter < n; { data1.mux.Lock() data2.mux.Lock() for counter < n && data1.done[counter] && data2.done[counter] { out <- data1.data[counter] + data2.data[counter] counter ++ } data1.mux.Unlock() data2.mux.Unlock() runtime.Gosched() } } () } func sender(in <-chan int, out chan<- int, n int) { for counter := 0; counter < n; counter++ { out <- <-in } } func Merge2Channels(f func(int) int, in1 <-chan int, in2 <- chan int, out chan<- int, n int) { go func() { pipe := make(chan int, n) data1 := mutedSliceConstructor(n) data2 := mutedSliceConstructor(n) go reader(f, in1, data1, n) go reader(f, in2, data2, n) go checker(pipe, data1, data2, n) go sender(pipe, out, n) } () }

package main import "sync" func Merge2Channels(f func(int) int, in1 <-chan int, in2 <-chan int, out chan<- int, n int) { results := make([]int, n) go func() { wg := new(sync.WaitGroup) wg.Add(n) for i := 0; i < n; i++ { x1 := <-in1 x2 := <-in2 index := i go func() { x1res := make(chan int) x2res := make(chan int) go func() { x1res <- f(x1) }() go func() { x2res <- f(x2) }() results[index] = (<-x1res) + (<-x2res) wg.Done() }() } wg.Wait() for i := 0; i < n; i++ { out <- results[i] } }() }

package main func Merge2Channels(f func(int) int, in1 <-chan int, in2 <-chan int, out chan<- int, n int) { type Sum struct { index int value int } go func() { sumChannel := make(chan Sum) go func() { sums := make(map[int]int) for i := 0; i < n; { if sum, ok := sums[i]; ok { out <- sum i++ } else { sum := <-sumChannel sums[sum.index] = sum.value } } }() for i := 0; i < n; i++ { var x1, x2 int x1 = <-in1 x2 = <-in2 go func(index int) { var fx1Ch = make(chan int) var fx2Ch = make(chan int) go func() { fx1Ch <- f(x1) }() go func() { fx2Ch <- f(x2) }() fx1 := <-fx1Ch fx2 := <-fx2Ch sumChannel <- Sum{index, fx1 + fx2} }(i) } }() }

package main import "sync" func Merge2Channels(f func(int) int, in1 <-chan int, in2 <-chan int, out chan<- int, n int) { go func(f func(int) int, in1 <-chan int, in2 <-chan int, out chan<- int) { results := make([]chan int, n) var totalWg sync.WaitGroup totalWg.Add(n) for i := 0; i < n; i++ { results[i] = make(chan int, 1) go work(f, <-in1, <-in2, results[i], &totalWg) } totalWg.Wait() for i := 0; i < n; i++ { out <- <-results[i] } }(f, in1, in2, out) } func work(f func(int) int, x1 int, x2 int, out chan int, wg *sync.WaitGroup) { var handlerWg sync.WaitGroup handlerWg.Add(2) numHandler := func(f func(int) int, num int, wg *sync.WaitGroup, res chan<- int) { res <- f(num) wg.Done() } resultChan := make(chan int, 2) go numHandler(f, x1, &handlerWg, resultChan) go numHandler(f, x2, &handlerWg, resultChan) handlerWg.Wait() out <- <-resultChan + <-resultChan wg.Done() }

package main func Merge2Channels(f func(i int) int, in1 <-chan int, in2 <-chan int, out chan<- int, n int) { go func() { c := make(chan pair, n) go func() { m := make(map[int]int, n/2) toSend := 0 for i := 0; i < 2*n; i++ { p := <-c if v, ok := m[p.Num]; !ok { m[p.Num] = p.Res } else { s := v + p.Res if toSend == p.Num { out <- s delete(m, p.Num) toSend++ } else { m[p.Num] = s } } } close(c) for i := toSend; i < n; i++ { a := m[i] out <- a } }() ff := func(v int, i int) { res := f(v) c <- pair{ Num: i, Res: res, } } run := func(in <-chan int) { for i := 0; i < n; i++ { a := <-in go ff(a, i) } } go run(in1) go run(in2) }() } type pair struct { Num int Res int }

package main type orderedNumber struct { pos int num int } //Merge2Channels merges 2 channels func Merge2Channels(f func(int) int, in1 <-chan int, in2 <-chan int, out chan<- int, n int) { fin1 := make(chan orderedNumber, n) fin2 := make(chan orderedNumber, n) go funcNum(f, in1, fin1, n) go funcNum(f, in2, fin2, n) go mergeFNums(fin1, fin2, out, n) } func funcNum(f func(int) int, in <-chan int, fin chan<- orderedNumber, n int) { pos := 0 for i := 0; i < n; i++ { num, ok := <-in if !ok { return } // fmt.Println("Read from in: ", num) onum := orderedNumber{pos, num} pos++ go func(f func(int) int, onum orderedNumber, fin chan<- orderedNumber) { onum.num = f(onum.num) // fmt.Println("Write to fin: ", onum) fin <- onum }(f, onum, fin) } } func mergeFNums(fin1 <-chan orderedNumber, fin2 <-chan orderedNumber, out chan<- int, n int) { nums1 := make(map[int]int, n) nums2 := make(map[int]int, n) sendPos := 0 for i := 0; i < n*2; i++ { select { case onum1 := <-fin1: nums1[onum1.pos] = onum1.num case onum2 := <-fin2: nums2[onum2.pos] = onum2.num } for j := sendPos; j < n; j++ { sendNum1, ok1 := nums1[sendPos] sendNum2, ok2 := nums2[sendPos] if ok1 && ok2 { out <- sendNum1 + sendNum2 sendPos++ } else { break } } } }

package main func Merge2Channels(f func(int) int, in1 <-chan int, in2 <-chan int, out chan<- int, n int) { type tresult struct { operandNum int i int sum int } results := make(chan tresult, n*2) for i := 0; i < n; i++ { x1 := <-in1 x2 := <-in2 go func(results chan tresult, index int, operandNum int, x int, f func(int) int) { result := f(x) results <- tresult{ operandNum: operandNum, i: index, sum: result, } }(results, i, 0, x1, f) go func(results chan tresult, index int, operandNum int, x int, f func(int) int) { result := f(x) results <- tresult{ operandNum: operandNum, i: index, sum: result, } }(results, i, 1, x2, f) } go func() { a1 := make([]int, n) a2 := make([]int, n) for i := 0; i < n*2; i++ { row := <-results if row.operandNum == 0 { a1[row.i] = row.sum } if row.operandNum == 1 { a2[row.i] = row.sum } } for i := 0; i < n; i++ { out <- a1[i] + a2[i] } }() }

package main import ( "sync" ) func Merge2Channels(f func(int) int, in1, in2 <-chan int, out chan<- int, n int) { var wg sync.WaitGroup wg.Add(n * 2) store := make([]int, n*2) go func() { wg.Wait() for i := 0; i < n; i++ { out <- store[i] + store[i+n] } }() worker := func(c <-chan int, wg *sync.WaitGroup, adder int) { i := 0 for i < n { num := <-c go func(ix int) { store[ix+adder] = f(num) wg.Done() }(i) i++ } } go worker(in1, &wg, 0) go worker(in2, &wg, n) }

package main func Merge2Channels(f func(int) int, in1 <-chan int, in2 <-chan int, out chan<- int, n int) { go func() { x1 := make(chan chan int, n) x2 := make(chan chan int, n) go func() { for i := 0; i < n; i++ { f1 := <-<-x1 f2 := <-<-x2 out <- f1 + f2 } }() inp := func(in <-chan int, a chan chan int) { for i := 0; i < n; i++ { w := make(chan int) a <- w x := <-in go func(w chan int, x int) { w <- f(x) }(w, x) } } go inp(in1, x1) go inp(in2, x2) }() }

package main import "sync" func Merge2Channels(f func(int) int, in1 <-chan int, in2 <-chan int, out chan<- int, n int) { fx1 := make([]int, n) fx2 := make([]int, n) go func() { wg := new(sync.WaitGroup) wg.Add(2*n) for i := 0; i < n; i++ { x1 := <-in1 x2 := <-in2 index := i go func() { fx1[index] = f(x1) wg.Done() }() go func() { fx2[index] = f(x2) wg.Done() }() } wg.Wait() for i := 0; i < n; i++ { out <- fx1[i] + fx2[i] } }() }