Concurrency 102

12.11.2014

Въпрос за мъфин #1

Как постигаме паралелизъм в Go?

• С go рутини създаваме конкурентна програма
• Ако сме били достатъчно добри тя ще се паралелизира лесно без да се налага да правим нищо.

Въпрос за мъфин #2

Какво е отношението между go рутини и нишки?

• Go рутините се пускат и спират от изпълнение върху нишките. Това го прави runtime-a.
• Прилича на това как процесите се пускат и спират върху процесора от OS-a
• Не е същото

Въпрос за мъфин #3

Каква е разликата между паралелизъм и конкурентност?

• Конкурентност е да се да се занимаваш с много неща едновременно. Става въпрос за структура на програмата.
• Паралелизъм е да правиш много неща едновременно. Става въпрос за execution.

Въпрос за мъфин #4

Как Дойчо ви препоръчва да изчакваме много go рутини да си свършат работата?

• sync.WaitGroup

Въпрос за мъфин #5

Какво става в go, когато всички go рутини блокират? Т.е. при deadblock?

• Програмата ви се панира и ви казва, че е имало deadlock

Cond

type Cond struct {
    // L is held while observing or changing the condition
    L Locker
    // contains filtered or unexported fields
}

• Нарича се condition-variable или monitor
• Синхронизира горутини на принципа на съобщаването за настъпване на дадено събитие
• Напълно в реда на нещата е да бъде част от наша структура
• Никога не трябва да го копирате след като решите да го ползвате!

• Demo code/08/cond.go

Communicate by sharing vs. Share by communicating

Channels

• Вграден тип, който се използва за комуникация между две горутини.
• Може да се използва и за синхронизация
• За тях има специален синтаксис
• Ползваме нещата от sync, ако не можем да ползваме канали

Употреба на канали

• Инстанцират се с make
• Подава се типа, който ще се пренася от канала
• Могат да бъдат буферирани и небуферирани
• В канал може да се изпраща и от него може да се получава
• Когато изпращаме или получаваме, това може да блокира

IO в канал

Операциите по изпращане и получаване се изпълняват с оператора <-

• chan <- стойност изпраща по канала
• [променлива] <- chan получава от канала

Канал може да бъде затворен

• Повече не може да бъде отворен
• Никога не блокира
• Писането в него води до паника
• Четенето връща нулевата стойност за типа

Създаване на канали

• Небуфериран канал за пренасяне на цели числа:

intChannel := make(chan int)

• Буфериран канал за пренасяне на стрингове:

stringBufferedChannel := make(chan string, 5)

• Канал само за четене:

readOnlyChannel := make(<-chan int)

• Канал само за писане:

writeOnlyChannel := make(chan<- int)

Каналите са първокласни обекти в Go

• По канал може да пренасяте канал

c := make(chan chan int)

• Каналите могат да се подават като параметри на функции

func doSomething(input <-chan string) {}

• Функциите могат да връщат канали като резултат.

func doSomethingElse() chan string {
  result := make(chan string)
  return result
}

nil channel

Никога не използвайте неинициализиран канал!

• Писането в него блокира завинаги

package main

func main() {
    var c chan string
    c <- "ping" // deadlock
}

• Четенето от него... блокира завинаги

package main

import "fmt"

func main() {
    var c chan string
    fmt.Println(<-c) // deadlock
}

Пример

c := make(chan int)

go func() {
    list.Sort()
    c <- 1
}()

doSomethingForAWhile()
<-c

• Не използвайте int или bool ако просто използвате канала за синхронизация.
• Използвайте struct{} за целта - безплатно от гледна точка на памет.

По-сложен пример

var sem = make(chan struct{}, MaxOutstanding)

func handle(r *Request) {
    <-sem
    process(r)
    sem <- struct{}{}
}

func init() {
    for i := 0; i < MaxOutstanding; i++ {
        sem <- struct{}{}
    }
}

func Serve(queue chan *Request) {
    for {
        req := <-queue
        go handle(req)
    }
}

Deadlock

func main() {
     c := make(chan int)
     c <- 42
     val := <-c
     println(val)
}

Затваряне на канали

package main

import "fmt"

func main() {
    ch := make(chan string)

    go func(output chan string) {
        for i := 0; i < 5; i++ {
            output <- fmt.Sprintf("sending N=%d", i)
        }
        close(output)
    }(ch)

    for i := 0; i < 7; i++ {
        val, ok := <-ch
        fmt.Printf("Recieved: %#v, %#v\n", val, ok)
    }

    ch <- fmt.Sprintf("where is your towel?")
}

range

Помните ли как ви казахме, че range е нещо супер яко?

• Може да чете и от канали
• Блокира, докато не получи следващата стойност
• Излизаме от for, когато каналът бъде затворен

for val := range ch {
    fmt.Printf("Recieved: %#v\n", val)
}

select

select {
case v1 := <-c1:
    fmt.Printf("received %v from c1\n", v1)
case v2 := <-c2:
    fmt.Printf("received %v from c2\n", v2)
default:
    fmt.Printf("no one was ready to communicate\n")
}

Накратко: switch за канали.
Надълго: Изчаква първия канал, по който е изпратена стойност

• Ако по никой от каналите не е изпратено нищо, изпълнява default
• Ако няма default блокира и изчаква
• Важна част от различни concurency patterns в Go

Concurrency patterns

Timeout

select {
case result := <-google:
    fmt.Printf("received %v from google\n", result)
case result := <-bing:
    fmt.Printf("received %v from bing\n", result)
case <- time.After(1 * time.Second):
    fmt.Printf("timed out\n")
}

Игра на развален телефон

• или колко точно са бързи и евтини горутините?

Игра на развален телефон

package main

import (
	"fmt"
)

func f(left, right chan int) {
    left <- 1 + <-right
}

func main() {
    const n = 100000
    leftmost := make(chan int)
    right := leftmost
    left := leftmost

    for i := 0; i < n; i++ {
        right = make(chan int)
        go f(left, right)
        left = right
    }

    go func(c chan int) { c <- 1 }(right)
    fmt.Println(<-leftmost)
}

Generators

package main

func fib() <-chan int {
    c := make(chan int)
    go func() {
        for a, b := 0, 1; ; a, b = b, a+b {
            c <- a
        }
    }()
    return c
}

func main() {
    fibonacci := fib()
    for i := 0; i < 10; i++ {
        println(<-fibonacci)
    }
}

Fan In

func talk(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
}

package main

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

func main() {
    kiro := talk("Kiro")
    ned := talk("Ned")
    for i := 0; i < 5; i++ {
        fmt.Println(<-kiro)
        fmt.Println(<-ned)
    }
}

// TALK START OMIT
func talk(msg string) <-chan string { // HL
	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
}

// TALK END OMIT

• Какъв е проблемът тук?

Fan In <- Concurrency

func fanIn(input1, input2 <-chan string) <-chan string {
    c := make(chan string)
    go func() {
        for {
            select {
            case s := <-input1:
                c <- s
            case s := <-input2:
                c <- s
            }
        }
    }()
    return c
}

package main

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

// TALK START OMIT
func talk(msg string) <-chan string { // HL
	c := make(chan string)
	r := rand.New(rand.NewSource(time.Now().UnixNano()))
	go func() {
		for i := 0; ; i++ {
			c <- fmt.Sprintf("%s: %d", msg, i)
			time.Sleep(time.Duration(r.Intn(1e3)) * time.Millisecond)
		}
	}()
	return c
}

// TALK END OMIT

// FANIN START OMIT
func fanIn(input1, input2 <-chan string) <-chan string { // HL
	c := make(chan string)
	go func() { // HL
		for {
			select { // HL
			case s := <-input1:
				c <- s // HL
			case s := <-input2:
				c <- s // HL
			} // HL
		}
	}()
	return c
}

// FANIN END OMIT

func main() {
    c := fanIn(talk("Ned"), talk("Kiril"))
    for i := 0; i < 10; i++ {
        fmt.Println(<-c)
    }
}

Finish channel

func fanIn(input1, input2 <-chan string, finish chan struct{}) <-chan string {
    c := make(chan string)
    go func() {
        defer wg.Done()
        defer close(c)
        for {
            select {
            case s := <-input1:
                c <- s
            case s := <-input2:
                c <- s
            case <-finish:
                return
            }
        }
    }()
    return c
}

Finish channel

package main

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

var wg sync.WaitGroup

// TALK START OMIT
func talk(msg string) <-chan string { // HL
	c := make(chan string)
	r := rand.New(rand.NewSource(time.Now().UnixNano()))
	go func() {
		for i := 0; ; i++ {
			c <- fmt.Sprintf("%s: %d", msg, i)
			time.Sleep(time.Duration(r.Intn(1e3)) * time.Millisecond)
		}
	}()
	return c
}

// TALK END OMIT

// FANIN START OMIT
func fanIn(input1, input2 <-chan string, finish chan struct{}) <-chan string { // HL
	c := make(chan string)
	go func() {
		defer wg.Done()
		defer close(c) // HL
		for {
			select {
			case s := <-input1:
				c <- s
			case s := <-input2:
				c <- s
			case <-finish: // HL
				return
			}
		}
	}()
	return c
}

// FANIN END OMIT

func main() {
    wg.Add(1)
    finish := make(chan struct{})
    c := fanIn(talk("Ned"), talk("Kiril"), finish)
    for value := range c {
        fmt.Println(value)
        if len(value) > 8 {
            close(finish)
        }
    }
    wg.Wait()
}

Въпроси?