5.3 Zenoh 세션(Session) 설정 및 라이프사이클 관리

5.3 Zenoh 세션(Session) 설정 및 라이프사이클 관리

모든 Zenoh 통신의 출발점은 라우터 혹은 다른 피어(Peer)와 손을 맞잡는 세션(Session) 의 확립이다.
세션은 단순한 TCP 소켓 껍데기가 아니다. 그 안에는 데이터 라우팅을 위한 정보 해시 테이블(DHT), 재연결(Backoff) 타이머, 비동기 I/O 이벤트 루프가 살아 숨 쉬고 있다.

따라서 세션을 언제 어떻게 열고(Open), 여러 백그라운드 스레드에서 어떻게 이 세션 객체를 공유하며(Share), 메인 프로세스가 꺼질 때 어떻게 우아하게 닫을 것인가(Graceful Shutdown)를 설계하는 것은 메모리 누수 방지의 첫 단추이자, 분산 시스템 개발자의 기본 소양이다. 이 장에서는 zenoh::open() 이면의 심연을 들여다본다.

1. Config 객체 생성과 네트워크 파라미터 튜닝

세션을 열기 전에 라우터에게 내 신원과 통신 방식을 알려주는 명함이 바로 Config 객체다.

1.0.1 [Runbook] 코드 레벨 Config 주입 전술

설정 파일(json5)을 외부에서 읽어올 수도 있지만, 동적으로 IP를 할당받는 엣지 디바이스라면 코드 내에서 Config를 튜닝하는 편이 훨씬 직관적이다.

1. 기본 클라이언트(Client) 모드 생성

use zenoh::config::Config;

// 디폴트 설정: 백그라운드에서 멀티캐스트 스카우팅이 켜지며, 
// 로컬 네트워크의 라우터를 자동 탐색한다.
let mut config = Config::default();

// 명시적으로 라우터에 하드코딩된 IP를 조준하고 싶을 때
config.insert_json5("connect/endpoints", "[\"tcp/192.168.10.10:7447\"]").unwrap();

2. 메모리 오버헤드를 막는 방어적 파라미터 주입
대역폭이 극단적으로 부족하거나, 연결 유지가 불안정한 무선망(LTE)을 위한 커스텀 설정 런북.

// Keepalive 확인 주기를 무선망에 맞게 10초로 연장시켜서 네트워크 단절 오탐지를 방지
config.insert_json5("transport/link_rx_keepalive", "10000").unwrap();

// 만약 연결이 죽더라도 즉각 0.1초만에 재접속(Backoff)을 치도록 튜닝
config.insert_json5("connect/backoff/initial_delay", "100").unwrap();

Config 객체의 내부는 사실상 JSON 노드 트리이므로, insert_json5 메서드 하나로 공식 문서에 있는 모든 zenohd.json5 파라미터를 메모리 상에서 덮어쓸(Override) 수 있다.

2. zenoh::open()을 이용한 비동기 세션 초기화

Config 명함이 준비되었다면, 이제 네트워크 스위치를 올릴 차례다. Rust 비동기 생태계에서 네트워크 I/O 병목을 피하기 위해 open 함수는 비동기 퓨처(Future) 구조를 띤다.

2.0.1 [Runbook] 비동기 세션 마운트

1. Tokio 런타임 위에서의 세션 개방

use zenoh;

#[tokio::main]
async fn main() {
    let config = zenoh::config::Config::default();
    
    println!("Zenoh 네트워크에 연결을 시도합니다...");
    
    // .res().await 를 호출하는 순간, TCP 3-way handshake가 백그라운드로 실행된다.
    // unwrap() 대신 match 로 네트워크 에러를 우아하게 잡아내는 것이 프로덕션 코드다.
    let session = match zenoh::open(config).res().await {
        Ok(s) => s,
        Err(e) => {
            eprintln!("[FATAL] 라우터를 찾을 수 없거나 인증에 실패했습니다: {}", e);
            std::process::exit(1);
        }
    };

    println!("세션 연결 성공! Zenoh UUID: {}", session.info().zid().await);
}

[내부 동작 인스펙션]
이 session 변수는 단순한 핸들이 아니다. 이 변수가 메모리에 살아있는 동안, Zenoh의 C/Rust 코어 엔진은 보이지 않는 곳에서 백그라운드 I/O 스레드를 띄워 라우터와의 Heartbeat(핑/퐁)를 무한 지속한다. 즉, 이 세션의 라이프사이클이 곧 당신의 노드의 수명 그 자체다.

3. Peer-to-Peer 모드와 Router 연결 모드의 코드 수준 차이점

당신의 디바이스는 클라이언트(Client)인가, 아니면 그 자체로 독립된 피어(Peer) 노드인가?
모드(Mode)에 따라 zenoh::open 이면에서 일어나는 DHT(분산 라우팅 테이블) 생성량과 코어 스레드의 점유율은 완전히 달라진다.

3.0.1 [Runbook] 모드 파라미터 강제 치환 매커니즘

1. 클라이언트(Client) 모드: 경량화 접속 최우선
단순히 클라우드 라우터(zenohd)에 붙어서 데이터만 상납하는 역할이라면 무조건 Client 모드다. 라우팅 연산 오버헤드와 스카우팅 트래픽 폭풍이 사라진다.

let mut config = Config::default();
// 나는 평민(클라이언트)이다. 라우팅 기능 따위는 끈다.
config.set_mode(Some(zenoh::config::WhatAmI::Client)).unwrap();
config.insert_json5("connect/endpoints", "[\"tcp/my.router.com:7447\"]").unwrap();

2. 피어(Peer) 모드: 라우터 없는 완전 P2P 망치기
전쟁터 한복판, 혹은 라이다(LiDAR) 센서와 자율주행 PC를 다이렉트로 랜 케이블로 묶는 환경에서는 라우터가 존재하지 않는다.
이때는 프로그램 자체가 피어(Peer)가 되어 라우팅 기능을 몸에 탑재해야 한다.

let mut config = Config::default();
// 내가 스스로 라우팅 기능을 흡수하는 피어(Peer)가 되겠다!
config.set_mode(Some(zenoh::config::WhatAmI::Peer)).unwrap();

// 다른 Peer들이 나를 찾아올 수 있도록 문(Listen)을 열어둔다!
config.insert_json5("listen/endpoints", "[\"tcp/0.0.0.0:7447\"]").unwrap();

정말 끔찍한 실수는, 공장 내 1,000대의 센서를 코딩하면서 WhatAmI::Peer로 일괄 복붙해버리는 경우다. 1,000대의 센서가 서로가 서로를 중계하겠다고 아우성치며 통신망 자체를 마비시키는 “네트워크 지옥(Broadcast Storm)“을 맛보게 될 것이다. 조심하라.

4. 스레드 간 안전한 세션 공유 전략 (Arc, Mutex 활용)

zenoh::open() 을 통해 한 번 맺어진 Session 객체는 소중한 자원이다.
그런데 백엔드 서버를 짤 때는 HTTP 요청 스레드, 데이터베이스 워커 스레드 10여 개가 동시에 session.put() 을 호출해야만 한다.
C/C++ 개발자라면 Session 포인터를 무지성으로 전역 변수(Global Variable)에 박고 세그폴트(Segfault)의 공포에 떨었겠지만, Rust는 이를 절대 허락하지 않는다.

4.0.1 [Runbook] 비동기 다중 태스크간 세션 복제 마법

만물의 근원은 Arc(Atomic Reference Counting)다.
하지만 놀랍게도 Zenoh의 Rust 컴파일 설계자들은 Session 껍데기 자체를 기본적으로 T: Clone + Send + Sync 특성으로 구워놓았다.
즉, Arc 포장재조차 직접 쓸 필요 없이 session.clone()만 치면 완벽하게 레퍼런스 카운팅이 되어서 다른 스레드로 비동기 락 없이 주입된다!

[우아한 스레드 분배 전술]

use std::sync::Arc;
use tokio;

#[tokio::main]
async fn main() {
    let session = zenoh::open(Config::default()).res().await.unwrap();
    
    // session 객체를 아크(Arc)에 담을 필요도 없다. 
    // 내부적으로 이미 비싸지 않은 스레드-세이프 핸들이기 때문!
    
    // 워커 1: 온도 센서 보고 스레드
    let s1 = session.clone(); 
    tokio::spawn(async move {
        loop {
            // 뮤텍스 락(Lock) 없이 바로 동시 호출 가능!
            s1.put("sensor/temp", "32.5").res().await.unwrap(); 
            tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
        }
    });

    // 워커 2: 백엔드 상태 전송 스레드
    let s2 = session.clone();
    tokio::spawn(async move {
        loop {
            s2.put("system/status", "OK").res().await.unwrap();
            tokio::time::sleep(tokio::time::Duration::from_secs(5)).await;
        }
    });

    // 메인 스레드 대기
    tokio::signal::ctrl_c().await.unwrap();
}

Zenoh Rust API 설계팀의 천재성이 빛을 발하는 순간이다. 개발자는 성능 저하를 일으키는 Mutex 나 RwLock 오버헤드를 걱정할 필요 없이, 단순히 session.clone()을 천 번 만 번 복사해 멀티 코어 스레드에 뿌리기만 하면 내부 I/O 큐가 완벽하게 다중 접근을 정리(Multiplexing)해 준다.

5. 세션 종료 및 리소스의 안전한 해제 메커니즘

서버나 백엔드 컨테이너가 K8s에 의해 종료 통보(SIGTERM)를 받았을 때, “그냥 프로세스를 킬(Kill)해버리면 안 되나요?“라고 묻는다면 하수다.
급작스럽게 소켓이 날아가면 네트워크 중간의 라우터 테이블에는 한동안 썩은(Stale) 구독 라우팅 정보가 쓰레기처럼 표류하게 되어 전체 네트워크 처리량에 악영향을 준다.

5.0.1 [Runbook] 우아한 종료(Graceful Shutdown) 전술

Rust는 변수가 스코프를 벗어나면 자동으로 Drop 트레이트가 호출되어 메모리를 회수한다. zenoh::Session 객체도 내부적으로 Drop 시에 상대방 라우터에게 “나 이제 떠난다!“라는 CLOSE 제어 패킷을 날리도록 설계되어 있다.
하지만 멀티 스레드 환경에서는 모든 clone() 껍데기들이 소거되어 레퍼런스 카운트(Ref Count)가 0이 떨어지길 기다려서는 안 된다. 강력한 강제 회수 API를 호출해야 한다.

명시적 세션 폐기 절차

use tokio::signal;

#[tokio::main]
async fn main() {
    let session = zenoh::open(Config::default()).res().await.unwrap();
    
    // ... 백그라운드 스레드들이 session.clone() 을 들고 돌아가는 중 ...

    // [데몬 종료 시그널 대기 (Ctrl+C 또는 SIGTERM)]
    signal::ctrl_c().await.unwrap();
    println!("종료 시그널 수신. 우아한 종료를 시작합니다...");

    // [핵심] 다른 스레드들의 참조 카운트를 무시하고, 
    // 즉각 하부 네트워크 소켓을 찢고 이웃 라우터에게 작별 인사(BYE) 패킷 방출!
    session.close().res().await.unwrap();
    
    println!("Zenoh 세션이 안전하게 해제되었습니다.");
}

이 session.close().res().await 단 한 줄이, 자율주행 모빌리티가 지하주차장에 들어가면서 중앙 관제 클라우드에 “정상 오프라인” 상태임을 통보하는 핵심 마침표가 된다.