NDS#37 ゴルーチンと並行性パターン

README.md

ゴルーチンと並行性パターン

第37回勉強会(2014/08/09) - 長岡 IT開発者 勉強会(NDS) 発表資料

Usage

• Basic

    $ go get code.google.com/p/go.tools/cmd/present
    $ present -orighost="localhost"
  

  Go to the following URL: http://localhost:3999/_main.slide

• Vagrant

    $ vagrant up
    $ vagrant ssh -c 'cd /vagrant && GOPATH=/usr/local present -http="0.0.0.0:3999" -orighost="localhost"'
  

  Go to the following URL: http://localhost:3999/_main.slide

• Docker

    $ docker build -t 'nds37:go-present' .
    $ docker run -d --net="host" --name="go-present" -v `pwd`:/mnt nds37:go-present
  

  Go to the following URL: http://localhost:3999/_main.slide

_main.slide

ゴルーチンと並行性パターン
長岡 IT開発者 勉強会(NDS) - 第37回勉強会
9 Aug 2014

@hayajo

* Goにおける並行性基礎

* ゴルーチン

並行性のための基本機能。スレッドのようなもので、とても軽量。

goステートメントで生成される。

- 関数
.code -edit goroutine.go /START1/,/END1/

- メソッド
.code -edit goroutine.go /START2/,/END2/

- クロージャ
.code -edit goroutine.go /START3/,/END3/

* ゴルーチン: Example

.play -edit goroutine.go HL1

* 同期

syncパッケージで提供される基本的な同期機能。

- sync.Mutex (ミューテックス)
- sync.Once (pthread_once())
- sync.WaitGroup (カウンタセマフォ)
- sync.Pool (フリーリスト）

* 同期: Example - sync.WaitGroup

.play -edit sync-waitgroup.go

* チャネル

ゴルーチン間の通信を提供。双方向パイプに近い。メッセージ型を指定して利用する。

.code -edit channel.go /START1/,/END1/

送受信はバッファサイズ分でブロックされる。（デフォルトはバッファなし）

.code -edit channel.go /START2/,/END2/ HL1

引数や戻り値にも使用できる。

* チャネル: Example

.play -edit channel.go HL1

* 並行性パターン

* Bockground jobs

.play -numbers -edit background-jobs.go /START/,/END/

* Unbuffered channel

.code -numbers -edit unbuffered-channel.go /START1/,/END1/
.play -numbers -edit unbuffered-channel.go /START2/,/END2/

* Buffered channel

.play -numbers -edit buffered-channel.go /START/,/END OMIT/

* Reading from multiple channels: select

.play -numbers -edit select.go /START/,/END/

* Consume all values from a channel: range

.play -numbers -edit range.go /START/,/END/

* Signaling

.play -numbers -edit signaling.go /START/,/END/

チャネルからの読み込みは値とチャネルの状態を返す。

    value, isOpen := <- channel

チャネルが閉じられている場合、値はゼロ値、状態はfalseとなり読み込みはブロックされない。

* Safe counter(1)

.code -numbers -edit safe-counter.go /START1/,/END1/

* Safe counter(2)

.play -numbers -edit safe-counter.go /START2/,/END2/

単純なカウンタであれば sync.atomic パッケージでも実現可能。

* Timer

.play -numbers -edit timer.go /START/,/END/

- time.After
- time.Tick
- time.Sleep

* Fan-out(1)

処理を分散させる

.code -numbers -edit fan-out.go /START1/,/END1/

* Fan-out(2)

.play -numbers -edit fan-out.go /START2/,/END2/

* Fan-in(1)

チャネルを多重化する

.code -numbers -edit fan-in.go /START1/,/END1/

* Fan-in(2)

.code -numbers -edit fan-in.go /START2/,/END2/
.play -numbers -edit fan-in.go /START3/,/END3/

* Coroutine

[[http://ja.wikipedia.org/wiki/%E3%82%B3%E3%83%AB%E3%83%BC%E3%83%81%E3%83%B3][コルーチン - Wikipedia]]

.code -numbers -edit coroutine.go /START1/,/END1/
.play -numbers -edit coroutine.go /START2/,/END2/

* Generator

[[http://ja.wikipedia.org/wiki/%E3%82%B8%E3%82%A7%E3%83%8D%E3%83%AC%E3%83%BC%E3%82%BF][ジェネレータ (プログラミング) - Wikipedia]]

.code -numbers -edit generator.go /START1/,/END1/
.play -numbers -edit generator.go /START2/,/END2/

* Future(1)

[[http://ja.wikipedia.org/wiki/Future][future - Wikipedia]]

.code -numbers -edit promise.go /START1/,/END1/

* Future(2)

.play -numbers -edit promise.go /START2/,/END2/

* Actor model(1)

[[http://ja.wikipedia.org/wiki/%E3%82%A2%E3%82%AF%E3%82%BF%E3%83%BC%E3%83%A2%E3%83%87%E3%83%AB][アクターモデル - Wikipedia]]

.code -numbers -edit actor-model.go /START1/,/END1/

* Actor model(2)

.code -numbers -edit actor-model.go /START2/,/END2/

* Actor model(3)

.play -numbers -edit actor-model.go /START3/,/END3/

* Appendix

* 並列度をあげる

デフォルトは1。変更は runtime.GOMAXPROCS(int) で行う。

    runtime.GOMAXPROCS(2)

CPUコア数は runtime.NumCPU() で取得する。

    runtime.GOMAXPROCS(runtime.NumCPU())

* 現在のゴルーチン数を取得

.play -edit numgoroutine.go /START/,/END/

* 現在のメモリ情報を取得

.play -edit memstats.go /START/,/END/

actor-model.go

package main

import "log"

// START1 OMIT
type actor chan<- interface{} // HL

type PutReq struct {
	key, val string
}

type GetReq struct {
	key string
	ack chan<- string // HL
}
// END1 OMIT

// START2 OMIT
func NewMapActor() actor {
	a := make(chan interface{})

	go func() {
		m := make(map[string]string)
		for {
			switch r := (<-a).(type) { // HL
			case PutReq: // HL
				m[r.key] = r.val
			case GetReq: // HL
				r.ack <- m[r.key]
			}
		}
	}()

	return actor(a) // HL
}
// END2 OMIT

// START3 OMIT
func main() {
	a := NewMapActor()

	a <- PutReq{key: "hoge", val: "HOGEHOGE"} // HL

	ack := make(chan string)
	a <- GetReq{key: "hoge", ack: ack} // HL
	log.Println(<-ack)
}
// END3 OMIT

background-jobs.go

package main

import (
	"fmt"
	"strings"
	"sync"
)

func main() {
	// START OMIT
	var wg sync.WaitGroup // HL
	args := []string{"foo", "bar", "baz"}
	for _, a := range args {
		wg.Add(1) // HL
		go func(str string) {
			fmt.Println(strings.ToUpper(str))
			wg.Done() // HL
		}(a)
	}
	wg.Wait() // waiting for all goroutines to finish // HL
	// END OMIT
}

buffered-channel.go

package main

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

func main() {
	// START OMIT
	limCh := make(chan struct{}, 3) // HL

	for i := 0; i < 10; i++ {
		limCh <- struct{}{} // HL
		go func(i int) {
			defer func() { <-limCh }() // HL
			log.Println("BEGIN", i)
			time.Sleep(time.Duration(rand.Intn(5)) * time.Second)
			log.Println("END", i)
		}(i)
	}

	for i := 0; i < cap(limCh); i++ { // HL
		limCh <- struct{}{} // HL
	} // HL
	// END OMIT
}

channel.go

package main

import "fmt"

func main() {
	name := "Gopher"
	// START1 OMIT
	doneCh := make(chan struct{}) // HL1
	// END1 OMIT
	// START2 OMIT
	go func() {
		fmt.Printf("Hello, %s\n", name)
		doneCh <- struct{}{} // blocking until ready to read // HL1
	}()
	<-doneCh // blocking until a value is received // HL1
	// END2 OMIT
}

coroutine.go

package main

import "log"

// START1 OMIT
func hello(yield chan<- string) {
	go func() {
		yield <- "Hello" // HL
		yield <- "Gopher" // HL
		yield <- "Hello NDS" // HL
	}()
}
// END1 OMIT

func main() {
	// START2 OMIT
	co := make(chan string)
	hello(co)
	log.Println(<-co, "World")
	log.Println("Hello", <-co)
	log.Println(<-co)
	// END2 OMIT
}

daisy-chain.go

package main

import "log"

func filter(in, out chan int) {
	out <- 1 + <-in
}

func main() {
	begin := make(chan int)
	in := begin
	var out chan int

	for i := 0; i < 10; i++ {
		out = make(chan int)
		go filter(in, out)
		in = out
	}

	begin <- 0

	log.Println(<-out)
}

Dockerfile

FROM ubuntu:trusty

RUN apt-get update && apt-get install -y golang mercurial
ENV GOPATH /usr/local
RUN go get code.google.com/p/go.tools/cmd/present

EXPOSE 3999
CMD cd /mnt && present -orighost="localhost"

fan-in.go

package main

import (
	"bytes"
	"log"
	"sync"
)

// START1 OMIT
func fanIn(ch ...<-chan byte) <-chan byte {
	var wg sync.WaitGroup
	outCh := make(chan byte) // HL
	for _, c := range ch { // HL
		wg.Add(1)
		go func(c <-chan byte) {
			for b := range c {
				outCh <- b // HL
			}
			wg.Done()
		}(c)
	}
	go func() {
		wg.Wait()
		close(outCh)
	}()
	return outCh // HL
}
// END1 OMIT

func main() {
	// START2 OMIT
	ch1 := make(chan byte)
	ch2 := make(chan byte)

	go func() {
		defer func() {
			close(ch1)
			close(ch2)
		}()
		for i := 0; i < 26; i++ {
			ch1 <- byte(65 + i) // A..Z
			ch2 <- byte(97 + i) // a..z
		}
	}()
	// END2 OMIT

	// START3 OMIT
	var b bytes.Buffer

	outCh := fanIn(ch1, ch2) // HL
	for v := range outCh { // HL
		b.WriteByte(v)
	}

	log.Println(b.String())
	// END3 OMIT
}

fan-out.go

package main

import (
	"log"
	"sync"
)

func main() {
	// START1 OMIT
	outCh := make(chan byte) // HL
	go func() {
		defer func() {
			close(outCh)
		}()
		for i := 0; i < 26; i++ {
			outCh <- byte(65 + i) // A..Z
		}
	}()
	// END1 OMIT

	var wg sync.WaitGroup

	// START2 OMIT
	startCh := make(chan struct{})
	for i := 0; i < 3; i++ {
		wg.Add(1)
		go func(i int) { // HL
			<-startCh
			for b := range outCh { // HL
				log.Printf("#%d %s", i, string(b))
			}
			wg.Done()
		}(i)
	}
	close(startCh)
	// END2 OMIT

	wg.Wait()
}

generator.go

package main

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

func init() {
	rand.Seed(time.Now().UnixNano())
}

// START1 OMIT
func generator(n int) <-chan int {
	yield := make(chan int) // HL
	go func() {
		for i := 0; i < n; i++ {
			yield <- rand.Int() // HL
		}
		close(yield)
	}()
	return yield
}
// END1 OMIT

func main() {
	// START2 OMIT
	for v := range generator(100) { // HL
		log.Println(v)
	}
	// END2 OMIT
}

goroutine.go

package main

import (
	"log"
	"os"
	"time"
)

func logging(fmt string, args ...interface{}) {
	log.Printf(fmt, args...)
}

func main() {
	name := "Gopher"
	// START1 OMIT
	go logging("Hello, %s\n", name) // HL1
	// END1 OMIT
	logger := log.New(os.Stdout, "", log.Ldate | log.Ltime)
	// START2 OMIT
	go logger.Printf("Hello, %s\n", name) // HL1
	// END2 OMIT
	// START3 OMIT
	go func() { // HL1
		log.Printf("Hello, %s\n", name) // HL1
	}() // HL1
	// END3 OMIT
	time.Sleep(100 * time.Millisecond)
}

memstats.go

package main

import (
	"encoding/json"
	"fmt"
	"runtime"
)

func main() {
	// START OMIT
	var m runtime.MemStats
	runtime.ReadMemStats(&m)
	// END OMIT
	b, _ := json.MarshalIndent(m, "", "  ")
	fmt.Println(string(b))
}

numgoroutine.go

package main

import (
	"log"
	"runtime"
)

func main() {
	// START OMIT
	log.Printf("running %d goroutines", runtime.NumGoroutine())
	// END OMIT
}

promise.go

package main

import (
	"fmt"
	"log"
	"strconv"
	"sync"
)

// START1 OMIT
type futureUint64 struct {
	value uint64
	ch    chan uint64
	once  sync.Once
}

func (f *futureUint64) String() string {
	f.once.Do(func() { f.value = <-f.ch }) // HL
	return strconv.FormatUint(f.value, 10)
}

func Fibonacci(n uint64) fmt.Stringer {
	f := new(futureUint64)
	f.ch = make(chan uint64)
	go func() { // HL
		f.ch <- fib(n) // HL
	}() // HL
	return f
}
// END1 OMIT

// START2 OMIT
func fib(n uint64) uint64 {
	switch n {
	case 0:
		return 0
	case 1, 2:
		return 1
	default:
		return fib(n-1) + fib(n-2)
	}
}

func main() {
	f := Fibonacci(20)
	log.Println("The 20th Fibonacci number is ", f)
}
// END2 OMIT

range.go

package main

import (
	"log"
	"time"
)

func main() {
	// START OMIT
	ch := make(chan int) // HL

	go func() {
		defer func() {
			close(ch) // HL
		}()
		for i := 0; i < 10; i++ {
			ch <- i
			time.Sleep(time.Second)
		}
	}()

	for v := range ch { // HL
		log.Println(v)
	}
	// END OMIT
}

safe-counter.go

package main

import (
	"log"
	"sync"
)

// START1 OMIT
func counter(ch chan<- int64) {
	go func() {
		var c int64 = 1
		for {
			ch <- c // HL
			c++
		}
	}()
}
// END1 OMIT

func main() {
	// START2 OMIT
	var wg sync.WaitGroup
	cntCh := make(chan int64) // HL
	counter(cntCh)
	start := make(chan struct{})
	for i := 0; i < 1000; i++ {
		wg.Add(1)
		go func(i int) {
			<-start
			log.Printf("#%03d: % 4d\n", i, <-cntCh) // HL
			wg.Done()
		}(i)
	}
	close(start)
	wg.Wait()
	// END2 OMIT
}

select.go

package main

import "log"

func main() {
	// START OMIT
	cntCh := make(chan int) // HL
	sumCh := make(chan int) // HL

	go func() {
		var s int
		for i := 0; i < 1000; i++ {
			cntCh <- i
			s += i
		}
		sumCh <- s
	}()

loop: // HL
	for { // HL
		select { // HL
		case c := <-cntCh: // HL
			log.Println(c)
		case s := <-sumCh: // HL
			log.Println("SUM", s)
			break loop // HL
		}
	}
	// END OMIT
}

signaling.go

package main

import (
	"log"
	"sync"
)

func main() {
	// START OMIT
	var wg sync.WaitGroup
	startCh := make(chan struct{})
	for i := 0; i < 1000; i++ {
		wg.Add(1)
		go func(i int) {
			<-startCh // blocking until the channel is closed // HL
			log.Println(i)
			wg.Done()
		}(i)
	}
	close(startCh) // closing the channel in order to start goroutines // HL
	wg.Wait()
	// END OMIT
}

sync-waitgroup.go

package main

import (
	"fmt"
	"sync"
)

func main() {
	name := "Gopher"
	var wg sync.WaitGroup // HL
	wg.Add(1) // HL
	go func() {
		defer wg.Done() // HL
		fmt.Printf("Hello, %s\n", name)
	}()
	wg.Wait() // HL
}

timer.go

package main

import (
	"container/ring"
	"fmt"
	"log"
	"time"
)

var aa []string = []string{
	"ヽ(ﾟ∀ﾟ)ﾉ",
	"( ﾟ∀)ﾉ",
	"(　　ﾟ)ﾉ",
	"ヽ(　　)ﾉ",
	"ヽ(ﾟ　　)",
	"ヽ(∀ﾟ )ﾉ",
}

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

	go func() {
		r := ring.New(len(aa))
		for _, s := range aa {
			r.Value = s
			r = r.Next()
		}
		for {
			r = r.Next()
			ch <- fmt.Sprintf("%v", r.Value)
			time.Sleep(100 * time.Millisecond)
		}
	}()

	// START OMIT
	timeoutCh := time.After(5 * time.Second) // HL

loop:
	for {
		select {
		case v := <-ch:
			fmt.Print("\x0c") // New Page
			fmt.Println(v)
		case <-timeoutCh: // HL
			log.Println("timed out")
			break loop
		}
	}
	// END OMIT
}

unbuffered-channel.go

package main

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

func init() {
	rand.Seed(time.Now().UnixNano())
}

// START1 OMIT
func from(conCh chan<- int) {
	conCh <- rand.Intn(100) // blocking until ready to read // HL
}

func to(conCh <-chan int) {
	fmt.Printf("Received %d\n", <-conCh) // blocking until a value is received // HL
}
// END1 OMIT

func main() {
	// START2 OMIT
	conCh := make(chan int) // HL
	go from(conCh)
	go to(conCh)
	// END2 OMIT
	time.Sleep(100 * time.Millisecond)
}

Vagrantfile

# -*- mode: ruby -*-
# vi: set ft=ruby :

Vagrant.configure(2) do |config|
  config.vm.box = "ubuntu/trusty64"

  config.vm.network "forwarded_port", guest: 3999, host: 3999

  config.vm.provision "shell", inline: <<-EOS
    set -x
    if [ ! `which go` ]; then
        apt-get update
        apt-get install -y golang mercurial
    fi
    export GOPATH=/usr/local
    go get code.google.com/p/go.tools/cmd/present
  EOS

end