Sequence

Sequence

Kotlin 에서는 Java의 Stream 과 유사한 Sequence 라는 기능을 지원한다.

Sequence 는 Iterator 를 잘 이용한 방법 중 하나이다.

Collection 과의 차이

Collection 에서 연산은 즉시(eager) 발생한다. 하지만, Sequence 에서 연산은 지연(lazy) 처리된다.

즉 map, filter 등이 호출될 때, Collection 에서는 전체 원소를 순회하지만, Sequence 는 순회하지 않을 수 있다.

이런 방식은 데이터가 많거나, first 와 같은 쇼트 서킷 연산의 경우 도움이 되고, 원하는 값을 찾았을 때, 시퀀스를 종료할 수 있게 도와준다.

실행흐름

1,2,3,4,5 로 이루어진 list 와 sequence 가 있다고 했을 때

collection 방식은 모든 원소들이 각 연산을 모두 순회한다.

sequence 방식은 지연평가 되어, 각 단계가 최종 연산에 다다랐을 때 동작이 수행된다.

쇼트 서킷 함수 사용

collection 의 map 과 filter 는 기본적으로 즉시 처리되므로 filter 의 경우 그 동작이 비효율적일 수 있다.

예를 들어, 결과로 단 한개의 결과만 필요한 경우 굳이 해당하는 결과를 찾았음에도 불구하고, 다른 원소에 대해서 반복을 수행하는 것은 불필요하다.

이 때, 사용하는 것이 쇼트 서킷 함수이다.

쇼트 서킷 함수는 특정 조건을 만족하는 경우 다음 연산을 굳이 진행시키지 않아도 되는 경우 사용한다.

// 총 205번 연산
(100 .. 200).map { it * 2 }
    .filter { it % 3 == 0 } // 여기서 101번의 불필요한 연산이 발생하게된다.
    .first()

// 총 104번 연산
(100 .. 200).map { it * 2 }
    .first { it % 3 == 0 } // 조건을 만족하는 경우 이후 원소는 불필요한 연산을 수행하지 않게된다.

코틀린에서는 이보다 더 좋은 방법을 제공한다.

// 총 6번 연산
(100 .. 2_000_000).asSequence() // 주목할만한 점은 200만으로 올렸음에도 실행속도에 영향이 없다.
    .map { println("doubling $it"); it * 2 }
    .filter { println("filtering $it"); it % 3 == 0 }
    .first() // 시퀀스가 비었다면 NoSuchElementException이 발생한다.

결과 (총 6번의 연산)

doubling 100
filtering 200
doubling 101
filtering 202
doubling 102
filtering 204

시퀀스가 비었을 때, first 를 사용한 경우 예외가 발생한다. 이렇게 시퀀스가 비는 경우가 발생할 수 있다면, first 대신 firstOrNull 을 사용하자. (응답 타입이 Nullable 로 변경된다.)

Exception in thread "main" java.util.NoSuchElementException: Sequence is empty.
	at kotlin.sequences.SequencesKt___SequencesKt.first(_Sequences.kt:111)
	at sequencetest.MainKt.main(Main.kt:26)
	at sequencetest.MainKt.main(Main.kt)

중간 연산과 최종 연산

시퀀스는 Collection API 에 들어있는 연산과 똑같은 함수를 가지고 있지만, 두가지 유형의 범주로 나뉜다.

바로 중간 연산(intermediate operation)과 최종 연산(terminal operation)이라는 범주다.

map 과 filter 와 같은 중간 연산은 새로운 시퀀스를 리턴한다.

first 와 toList, forEach 와 같은 최종 연산은 시퀀스가 아닌 다른 것을 리턴한다.

중요한 점은 시퀀스는 최종 연산 없이는 데이터를 처리하지 않고, 지연 평가를 한다는 것이다.

Java의 Stream 과의 차이점

코틀린의 일부 시퀀스는 여러 번 순회할 수 있고, 그렇지 못한 시퀀스는 여러 번 순회가 불가능하다고 문서화 되어있다. (근데, 개인적인 생각으론 불변 문제도 있고..)

자바의 Stream 은 병렬처리가 가능하기 때문에, 이 부분을 사용하려면 Stream 을 사용해야 한다.

Sequence 생성

이미 원소가 있다면 sequenceOf 를 사용하고 Iterable 이 있다면 asSequence 를 사용한다.

asSequence 파헤쳐보기

Iterable 을 받아서 어떻게 sequence 를 만드는 것일까?

public fun <T> Iterable<T>.asSequence(): Sequence<T> {
    return Sequence { this.iterator() }
}

Sequence 인터페이스는 다음과 같이 정의되어 있다.

public interface Sequence<out T> {
    /**
     * Returns an [Iterator] that returns the values from the sequence.
     *
     * Throws an exception if the sequence is constrained to be iterated once and `iterator` is invoked the second time.
     */
    public operator fun iterator(): Iterator<T>
}

map 은 확장함수로 다음과 같이 정의되어 있다.

public fun <T, R> Sequence<T>.map(transform: (T) -> R): Sequence<R> {
    return TransformingSequence(this, transform)
}

TransformingSequence 클래스는 다음과 같이 정의되어 있다.

internal class TransformingSequence<T, R>
constructor(private val sequence: Sequence<T>, private val transformer: (T) -> R) : Sequence<R> {
    override fun iterator(): Iterator<R> = object : Iterator<R> {
        val iterator = sequence.iterator()
        override fun next(): R {
            return transformer(iterator.next())
        }

        override fun hasNext(): Boolean {
            return iterator.hasNext()
        }
    }

    internal fun <E> flatten(iterator: (R) -> Iterator<E>): Sequence<E> {
        return FlatteningSequence<T, R, E>(sequence, transformer, iterator)
    }
}

first 는 확장함수로 다음과 같이 정의되어 있다.

public fun <T> Sequence<T>.first(): T {
    val iterator = iterator()
    if (!iterator.hasNext())
        throw NoSuchElementException("Sequence is empty.")
    return iterator.next()
}

firstOrNull 은 확장함수로 다음과 같이 정의되어 있다.

public fun <T> Sequence<T>.firstOrNull(): T? {
    val iterator = iterator()
    if (!iterator.hasNext())
        return null
    return iterator.next()
}

따라서 iterator 를 최종 연산에서 호출해서 지연 연산이 이뤄지는 원리다.

시퀀스 생성 함수 generateSequence

2개의 인자(초기값, 시퀀스에 속하게 될 다음 값을 생산하는 함수)를 받는다.

public fun <T : Any> generateSequence(seed: T?, nextFunction: (T) -> T?): Sequence<T> =
    if (seed == null)
        EmptySequence
    else
        GeneratorSequence({ seed }, nextFunction)

EmptySequence 클래스

private object EmptySequence : Sequence<Nothing>, DropTakeSequence<Nothing> {
    override fun iterator(): Iterator<Nothing> = EmptyIterator
    override fun drop(n: Int) = EmptySequence
    override fun take(n: Int) = EmptySequence
}

GeneratorSequence 클래스

private class GeneratorSequence<T : Any>(private val getInitialValue: () -> T?, private val getNextValue: (T) -> T?) : Sequence<T> {
    override fun iterator(): Iterator<T> = object : Iterator<T> {
        var nextItem: T? = null
        var nextState: Int = -2 // -2 for initial unknown, -1 for next unknown, 0 for done, 1 for continue

        private fun calcNext() {
            nextItem = if (nextState == -2) getInitialValue() else getNextValue(nextItem!!)
            nextState = if (nextItem == null) 0 else 1
        }

        override fun next(): T {
            if (nextState < 0)
                calcNext()

            if (nextState == 0)
                throw NoSuchElementException()
            val result = nextItem as T
            // Do not clean nextItem (to avoid keeping reference on yielded instance) -- need to keep state for getNextValue
            nextState = -1
            return result
        }

        override fun hasNext(): Boolean {
            if (nextState < 0)
                calcNext()
            return nextState == 1
        }
    }
}

최종 연산이 수행될 때 부터 연산이 완벽히 끝날 때까지 해당 Sequence Generator 는 계속해서 Sequence 원소를 생성한다. (무한루프가 가능하다.)

시퀀스의 일부 가져오기

take(count) 를 사용하거나, takeWhile(predicate) 를 사용한다. generateSequence 에 null 을 리턴하는 람다를 사용하면 null 이 발생했을 때, 해당 시퀀스가 종료된다.(하지만 takeWhile 을 사용하는 것이 가독성이 더 좋다.)

• 처음 N개의 소수 찾기(take 사용)

  fun firstNPrimes(count: Int) = generateSequence(2, ::nextPrime)
      .take(count)
      .toList()

• 주어진 수보다 작은 소수(생성 함수에서 null 리턴)

  fun primesLessThan(max: Int): List<Int> =
      generateSequence(2) { n -> if (n < max) nextPrime(n) else null }
          .toList()
          .dropLast(1) // 한계 값을 넘는 소수를 하나 포함하므로 리턴하기 전에 잘라냄

• 주어진 수보다 작은 소수(takeWhile 사용)

  fun primesLessThan(max: Int): List<Int> = generateSequence(2, ::nextPrime)
      .takeWhile { it < max }
      .toList()

filter 와 takeWhile 의 차이

둘은 둘 다 Sequence 를 반환하는 중간 연산이다.

다만, filter 는 전체 원소를 모두 순회하는 반면, takeWhile 은 특정 조건을 만족하지 않는 경우(predicate 의 결과가 false 인 경우) 순회를 멈추고, Sequence 를 반환합니다.

또한, 무한시퀀스를 사용한 경우 filter 는 종료되지 않을 가능성이 있습니다.

Difference between takeWhile() and filter() method in Kotlin

sequence 에서 yield

sequence 에서는 yield 중단(suspend) 함수를 통해 값을 생성하는 람다를 제공해야 한다.

• sequence() 를 사용해 피보나치 수 생성하기

  fun fibonacciSequence() = sequence {
      var terms = Pair(0, 1)
  
      while (true) {
          yield(terms.first) // 여기서 sequence 의 원소가 반환된다.
          terms = terms.second to terms.first + terms.second
      }
  }

yieldAll 은 다수의 값을 Iterator 에 넘겨준다.

참고자료

• 코틀린 쿡북(저자: 켄 코우젠, 옮긴이: 김도남, ISBN: 9791189909147(1189909146))