[coroutine] Channels

[coroutine] Channels

https://github.com/Kotlin/kotlinx.coroutines/blob/master/docs/channels.md#channels

Channel basics

-

Channel 은 BlockingQueue 와 매우 비슷한 개념이다.

한가지 큰 차이는 blocking 하는 put 대신 suspending function 인 send 가, 

blocking 하는 take 대신 suspending function 인 receive 가 있다는 것이다.

fun main() = runBlocking{
    val channel = Channel<Int>()
    launch{
        for( x in 1..5) channel.send( x * x )
    }
    repeat(5) { println( channel.receive() ) }
    println("Done!")
}

// 1
// 4
// 9
// 16
// 25
// Done!

Closing and iteration over channels

-

queue 와 달리 channel 은 더 이상의 element 가 없음을 close 로 표현할 수 있다.

이렇게 하면 receiver side 에서는 channel 을 통해 element 를 받기 위해 for loop 을 쉽게 적용할 수 있다.

개념상 close 는 특별한 close token 을 channel 에 보내는 것과 같다.

반복(iteration)은 close token 이 전달되는 순간 끝나고, close 가 받아지기 전 모든 보내진 element 를 받는 것은 보장된다.

fun main() = runBlocking{
    val channel = Channel<Int>()
    launch{
        for( x in 1..5) channel.send( x * x )
        channel.close()
    }
    for( y in channel ) println(y)
    println("Done!")
}

// 1
// 4
// 9
// 16
// 25
// Done!

Building channel producers

-

coroutine 이 element sequence 를 만들어 내는 패턴은 아주 일반적이다.

producer-consumer pattern 은 동시성 코드에서 자주 찾아볼 수 있다.

이 producer 를 function 으로 만들 수 있고, channel 을 param 으로 받게 할 수 있다. 

반면에 결과는 반드시 function 으로부터 return 되어야 한다.

produce 라는 coroutine builder 를 사용하면 쉽게 producer side 를 구현할 수 있다.

그리고 extension function 인 "consumeEach" 를 통해 consume 의 for loop 를 대체할 수 있다.

// produce block 의 return 이 channel 에 대한 close 를 불러준다고 이해하면 되겠다.
fun CoroutineScope.produceSquares(): ReceiveChannel<Int> = produce{
    for (x in 1..5) send(x * x)
}

fun main() = runBlocking{
    val squares = produceSquares()
    squares.consumeEach { println( it ) }
    println("Done!")
}

// 1
// 4
// 9
// 16
// 25
// Done!

Pipelines

-

pipeline 은 하나의 coroutine 이 produce 하는 경우에 사용하는 pattern 이다. (영구적으로 produce 할수도)

그리고 다른 coroutine(s) 가 해당 stream 을 consume 하면서 process 해서 또 다른 producer를 만들어내는 경우에 사용한다.

fun main() = runBlocking{
    val numbers = produceNumbers()
    val squares = square(numbers)
    for (i in 1..5) println(squares.receive())
    println("Done!")
    coroutineContext.cancelChildren()
}

fun CoroutineScope.produceNumbers() = produce<Int> {
    var x = 1
    while (true) send(x++)
}

fun CoroutineScope.square(numbers: ReceiveChannel<Int>) : ReceiveChannel<Int> = produce{
    for ( x in numbers ) send( x * x )
}

// 1
// 4
// 9
// 16
// 25
// Done!

coroutine 을 만드는 모든 함수들은 CoroutineScope 의 extension 으로 정의되어 있다.

그래서 우리는 structured concurrency 를 사용하기 위해서 오래 지속되는 global coroutine 을 도입하지 않았다.

Prime numbers with pipeline

-

예제를 통해 pipeline 을 극단으로 써보자.

예제는 coroutine pipeline 을 이용해서 소수를 생성한다.

fun main() = runBlocking{
    var cur = numbersFrom(2)
    for ( i in 1..10 ){
        val prime = cur.receive()
        println(prime)
        cur = filter(cur, prime)
    }
    coroutineContext.cancelChildren()
}

fun CoroutineScope.numbersFrom(start:Int) = produce<Int>{
    var x = start
    while(true) send(x++)
}

fun CoroutineScope.filter(numbers: ReceiveChannel<Int>, prime:Int) = produce<Int>{
    for(x in numbers) if (x % prime != 0) send(x)
}

// 2
// 3
// 5
// 7
// 11
// 13
// 17
// 19
// 23
// 29

iterator coroutine builder 를 사용해서 같은 pipeline 을 만들 수 있다.

produce 대신 iterator 로 교체하고, send 대신 yield, receive 대신 next, ReceiveChannel 대신 Iterator 를 사용하고, coroutine scope 를 제거하면 된다.

runBlocking 도 필요없어진다.

하지만 channel 을 사용한 pipeline 의 장점은 Dispatchers.Default context 를 사용하면 multi core 를 활용할 수 있다는 점이다.

어쨌든 위의 방법은 소수를 찾는 현실적인 방법은 아니다.

현실에서는 pipeline 이 suspending function 을 호출하기도 하는데, suspending functino 을 사용하는 pipeline 은 sequence 나 iterator 로 대체할 수 없다.

왜냐하면 그것들은 async 할 수 있는 produce 와는 다르게 suspension 을 허락하지 않기 때문이다.

Fan-out

-

여러개의 coroutine 이 동일한 channel 에서 값을 받을 수 있다.

이를 통해 일을 분배할 수 있다.

fun main() = runBlocking<Unit>{
    val producer = produceNumbers()
    repeat(5) { launchProcessor(it, producer) }
    delay(950)
    producer.cancel()
}

fun CoroutineScope.produceNumbers() = produce<Int>{
    var x = 1
    while(true){
        send(x++)
        delay(100)
    }
}

fun CoroutineScope.launchProcessor(id: Int, channel: ReceiveChannel<Int>) = launch{
    for (msg in channel){
        println("Processor #$id received $msg")
    }
}

// Processor #0 received 1
// Processor #0 received 2
// Processor #1 received 3
// Processor #2 received 4
// Processor #3 received 5
// Processor #4 received 6
// Processor #0 received 7
// Processor #1 received 8
// Processor #2 received 9
// Processor #3 received 10

위의 결과에서 id 의 순서는 바뀔 수 있다.

producer coroutine 의 취소는 channel 의 close 와 연결된다.

따라서 결국 channel 을 처리하는 iteration 도 종료된다.

그리고 channel 에 대한 for loop 가 fan-out 을 어떻게 처리하는지 iterate 가 어떻게 도는지 보고파악해 보자.

consumeEach 와 달리 for loop pattern 은 multiple coroutine 에 대해서 완벽하게 안전하다.

만약에 한 processor 의 coroutine 이 실패해도 다른 녀석들은 여전히 channel 을 처리한다.

반면에 consumeEach 로 하는경우는 보통의 혹은 에러로 종료되었을 때 관련된 channel 을 취소시킨다.

Fan-in

-

여러개의 coroutine 이 하나의 channel 에 보낼 수도 있다.

fun main() = runBlocking{
    val channel = Channel<String>()
    launch { sendString(channel, "foo", 200L) }
    launch { sendString(channel, "BAR!", 500L) }
    repeat(6) {
        println(channel.receive())
    }
    coroutineContext.cancelChildren()
}

suspend fun sendString(channel: SendChannel<String>, s:String, time:Long){
    while(true){
        delay(time)
        channel.send(s)
    }
}

// foo
// foo
// BAR!
// foo
// foo
// BAR!

Buffered channels

-

지금까지 본 channel 은 buffer 가 없었다.

buffer 가 없는 channel 은 sender 와 receiver 가 서로 맞아야만 가능하다. (약속의 광장?)

만약 send 가 먼저 불리면, receive 가 불릴 때까지 suspend 하고,

만약 receive 가 먼저 불리면, send 가 불릴때까지 suspend 한다.

-

Channel() factory function 과 produce builder 는 capacity 라는 parameter 를 받는데 이는 buffer size 이다.

Buffer 는 BlockingQueue 가 capacity 를 가진 것처럼, sender 가 suspending 하기 전에 보낼 수 있고, buffer 가 다 차면 suspending 된다.

fun main() = runBlocking{
    val channel = Channel<Int>(4) // buffer size 4 channel
    val sender = launch {
        repeat(10){
            println("Sending $it")
            channel.send(it) // buffer 가 full 되기 전까지는 suspend 하지 않는다.
        }
    }
    delay(1000)
    sender.cancel()
}

// Sending 0
// Sending 1
// Sending 2
// Sending 3
// Sending 4

첫 4개의 element 는 buffer 에 담겨지고, 다섯번째 녀석을 보내기 전에 suspend 된다.

Channels are fair

-

channel  에 대한 send 와 receive 는 여러 코루틴으로부터의 호출로부터 순서적 입장에서 fair 하다.

선착순이다.

data class Ball(var hits: Int)

fun main() = runBlocking{
    val table = Channel<Ball>()
    launch { player("ping", table) }
    launch { player("pong", table) }
    table.send(Ball(0))
    delay(1000)
    coroutineContext.cancelChildren()
}

suspend fun player(name:String, table:Channel<Ball>){
    for (ball in table){
        ball.hits++
        println("$name $ball")
        delay(300)
        table.send(ball)
    }
}

// ping Ball(hits=1)
// pong Ball(hits=2)
// ping Ball(hits=3)
// pong Ball(hits=4)

"ping" coroutine 이 먼저 시작되어서 먼저 ball 을 받는다.

"ping" coroutine 이 send 를 호출한 후 바로 다시 ball 받을 수도 있지만, ball 은 "pong" coroutine 으로 간다.

왜냐하면 "pong" 이 먼저 기다리고 있었고, channel 은 fair 하기 때문이다.

가끔 channel 이 사용되는 executor 의 종류에 따라 unfair 한듯한 동작을 하는 것처럼 보일 수 있다.

Ticker channels

-

Ticker channel 은 특별한 만남의 광장(?) channel 로 매번 마지막 consume time 으로부터 특정 delay 후에 Unit 을 만들어 낸다.
( 돼왕 : 아래 실험 결과를 보면 마지막 consume time 으로부터 특정 delay 가 아니라, 그냥 특정 delay 로 Unit 을 만들어내며, last value 에 접근할 수 있다. 기본값인 FIXED_PERIOD 의 역할이며, 글에서 설명하는 마지막 consume 시점으로부터의 time interval 은 FIXED_DLEAY 라고 볼 수 있다.)

이것 자체만 보면 별로 쓸모없어 보이지만, 복잡한 시간 기반 produce pipeline 과 windowing 을 하거나, 시간 기반 operator 를 만들 때 유용하다.

Ticker channel 은 select 에서 on tick 동작을 위해 사용될 수 있다.

ticker factory method 를 사용해서 해당 channel 을 만들 수 있다.

더 이상 사용하지 않을때는 ReceiveChannel.cancel 을 호출하면 된다.

fun main() = runBlocking{
    val tickerChannel = ticker(delayMillis = 100, initialDelayMillis = 0)
    var nextElement = withTimeoutOrNull(1) { tickerChannel.receive() }
    println("Initial element is available immediately: $nextElement")

    nextElement = withTimeoutOrNull(50) { tickerChannel.receive() }
    println("Next element is ready in 50 ms: $nextElement")

    nextElement = withTimeoutOrNull(50) { tickerChannel.receive() }
    println("Next element is ready in 100 ms: $nextElement")

    println("Consumer pauses for 150ms")
    delay(150)

    nextElement = withTimeoutOrNull(1) { tickerChannel.receive() }
    println("Next element is available immediately after large consumer delay: $nextElement")

    nextElement = withTimeoutOrNull(50) { tickerChannel.receive() }
    println("Next element is ready in 50 ms after consumer pause in 150ms: $nextElement")

    tickerChannel.cancel()
}

// Initial element is available immediately: kotlin.Unit
// Next element is not ready in 50 ms: null
// Next element is ready in 100 ms: kotlin.Unit
// Consumer pauses for 150ms
// Next element is available immediately after large consumer delay: kotlin.Unit
// Next element is ready in 50ms after consumer pause in 150ms: kotlin.Unit

ticker 는 consumer 의 pause 를 기본적으로 인지하고, 생성된 element 간의 fixed rate 를 유지하기 위해 다음 produced element 를 delay 하여 전달해준다.

optional param 인 mode 에 TickerMode.FIXED_DELAY 값을 줄 수 있는데, 이는 element 간의 fixed delay 를 유지해준다.

buffered channels, chanel receive suspend function, channel capacity param, channel fan in, channel fan-out, channel pipeline, channel producers, channel receive, channel receive suspending function, channel send, channel send suspendi funcation, channel send suspending function, channels are fair, close token, consumeEach, consumeeach vs launch foor loop, coroutine channel, coroutine channel basics, coroutine channel close, coroutine receivechannel, coroutinecontext cancelchildren, coroutinescope extension, iteration, iterator coroutine builder, iterator yield next, prime numbers with pipeline, produce coroutine builder, produce vs iterator, producer cancel, producer consumer pattern, producer consumer pattern with channel, producer consumer pattern with coroutine, ticker based on last consume time, ticker channel, ticker mode, tickermode fixed_delay, [coroutine] Channels