RxSwift의 기본 - 스케쥴러

스케쥴러(Scheduler)는 스레드(Thread)의 리엑티브엑스적 개념이다.
하지만, 스레드와 동일한 개념은 아니다.

일반적인 스레드의 설명을 먼저하자면

스레드는 크게 두가지로 나뉜다.
- 직렬 처리 스레드
- 병렬 처리 스레드

웹 Api를 리퀘스트A와 리스폰스B가 있다고 해보자
리퀘스트A와 리스폰스B는 직렬 처리이다.
만약 리퀘스트 A가 여러번 발생할 수 있다면 이것은 병렬처리로 해야한다.

즉, 하나의 처리는 대부분 직렬처리와 병렬처리가 병행된다.
일반적으로 하나의 큰 직렬처리 안에 여러개의 병렬처리가 쪼개져 있다.


GCD에서의 queue
- main: app의 main thread에서 동작 병렬 Queue
- global: 전체 시스템에서 공유하는 병렬 Queue
- custom: 개발자가 만들 수 있음.

iOS에서 UI의 변경 작업은 반드시 main 쓰레드에서 동작한다. (안드로이드도 동일)
또한 한번에 여러가지 UI이벤트가 있을 수 있으므로 동일하게 병렬 처리 큐이다.
또한 UX의 이유로 main thread는 정지하면 안된다.
UI animation 처리가 아닌 연산 처리로 main thread를 점유하면 안된다.


GCD의 큐에는 총 4가지의 QoS(Quality of Service: 우선 순위)가 존재한다.
- userinterative: 즉각적인 반응이 필요한 queue. main 쓰레드가 이 QoS를 사용한다.
- userinitiated: 유저의 UI 중 중요도가 떨어지는 element를 async로 작업할때 쓴다.
- utility: 유저의 프로세스에 필요한 연산작업에 사용한다. 프로그래스바, I/O, Networking 등
- background: 유저에게 직접적으로 필요하지 않은 작업들. logging 등

병렬처리에 있어서 주의 해야할 것들
공유자원
상호배제
데드락
starvation
우선순위 역전

멀티 스레딩에 있어서 가장 걱정해야 하는 것은 자원의 공유이다.
하나의 자원이 공유되는데 이 자원이 비동기(async) 병렬 작업으로 여러곳에서 순서없이 동시에 접근하여 데이터가 변경되는데 이 변경된 데이터를 다른곳에서 참조한다고 할때 데이터의 정합성을 보증할 수 없다. 이 것을 위해 세마포어(Semaphore) 와 뮤텍스(Mutex)라는 개념이 있다.

Semaphore(세마포어) - 원래뜻은 기차등에서 사용하는 까치발 신호기
공유된 자원데이터를 여러 `프로세스`가 접근하는 것을 막음
자원데이터를 카운팅하여 몇가지의 쓰레드가 접근 할 수 있는지 확인할 수 있음
지정한 수 이상 쓰레드가 접근하면 대기하도록 함.

Mutex(Mutual Exclusion: 상호 배제)
공유된 자원 데이터를 여러 `쓰레드`가 접근하는 것을 막음
즉, 멀티쓰레드중 한가지의 쓰레드만 사용가능한 리소스
1과 0만을 갖고 있는, 즉 한계치가 1인 Binary Semaphore(이진 세마포어)

RxSwift에 있어서 스케쥴러 클래스

ConcurrentDispatchQueueScheduler
큐의 종류와 QoS를 지정할 수 있는 스케쥴러
background에서 작업할때 사용한다.


ConcurrentMainScheduler  
메인스레드를 사용하는 스케쥴러
만약이 스케쥴러가 메인스레드에서 호출되면 별도의 스케쥴링 없이 바로 실행된다.
이 스케쥴러는 SubscribeOn에 최적화되어 있으므로 ObserveOn을 할 경우에는 MainScheduler를 사용할 것.


CurrentThreadScheduler  
현재 스레드에서 실행되는 스케줄러.
기본 오퍼레이터는 이 스케줄러를 사용한다.

VirtualTimeScheduler  
가상의 시간을 만들어내고 그 위에 스케쥴러를 돌린다.

HistoricalScheduler  
절대시간(Date)이나 상대시간(TimeInterval)을 이용하여 VirtualTimeScheduler를 사용하는 클래스


MainScheduler  
DispatchQueue.main에서 작동하는 스케쥴러.
이 스케쥴러가 DispatchQueue.main에서 사용되면 별도의 스케쥴링 없이 바로 실행된다.
UI작업에서 주로 사용된다.
SerialDispatchQueueScheduler가 특화된 클래스.
ObserveOn에 특화된 스케줄러
SubscribeOn을 할 경우에는 ConcurrentMainScheduler을 사용할 것.


OperationQueueScheduler  
특정 OperationQueue에서 작업하는 스케줄러
maxConcurrentOperationCount을 사용하여 대규모 작업을 백그라운드에서 실행할 경우 사용


SerialDispatchQueueScheduler  
특정 큐를 지정할 수 있는 스케줄러.
단, 병렬 큐를 지정해도 직렬로 처리하므로 주의.
사용하기전에 serialQueueConfiguration를 통해 몇가지 설정이 가능함

* thread-safe

쓰레드 자체를 안전하게 사용하는지에 관한 이야기.

iOS에서는 GCD(Grand Central Dispatch)에 의해서 자동적으로 thread-safe가 된다.

thread-safe의 조건

1. Re-entrancy

어떤 함수가 한 스레드에 의해 호출되어 실행 중일 때, 다른 스레드가 그 함수를 호출하더라도 그 결과가 각각에게 올바로 주어져야 한다.

2. Thread-local storage

공유 자원의 사용을 최대한 줄여 각각의 스레드에서만 접근 가능한 저장소들을 사용함으로써 동시 접근을 막는다.

이 방식은 동기화 방법과 관련되어 있고, 또한 공유상태를 피할 수 없을 때 사용하는 방식이다.

3. Mutual exclusion

공유 자원을 꼭 사용해야 할 경우 해당 자원의 접근을 세마포어 등의 락으로 통제한다.

4. Atomic operations

공유 자원에 접근할 때 원자 연산을 이용하거나 '원자적'으로 정의된 접근 방법을 사용함으로써 상호 배제를 구현할 수 있다.

RxSwift의 기본 - 실수편

일주일동안 메모리 누수를 잡았다.
하루 8시간씩 변수 하나하나 바꿔가며 테스트한다는게 여간 에너지를 소모하는 일이다.

결국. 조심해야 할것은 역시나 ARC(Auto Reference Counting)
drive나 subscribe의 경우는 CompositeDisposable()에 들어간다.

Variable, PublishSubject 또한 CompositeDisposable에 넣어줄 수 있다.

그런데. .do(onNext: { _ in } ... ) <- 요 메소드는 뭔가 어렵다...
만약 .do() 메소드 안에 self를 써야한다면 반드시 [weak] self를 사용해야한다.

RxSwift의 기본 - Disposable

Disposable.

RxSwift에서 Disposable을 보면서 이것 참 C 스럽다는 생각이 많이 든다.
C언어를 해본사람들은 알겠지만 C에서 가장 귀찮은게 뭐냐고 물어본다면 포인터와 메모리 관리다.
이 배열의 첫 아이템이 배열의 주소니까 그걸 따라가서 복사한다음에 어쩌구 저쩌구...
그리고 그걸 자연스럽게 해제해주지 않으면 프로그램이 shut down!

사실 RxSwift에서도 동일한 작업이 이뤄진다.
이 시퀀스(Observable)이 저쪽에서 구독(subscribe)되고 또 그게 다른 변수로 복사된다음 합쳐지고 다시 이건 해제(Disposable)되는데 다른건 아직 안되서 결국은 메모리 누수가 발생한다.

제대로 해제해주지 않아 발생하는 에러...  RxSwift가 발전하면 언젠가는 이게 편하게 쓸수 있는 날이 올거라 믿는다.

아무튼, 이러한 이유로 어떻게 메모리 해제를 해야하는가에 대해 정리하고자 한다.
원래는 DisposeBag에 전부 넣어서 관리를 하려 했으나... 메인스케쥴러가 아닌 곳에서 시퀀스가 생성되고 돌아가다보니 뷰 컨트롤러가  Dismiss되어도 메모리에 찌꺼기처럼 계속 남는다.

ARC에서도 상호참조가 문제였는데 여기는 그 문제가 더욱 깊어진다.
좋다고 덮어놓고 쓰다보면 밤샘지옥이 기다리고 있을 것이다.

일단 RsSwift에서 사용하는 Disposable의 종류를 알파벳순으로 정리해보면 다음과 같다.

AnonymousDisposable
 - action base의 Disposable
 - Dispose 된 이후 실행시킬 Action을 지정할 수 있다.

BinaryDisposable
 - 동시에 Dispose될 2개의 Disposable을 지정한다.

BooleanDisposable
 - dispose되었는지에 대한 스테이터스를 확일 할 수 있는 Disposable

CompositeDisposable
 - 여러개의 disposable을 하나로 묶을 수 있는 Disposable

NopDisposable
 - Nop: No operation
 - Dispose를 해도 Dispose하지 않고 아무것도 하지 않는다.

RefCountDisposable
 - ARC(Auto Reference Counting)를 해주는 Disposable

ScheduledDisposable
 - 지정한 스케쥴러에 의해 Dispose처리를 함

SerialDisposable
 - 새로운 Disposable을 넣으면 기존에 있던 Disposable이 Dispose됨. -> 병렬처리

SingleAssignmentDisposable
 - 1개만을 집어 넣을 수 있는 Disposable
 - 2번째를 넣으면 Exception을 발생시킨다

DisposeBag
 - 자기 자신이 메모리에서 해제될때 같이 Dispose되는 Disposable


** Disposable은 프로토콜 타입이다. 따라서 Disposable만을 생성하는 것은 불가능하다.

RxSwift의 기본 - RxCocoa - 작성중

UIKit에 맞춰 편리하게 만들어 놓은 라이브러리.
Driver 라던가 control event 등이 추가되어 있음.

RxSwift의 기본 - RxSwift란 무엇인가 - 작성중

어느 책이나 기본 가이드를 보더라도 가장 처음에 있는 내용이다.

이 글을 보는 사람들은 대부분 RxSwift가 무엇인지 알겠지만 그래도 다시 한번 글로서 정리해서 메타 지식을 명확히 하기 위해 이번 장을 마련했다.



설치 방법은 github에서 확인하는 것이 가장 정확하고 빠르다.
https://github.com/ReactiveX/RxSwift


ReactiveX의 공식 페이지 

RxSwift의 기본 - RxSwift의 3대 기둥(Observable, Operator, Scheduler) - 작성중

RxSwift는 아래와 같은 3개의 기둥이 있다.

- Observable

- Operator

- Scheduler

Observable

- Observable은 시퀀스의 타입이다.
- 스트림과 비슷한 개념이지만 시퀀스라고 이름이 붙어 있다.
- 뜻 그대로 옵저버 패턴에 있어서 옵저버가 가능한 시퀀스를 만들어내는 타입이다.

이 시퀀스에는 크게 3가지의 이벤트가 있다.
 - onNext: 시퀀스에 아이템을 흘려보낸다.
 - onError: 시퀀스에 에러가 발생했으므로 에러를 넘기고 종료한다.
 - onComplete: 시퀀스를 종료한다.

하나의 시퀀스가 종료되면 다시 생성하기 전까지 아이템이 흘러가지 않는다.


Observable(시퀀스)를 선언하고 subscribe(구독)을 하므로 rxSwift를 사용하게 된다. 그리고 마지막에 dispose(처리)하므로 시퀀스가 메모리에서 삭제된다.

간단한 예를들어보면

1 2 3 4

Observable<Int>.just(1)     .subscribe { event in         print(event)     }.dispose()

cs

위와 같은 코드가 있다고 할때 .just(1)이 observable을 생성하는 operator가 된다. 그리고 1을 만드는 시퀀스를 구독하여 print하고 그대로 dispose()하고 있는 간단 심플한 코드이다.

Operator

rxSwift를 쓰는 이유는 하나의 이벤트를 기준으로 여러가지 기능을 하기 위해서다. 따라서 여러가지 이벤트를 하나의 이벤트로 합치거나, 시간을 조절하여 이벤트를 받거나 발생시킬 필요가 있다. 그 때 쓰는 것이 Operator이다.
그리고 이 operator가 함수형 프로그래밍의 꽃이 된다.
예를들면, map(), merge(), zip(), concat() 등이 대표적인 Operator이다.
map, filter, reduce는 일반적인 배열에서의 사용법과 같다.


Scheduler
시퀀스의 이벤트를 어떤 큐에서 어떤 타이밍에 발생시킬가에 대한 문제.

RxSwift의 기본 - Hot/Cold Observable

Hot/ Cold Observable이란 무엇인가?

우선 RxSwift의 깃허브를 번역해 보겠다.


Hot/Cold 의개념은 별도의 타입이 아니다. 시퀀스의 속성중 하나로 생각하는 것이 쉽고 바른 생각이다.

ReactiveX.io에 나온 정의를 적어보도록 하겠다.

Observable은 시퀀스를 언제 발생하기 시작하죠? 이것은 Observable의 속성에 따라 다릅니다. Hot Observable은 만들어지자 마자 시퀀스에 신호를 보내기 시작합니다. 따라서 나중에 구독을 시작하는 Observer라든지 어떠한 Observer라도 중간부터 Observing을 시작할 수 있습니다. 반면에 “Cold” Observable은 Observer가 구독할때까지 기다렸다가 시퀀스에 신호를 보냅니다. 따라서 Cold Observable은 처음부터 전체의 시퀀스를 구독하는 것을 보증할 수 있습니다. 

간단하게 정리를 해보자면

- Hot Observable

생성과 동시에 시퀀스에 신호를 보내는 Observable
Observer가 중간부터 신호를 받을 수 있음.

- Cold Observable

구독과 동시에 시퀀스에 신호를 보내는 Observable
생성하더라도 시퀀스에 신호를 보내지 않음
Observer는 처음부터 모든 신호를 받을 수 있다고 보장함

[One More Step]

한발짝 더 다가가보자면
Hot/Cold에 관련한 operater들을 설명해보도록 하겠다.

publish

Observable을 ConnectableObservable로 변환해 준다. 따라서 Observable의 속성이 Hot이 된다.

refCount

hot속성을 갖고 있는 Observable의 레퍼런스(구독자)를 카운팅 한다.
구독자가 제로(0)가 되면 Observable이 dispose된다.

replay

Observable을 ConnectableObservable로 변환해 준다. 따라서 Observable의 속성이 Hot이 된다.
구독을 시작하면 발생했던 아이템을 처음부터 전부 반환해준다.

share

publish + refCount

RxSwift의 기본 - 서문

리엑티브 익스텐션(Reactive eXtension, 이하 리엑티브) 프로그래밍을 한지 어언 1년이 다되어감에도 불구하고 여전히 개념이 확실히 잡혀있지 않다.

기본적인 observable과 driver, 그리고 map, distinct같은 연산자만 쓰다보니 실력이 늘지를 않는다.

역시 이게 문제였는지 에러가 발생한다.

한 유저가 1초에 30건씩 request를 보내고 있는 것이다.

1초에 30건씩 한 유저니 동시접속 100유저면 3천건 1분에 18만건 10분에 180만건 한시간에 1억건이 넘는 리퀘스트를 처리해야 하니 서버가 뻗어버린다.

다행히도 인프라팀에서 이걸 일찍 발견하여 그 리퀘스트를 날리는 부분을 막는 핫픽스를 업데이트 하고 트래픽도 늘렸기에 별문제는 없었다.

그렇게 한숨돌리고 원인을 찾아보니.

rxswift에서 중요한 개념인 hot/cold의 개념이 역시나 발목을 잡았다.

물론 저걸 몰라도 풀수 있는 문제였다.

핵심적인 문제는 rxSwift에서 dispose를 할때 어플 전반적으로 DisposeBag을 활용했는데 이게 viewmodel이 상호참조가 되어버려 인스턴스가 종료되지 않아 결국 DisposeBag도 dispose되지 않은채 영영 살아있고 Observable.Interval()을 사용하고 있었기에 설정 한 시간만큼 기하급수로 리퀘스트가 늘어나고 있었다.

따라서 이 기회에 hot과 cold의 개념을 포함한 rxSwift의 기본기를 좀더 확실하게 다지고자 포스트를 써 본다.