Node.js와 libuv

Node.js가 비동기 작업을 처리하는 방법

2024-11-05


이번 글은 Node.js가 libuv를 통해 비동기 작업을 처리하는 방법에 대한 글입니다.
최근 Spring Framework의 비동기 및 논블로킹(Non-blocking) 모듈인 Spring WebFlux에 대해 깊게 공부하게 되었습니다.
저는 Spring WebFlux 뿐만 아니라 대표적인 JavaScript 런타임인 Node.js는 어떻게 비동기 처리를 구현하는 지에 대해 궁금했는데요.
사실 제가 비동기라는 개념을 처음 배운 기술이 Node.js라서 더욱 흥미가 생겼었습니다.

Event Loop

이벤트 루프란 I/O 멀티플렉싱(I/O Multiplexing)을 리액터 패턴(Reactor Pattern)으로 구현한 모델입니다.
이벤트 루프 모델을 사용하면 I/O 멀티플렉싱을 활용해 단일 스레드로도 비동기 처리를 할 수 있다는 장점이 있습니다.
while there are still events to process:
    e = get the next event
    if there is a callback associated with e:
        call the callback
위 코드는 libuv의 공식 문서에 있는 이벤트 루프의 의사 코드(Pseudo Code)인데요.
이벤트 루프가 발생한 이벤트들을 가져오고 처리하는 작업을 반복 수행한다는 것을 확인할 수 있습니다.

Spring WebFlux의 WAS(Web Application Server)인 Netty나 Node.js는 이벤트 루프를 통해 비동기 작업을 처리하는데요.
Netty는 Java NIO(New I/O)의 Selector를 통해, Node.js는 libuv라는 라이브러리를 통해 이벤트 루프를 구현합니다.
사실 추상화의 차이일 뿐, 내부적으로는 Selector나 libuv 모두 epoll()이나 kqueue() 등의 이벤트 관리 시스템 콜(System Call)을 통해 구현되어 있습니다.

libuv

여러 라이브러리 중 Node.js의 libuv에 대해 자세히 알아보겠습니다.
공식 문서에 따르면 libuv는 다중 플랫폼에 기반한 비동기 I/O 라이브러리입니다.
사실 이벤트 루프를 구현하는데 사용하는 epoll()이나 kqueue() 등의 시스템 콜들은 특정 플랫폼에 종속적인데요.
이때, libuv는 이들을 모두 통합해 추상화한 이벤트 루프 API(Application Programming Interface)를 제공함으로써 다중 플랫폼을 지원합니다.

Event Loop with libuv

libuv의 주요 기능은 이벤트 루프입니다.
auto* loop = new uv_loop_t();
 
uv_loop_init(loop);
uv_run(loop, UV_RUN_DEFAULT);
uv_loop_close(loop);
 
delete loop;
libuv에서 이벤트 루프는 uv_loop_t 타입을 가지고 있는데요.
위에서는 동적 메모리 할당을 통해서 이벤트 루프를 생성했지만, uv_default_loop()를 통해서도 이벤트 루프를 생성할 수도 있습니다.
uv_default_loop()를 사용하면 동적 메모리 할당 및 해제 과정을 생략할 수 있습니다.
core.c
int uv_run(uv_loop_t* loop, uv_run_mode mode) {
  int timeout;
  int r;
  int can_sleep;
 
  r = uv__loop_alive(loop);
  if (!r)
    uv__update_time(loop);
 
  if (mode == UV_RUN_DEFAULT && r != 0 && loop->stop_flag == 0) {
    uv__update_time(loop);
    uv__run_timers(loop);
  }
 
  while (r != 0 && loop->stop_flag == 0) {
    can_sleep =
        uv__queue_empty(&loop->pending_queue) &&
        uv__queue_empty(&loop->idle_handles);
 
    uv__run_pending(loop);
    uv__run_idle(loop);
    uv__run_prepare(loop);
 
    timeout = 0;
    if ((mode == UV_RUN_ONCE && can_sleep) || mode == UV_RUN_DEFAULT)
      timeout = uv__backend_timeout(loop);
 
    uv__metrics_inc_loop_count(loop);
 
    uv__io_poll(loop, timeout);
 
    for (r = 0; r < 8 && !uv__queue_empty(&loop->pending_queue); r++)
      uv__run_pending(loop);
 
    uv__metrics_update_idle_time(loop);
 
    uv__run_check(loop);
    uv__run_closing_handles(loop);
 
    uv__update_time(loop);
    uv__run_timers(loop);
 
    r = uv__loop_alive(loop);
    if (mode == UV_RUN_ONCE || mode == UV_RUN_NOWAIT)
      break;
  }
 
  if (loop->stop_flag != 0)
    loop->stop_flag = 0;
 
  return r;
}
uv_run()은 실제로 이벤트 루프가 수행하는 반복문이 있는 함수입니다.
이벤트 루프는 여러 페이즈(Phase)로 나뉘게 되는데요.
그래서 uv_run()의 반복문 내부를 보면 uv__run_pending()이나 uv__run_idle() 등 해당 페이즈에 대한 작업을 순서대로 호출하는 것을 확인할 수 있습니다.
대표적으로 uv__run_timers()인 Timer 페이즈에서는 setTimeout() 등의 시간 관련 JavaScript 함수의 콜백을 처리하게 됩니다.

Handle

libuv에서는 이벤트 루프에 등록하는 콜백을 핸들(Handle)이라고 합니다.
핸들은 이벤트 종류에 따라 여러 타입이 있으며, 각 핸들은 특정 페이즈에 처리됩니다.
main.cpp
void on_idle(uv_idle_t* handle) {
    std::cout << "Event loop is idle." << std::endl;
}
 
int main() {
    uv_loop_t *loop = uv_default_loop();
    uv_idle_t handle;
 
    uv_idle_init(loop, &handle);
    uv_idle_start(&handle, on_idle);
 
    uv_run(loop, UV_RUN_DEFAULT);
 
    uv_loop_close(loop);
 
    return 0;
}
Event loop is idle.
Event loop is idle.
Event loop is idle.
...
위 코드는 유휴 핸들을 통해 이벤트 루프가 유휴 상태일 때 로그를 출력하도록 하는 코드인데요.
실제로 이벤트 루프가 실행되자마자 유휴 상태로 전환되고, 유휴 핸들의 콜백인 on_idle()이 호출되는 것을 확인할 수 있습니다.

이번엔 소켓 프로그래밍의 예시도 한 번 보겠습니다.
main.cpp
void on_connect(uv_stream_t *server, int status) {
    auto *client = new uv_tcp_t;
 
    uv_tcp_init(server->loop, client);
 
    if (uv_accept(server, reinterpret_cast<uv_stream_t *>(client)) == 0) {
        int address_len = sizeof(sockaddr_in);
        char address[address_len];
        sockaddr_in client_address{};
        inet_ntop(AF_INET, &client_address.sin_addr, address, address_len);
        uv_tcp_getpeername(client, reinterpret_cast<sockaddr *>(&client_address), &address_len);
 
        std::cout << "New connection from " << address << ":" << ntohs(client_address.sin_port) << std::endl;
    } else {
        std::cerr << "accept() failed";
        uv_close(reinterpret_cast<uv_handle_t *>(client), nullptr);
        delete client;
    }
}
 
int main() {
    uv_loop_t *loop = uv_default_loop();
    uv_tcp_t server;
    sockaddr_in address{};
 
    address.sin_family = AF_INET;
    address.sin_addr.s_addr = INADDR_ANY;
    address.sin_port = htons(PORT);
 
    uv_tcp_init(loop, &server);
    uv_tcp_bind(&server, reinterpret_cast<const sockaddr*>(&address), 0);
    uv_listen(reinterpret_cast<uv_stream_t *>(&server), 128, on_connect);
 
    std::cout << "Server listening on " << ntohs(address.sin_port) << std::endl;
 
    uv_run(loop, UV_RUN_DEFAULT);
 
    uv_loop_close(loop);
 
    return 0;
}
위 코드는 TCP(Transmission Control Protocol) 핸들을 통해서 클라이언트가 연결 요청을 보내면 로그를 출력하도록 하는 코드입니다.
libuv의 이벤트 루프를 사용했기 때문에 하나의 스레드로 여러 클라이언트 소켓의 요청을 받을 수 있습니다.

Node.js의 Event Loop

이제 Node.js 내부에서 libuv의 이벤트 루프가 어떻게 실행되는지 살펴보겠습니다.
node_main.cc
int main(int argc, char* argv[]) {
  return node::Start(argc, argv);
}
node.cc
int Start(int argc, char** argv) {
  #ifndef DISABLE_SINGLE_EXECUTABLE_APPLICATION
    std::tie(argc, argv) = sea::FixupArgsForSEA(argc, argv);
  #endif
  return static_cast<int>(StartInternal(argc, argv));
}
 
static ExitCode StartInternal(int argc, char** argv) {
  ...
  NodeMainInstance main_instance(snapshot_data,
                                 uv_default_loop(),
                                 per_process::v8_platform.Platform(),
                                 result->args(),
                                 result->exec_args());
  return main_instance.Run();
}
Node.js는 구동 시 Start()를 호출합니다.
Start() 내에서 호출되는 StartInternal()에서는 uv_default_loop()으로 생성한 이벤트 루프를 NodeMainInstance에 전달하는 것을 확인할 수 있습니다.
또 다시 StartInternal()에서는 NodeMainInstanceRun()을 호출하는데요.
node_main.cc
ExitCode NodeMainInstance::Run() {
  Locker locker(isolate_);
  Isolate::Scope isolate_scope(isolate_);
  HandleScope handle_scope(isolate_);
 
  ExitCode exit_code = ExitCode::kNoFailure;
  DeleteFnPtr<Environment, FreeEnvironment> env = CreateMainEnvironment(&exit_code);
  CHECK_NOT_NULL(env);
 
  Context::Scope context_scope(env->context());
  Run(&exit_code, env.get());
  return exit_code;
}
 
void NodeMainInstance::Run(ExitCode* exit_code, Environment* env) {
  if (*exit_code == ExitCode::kNoFailure) {
    if (!sea::MaybeLoadSingleExecutableApplication(env)) {
      LoadEnvironment(env, StartExecutionCallback{});
    }
 
    *exit_code = SpinEventLoopInternal(env).FromMaybe(ExitCode::kGenericUserError);
  }
 
  #if defined(LEAK_SANITIZER)
    __lsan_do_leak_check();
  #endif
}
Run()은 핸들 설정 및 이벤트 루프 실행을 위해 LoadEnvironment()SpinEventLoopInternal()를 호출합니다.
environment.cc
MaybeLocal<Value> LoadEnvironment(Environment* env,
                                  StartExecutionCallback cb,
                                  EmbedderPreloadCallback preload) {
  env->InitializeLibuv();
  env->InitializeDiagnostics();
  if (preload) {
    env->set_embedder_preload(std::move(preload));
  }
  env->InitializeCompileCache();
 
  return StartExecution(env, cb);
}
env.cc
void Environment::InitializeLibuv() {
  HandleScope handle_scope(isolate());
  Context::Scope context_scope(context());
 
  CHECK_EQ(0, uv_timer_init(event_loop(), timer_handle()));
  uv_unref(reinterpret_cast<uv_handle_t*>(timer_handle()));
 
  CHECK_EQ(0, uv_check_init(event_loop(), immediate_check_handle()));
  uv_unref(reinterpret_cast<uv_handle_t*>(immediate_check_handle()));
 
  CHECK_EQ(0, uv_idle_init(event_loop(), immediate_idle_handle()));
 
  CHECK_EQ(0, uv_check_start(immediate_check_handle(), CheckImmediate));
 
  CHECK_EQ(0, uv_prepare_init(event_loop(), &idle_prepare_handle_));
  CHECK_EQ(0, uv_check_init(event_loop(), &idle_check_handle_));
 
  CHECK_EQ(0, uv_async_init(
      event_loop(),
      &task_queues_async_,
      [](uv_async_t* async) {
        Environment* env = ContainerOf(
            &Environment::task_queues_async_, async);
        HandleScope handle_scope(env->isolate());
        Context::Scope context_scope(env->context());
        env->RunAndClearNativeImmediates();
      }));
  uv_unref(reinterpret_cast<uv_handle_t*>(&idle_prepare_handle_));
  uv_unref(reinterpret_cast<uv_handle_t*>(&idle_check_handle_));
  uv_unref(reinterpret_cast<uv_handle_t*>(&task_queues_async_));
 
  {
    Mutex::ScopedLock lock(native_immediates_threadsafe_mutex_);
    task_queues_async_initialized_ = true;
    if (native_immediates_threadsafe_.size() > 0 ||
        native_immediates_interrupts_.size() > 0) {
      uv_async_send(&task_queues_async_);
    }
  }
 
  StartProfilerIdleNotifier();
  env_handle_initialized_ = true;
}
LoadEnvironment()는 Node.js의 환경을 설정하는 함수인데요.
실제로 이벤트 루프의 각 페이즈에 핸들을 등록하는 부분은 EnvironmentInitializeLibuv()입니다.
embedded_helpers.cc
Maybe<ExitCode> SpinEventLoopInternal(Environment* env) {
  ...
  do {
    if (env->is_stopping()) break;
    uv_run(env->event_loop(), UV_RUN_DEFAULT);
    ...
  } while (more == true && !env->is_stopping());
  ...
}
SpinEventLoopInternal()에서는 최종적으로 uv_run()을 통해 이벤트 루프를 실행시키게 됩니다.