[NODE] 📚 Worker_Threads 모듈 (멀티 쓰레드 구현)

Worker Thread 이해하기

Node.js가 시작되면, 다음이 실행됩니다.

• 하나의 프로세스 : 어디서든 접근 가능한 전역 객체이자 그 순간 실행되고 있는 것들의 정보를 가지고 있는 프로세스
• 하나의 스레드 : 단일 스레드는 주어진 프로세스에서 오직 한 번에 하나의 명령만이 실행된다는 뜻.
• 하나의 이벤트 루프 : 노드를 이해하기 위해 가장 중요한 부분 중 하나입니다.
  이는 자바스크립트가 단일 스레드라는 사실에도 불구하고, 언제든 가능한 callback, promise, async/await 를 통해 시스템 커널에 작업을 offload 하게 합니다. 이로서 노드가 비동기식, 비차단 I/O 의 특성을 가집니다.
• 하나의 js 엔진 인스턴스 :  js 코드를 실행하는 컴퓨터 프로그램입니다.
• 하나의 노드js 인스턴스 : 노드js 코드를 실행하는 컴퓨터 프로그램입니다.

 

즉, 노드는 단일 스레드에서 실행되고, 이벤트 루프에는 한 번에 하나의 프로세스만 발생합니다.

하나의 코드, 하나의 실행, (코드는 병렬로 실행되지 않습니다).

 

이 점은 매우 유용한데, 왜냐하면 당신이 동시성 문제에 대한 걱정 없이 자바스크립트를 사용하는 방법을 단순하게 만들어주기 때문입니다.

 

그러나 모든 것과 마찬가지로 단점도 있습니다.

만약 인 메모리에서 발생하는 대규모 데이터 세트의 복잡한 계산같은 CPU 자원을 많이 사용하는 코드가 있으면, 이 코드가 다른 프로세스가 실행되는걸 차단할 수도 있습니다.

마찬가지로, CPU 자원을 많이 사용하는 코드가 있는 서버에 요청하는 경우, 이 코드가 이벤트 루프를 차단하고 다른 요청들이 처리되지 않게 할 수도 있습니다.

 

CPU 작업과 I/O 작업을 구분하는 게 중요합니다.

앞에서 말했듯이, Node.js 코드는 병렬로 실행되지 않습니다. 

오직 I/O 작업만 비동기식으로 실행되므로, 병렬로 실행됩니다.

 

그래서 워커 스레드는 I/O 집약적인 일에는 별로 효과적이지 못한데, 왜냐하면 비동기적 I/O 작업이 워커가 하는 것보다 더 효율적이기 때문입니다.

워커의 가장 중요한 목표는 I/O 작업이 아닌 CPU 집약적인 작업의 퍼포먼스를 향상시키는 겁니다.

---

워커 솔루션

CPU 퍼포먼스를 위한 최고의 솔루션은 워커 스레드입니다.

 

기존 브라우저 특징

• 하나의 프로세스
• 하나의 스레드
• 하나의 이벤트 루프
• 하나의 js 엔진 인스턴스
• 하나의 node.js 인스턴스

대신, 워커 스레드는 아래와 같은 특징을 가집니다.

• 하나의 프로세스
• 여러개의 스레드
• 스레드 별 하나의 이벤트 루프
• 스레드 별 하나의 js 엔진 인스턴스
• 스레드 별 하나의 node.js 인스턴스

아래와 같은 이미지로 표현할 수 있습니다.

---

Worker Thread 사용

• const { worker, parentPort } = require('worker_threads') 
  -> worker 클래스는 독립적인 자바스크립트 실행 스레드를 의미하고, parentPort는 메세지 포트의 인스턴스입니다.
• new Worker(filename) 이나 new Worker(code, {eval: true}) 
  -> 워커를 시작하는 두 가지 메인 방법입니다(파일명을 넘기거나, 실행하고자 하는 코드를 작성하거나). 실제 제작시 파일명을 사용하는 편이 권장됩니다.
• worker.on('message'), worker.postMessage(data)
  -> 다른 스레드간 메세지를 주고받을 때 사용합니다.
• parentPort.on('message'), parentPort.postMessage(data)
  -> parentPort.postMessage(data) 를 통해 보내진 메세지는 worker.on('message') 를 사용한 부모 스레드에서 사용 가능합니다. 그리고 worker.postMessage(data) 를 사용한 부모 스레드로부터 보내진 메세지는 parentPort.on('message') 를 사용한 스레드에서 사용 가능합니다.

워크 쓰레드 통신 로직은 iframe통신 로직과 유사합니다.
iframe통신을 공부하셨다면 수월하게 익히실수 있습니다.

 

메인 스레드 와 Worker 스레드

• isMainThread : 현재 코드가 메인 스레드에서 실행되는지, 워커 스레드에서 실행되는지 구분
• 메인 스레드에서는 new Worker를 통해 현재 파일(__filename)을 워커 스레드에서 실행시킴

const { Worker, isMainThread, parentPort } = require('worker_threads');

if (isMainThread) { // 메인 스레드
	const worker = new Worker(__filename); // 같은 dir폴더에 워커를 생성
    
} else { // 워커스레드
	// 위에서 생성한 worker는 여기서 동작
}

 

메인 스레드 <-> Worker 데이터 송수신

• worker.postMessage로 부모에서 워커로 데이터를 보냄
• parentPort.on('message')로 부모로부터 데이터를 받고, postMessage로 데이터를 보냄

const { Worker, isMainThread, parentPort } = require('worker_threads');

if (isMainThread) { // 메인 스레드
   const worker = new Worker(__filename);

   worker.on('message', (value) => {
      console.log('워커로부터', value)
   })
   worker.on('exit', (value) => { // parentPort.close()가 일어나면 이벤트 발생
      console.log('워커 끝~');
   })

   worker.postMessage('ping'); // 워커스레드에게 메세지를 보낸다.

} else { // 워커스레드

   parentPort.on('message', (value) => {
      console.log("부모로부터", value);
      parentPort.postMessage('pong');
      parentPort.close(); // 워커스레드 종료라고 메인스레드에 알려줘야 exit이벤트 발생
   })
}

 

Worker를 모듈로 관리

new Worker(경로)를 통해 모듈 적으로 워커들을 관리할 수도 있습니다.

// main.js
const { Worker } = require('worker_threads')

const worker = new Worker('./worker.js')

worker.on('message', (value) => {
   console.log('워커로부터', value)
})
worker.on('exit', (value) => {
   console.log('워커 끝~');
})
worker.postMessage('ping'); // 워커스레드에게 메세지를 보낸다.

// worker.js
const { parentPort } = require('worker_threads')

parentPort.on('message', (value) => {
   console.log("부모로부터", value);
   parentPort.postMessage('pong');
   parentPort.close(); // 워커스레드 종료라고 메인스레드에 알려줘야 exit이벤트 발생
})

 

MessageChannel 을 사용한 Worker 간 데이터 송수신

위의 예시에선 메인 스레드가 메시지를 중계합니다.

이 방법으로 Worker 간 데이터 송수신을 구현했습니다.

 

Worker 간에서 데이터 송수신하는 다른 방법으론 MessageChannel 활용이 있습니다.

메인 측에서 MessageChannel 을 만들 때 MessagePort 2개가 생성됩니다. 각각을 각 Worker 에 전달하도록 합니다.

// main.js
const {Worker, MessageChannel} = require('worker_threads')

const {port1, port2} = new MessageChannel()

const worker1 = new Worker('./worker1.js')
const worker2 = new Worker('./worker2.js')

worker1.postMessage({worker2: port1}, [port1])
worker2.postMessage({worker1: port2}, [port2])

 

MessagePort 를 받은 Worker 는 이후 MessagePort 를 통해 Worker 간 데이터 송수신을 할 수 있게 됩니다.

// worker1.js
const assert = require('assert')
const {parentPort, MessagePort} = require('worker_threads')

parentPort.once('message', ({worker2}) => {
  assert(worker2 instanceof MessagePort)
  worker2.postMessage('message from worker1')
})

// worker2.js
const assert = require('assert'); // 값을 비교하여 프로그램이 제대로 동작하는지 테스트 하는데 사용하는 모듈
const {parentPort, MessagePort} = require('worker_threads')

parentPort.once('message', ({worker1}) => {
  assert(worker1 instanceof MessagePort)
  worker1.on('message', message => {
    console.log('worker2 received message: %o', message)
  })
})

 

여러개의 Worker 스레드

• WorkerData: startup 데이터를 전달하기 위해 사용됩니다. 워커 생성자에게 전달된 데이터의 복제본이 포함된 임의의 js 값입니다. postMessage()를 사용하는 것처럼 데이터가 복제됩니다.

const { Worker, isMainThread, parentPort , workerData} = require('worker_threads');

if (isMainThread) { // 메인 스레드

   const threads = new Set(); // 자료형 set

   threads.add(new Worker(__filename, { // Worker객체를 생성하고 workerData객체를 주어 초기 데이터를 전송
      workerData: { start: 1 }, // 초기 데이터
   }));
   threads.add(new Worker(__filename, {
      workerData: { start: 2 },
   }));
   
   // Woker가 들어있는 set을 순회하여 각 Worker에 이벤트를 등록
   for(let worker of threads) {
      worker.on('message', (value) => {
         console.log('워커로부터', value)
      })

      worker.on('exit', (value) => {
         threads.delete(worker); // set에서 쓰레드 삭제
         if(threads.size === 0 ){ // set이 모두 비웠졌을 경우, 워커쓰레드가 모두 끝난거니까
            console.log('워커 끝~');
         }
      })
   }

} else { // 워커스레드

   const data = workerData;
   parentPort.postMessage(data.start + 100);

}

---

Worker 스레드 성능

다음은 2에서 10000000 까지 소수를 구하는 예제 입니다.

알고리즘은 아리토스테네스의 체 소수 구하기 를 사용하였습니다.

let
   min = 2,
   max = 10_000_000,
   primes = [];

// 아리토스테네스의 체 소수구하기
function generatePrimes(start, range) {
   let isPrime = true;
   const end = start + range;
   for (let i = start; i < end; i++) {
      for (let j = min; j < Math.sqrt(end); j++) {
         if (i !== j && i % j === 0) {
            isPrime = false;
            break;
         }
      }

      if (isPrime) {
         primes.push(i)
      }
      isPrime = true;
   }
}

console.time('prime');
generatePrimes(min, max)
console.timeEnd('prime');
console.log(primes.length);

총 소수가 663479개가 나왔고 걸린 시간은 7.5초나 걸렸습니다.

하나의 쓰레드에서 하드한 cpu작업을 돌려서 그렇습니다.

 

워커를 이용해서 하드한 작업을 분배해 봅시다.

const { Worker, isMainThread, parentPort, workerData } = require('worker_threads');

let
   min = 2,
   max = 10_000_000,
   primes = [];

// 아리토스테네스의 체 소수구하기
function generatePrimes(start, range) {
   let isPrime = true;
   const end = start + range;
   for (let i = start; i < end; i++) {
      for (let j = min; j < Math.sqrt(end); j++) {
         if (i !== j && i % j === 0) {
            isPrime = false;
            break;
         }
      }

      if (isPrime) {
         primes.push(i)
      }
      isPrime = true;
   }
}


if (isMainThread) {
   const threadCount = 8;
   const threads = new Set()
   const range = Math.ceil((max - min) / threadCount); // 10_000_000 max를 8개의 쓰레드에 분배를 해서 처리하기 위해서

   let start = min;
   console.time('prime2');

   // 우리가 워커가 일을 할수있게 분배하고 직접 짜야 한다. 여간 복잡한게 아니다..
   for (let i = 0; i < threadCount - 1; i++) {
      const wStart = start;
      threads.add(new Worker(__filename, { workerData: { start: wStart, range: range } }))
      start += range;
   }
   // 7개만 for돌고 마지막 워커는 특별해서 따로 지정
   threads.add(new Worker(__filename, { workerData: { start: start, range: range + ((max - min + 1) % threadCount) } }));

   // 워커들 이벤트 등록
   for (let worker of threads) {

      worker.on('error', (err) => {
         throw err;
      })

      worker.on('exit', () => {

         threads.delete(worker);

         if (threads.size ===0){
            console.timeEnd('prime2')
            console.log(primes.length);
         }
      });

      // 워커들이 일한 결과를 메시지 받아서 정리해주는 동작도 직접 구현
      worker.on('message', (msg) => {
         primes = primes.concat(msg);
      })
   }

} else {
   // 워커들 일 등록
   generatePrimes(workerData.start, workerData.range);
   parentPort.postMessage(primes);
}

워커 작업의 핵심은 무언가 하드한 코드가 있으면,
이를 개발자가 이 코드를 잘 쪼개서 나뉘어 할당하는 로직을 구현하여 나누는 능력이 필요하고, 워커에서 돌리고 나온 결과나 설정 등등 하나하나 개발자가 세세하게 직접 코딩해야 한다는 특징이 있습니다.