E. 2 канала

package main import ( "sync" ) type res struct { value int } type calcResults struct { mx sync.RWMutex m map[int]int } func newCalcResults() *calcResults { return &calcResults{m: make(map[int]int, 0)} } func (c *calcResults) Load(key int) (int, bool) { c.mx.RLock() defer c.mx.RUnlock() val, ok := c.m[key] return val, ok } func (c *calcResults) Store(key int, value int) { c.mx.Lock() defer c.mx.Unlock() c.m[key] = value } func Merge2Channels(f func(int) int, in1 <-chan int, in2 <-chan int, out chan<- int, n int) { go func() { var calcResults = newCalcResults() var results = make([]*res, n) for i := 0; i < n; i++ { x1 := <-in1 x2 := <-in2 go func(x1 int, x2 int, idx int) { var calcWg sync.WaitGroup if x1 == x2{ resX1, ok := calcResults.Load(x1) if !ok { calcWg.Add(1) go func() { resX1 = f(x1) calcResults.Store(x1, resX1) calcWg.Done() }() } calcWg.Wait() results[idx] = &res{value: resX1 *2} } else { resX1, ok := calcResults.Load(x1) if !ok { calcWg.Add(1) go func() { resX1 = f(x1) calcResults.Store(x1, resX1) calcWg.Done() }() } resX2, ok := calcResults.Load(x2) if !ok { calcWg.Add(1) go func() { resX2 = f(x2) calcResults.Store(x2, resX2) calcWg.Done() }() } calcWg.Wait() results[idx] = &res{value: resX1 + resX2} } }(x1, x2, i) } index := 0 for { if results[index] != nil { out <- results[index].value index++ } if index >= n { break } } close(out) }() }

package main func Merge2Channels(f func(int) int, in1 <-chan int, in2 <-chan int, out chan<- int, n int) { // type to store f result type Tres struct { Index int Value int Count int8 } // func to evaluate f in goroutine evalF := func(index int, val int, output chan<- Tres) { output <- Tres{index, f(val), 0} } evalFNTimes := func(in <-chan int, output chan<- Tres) { for i := 0; i < n; i++ { go evalF(i, <-in, output) } } go func() { // chan for f results fChan := make(chan Tres) // map to store result for out resMap := make(map[int]Tres) go evalFNTimes(in1, fChan) go evalFNTimes(in2, fChan) for i := 0; i < n; { if r := resMap[i]; r.Count == 2 { out <- r.Value delete(resMap, i) i++ continue } select { case res := <-fChan: // save to map, sum result and increase counter r := resMap[res.Index] r.Value += res.Value r.Count++ resMap[res.Index] = r default: } } }() }

package main 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, n int) { f1res := make([]*int, n) f2res := make([]*int, n) go func() { i := 0 for true { if f1res[i] != nil && f2res[i] != nil { res := *f1res[i] + *f2res[i] out <- res if i++; i == n { break } } } }() input := func(input <-chan int, results []*int) { for i := 0; i < n; i++ { x := <-input go func(i int, x int) { res := f(x) results[i] = &res }(i, x) } } go input(in1, f1res) go input(in2, f2res) }(f, in1, in2, out, n) }

package main import ( "sync" ) func ExecuteFunc(index int, f func(int) int, x int, mut *sync.Mutex, resultArray []int, finish chan<- int) { result := f(x) mut.Lock() resultArray[index] = result mut.Unlock() finish <- index } func readChannel(chanNum int, f func(int) int, in1 <-chan int, muResult *sync.Mutex, resultArray []int, n int, finish chan<- int) { for i := 0; i < n; i++ { value := <-in1 go ExecuteFunc(i, f, value, muResult, resultArray, finish) } } func Iterate(f func(int) int, in1 <-chan int, in2 <-chan int, out chan<- int, n int) { result1Array := make([]int, n, n) result2Array := make([]int, n, n) done := make(chan int, 2*n) muResult1 := &sync.Mutex{} muResult2 := &sync.Mutex{} go readChannel(1, f, in1, muResult1, result1Array, n, done) go readChannel(2, f, in2, muResult2, result2Array, n, done) i := 0 doneArray := make([]int, n, n) for i < n { select { case val := <-done: doneArray[val]++ default: break } if doneArray[i] == 2 { muResult1.Lock() muResult2.Lock() out <- result1Array[i] + result2Array[i] muResult1.Unlock() muResult2.Unlock() i++ } } } func Merge2Channels(f func(int) int, in1 <-chan int, in2 <-chan int, out chan<- int, n int) { go Iterate(f, in1, in2, out, n) }

package main import ( "sync" ) var ( mtx = new(sync.Mutex) tasksQueue = make(chan task, 100000) ) type task struct{ function func(int) int inputChannel1 <-chan int inputChannel2 <-chan int outputChannel chan<- int iterations int } type ConcurrentSlice struct { sync.RWMutex slice []*int } func NewConcurrentSlice (size int) *ConcurrentSlice { return &ConcurrentSlice{ slice: make([]*int, size), } } func (cs *ConcurrentSlice) Get(key int) (*int,) { cs.RLock() value := cs.slice[key] cs.RUnlock() return value } func (cs *ConcurrentSlice) Set(key int, value *int) { cs.Lock() cs.slice[key] = value cs.Unlock() } func calculateFunction(function func(int) int, inputChannel <-chan int, outputSlice *ConcurrentSlice, iterationsCount int) { for i := 0; i < iterationsCount; i ++ { ii := i select { case curNumber := <-inputChannel: go func(ni, value int) { curResult := function(curNumber) outputSlice.Set(ni, &curResult) }(ii, curNumber) } } } func Merge2Channels(f func(int) int, in1 <-chan int, in2 <-chan int, out chan<- int, n int) { tasksQueue <- task{ function: f, inputChannel1: in1, inputChannel2: in2, outputChannel: out, iterations: n, } go func() { var ( passedIterations = 0 functionResults1 = NewConcurrentSlice(n) functionResults2 = NewConcurrentSlice(n) ) mtx.Lock() currentTask := <- tasksQueue go calculateFunction(currentTask.function, currentTask.inputChannel1, functionResults1, currentTask.iterations) //, wg) go calculateFunction(currentTask.function, currentTask.inputChannel2, functionResults2, currentTask.iterations) //, wg) for { firstResult := functionResults1.Get(passedIterations) secondResult := functionResults2.Get(passedIterations) if firstResult != nil && secondResult != nil { currentTask.outputChannel <- *firstResult + *secondResult passedIterations += 1 if passedIterations >= n { mtx.Unlock() return } } } }() }

package main import ( "sync" ) var nMutex sync.Mutex type Result struct { out chan<- int value int } func Merge2Channels(f func(int) int, in1 <-chan int, in2 <-chan int, out chan<- int, n int) { go func() { nMutex.Lock() var waitingChans []chan Result for i := 0; i < n; i++ { x1, x2 := <-in1, <-in2 waitingChan := make(chan Result, 1) waitingChans = append(waitingChans, waitingChan) go func(x1, x2 int, waitingChan chan Result) { waitingChan <- Result{out, f(x1) + f(x2)} }(x1, x2, waitingChan) } for _, waitingChan := range waitingChans { result := <-waitingChan result.out <- result.value } nMutex.Unlock() }() }

package main import ( "sync" ) var jobsMutex, nMutex sync.Mutex type Pair struct { x1, x2 int out chan<- int f func(int) int } type Result struct { out chan<- int value int x1, x2 int } var jobs = make(chan Pair) func worker(n int, done chan bool) { jobsMutex.Lock() var waitingChans []chan Result for i := 0; i < n; i++ { pair := <-jobs waitingChan := make(chan Result, 1) waitingChans = append(waitingChans, waitingChan) go func(pair Pair, waitingChan chan Result) { waitingChan <- Result{pair.out, pair.f(pair.x1) + pair.f(pair.x2), pair.x1, pair.x2} }(pair, waitingChan) } for _, waitingChan := range waitingChans { result := <-waitingChan result.out <- result.value } jobsMutex.Unlock() done <- true } func Merge2Channels(f func(int) int, in1 <-chan int, in2 <-chan int, out chan<- int, n int) { done := make(chan bool) go worker(n, done) go func() { nMutex.Lock() for i := 0; i < n; i++ { x1, x2 := <-in1, <-in2 pair := Pair{x1, x2, out, f} jobs <- pair } nMutex.Unlock() <-done }() }

package main type Terms struct { x int y int } func MergeChannels(in1 <-chan int, in2 <-chan int, repeat int) chan Terms { var out = make(chan Terms, repeat) go func() { for i := 0; i < repeat; i++ { out <- Terms{x: <-in1, y: <-in2} } }() return out } func CalcFunction(merged chan Terms, f func(int) int, repeat int) []chan int { var outChans = make([]chan int, repeat) for i := 0; i < repeat; i++ { out := make(chan int) outChans[i] = out terms := <- merged go func() { out <- f(terms.x) + f(terms.y) }() } return outChans } var syncChan = make(chan int, 1) func Merge2Channels(f func(int) int, in1 <-chan int, in2 <-chan int, out chan<- int, n int) { go func() { syncChan <- 1 merge := MergeChannels(in1, in2, n) calc := CalcFunction(merge, f, n) for i := 0; i < n; i++ { out<- <-calc[i] } <- syncChan }() }

package main import ( "sync" ) var gmtx sync.Mutex func Merge2Channels(f func(int) int, in1 <-chan int, in2 <-chan int, out chan<- int, n int) { go func(mf func(int) int, min1 <-chan int, min2 <-chan int, mout chan<- int, mn int) { gmtx.Lock() c := make(chan struct{}) res := make(chan chan int, 50000) go func(am int) { for i := 0; i < am; i++ { mout <- <-<-res } c <- struct{}{} close(res) }(mn) calc := func( iMf func(int) int, val1 int, val2 int, ) chan int { r := make(chan int, 1) inCalc := func( iMf func(int) int, val1 int, val2 int, r chan int, ) { r <- iMf(val1) + iMf(val2) close(r) } go inCalc(iMf, val1, val2, r) return r } for i := 0; i < mn; i++ { res <- calc(mf, <-min1, <-min2) } <-c close(c) gmtx.Unlock() }(f, in1, in2, out, n) }

package main import ( "container/list" "sync" "sync/atomic" ) var jobsMutex, nMutex sync.Mutex type Pair struct { x1, x2 int out chan<- int } type Result struct { out chan<- int value int } var jobs = make(chan Pair) var count int64 func master(f func(int) int) { jobsMutex.Lock() var q = list.New() var w sync.WaitGroup var wcc []chan Result loop: for { select { case pair := <-jobs: q.PushBack(pair) atomic.AddInt64(&count, -1) default: if atomic.LoadInt64(&count) <= 0 { break loop } } } for q.Len() > 0 { e := q.Front() pair := e.Value.(Pair) wc := make(chan Result, 1) wcc = append(wcc, wc) w.Add(1) go func(pair Pair, wc chan Result) { wc <- Result{pair.out, f(pair.x1) + f(pair.x2)} }(pair, wc) q.Remove(e) } for _, wc := range wcc { r := <-wc r.out <- r.value w.Done() } w.Wait() jobsMutex.Unlock() } func Merge2Channels(f func(int) int, in1 <-chan int, in2 <-chan int, out chan<- int, n int) { atomic.AddInt64(&count, int64(n)) go master(f) go func() { nMutex.Lock() for i := 0; i < n; i++ { x1, x2 := <-in1, <-in2 pair := Pair{x1, x2, out} jobs <- pair } nMutex.Unlock() }() }