go routine

Go루틴(goroutine)은 Go 런타임이 관리하는 Lightweight 논리적 (혹은 가상적) 쓰레드(주1)이다. Go에서 "go" 키워드를 사용하여 함수를 호출하면, 런타임시 새로운 goroutine을 실행한다. goroutine은 비동기적으로(asynchronously) 함수루틴을 실행하므로, 여러 코드를 동시에(Concurrently) 실행하는데 사용된다.

goroutine은 OS 쓰레드보다 훨씬 가볍게 비동기 Concurrent 처리를 구현하기 위하여 만든 것으로, 기본적으로 Go 런타임이 자체 관리한다. Go 런타임 상에서 관리되는 작업단위인 여러 goroutine들은 종종 하나의 OS 쓰레드 1개로도 실행되곤 한다. 즉, Go루틴들은 OS 쓰레드와 1 대 1로 대응되지 않고, Multiplexing으로 훨씬 적은 OS 쓰레드를 사용한다. 메모리 측면에서도 OS 쓰레드가 1 메가바이트의 스택을 갖는 반면, goroutine은 이보다 훨씬 작은 몇 킬로바이트의 스택을 갖는다(필요시 동적으로 증가). Go 런타임은 Go루틴을 관리하면서 Go 채널을 통해 Go루틴 간의 통신을 쉽게 할 수 있도록 하였다.

기본 사용방법

package main

import (
	"fmt"
    "time"
)

func say(s string) {
	for i := 0; i < 10; i++ {
    	fmt.Printf("%s-%d", i)
    }
}

func main() {
	say("Sync")
    
    go say("Async1")
    go say("Async2")
    go say("Async3")
    
    time.Sleep(time.Second * 3)
}

Go에서는 일반 함수 앞에 go keyword를 추가하여 호출하면 비동기적으로 호출된다.
비동기적으로 동작하도록 만드는데는 go keyword 하나면 충분합니다.
함수를 설계하는 단계에서는 해당 함수를 비동기로 사용할 것인지 아니면 동기로 사용할 것인지를 전혀 고민할 필요가 없다.
함수의 기능구현에만 집중히면 된다.
그리고 동기 또는 비동기로 사용여부는 사용자가 결정하면 된다.

익명함수 Go Routine

package main

import (
	"fmt"
    "sync"
)

func main() {
	var wait sync.WaitGroup  	// 1
    wait.Add(2)					// 2
    
    go func() {
    	defer wait.Done()		// 3
        fmt.Println("Hello")
    }()
    
    go func(msg string) {
    	defer wait.Done()
        fmt.Println(msg)
    }("Hi")
    
    //go func() {				// 5
	//	defer wait.Done()
	//	time.Sleep(time.Millisecond * 3)
	//	fmt.Println("Async")
	//}()
    
    wait.Wait()					// 4
}

1. WaitGroup 생성
2. 2개의 Go routine 기다림
3. Go routine이 끝나면 .Done 호출
4. Go routine이 모두 끝날때까지 대기
5. "Async" 문자열은 출력이 될까요?

type WaitGroup struct {
	noCopy noCopy

	// 64-bit value: high 32 bits are counter, low 32 bits are waiter count.
	// 64-bit atomic operations require 64-bit alignment, but 32-bit
	// compilers only guarantee that 64-bit fields are 32-bit aligned.
	// For this reason on 32 bit architectures we need to check in state()
	// if state1 is aligned or not, and dynamically "swap" the field order if
	// needed.
	state1 uint64
	state2 uint32
}

다중 CPU 처리

Go는 디폴트로 1개의 CPU를 사용한다. 즉, 여러 개의 Go 루틴을 만들더라도, 1개의 CPU에서 작업을 시분할하여 처리한다 (Concurrent 처리). 만약 머신이 복수개의 CPU를 가진 경우, Go 프로그램을 다중 CPU에서 병렬처리 (Parallel 처리)하게 할 수 있는데, 병렬처리를 위해서는 아래와 같이 runtime.GOMAXPROCS(CPU수) 함수를 호출하여야 한다 (여기서 CPU 수는 Logical CPU 수를 가리킨다).
Concurrency (혹은 Concurrent 처리)와 Parallelism (혹은 Parallel 처리)는 비슷하게 들리지만, 전혀 다른 개념이다.

package main
 
import (
    "runtime"  
)
 
func main() {
    // 3개의 CPU 사용
    runtime.GOMAXPROCS(3)
 
    ...
}

Go channel

channel이란 주로 Go Routine끼리 데이터를 주고 받기 위해 사용하는 통신 mechanism이다.
channel을 새로 선언하는 문장은 chan keyword로 표시하고, channel을 닫으려면 close() 함수를 호출해야 한다.

Channel을 통해 데이터를 주고 받는 데에는 몇가지 규칙이 있다.

1. 각 channel마다 특정한 데이터 타입으로만 데이터를 교환할 수 있다. 이를 그 채널의 element type이라고 한다.
2. channel이 정상적으로 작동하려면 channel로 데이터를 보내는 상대방이 있어야 한다.
3. channel을 함수의 매개변수로 사용할 때 반드시 channel의 방향, 다시 말해 데이터를 보낼 channel인지, 아니면 받을 channel인지를 명시해야 한다. channel의 용도를 미리 알고 있다면 최대한 이 규칙에 따라 작성해야 한다. 그래야 데이터를 받는 채널에 실수로 데이터를 보내거나 그 반대로 하는 일을 방지할 수 있어서 프로그램을 보다 안전하고 견고하게 만들 수 있다.

channel에 데이터 쓰기

package main

import (
	"fmt"
)

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

	go writeToChannel(c, 42)

	fmt.Println(3, <-c)						// 5
}

func writeToChannel(c chan int, x int) {	// 1
	fmt.Println(1, x)

	c <- x									// 2
	close(c)								// 3

	fmt.Println(2, x)
}
/*
1 42
2 42
3 42
*/

1. 매개변수를 channel로 선언하기 위해 chan keyword를 붙인다. 그리고 그 뒤에 반드시 channel의 type을 명시해야 한다.
2. c <- x 구문으로 x라는 값을 c라는 channel로 쓴다.
3. channel를 닫는다. channel에 데이터를 쓰고 닫지 않으면 또 다른 쓰기로 데이터를 덮어쓸 수 있다. 때문에 데이터를 쓴 후 항상 닫아주는 것이 좋다.
4. channel 변수를 선언한다.
5. <- c 구문으로 channel에 쓰여진 값을 사용한다.

channel에서 데이터 읽기

package main

import (
	"fmt"
	"time"
)

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

	go writeToChannel(c, 42)
	time.Sleep(time.Second)
	v := <-c					// 1
	fmt.Println(v)
	time.Sleep(time.Second)
    
	_, ok := <-c				// 2
	if ok {
		fmt.Println("Channel is open")
	} else {
		fmt.Println("Channel is closed")
	}
}

func writeToChannel(c chan int, x int) {
	c <- x
	close(c)
}

1. channel에서 데이터를 읽는다.
2. channel이 열려있는지 확인한다.

함수 매개변수로 지정한 채널

Go 언어에서는 channel을 함수 매개변수로 사용할 때 반향을 지정하는 기능을 제공한다. 따라서 읽기 용도인지 아니면 쓰기 용도인지를 지정할 수 있다. 이런 타입의 channel을 unidirectional channel이라고 한다.
별도로 지정하지 않으면 channel은 기본적으로 양방향이다.

func f1(c chan int, x int) {}	// 1

func f2(out <-chan int, in chan<- int) {}	// 2

1. 양방향 channel
2. 단방향 channel

Pipeline

Pipeline이란 Go routine과 channel을 연결하는 기법으로 channel로 데이터를 전송하는 방식으로 한쪽 Go routine의 출력을 다른 Go routine의 입력으로 연결할 수 있다.
Pipeline 사용의 장점

• Go routine이나 channel은 다른 작업이 끝날 때까지 기다릴 필요가 없이 실행을 시작할 수 있기 때문에 데이터 흐름을 일정하게 구현할 수 있다.
• 주고 받는 값을 일일이 변수에 저장할 필요가 없기 때문에 변수 사용 횟수를 줄일 수 있어 메모리 공간도 절약할 수 있다.
• 프로그램 설계를 간결하게 구성하여 유지 보수정이 높아진다.

package main

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

var CLOSEA = false
var DATA = make(map[int]bool)

func random(min, max int) int {
	return rand.Intn(max-min) + min
}

func first(min, max int, out chan<- int) {
	for {
    	if CLOSEA {
        	close(out)
            return
        }
        out <- random(min, max)
    }
}

func second(out chan<- int, in <-chan int) {
	for x := range in {
    	fmt.Print(x, " ")
        _, ok := DATA[x]
        if ok {
        	CLOSEA = true
        } else {
        	DATA[X] = true
            out <- x
        }
    }
    fmt.Println()
    close(out)
}

func third(in <-chan int) {
	var sum int
    sum = 0
    for x2 := range in {
    	sum = sum + x2
   	}
    fmt.Printf("The sum of the random numbers is %d\n", sum)
}

func main() {
	if len(os.Args) != 3 {
    	fmt.Println("Need two integer parameters!")
        os.Exit(1)
    }
    
    n1, _ := strconv.Atoi(os.Args[1])
    n2, _ := strconv.Atoi(os.Args[2])
    
    if n1 > n2 {
    	fmt.Printf("%d should be smaller than %d\n", n1, n2)
    	return
    }
    
    rand.Seed(time.Now().UnixNano())
    A := make(chan int)
    B := make(chan int)
    
    go first(n1, n2, A)
    go second(B, A)
    third(B)
}

Buffered channel

Go scheduler에서 더 많은 요청을 처리할 수 있도록 작업을 Queue 재빨리 저장할 때 이 타입의 channel을 사용한다. 또한 buffered channel을 semaphore처럼 사용해 애플리케이션의 처리량을 제한할 수도 있다.
Buffered channel의 동작 방식은 다음과 같다.

1. 들어온 요청은 모두 channel에 전달되고, 각각을 하나씩 처리한다.
2. Channel이 어떤 요청에 대한 처리 작업을 끝내면 호출한 측으로 새로운 작업을 처리할 준비가 됐다는 메시지를 보낸다.
3. Channel에서 사용하는 버퍼의 크기에 따라 동시에 처리할 수 있는 요청의 수가 결정된다.

package main

import (
	"fmt"
)

func main() {
	numbers := make(chan int, 5)
    counter := 10
    
    for i := 0; i < counter; i++ {
    	select {
        case numbers <- i:
        default:
           fmt.Println("Not enough space for", i)
        }
    }
    
    for i := 0; i < counter+5; i++ {
    	select {
        case num := <-numbers:
            fmt.Println(num)
        default:
            fmt.Println("Nothing more to be done!")
            break
        }
    }
}
/*
Not enough space for 5
Not enough space for 6
Not enough space for 7
Not enough space for 8
Not enough space for 9
0
1
2
3
4
Nothing more to be done!
Nothing more to be done!
Nothing more to be done!
Nothing more to be done!
Nothing more to be done!
Nothing more to be done!
Nothing more to be done!
Nothing more to be done!
Nothing more to be done!
Nothing more to be done!
*/

Go routine의 실행 순서 지정하기

Go routine이 실행되는 순서는 예측할 수 없지만, 간혹 실행 순서를 제어해야 할 때가 있다. 이때 signal channel을 이용해 Go routine의 실행 순서를 제어할 수 있다.

간단한 함수만으로도 쉽게 처리할 수 있는 일을 왜 굳이 Go routine으로 구성해 특정한 순서로 실행하는지 궁금할 수 있다.
Go Routine은 각자 동시에 실행될 수 있고, 다른 Go routine이 끝날 때까지 기다릴 수 있는 반면, 순차적으로 실행되는 일반 함수는 이렇게 할 수 없기 때문이다.

package main

import (
	"fmt"
    "time"
)

func A(a, b chan struct {}) {
	<-a
    fmt.Println("A()!")
    time.Sleep(time.Second)
    close(b)
}

func B(a, b chan struct{}) {
	<-a
    fmt.Println("B()!")
    close(b)
}

func C(a chan struct{}) {
	<-a
    fmt.Println("C()!")
}

func main() {
	x := make(chan struct{})
    y := make(chan struct{})
    z := make(chan struct{})
    
    go C(z)
    go A(x, y)
    go C(z)
    go B(y, z)
    go C(z)
    
    close(x)
    time.Sleep(time.Second * 3)
}