GoLang: Concurency patterns (from Rob Pike Slides IO2010)

fanin--concurrency-pattern.go

package main
//fanIn channel multiplexer (concurrency patterns)
// Rob Pike: Concurency patterns example from  https://www.youtube.com/watch?v=f6kdp27TYZs
// playground https://play.golang.org/p/zmqbIwAy6YH

import (
	"fmt"
	"math/rand"
	"time"
)

// Генератор, ,пишен строки в канал внутренней горутиной
// Возвращает канал string
func say(msg string) <- chan string {
	c := make(chan string)
	go func() {
		for i:=0; ;i++ {
			// кладем в канал сгенерированные значения
			c <- fmt.Sprintf("%s %d", msg, i)
      // Случайная задержка
			time.Sleep(time.Duration(rand.Intn(1e3)) * time.Millisecond)
		}
	}()
	return c 
}


// fanIn функция мультиплексор каналов, читает из двух канало и пишет в выходной канал, данные с первого готового канала.
// Возвращает канал только для отправки `<-chan string`
func fanIn(in1, in2 <-chan string) <-chan string {
	out:=make(chan string)
	// создаем горутину вычитывающую из канала 1 и пишушую в канал вывода
	go func() { 
		// бесконечно читаем из канала in1
		for {
			// поместить в канал прочитанное из in1
			in1Val := <-in1
			out <- in1Val // тоже самое что out <- <-in1
		} 
	}()
	// создаем горутину вычитывающую из канала 2 и пишушую в канал вывода
	go func() { 
		// бесконечно читаем из канала in2
		for {
			// поместить в канал прочитанное из in2			
			out <- <-in2 // тоже самое что ut <- <-in2
		} 
	}()
	return out
}

func main() {
  // вызов без мультиплексора каналов fanIn
  a1 := say("A1")
  b1 := say("B1")
  for i := 0; i <5; i++ {
		fmt.Println(<-a1)	// если a1 не готов, то тут будет блокировка
    fmt.Println(<-b1) // если b1 не готов, то тут будет блокировка
	}
  
	rand.Seed(time.Now().UnixNano())
	c := fanIn(say("A"), say("B"))
	for i := 0; i <5; i++ {
		fmt.Printf("You say %q\n", <-c)	
	}
	fmt.Println("Exit.")
}

// если

fanin-with-select--concurrency-pattern.go

//fanIn channel multiplexer (concurrency patterns) версия с select 
// Rob Pike: Concurency patterns example from  https://www.youtube.com/watch?v=f6kdp27TYZs
// Вместо двух горутин используем одну, а для вычитки из канала по готовносит используем select
// Запсустить онлайн https://play.golang.org/p/6XtzoYhVrps
package main

import (
	"fmt"
	"math/rand"
	"time"
)

// Генератор, ,пишен строки в канал внутренней горутиной
// Возвращает канал string
func say(msg string) <- chan string {
	c := make(chan string)
	go func() {
		for i:=0; ;i++ {
			// кладем в канал сгенерированные значения
			c <- fmt.Sprintf("%s %d", msg, i)
      			// Случайная задержка
			time.Sleep(time.Duration(rand.Intn(1e3)) * time.Millisecond)
		}
	}()
	return c 
}


// fanIn функция мультиплексор каналов, читает из двух канало и пишет в выходной канал, данные с первого готового канала.
// Возвращает канал только для отправки `<-chan string`
func fanIn(in1, in2 <-chan string) <-chan string {
	out:=make(chan string)
	// создаем горутину вычитывающую из канала 1 и пишушую в канал вывода
	go func() { 
		// бесконечно читаем из канала in1
		for {
      			select {
        		// если сработала выборка из канала in1, то запишем полученное значение в канал результатов out
        		case s := <- in1: 
          			out <- s
        		case s := <- in2:
          			out <- s
      			}				
		} 
	}()
	return out
}

func main() {
  // вызов без мультиплексора каналов fanIn
  a1 := say("A1")
  b1 := say("B1")
  for i := 0; i <5; i++ {
    fmt.Println(<-a1)	// если a1 не готов, то тут будет блокировка
    fmt.Println(<-b1) // если b1 не готов, то тут будет блокировка
	}
  
	rand.Seed(time.Now().UnixNano())
	c := fanIn(say("A"), say("B"))
	for i := 0; i <5; i++ {
		fmt.Printf("You say %q\n", <-c)	
	}
	fmt.Println("Exit.")
}

fanin-with-wait-concurrency-pattern.go

// Это измененный пример fanin-concurrency-pattern.go
// В нем канал chan string заменен на канал из Message {str string;	wait chan bool} 
// Канал wait внутри Message служит семафором блокировки внутри функции `say`
package main

import (
	"fmt"
	"math/rand"
	"time"
)

type Message struct {
	str string
	wait chan bool
}
func fanIn(in1, in2 <-chan Message ) <-chan Message {
	out:=make(chan Message )
	// создаем горутину вычитывающую из канала 1 и пишушую в канал вывода
	go func() { 
		// бесконечно читаем из канала in1
		for {
			// поместить в канал прочитанное из in1
			in1Val := <-in1
			out <- in1Val // тоже самое что out <- <-in1
		} 
	}()
	return out
}

func say(msg string) <- chan Message {
	waitForIt := make(chan bool)
	c := make(chan Message )
	go func() {
		for i:=0; ;i++ {
			// кладем в канал сгенерированные значения
			c <- Message{ fmt.Sprintf("%s %d", msg, i),  waitForIt}
      // Случайная задержка
			time.Sleep(time.Duration(rand.Intn(1e3)) * time.Millisecond)
			<- waitForIt // блокирует выполнение горутины до запси в канал waitForIt
		}
	}()
	return c 
}

func main() {
	c := fanIn(say("A"), say("B"))
	for i:=0; i<5;i++ {
		msg1:=<-c; fmt.Println(msg1.str)
		msg2:=<-c; fmt.Println(msg2.str)
		msg1.wait <- true // разблокирует исполнение горутины в которой создано msg1
		msg2.wait <- true // разблокирует исполнение горутины в которой создано msg2
	}
	fmt.Println("Exit.")

}

timer-using-select--concurrency-pattern.go

// Таймер на основе select
// Rob Pike: Concurency patterns example from  https://youtu.be/f6kdp27TYZs?t=1389
// В видео есть ошибка, если указывать как в видео case <-time.After(1 * time.Second), 
// то таймер создается в каждом шаге цикла и никогда не выдаст значение, т.е. не сработает
// он должен быть задан до цикла for как переменная, чтобы сработать один раз

// Посмотреть в работе https://play.golang.org/p/SWjEhsmyjkq


package main

import (
	"fmt"
	"math/rand"
	"time"
)

// Генератор, ,пишен строки в канал внутренней горутиной
// Возвращает канал string
func say(msg string) <- chan string {
	c := make(chan string)
	go func() {
		for i:=0; ;i++ {
			// кладем в канал сгенерированные значения
			c <- fmt.Sprintf("%s %d", msg, i)
      			// Случайная задержка
			time.Sleep(time.Duration(rand.Intn(1e3)) * time.Millisecond)
		}
	}()
	return c 
}


// fanIn функция мультиплексор каналов, читает из двух канало и пишет в выходной канал, данные с первого готового канала.
// Возвращает канал только для отправки `<-chan string`
func fanIn(in1, in2 <-chan string) <-chan string {
	out:=make(chan string)
	// создаем горутину вычитывающую из канала 1 и пишушую в канал вывода
	go func() { 
		// бесконечно читаем из канала in1
		for {
      			select {
        		// если сработала выборка из канала in1, то запишем полученное значение в канал результатов out
        		case s := <- in1: 
          			out <- s
        		case s := <- in2:
          			out <- s
      			}				
		} 
	}()
	return out
}

func main() {
	c1 := say("A")
    	timer := time.After(time.Second * 1) 
	for {
		select {
		    case res := <-c1:
		        fmt.Println(res)
		    case <-timer :
		        fmt.Println("timeout 1")
			return
		}
	}

}