Concurrency 102

22.11.2016

Но преди това...

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

Отговори:

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

Можем ли да дефинираме метод на функция? А на канал?

Никакъв проблем, стига да са "наш" тип: 

package main

import "fmt"


type queue chan string
type callableStriner func() string

func (q queue) add(s string) {
    q <- s
}
func (q queue) poll() string {
    return <-q
}
func (cs callableStriner) String() string {
    return cs()
}

func main() {
    q := make(queue, 10)
    q.add("test")
    fmt.Printf("%s\n", callableStriner(q.poll))
}

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

Как ще укажем, че типа Student имплементира интерфейса Person?

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

Как проверяваме дали дадена променлива foo е от тип bar?

Type assertions и interface conversions 

if b, ok := foo.(bar);ok {
    // b is a bar
}

и 

switch b := foo.(type) {
case bar:
    // b is a bar
case sort.Interface:
    // b is sortable!
default:
    // b is something else
}

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

Как може да преизползваме вече имплементирани методи и атрибути на някой тип в нов? 

type Student struct {
    Person
}

Допълнение на интерфейси и sneak-peak към error handling

type error interface {
    Error() string
}
package main

import (
	"fmt"
	"io/ioutil"
)


func main() {
    if contents, err := ioutil.ReadFile("/etc/hosts"); err != nil {
        fmt.Println(err.Error())
        fmt.Printf("%#v\n", err)
    } else {
        fmt.Printf("%s", contents)
    }
}

Създаване на грешки

func someFunc(a int) (someResult, error) {
    if a <= 0 {
        return nil, errors.New("a must be positive!")
    }
    // ...
}
func someFunc(a int) (someResult, error) {
    if a <= 0 {
        return nil, fmt.Errorf("a is %d, but it must be positive!", a)
    }
    // ...
}

Concurrency 102

Ще си говорим за

Предизвикателството

Виждате ли проблем с този код? 

type PubSub struct {
    // ...
    subscribers []chan string
}
func (ps *PubSub) Subscribe() <-chan string {
    subscriber := make(chan string)
    ps.subscribers = append(ps.subscribers, subscriber)
    return subscriber
}

Какво ще направи следния код: 

ps := NewPubSub()
for i := 0; i < 20; i++ {
    go func() {
        fmt.Println(<-ps.Subscribe())
    }()
}
// ...

Race condition: в subscribers има някъде от 1 до 20 елемента :)

Locking с канали

type PubSub struct {
    // ...
    subscribers []chan string
    lock        chan struct{}
}
func NewPubSub() *PubSub {
    ps := new(PubSub)
    ps.lock = make(chan struct{}, 1)
    // ...
}
func (ps *PubSub) Subscribe() <-chan string {
    subscriber := make(chan string)
    ps.lock <- struct{}{}
    ps.subscribers = append(ps.subscribers, subscriber)
    <-ps.lock
    return subscriber
}

Обратно към днешната лекция: да си спомним скуката

package main

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


func main() {
    boring := func(msg string) {
        for i := 0; ; i++ {
            fmt.Println(msg, i)
            time.Sleep(time.Duration(rand.Intn(1e3)) * time.Millisecond)
        }
    }

    go boring("boring!")
    fmt.Println("Listening.")
    time.Sleep(2 * time.Second)
    fmt.Println("You are way too boring. I am leaving.")
}

sync

sync е пакет, който ни дава синхронизационни примитиви от сравнително ниско ниво:

един полезен интерфейс: 

type Locker interface {
    Lock()
    Unlock()
}

и подпакет atomic с low-level memory примитиви

WaitGroup

Изчаква колекция от горутини да приключат и чак тогава продължава с изпълнението.
Така не правим простотии със time.Sleep, както преди. 

type WaitGroup struct {}

func (*WaitGroup) Add(delta int)
func (*WaitGroup) Done()
func (*WaitGroup) Wait()

Пример

package main

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


func main() {
    var wg sync.WaitGroup

    boring := func(msg string) {
        for i := 0; i < 8; i++ {
            fmt.Println(msg, i)
            time.Sleep(time.Duration(rand.Intn(1e3)) * time.Millisecond)
        }
        wg.Done()
    }

    wg.Add(1)
    go boring("boring!")
    fmt.Println("Waiting for the boring function to do its work")
    wg.Wait()
}

Стига вече с тая скука!

По - интересен пример

package main

import (
	"fmt"
	"net/http"
	"sync"
)


func main() {
    var wg sync.WaitGroup
    var urls = []string{
        "http://www.golang.org/",
        "http://www.google.com/",
        "http://www.somestupidname.com/",
    }
    for _, url := range urls {
        wg.Add(1) // Increment the WaitGroup counter.
        // Launch a goroutine to fetch the URL.
        go func(url string) {
            content, err := http.Get(url) // Fetch the URL.
            if err == nil {
                fmt.Println(url, content.Status)
            } else {
                fmt.Println(url, "has failed")
            }
            wg.Done() // Decrement the counter when the goroutine completes.
        }(url)
    }
    // Wait for all HTTP fetches to complete.
    wg.Wait()
}

Mutex

type Mutex struct {}

func (*Mutex) Lock()
func (*Mutex) Unlock()

В код

package main

import (
	"fmt"
	"sync"
)


func main() {
    var (
        count    int
        countMtx sync.Mutex
        countWg  sync.WaitGroup
    )

    worker := func() {
        countMtx.Lock()
        count += 1
        countMtx.Unlock()
        countWg.Done()
    }

    for i := 0; i < 8; i++ {
        countWg.Add(1)
        go worker()
    }

    countWg.Wait()
    fmt.Println("counter:", count)
}

На каналски

package main

import (
	"fmt"
	"sync"
)


func main() {
    var (
        count   int
        countWg sync.WaitGroup
    )
    ch := make(chan struct{}, 1)

    worker := func() {
        ch <- struct{}{}
        count += 1
        <-ch
        countWg.Done()
    }

    for i := 0; i < 8; i++ {
        countWg.Add(1)
        go worker()
    }

    countWg.Wait()
    fmt.Println("counter:", count)
}

RWMutex

type RWMutex struct {}
func (*RWMutex) Lock()
func (*RWMutex) Unlock()
func (*RWMutex) RLock()
func (*RWMutex) RUnlock()
func (*RWMutex) RLocker() sync.Locker

Once

Обект от този тип ще изпълни точно една функция. 

package main

import (
	"fmt"
	"sync"
)


func main() {
    var once sync.Once
    var wg sync.WaitGroup

    onceBody := func() {
        fmt.Println("Only once")
    }
    anotherBody := func() {
        fmt.Println("Another")
    }

    for i := 0; i < 10; i++ {
        wg.Add(1)
        go func() {
            once.Do(onceBody)
            once.Do(anotherBody)
            wg.Done()
        }()
    }
    wg.Wait()
}

Cond

type Cond struct {
    L Locker // this is held while observing or changing the condition
    // other unexported fields ...
}

func NewCond(l Locker) *Cond

func (c *Cond) Wait()
func (c *Cond) Broadcast()
func (c *Cond) Signal()

select

select

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

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

Въпрос

Какво би направил следния код? 

package main

import "fmt"
import "time"


func main() {
    c := make(chan int)

    go func() {
        c <- 42
    }()

    select {
    case v1 := <-c:
        fmt.Printf("received %d in v1\n", v1)
    case v2 := <-c:
        fmt.Printf("received %d in v2\n", v2)
    case <-time.After(1 * time.Nanosecond):
        fmt.Printf("timeout\n")
    default:
        fmt.Printf("nothing!\n")
    }
}

Who knows :)

Concurrency patterns

Timeout

package main

import (
	"fmt"
	"net/http"
	"time"
)


func fetch(url string) <-chan *http.Response {
    ch := make(chan *http.Response)

    go func() {
        if response, err := http.Get(url); err == nil {
            ch <- response
        }
    }()

    return ch
}

func main() {
    select {
    case resp := <-fetch("https://www.google.com/search?q=golang"):
        fmt.Printf("received %v from google\n", resp.Status)
    case resp := <-fetch("https://www.bing.com/search?q=golang"):
        fmt.Printf("received %v from bing\n", resp.Status)
    case <-time.After(500 * time.Millisecond):
        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

package main

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


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
}

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

Fan In

Fan In <- Concurrency

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 {
    c := make(chan string)
    go func() {
        for {
            select {
            case s := <-input1:
                c <- s
            case s := <-input2:
                c <- s
            }
        }
    }()
    return c
}

func main() {
    c := fanIn(talk("Ned"), talk("Doycho"))
    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() {
        for {
            select {
            case s := <-input1:
                c <- s
            case s := <-input2:
                c <- s
            case <-finish:
                close(c)
                wg.Done()
                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() {
		for {
			select {
			case s := <-input1:
				c <- s
			case s := <-input2:
				c <- s
			case <-finish: // HL
				close(c) // HL
				wg.Done()
				return
			}
		}
	}()
	return c
}

// FANIN END OMIT


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

Words of wisdom

И отново:

Въпроси?

fmi@golang.bg

http://fmi.golang.bg/

@fmi_golang