ForkJoinPool의 작업 도둑질(Work Stealing) 알고리즘: 스레드 풀 부하 균형 조정 방법
ForkJoinPool에서 중요한 작업 도둑질 알고리즘에 대해 자세히 알아보겠습니다. ForkJoinPool은 Java의 java.util.concurrent 패키지에서 제공하는 분할 정복(Divide and Conquer) 방식을 사용하여 작업을 처리하는 스레드 풀입니다. 이 알고리즘 덕분에 다중 스레드 환경에서 작업 부하를 효과적으로 분산시키고 CPU 리소스를 최대한 활용할 수 있습니다.
1. ForkJoinPool 기본 구조
작업 도둑질에 들어가기 전, 먼저 ForkJoinPool의 기본 구조를 간단히 살펴보겠습니다.
- ForkJoinPool: 전체 스레드 풀을 관리하며 Worker 스레드들을 통제합니다.
- ForkJoinWorkerThread: 실제로 작업을 수행하는 스레드로 각 스레드는 자신의 Deque를 유지합니다.
- ForkJoinTask: ForkJoinPool에서 실행될 수 있는 작업을 나타냅니다.
- Deque (이중 연결 큐): 각 Worker 스레드는 대기 중인 ForkJoinTask를 저장하기 위한 Deque를 유지합니다.
- 작업 도둑질 큐(Work-Stealing Queue): 위에서 언급된 Deque와 동일하며 각 스레드는 자신만의 큐를 소유합니다.
2. 분할 정복 모델과 ForkJoinTask
ForkJoinPool은 작업을 더 작은 독립적인 하위 작업으로 나눌 수 있는 문제에 적합합니다. ForkJoinTask는 ForkJoinPool에서 실행될 모든 작업의 기반이며 일반적으로 반환값이 필요한 경우 RecursiveTask, 그렇지 않은 경우 RecursiveAction를 상속받아 구현합니다.
다음은 배열의 합계를 계산하는 예제입니다:
import java.util.concurrent.RecursiveTask;
class ArraySum extends RecursiveTask<Long> {
private static final int THRESHOLD = 1000;
private final long[] data;
private final int start;
private final int end;
public ArraySum(long[] data, int start, int end) {
this.data = data;
this.start = start;
this.end = end;
}
@Override
protected Long compute() {
int size = end - start;
if (size <= THRESHOLD) {
long total = 0;
for (int i = start; i < end; i++) {
total += data[i];
}
return total;
} else {
int middle = (start + end) / 2;
ArraySum leftTask = new ArraySum(data, start, middle);
ArraySum rightTask = new ArraySum(data, middle, end);
leftTask.fork();
rightTask.fork();
return leftTask.join() + rightTask.join();
}
}
}
위 코드에서는 ArraySum 클래스가 데이터를 작은 단위로 나누어 작업을 처리하고, THRESHOLD보다 작으면 실제 합계를 계산합니다. fork() 메서드는 하위 작업을 해당 스레드의 Deque에 추가하고 비동기적으로 실행하도록 합니다. join() 메서드는 하위 작업의 완료를 기다리며 결과를 반환합니다.
3. 작업 도둑질 원리
작업 도둑질의 주요 아이디어는 한 스레드가 자신의 큐의 모든 작업을 완료했을 때 다른 스레드의 큐에서 작업을 "도둑질"하여 실행하는 것입니다. 이를 통해 일부 스레드가 과부하 상태이고 다른 스레드가 유휴 상태인 것을 방지하여 병렬 처리 효율성을 높입니다.
주요 단계는 다음과 같습니다:
- 작업 제출: 초기 작업을 ForkJoinPool에 제출합니다. 일반적으로 특정 Worker 스레드의 Deque에 배치됩니다.
- 작업 분할(Fork): 작업이 더 작은 하위 작업으로 나눌 수 있다면
fork()메서드를 사용해 자신의 Deque에 추가합니다. - 작업 실행: Worker 스레드는 자신의 Deque에서 가장 최근에 추가된 작업(LIFO 방식)을 가져와 실행합니다.
- 작업 도둑질(Steal): 자신의 Deque가 비어 있으면 다른 Worker 스레드의 Deque에서 가장 오래된 작업(FIFO 방식)을 가져옵니다.
- 결과 결합(Join): 하위 작업의 완료를 기다리고 그 결과를 결합하기 위해
join()메서드를 호출합니다.
4. 작업 도둑질의 장점
- 부하 균형: 작업 도둑질은 스레드 풀 내에서 작업 부하를 동적으로 균형 있게 분배합니다.
- CPU 활용률 증가: 유휴 스레드가 작업을 도둑질함으로써 CPU 리소스를 최대한 활용합니다.
- 경쟁 감소: 각 스레드가 고유한 Deque를 가지므로 공유 자료구조에 대한 경쟁을 줄여 병렬 처리 성능을 향상시킵니다.
- 적응성: 다양한 작업 크기와 컴퓨팅 환경에 유연하게 적응할 수 있습니다.
5. 작업 도둑질 알고리즘 세부사항
- 타겟 선택: 작업을 도둑질할 때 어떤 스레드를 선택해야 할까요? ForkJoinPool은 의사 난수 생성기를 사용해 타겟 스레드를 선택합니다.
- 도둑질 동작: 여러 스레드가 동시에 같은 작업을 도둑질하지 않도록 CAS(비교 및 교환) 연산을 사용해 원자성을 보장합니다.
- 유휴 스레드 처리: 도둑질할 작업이 없는 경우 스레드는 잠들어 있으며 일정 시간마다 새로운 작업 여부를 확인합니다.
6. 작업 도둑질의 Java 코드 구현 (간략화된 버전)
작업 도둑질의 원리를 더 명확히 이해하기 위해 간단화된 Java 코드를 살펴보겠습니다:
import java.util.ArrayDeque;
import java.util.Deque;
import java.util.Random;
import java.util.concurrent.atomic.AtomicBoolean;
class TaskWorker implements Runnable {
private final Deque<Runnable> queue = new ArrayDeque<>();
private final TaskWorker[] pool;
private final Random random = new Random();
private final AtomicBoolean running = new AtomicBoolean(true);
public TaskWorker(TaskWorker[] pool) {
this.pool = pool;
}
public void enqueue(Runnable task) {
synchronized (queue) {
queue.addFirst(task);
queue.notify();
}
}
@Override
public void run() {
while (running.get()) {
Runnable task = null;
synchronized (queue) {
while (queue.isEmpty()) {
try {
queue.wait();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
return;
}
}
task = queue.pollFirst();
}
if (task != null) {
try {
task.run();
} catch (Exception e) {
e.printStackTrace();
}
} else {
stealTask();
}
}
}
private void stealTask() {
int targetIndex = random.nextInt(pool.length);
TaskWorker target = pool[targetIndex];
if (target != this) {
Runnable stolenTask = null;
synchronized (target.queue) {
if (!target.queue.isEmpty()) {
stolenTask = target.queue.pollLast();
}
}
if (stolenTask != null) {
try {
stolenTask.run();
} catch (Exception e) {
e.printStackTrace();
}
} else {
try {
Thread.sleep(1);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
return;
}
}
}
}
public void stop() {
running.set(false);
synchronized (queue) {
queue.notifyAll();
}
}
}
class BasicThreadPool {
private final TaskWorker[] workers;
private final Thread[] threads;
public BasicThreadPool(int poolSize) {
workers = new TaskWorker[poolSize];
threads = new Thread[poolSize];
for (int i = 0; i < poolSize; i++) {
workers[i] = new TaskWorker(workers);
threads[i] = new Thread(workers[i]);
threads[i].start();
}
}
public void submit(Runnable task) {
int workerIndex = new Random().nextInt(workers.length);
workers[workerIndex].enqueue(task);
}
public void shutdown() {
for (TaskWorker worker : workers) {
worker.stop();
}
for (Thread thread : threads) {
try {
thread.join();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}
}
public class SimplifiedWorkStealingDemo {
public static void main(String[] args) throws InterruptedException {
int poolSize = 4;
BasicThreadPool pool = new BasicThreadPool(poolSize);
for (int i = 0; i < 20; i++) {
final int taskId = i;
pool.submit(() -> {
System.out.println("Task " + taskId + " executed by " + Thread.currentThread().getName());
try {
Thread.sleep(100);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
});
}
Thread.sleep(1000);
pool.shutdown();
}
}
이 간단한 예제는 Worker 스레드가 자신의 Deque에서 작업을 가져와 실행하고, Deque가 비어 있으면 다른 Worker 스레드의 Deque에서 작업을 도둑질하는 기본 프로세스를 보여줍니다.
7. ForkJoinPool 구성
ForkJoinPool의 성능은 스레드 풀 크기, 작업 단위, 하드웨어 구성 등 여러 요인에 영향을 받습니다. 적절한 설정을 통해 성능을 최적화할 수 있습니다.
8. 작업 도둑질의 비용
작업 도둑질은 병렬 처리 효율성을 높이지만 일부 비용이 발생할 수 있습니다:
- 스레드 간 통신: 작업 도둑질은 스레드 간 통신을 필요로 하며 이는 특정 비용을 초래합니다.
- 경쟁: 여러 스레드가 동시에 동일한 작업을 도둑질하려 할 수 있습니다.
- 잘못된 공유(False Sharing): 여러 스레드가 동일한 캐시 라인의 다른 변수에 접근하면 잘못된 공유가 발생하여 성능이 저하될 수 있습니다.