핸즈 온 머신러닝 :: 12. 다중 머신과 장치를 위한 분산 텐서플로

핸즈 온 머신러닝 (Hands-On Machine Learning with Scikit-Learn & TensorFlow) / 오렐리앙 제론 지음 , 박해선 옮김

을 읽고, 추후 기억을 되살릴 수 있게끔 나만의 방법으로 내용을 리뷰한다. 따라서 리뷰의 내용 별 비중이 주관적일 수 있다.

 

챕터 12. 다중 머신과 장치를 위한 분산 텐서플로

11장에서는 훈련 속도를 높일 수 있는 여러 기술을 알아보았다면, 12장에서는 텐서플로를 사용해 연산을 여러 개의 장치에 분산시켜서 병렬로 실행시키는 법을 배운다. 텐서플로의 분산 컴퓨팅 기능은 주요한 장점이다. 계산 그래프를 여러 장치와 머신에 어떻게 분할 시킬지 완전히 제어할 수 있고 연산을 다양한 방식으로 병렬화하고 동기화할 수 있어서 모든 종류의 병렬화 방식이 사용될 수 있다.

 

12.1 단일 머신의 다중 장치

우선 단일 머신, 여러 GPU의 경우를 알아본다. 네트워크 지연이 있기 때문에 한 대에 여러 GPU를 넣는 것이 여러 대에 더 많은 GPU를 나누어 넣는 것 보다 빠를 수 있다.

 

12.1.1 설치

nvidia compute capability를 가진 GPU를 사용해야 한다. 그리고 CUDA (Compute Unified Device Architecture) 라이브러리와 cuDNN (CUDA Deep Neural Network) 를 설치해야 한다. nvidia-smi 명령으로 CUDA 설치를 확인할 수 있다. 마지막으로 virtualenv에 tensorflow-gpu 를 설치해야 한다.

 

12.1.2 GPU RAM 관리

기본적으로 계산 그래프가 실행될 때 GPU의 모든 RAM을 확보하므로, 두 개 이상의 텐서플로 프로그램을 시작할 수 없다. 따라서 CUDA_VISIBLE_DEVICES 환경 변수를 설정해서 특정 GPU 카드만 쓰거나, ConfigProto 객체의 gpu_options.per_process_gpu_memory_fraction 옵션을 설정하여 GPU의 일부만 사용해야 한다. 텐서플로가 필요할 때만 메모리를 점유하도록 config.gpu_options.allow_growth 를 True 설정하여도 되지만, 메모리를 반납하지 않으며 예측이 안되기 때문에 권장되지 않는다.

 

12.1.3 장치에 연산 배치하기

모든 장치에 연산을 완전히 자동으로 분산하는 dynamic placer 알고리즘이 소개되어있지만 공개되어있지는 않다. 따라서 텐서플로는 매우 기본적인 simple placer에 의존한다. 텐서플로가 배치되지 않는 노드를 평가해야한다면 특정 규칙에 따라 단순 배치 한다.

 

- 이미 배치되어 있는 노드는 그 장치에 그대로 둔다.

- 사용자가 노드를 장치에 할당했다면 그 장치에 배치한다.

- 그 외에는 GPU #0, 혹은 CPU에 배치한다.

 

사용자는 with tf.device() 를 이용하여 직접 배치 할 수 있다. 배치 로그는 ConfigProto의 log_device_placement 옵션을 True로 설정하면 된다. tf.device를 이용할 때에 정적으로 이름을 적을 수도 있지만, 장치를 반환하는 동적 함수를 적을 수도 있다.

 

텐서플로 연산을 장치에서 실행하려면 그 장치에 맞는 구현, 즉 커널이 있어야 하는데 예를 들어 정수 변수에 대한 GPU 커널은 없으므로 이 경우 오류를 출력한다. 오류를 출력하는 대신 CPU를 사용하도록 하려면 allow_soft_placement 환경 설정을 True로 지정하면 된다.

 

12.1.4 병렬 실행

특정 노드가 의존하고 있는 노드 수에 대한 카운터를 작성한 후, 노드를 평가할 때마다 그 노드에 의존하고 있는 노드들의 카운터를 감소시킨다. 이후 카운터가 0이 된 노드를 평가 Queue에 추가한다. 모든 노드가 평가되면 출력을 반환한다.

 

CPU의 평가 큐에 있는 연산은 inter-op thread pool로보낸다. 이 중 멀티스레드 CPU kernel이 존재하는 연산은 intra-op theard pool로 보내지며 연산들이 병렬로 평가된다.

 

GPU는 각 연산이 대부분의 thread를 사용하므로 inter-op thread pool이 필요없고, CUDA나 cuDNN같은 라이브러리로 구현된 멀티스레드 GPU 커널을 통해 부분적으로 병렬 평가된다. 

 

12.1.5 제어 의존성

카운터가 0이 된 노드가 있더라도, 후반에 필요한 연산이라면 메모리나 통신 대역폭 등의 문제로 평가를 지연시킬 수 있다. 이를 제어 의존성을 추가한다고 한다. tf의 control_dependencies()를 사용하면 된다.

 

https://www.tensorflow.org/api_docs/python/tf/control_dependencies

tf.control_dependencies  |  TensorFlow Core r2.1

 

12.2 다중 머신의 다중 장치

여러 대의 머신에서 그래프를 실행하려면 cluster를 정의해야한다. cluster는 task라고 하는 하나 이상의 tensorflow 서버로 구성되며, 각 task는 하나의 job에 속해있다. 하나의 job은 task의 그룹이며 일반적으로 'ps' (parameter server) 라는 이름의 모델 파라미터를 저장하거나 일반적으로 'worker'라는 이름의 계산을 수행하는 공통된 역할을 가진다.

 

tensorflow.train.ClusterSpec을 통해 해당 cluster의 job들과 task들을 정의할 수 있다. 텐서플로 서버를 시작하기 위해서는 Server 객체를 만들고 ClusterSpec, job_name, task_index를 전달하면 된다. server.join()을 사용하면 서버가 멈출 때 까지 (영원히) 블록된다.

 

12.2.1 세션 열기

모든 태스크가 시작되면 어떤 머신의 프로세스에 있는 클라이언트에서도 다른 모든 서버에 대해 보통의 로컬 세션처럼 세션을 열 수 있다. 클러스터를 정의하고 서버를 열고 나면 나중에 클라이언트에서 클러스터의 서버들에 접근하여 세션을 실행한다.

 

12.2.2 마스터와 워커 서비스

클라이언트는 gPRC (Google Remote Procedure Call)을 사용하여 서버와 통신한다. 데이터는 구글의 또 다른 오픈소스 기술인 protocol buffer 형태로 전달된다.

 

모든 텐서플로 서버는 master service와 worker serivce를 제공한다. master service는 클라이언트가 세션을 열고 그래프를 실행할 수 있게 도와주고, worker service를 통해 실제로 로컬 장치에서 계산을 실행하고 결과를 받는다. 이 구조는 많은 유연성을 제공하여, 한 클라이언트가 각기 다른 스레드에서 여러 세션을 열어 여러 대의 서버에 연결할 수 있다. 한 서버는 한 개 이상의 클라이언트에서 오는 여러 개의 세션을 동시에 처리할 수 있다. 태스크마다 하나의 클라이언트를 실행하거나 하나의 클라이언트로 모든 태스크를 제어하는 등 모든 경우가 가능하다.

 

12.2.3 여러 태스크에 연산 할당하기

tf.device를 사용하여 연산을 특정 태스크에서 관리하는 장치에 할당할 수 있다.

("/job:ps/task:0/cpu:0" 등)

장치 유형이나 번호를 빼면 태스크의 기본 장치를 사용한다.

 

12.2.4 여러 대의 파라미터 서버에 변수를 나누어 분산하기

파라미터 서버 한 대의 네트워크 카드가 포화되는 것을 피하기 위해 파라미터들을 여러 대의 파라미터 서버에 나누는 것이 좋은데, replica_device_setter() 함수를 사용하면 자동으로 round robin 방식으로 할당하게 된다. 또한 with tf.device 구문은 중첩하여 사용할 수 있다.

 

12.2.5 리소스 컨테이너를 사용해 여러 세션에서 상태 공유하기

평범한 로컬 세션에서는 세션이 종료되면 변수가 사라지지만, 분산 세션에서는 변수의 상태가 세션이 아니라 클러스터의 resource container에 의해 관리되므로, 클러스터에 있는 다른 세션에서도 자동으로 사용할 수 있다. 클러스터의 다른 세션에서 변수 명이 겹칠 수 있으므로 variable_scope, 혹은 container를 사용하여 변수들을 관리할 수 있다. 

 

12.2.6 텐서플로 큐를 사용한 비동기 통신

여러 세션 사이에 데이터를 교환하기 좋은 다른 방법은 Queue를 이용하는 것이다. 한 클라이언트에서는 데이터를 로드하여 큐에 저장하는 그래프를 만들고, 다른 클라이언트는 큐에서 데이터를 추출하여 모델을 훈련시키는 그래프를 만든다면, 매 스텝마다 데이터를 기다리지 않아도 되기 때문에 훈련 속도가 빨라진다. 기본적으로 FIFOQueue class가 존재한다.

 

enqueue 혹은 enqueue_many 연산을 사용해 데이터를 넣고, dequeue 혹은 dequeue_many 연산을 사용해 데이터를 추출한다.

 

큐에는 텐서 하나 대신 텐서의 튜플이 아이템으로 들어갈 수 있다. (dtypes = [tf.int32, tf.float32] 등). enqueue와 dequeue는 마찬가지로 사용할 수 있다.

 

더 이상 아이템을 enqueue 하지 않는다면 queue.close() 텐서를 만들어 평가해주어야 한다. 이 때 dequeue_many의 배치 사이즈보다 적게 남아도 아이템을 추출해주는 dequeue_up_to 함수도 있다.

 

그 외에 랜덤으로 아이템을 반환하는 RandomShuffleQueue, shapes가 지정되지 않아도 dequeue_many로 아이템을 꺼낼 때 가장 큰 사이즈에 맞추어 모든 데이터가 padding되는 PaddingFifoQueue가 있다.

 

12.2.7 그래프에서 직접 데이터 로드하기

클라이언트가 훈련 데이터를 로드하고 placeholder를 사용해 클러스터에 데이터를 주입하면 간단하지만 데이터를 여러 번 전송하기 때문에 비효율 적이다.

 

데이터셋이 메모리 크기에 맞는다면 훈련 데이터를 한번에 로드해서 변수에 할당하고 그래프에서 바로 사용하는 것이 더 좋은 방법인데, 이를 preloading 한다고 부른다. 반드시 trainable = False로 지정하고, collections=[]로 지정하여 체크포인트 저장이나 복원에 사용되는 GraphKeys.GLOBAL_VARIABLES에 추가되지 않도록 해야 한다.

 

데이터셋이 메모리 크기에 맞지 않으면 reader operation을 사용하는 것이 좋다. 이 연산을 파일시스템에서 직접 데이터를 읽을 수 있다. 이 때 파일 명을 전달하는 Queue를 만들어서 플레이스홀더를 통해 파일 이름을 큐에 넣고 더 이상 읽을 파일이 없을 경우를 위해 종료 연산을 만든다. 마지막으로 랜덤 큐 등을 사용하여 훈련 샘플 Queue에 데이터를 넣는다. 훈련 시에는 샘플 Queue에서 미니배치만큼 꺼내서 사용하면 된다. 데이터를 읽어서 샘플 Queue에 넣는 과정을 여러 스레드에서 수행하면 더 높은 처리량을 낼 수 있다. 이를 위해서는 파이썬 스레드를 만들고 관리해야 하는데, Coordinator와 QueueRunner Class를 사용한다.

 

Coordinator Class는 스레드들의 중지를 조정하는 것이고, QueueRunner Class는 특정 Queue에 enqueue하는 연산을 멀티스레드에서 실행해 주는 것이다. enqueue하는 연산을 filename에 대해 실행하게 만들어, 여러 파일에 대해 스레드가 각각 하나의 파일을 열어 sample queue를 채우게 할 수 있다.

 

훈련 샘플을 읽을 때 공통된 작업들을 간단하게 처리해주는 몇 가지 편리한 함수가 있다.

 

string_input_producer()는 파일 이름 리스트가 담긴 1D 텐서를 받아서 스레드를 만들고 파일 이름 큐에 한 번에 하나씩 파일 이름을 넣은 후 큐를 닫는다. 이 함수는 스레드를 관리하기 위해 QueueRunner를 만들고 GraphKeys.QUEUE_RUNNERS 컬렉션에 추가하는데 tf.train.start_queue_runner() 함수를 호출하지 않으면 파일 이름 큐가 열린 채로 비어 있을 것이다.

 

https://www.tensorflow.org/api_docs/python/tf/compat/v1/train/string_input_producer

tf.compat.v1.train.string_input_producer  |  TensorFlow Core r2.1

 

enqueue 연산을 실행하기 위해 큐와 이에 상응하는 QueueRunner를 만들어주는 producer 함수가 있다. (input_producer(), range_input_producer(), slice_input_producer() 등)

 

shuffle_batch() 함수는 텐서 리스트를 받아 RandomShuffleQueue를 만들고 텐서를 넣기 위한 QueueRunner와 큐에서 미니배치를 꺼내기 위한 dequeue_many 연산을 만든다. 비슷한 기능의 batch(), batch_join(), shuffle_batch_join() 함수도 있다.

 

https://www.tensorflow.org/api_docs/python/tf/compat/v1/train/shuffle_batch

tf.compat.v1.train.shuffle_batch  |  TensorFlow Core r2.1

 

위 함수들은 v1에서 지원이 종료된 듯 하지만, 내용을 알면 document에 언급된 바뀐 함수들을 사용하는데에 도움이 될 것이다.

 

12.3 텐서플로 클러스터에서 신경망 병렬화하기

12.3.1 장치마다 하나의 신경망

여러 클라이언트 세션을 병렬로 실행해서 각기 다른 서버에 적용해 각 장치마다 하나의 신경망을 훈련하고 실행시키면 많은 신경망을 훈련시키고 실행할 수 있다. 이 방식은 하이퍼파라미터 튜닝을 할 때에 완벽하다. 또한 대량의 쿼리를 받아 예측을 수행할 때에도 좋다.

 

12.3.2 그래프 내 복제와 그래프 간 복제

여러 신경망을 다른 장치에 배치하여 대규모 앙상블 훈련을 진행할 때에는 개개의 예측을 모으는 약간의 조율이 필요하다. 이 과정을 진행하는 방법은 크게 두 가지가 있다.

 

in-graph replication : 각각 다른 장치에 할당된 모든 신경망을 담은 하나의 큰 그래프를 만들어서, 한 서버에 세션을 만들고 모든 계산을 위임한다.

 

between-graph replication : 각 신경망을 독립된 그래프로 만들어서 그래프 사이의 동기화를 직접 관리하고, 큐를 사용하여 그래프를 조율한다. 입력 큐에서 데이터를 읽어 예측을 출력 큐에 넣는 많은 신경망 클라이언트들과 입력 클라이언트, 출력 및 앙상블 클라이언트가 존재한다.

 

그래프 내 복제는 구현하기 간단하고, 그래프 간 복제는 목표 성능을 맞추고 테스트하기 쉬운 모듈을 구성하기 편리하며 유연한 모델을 만들 수 있다. 예를 들어 각 신경망에 RunOptions.timeout_in_ms 나 세션의 operation_timeout_in_ms 등으로 타임아웃을 조절할 수 있다.

 

12.3.3 모델 병렬화

하나의 장치에서 하나의 신경망을 실행하지 않고 하나의 신경망을 여러 개의 장치에서 실행하는 것을 model parallelism 이라고 한다. 모델 병렬화는 매우 어렵고 신경망 모델에 의존적이여서, 완전 연결 신경망의 경우에는 이득을 보기 어렵다. 모델의 각 층을 기준으로 병렬화한다면 이전 층의 출력을 기다려야 하기 때문에 이득이 없고, 그렇다고 수직으로 분할하자면 양쪽의 출력을 모두 필요로 하기 때문에 통신에 의해 손해가 더 클것이다.

 

하지만, 합성곱 신경망은 이전 층에 부분적으로만 연결된 층을 가지며 deep RNN은 메모리 셀의 층으로 이루어져 있어 (시간 t 셀 출력이 시간 t+1 셀 입력으로 재사용) 각 층을 다른 장치에 배치한다면 장점이 크다.

 

12.3.4 데이터 병렬화

하나의 신경망을 병렬화 하는 또 다른 방법은 미니배치 별로 병렬화 하는 data parallelism 이다. 이 때 그래디언트를 취합하여 모델 파라미터를 업데이트 할 필요성이 있다. 이 방식에는 synchronous update와 asynchronous update가 있다.

 

synchronous update는 모든 그래디언트가 계산될 때 까지 기다려서 평균을 계산하고 결과를 반영한다. 모델들은 결과가 반영될 때 까지 대기하였다가 파라미터를 반환받고 다음 미니배치에 대해 계산한다. 계산이 느린 장치가 있으면 그 장치를 기다려야하고, 파라미터를 반환받는 과정에서 대역폭이 포화된다는 단점이 잇다. 매 스텝에서 대기 시간을 줄이기 위해 가장 느린 모델 (보통 10%)를 무시하는 방법이 있다. 이 때 느린 모델을 sparse replica라고 표현한다.

 

asynchronous update는 모델이 그래디언트 계산을 끝낼 때 마다 즉시 파라미터를 업데이트하여 다음 미니배치를 적용한다. 모든 모델들은 독립적으로 작동한다. 시간당 더 많은 훈련을 실행할 수 있으며 단순하고 대역폭을 효율적으로 사용한다. 하지만 그래디언트 계산을 마칠 때 파라미터는 이미 평균적으로 N-1번의 업데이트를 진행하여 부정확할 수 있다는 단점이 있다. 오래된 그래디언트를 stale gradient라고 부른다. 이를 해결하기 위해서는 학습률을 감소시키거나, 낡은 그래디언트를 버리거나, 미니배치크기를 조절하거나, 처음 몇 번의 에포크에서는 하나의 모델만 사용하는 방법이 있다.

 

하지만 두 방법 모두 파라미터 및 그래디언트 잔송 시 대역폭이 사용됨은 어쩔 수 없는데, 따라서 규모가 작고 훈련 데이터가 많은 경우 단일 GPU 사용이 더 빠를 수 있다. 하지만 모델이 크고 희소한 모델의 경우 데이터 병렬화의 이득을 크게 볼 수 있다. 대역폭 포화를 감소시키기 위해서는 몇 대의 서버에 GPU를 모으거나, 여러 대의 파라미터 서버에 파라미터를 분산시키거나, 실수 정밀도를 32비트에서 16비트로 줄이는 등의 해결법을 취할 수 있다.

 

텐서플로에서 구현하려면 in-graph 혹은 between-graph를 사용할지, synchronous update 혹은 asynchronous update를 사용할지 총 2*2 = 4 가지 경우의 수가 있다.

 

(in-graph, synchronous update) : 모든 복제 모델을 담는 하나의 큰 그래프를 만들고 그래디언트를 수집할 노드를 만들어 optimizer에 전달하는 연산을 반복적으로 실행하면 된다.

 

(in-graph, asynchronous update) : 하나의 큰 그래프를 만들지만, 하나의 optimizer를 두고 개별적인 스레드에서 optimizer를 실행시킨다.

 

(between-graph, asynchronous update) : 독립적인 클라이언트를 여러 개 실행하고 독립된 모델인 양 optimizer를 실행한다. 하지만 파라미터는 실제로 리소스 컨테이너를 사용해 공유된다.

 

(between-graph, synchronous update) : 여러 개의 클라이언트를 실행해서 공유된 파라미터를 기반으로 모델을 실행한다. SyncReplicasOptimizer로 옵티마이저를 감싼 후, 이 옵티마이저를 사용한다. 옵티마이저 내부에서는 옵티마이저가 Queue에 그래디언트를 보내고 Chief라는 모델의 SyncReplicasOptimizer 중 하나가 읽어서 취합하고 복제 모델들을 위해 token queue에 토큰을 작성하여 다음번 작업을 진행해도 좋다고 신호를 보낸다. 이 과정에서 sparse replica를 지원한다.