성능과 보안 - 지연 측정·DDS 보안

이 장에서 배우는 것

앞 장에서 두리의 노드가 의도한 동작을 하는지 테스트했다. 그러나 테스트가 통과해도 실외 주행 중에는 장애물 정보가 늦게 도착하거나, 여러 로봇의 실험용 노드가 같은 통신 공간에 나타날 수 있다. 이제는 메시지가 언제 처리되는지 수치로 확인하고, 어떤 참여자가 통신할 수 있는지 구분해야 한다.

이 장에서는 지연과 지터를 측정하는 작은 Python 프로그램을 만들고, DDS 도메인과 보안 설정이 각각 해결하는 문제를 살펴본다. ROS가 없는 환경에서도 통계 계산과 시계 오차는 직접 재현할 수 있게 구성한다.

• 측정의 시작점과 끝점을 정하고 왕복 지연, 수신 간격, 시간 초과를 구분한다.
• 단조 시계와 백분위수를 사용해 재현 가능한 측정 결과를 만든다.
• DDS 도메인을 통신 탐색 범위로 사용하고 보안 경계와 구별한다.
• SROS2의 인증, 접근 제어, 암호화와 보안 설정 실패 시의 동작을 이해한다.

문제 상황

두리가 건물 밖 보행로에서 장애물을 발견했다. 센서는 계속 데이터를 보내지만 제어 노드가 반응하는 시점은 일정하지 않다. 평소에는 반응이 빠르고, 녹화 프로그램을 함께 실행하면 간헐적으로 늦어진다. 개발자는 토픽의 평균 수신 빈도가 정상이라는 이유로 통신 문제를 제외하려 한다.

그러나 평균 빈도는 개별 메시지의 대기 시간을 설명하지 못한다. 잠시 쌓인 메시지를 연달아 처리해도 전체 평균 빈도는 비슷하게 나올 수 있다. 두리에게 필요한 정보는 초당 메시지 개수뿐 아니라, 처리된 정보가 얼마나 오래되었는지와 지연이 얼마나 흔들리는지다.

같은 날 다른 팀이 같은 무선망에서 시험용 노드를 실행했다. 도메인 번호가 같고 토픽 이름과 자료형도 맞으면 두 팀의 노드가 서로 발견될 수 있다. 도메인 번호를 바꾸면 우발적인 혼선을 줄일 수 있지만, 번호를 아는 참여자의 접근을 차단하지는 못한다. 성능 측정과 접근 통제는 별도의 기준으로 확인해야 한다.

지연의 경계와 지터의 정의

무엇부터 무엇까지 잴 것인가

지연(latency)은 두 사건 사이의 경과 시간이다. 센서 촬영부터 모터 출력까지인지, 발행 함수 호출 직전부터 구독 콜백 시작까지인지에 따라 같은 로봇에서도 값이 달라진다. 따라서 숫자 앞에는 반드시 측정 경계가 있어야 한다. 이 장의 프로그램은 요청 발행 직전부터 응답 콜백 진입까지의 왕복 시간(round-trip time, RTT)을 측정한다.

왕복 측정에는 요청과 응답의 전송, 양쪽 미들웨어 처리, 응답 노드의 콜백 대기와 실행, 요청 노드의 응답 콜백 대기가 함께 들어간다. 따라서 결과를 네트워크 지연이라고 부르면 범위를 잘못 설명하게 된다. 측정용 메시지가 작으므로 큰 영상 메시지의 처리 비용까지 대표하지도 않는다.

측정 경계에 따라 같은 시간 값의 의미가 달라진다

지표	시작과 끝	해석할 때의 제한
단방향 지연	송신 측 기록부터 수신 측 기록까지	서로 다른 장치라면 시계 동기화 오차가 섞인다
왕복 지연	요청 발행 직전부터 응답 콜백 진입까지	왕복 경로와 응답 처리 시간을 포함한다
수신 간격	이전 콜백 진입부터 다음 콜백 진입까지	개별 메시지가 얼마나 오래 걸렸는지는 모른다
시간 초과 비율	전체 요청 중 제한 시간 안에 응답하지 않은 비율	실제 패킷 손실률과 같지 않다

경과 시간에는 Python의 time.monotonic_ns()를 사용한다. 단조 시계(monotonic clock)는 시스템 날짜와 시각을 보정해도 뒤로 가지 않는 시계다. 반환값의 기준 시점에는 의미를 부여하지 않고 두 값의 차이만 사용한다. 이름의 나노초는 반환 단위이며, 실제 측정 정확도가 나노초라는 뜻은 아니다.

서로 다른 컴퓨터의 단조 시계 값을 직접 빼서는 안 된다. 각 컴퓨터가 사용하는 기준이 같다는 보장이 없기 때문이다. 시스템 시각을 사용하더라도 수신 장치의 시계가 40 ms 앞서 있으면 실제 8 ms인 단방향 지연이 48 ms로 관측된다. 단방향 측정에는 시계 동기화와 그 오차를 따로 관리해야 한다.

왕복 시간을 둘로 나누면 단방향 지연이 된다는 주장도 조건이 필요하다. 양방향 경로가 비슷하고 응답 처리 시간이 충분히 작다는 가정이 있어야 한다. 무선망의 방향별 혼잡이나 응답 노드의 부하가 다르면 이 가정은 성립하지 않는다.

평균 옆에 흔들림과 누락을 둔다

지터(jitter)는 시간 값의 변동을 뜻하지만 계산 방식은 하나로 고정되어 있지 않다. 이 장에서는 관측한 왕복 지연 전체의 모표준편차를 왕복 지터로 정의한다. 수신 간격의 모표준편차는 별도의 수신 간격 지터다. 서로 다른 대상을 같은 이름으로 보고하지 않는다.

95백분위수(p95)는 정렬한 표본에서 낮은 값부터 약 95%가 포함되는 위치의 값이다. 여기서는 표본 수에 0.95를 곱해 올림한 순번을 선택한다. 표본이 다섯 개라면 마지막 값이 선택된다. 보간하는 계산법과 결과가 다를 수 있으므로 보고서에 계산법을 남긴다.

제한 시간 안에 돌아온 응답만 평균에 넣으면 느린 요청이 통계에서 빠진다. 통신 상태가 나빠졌는데 평균 지연은 오히려 작아지는 결과도 가능하다. 따라서 요청 수, 유효 응답 수, 시간 초과 수를 함께 저장한다. 시간 초과는 늦은 응답, 응답 노드 부재, 처리 지연 등을 포함하므로 원인을 곧바로 패킷 손실로 단정하지 않는다.

아래 프로그램은 발견 대기 시간을 두고 스무 번만 요청한다. 실행 여부를 확인하는 작은 측정이며 주행 성능을 판정할 표본 수는 아니다. 실제 비교에서는 측정 시간, 발행 주기, 메시지 크기, 부하, 보안 설정을 고정하고 반복한다. 지연이 몰리는 시간대가 있다면 전체 통계와 함께 시간 구간별 통계를 살펴본다.

DDS 도메인과 보안 경계

ROS 2에서 널리 사용하는 DDS(Data Distribution Service)는 같은 도메인에 속한 참여자들이 서로 발견하고 통신할 수 있는 기반을 제공한다. ROS_DOMAIN_ID는 이 도메인의 번호를 정하는 환경 변수다. 동일한 번호는 통신을 위한 조건 중 하나이며, 실제 연결에는 네트워크 도달 가능성, 발견 설정, 토픽 자료형과 통신 정책도 맞아야 한다.

두리의 실외 시험은 42, 다른 팀의 시험은 별도 번호처럼 운영 규칙을 정할 수 있다. 같은 로봇의 요청 노드와 응답 노드에는 같은 번호를 설정한다. 이미 실행 중인 프로세스는 나중에 셸 환경 변수를 바꿔도 설정이 바뀌지 않는다. 새 설정을 적용하려면 해당 프로세스를 다시 시작해야 한다.

도메인 번호는 비밀번호가 아니다. 같은 번호를 지정할 수 있는 참여자가 통신망에 접근하면 그 번호 자체로 신원을 검증하지 않는다. 별도의 DDS 라우터나 연결 구성을 만들면 도메인 사이에 메시지를 전달할 수도 있다. 도메인 분리는 운영상 탐색 범위를 나누는 수단이며 인증과 접근 제어를 대신하지 않는다.

도메인 번호 선택에는 구현과 운영체제의 포트 사용 조건도 관련된다. 이 장의 42는 예제용 선택이며 모든 배치에 사용할 공통 번호라는 뜻은 아니다. 현장에서는 다른 프로세스와의 충돌, 로봇 수, 방화벽 규칙까지 확인한다. 도메인 번호를 바꾼 직후 명령행 도구의 관찰 결과가 혼란스럽다면 기존 ROS 2 데몬의 실행 환경도 확인한다.

SROS2로 신원과 권한을 다룬다

SROS2는 ROS 2에서 DDS 보안 기능을 사용하도록 키와 인증서, 정책 자료를 준비하는 도구와 작업 흐름을 제공한다. 실제 통신 보호는 선택한 미들웨어의 DDS 보안 기능이 수행한다. 암호화는 전송 내용을 보호하고, 인증은 통신 상대의 신원을 확인하며, 접근 제어는 허용한 동작만 수행하도록 제한한다.

보안 구획(enclave)은 보안 신원과 권한 자료를 묶어 적용하는 단위다. 노드 이름이나 네임스페이스를 바꾸는 것만으로 별도 보안 신원이 만들어지지는 않는다. 이 장에서는 두 Python 프로세스에 각각 /duri/perf_probe와 /duri/perf_echo라는 구획을 지정한다. 구획 이름은 정책과 자료를 찾는 기준이며 비밀 문자열이 아니다.

보안 관련 설정과 자료는 서로 다른 책임을 맡는다

항목	맡는 일	혼동하지 않을 것
DDS 도메인	발견과 통신의 논리적 범위를 구분한다	번호만으로 참여자를 인증하지 않는다
인증서와 개인 키	신원을 증명하는 데 사용한다	개인 키는 배포 예제나 저장소에 넣지 않는다
거버넌스 정책	도메인의 보안 동작과 보호 요구를 정한다	보안을 켰다는 사실만으로 모든 보호 수준이 결정되지 않는다
권한 정책	신원별 발행·구독 등의 허용 범위를 정한다	유효한 인증서가 모든 토픽의 사용 권한을 뜻하지 않는다

요청 프로세스는 /perf/request를 발행하고 /perf/reply를 구독한다. 응답 프로세스는 반대 권한이 필요하다. 이 관계를 정책의 출발점으로 삼되 실제 노드가 사용하는 매개변수 서비스와 로그 등 부수 인터페이스도 확인한다. 허용할 인터페이스만 남기는 작업은 노드의 전체 동작을 관찰하며 진행해야 한다.

보안 자료 저장소에는 신원과 정책을 만드는 데 쓰는 인증 기관 자료도 포함될 수 있다. 개발용 저장소 전체를 모든 로봇에 복사하는 방식은 피한다. 운용 장치에는 필요한 구획의 자료와 신뢰 자료를 배포하고, 서명에 사용하는 개인 키는 발급 작업을 하는 환경에서 관리한다.

ROS_SECURITY_ENABLE=true는 보안 사용을 요청한다. 여기에 ROS_SECURITY_STRATEGY=Enforce를 설정하면 필요한 보안 자료를 사용할 수 없는 경우 보안 없이 계속하는 대신 초기화가 실패하도록 요구한다. 이 설정은 최소 권한 정책을 자동으로 만들어 주지 않는다. 초기화 실패를 막는 정책과 통신 상대의 권한을 제한하는 정책은 구분해야 한다.

ROS 2 Jazzy라는 배포판 이름만으로 보안 지원이 보장되지는 않는다. 선택한 미들웨어와 그 빌드에 DDS 보안 지원이 있어야 한다. Linux에서는 Jazzy가 지원하는 배포 환경과 패키지 구성을 확인하고, macOS에서는 ROS 2 설치 방식과 미들웨어 빌드 조건을 추가로 확인한다. 순수 Python 보조 프로그램은 이 조건에 의존하지 않는다.

설정의 사실 관계는 Jazzy의 도메인 설명, ROS 2 보안 개념, SROS2 프로젝트에서 확인할 수 있다. 문서의 예제를 옮기기보다 두리의 프로세스와 토픽에 맞게 정책을 설계한다.

완성 코드

두 파일을 같은 디렉터리에 저장한다. 첫 파일은 표준 라이브러리만 사용하며 통계 계산과 시계 오차를 재현한다. 두 번째 파일은 첫 파일의 계산 함수를 가져와 실제 ROS 2 요청과 응답을 측정한다. ROS 환경에서는 rclpy와 std_msgs를 가져올 수 있는 Python 인터프리터로 실행해야 한다.

perf_math.py

import math
from statistics import fmean, pstdev


def summarize(samples_ms, sent):
    values = sorted(float(value) for value in samples_ms)
    if sent < 0 or sent < len(values):
        raise ValueError("요청 수가 응답 수보다 작다.")
    if any(not math.isfinite(value) or value < 0 for value in values):
        raise ValueError("지연은 유한한 음이 아닌 값이어야 한다.")

    received = len(values)
    timed_out = sent - received
    rank = math.ceil(0.95 * received) - 1
    return {
        "sent": sent,
        "received": received,
        "timed_out": timed_out,
        "timeout_percent": 100.0 * timed_out / sent if sent else 0.0,
        "rtt_mean_ms": fmean(values) if values else None,
        "rtt_p95_ms": values[rank] if values else None,
        "rtt_jitter_ms": pstdev(values) if values else None,
    }


def main():
    report = summarize([8, 10, 12, 14, 26], sent=6)
    arrivals_ms = [0, 100, 200, 310, 400]
    intervals = [
        later - earlier
        for earlier, later in zip(arrivals_ms, arrivals_ms[1:])
    ]

    print(
        f"응답: {report['received']}/{report['sent']}, "
        f"시간 초과: {report['timed_out']} "
        f"({report['timeout_percent']:.2f}%)"
    )
    print(
        f"왕복 지연: 평균 {report['rtt_mean_ms']:.2f} ms, "
        f"p95 {report['rtt_p95_ms']:.2f} ms"
    )
    print(f"왕복 지터(모표준편차): {report['rtt_jitter_ms']:.2f} ms")
    print(
        f"수신 간격: 평균 {fmean(intervals):.2f} ms, "
        f"지터 {pstdev(intervals):.2f} ms"
    )

    actual_one_way_ms = 8.0
    receiver_clock_offset_ms = 40.0
    observed_ms = actual_one_way_ms + receiver_clock_offset_ms
    print(
        "시계가 40 ms 앞선 수신자의 단방향 관측값: "
        f"{observed_ms:.2f} ms"
    )


if __name__ == "__main__":
    main()

latency_probe.py

import argparse
import json
import time
from pathlib import Path

import rclpy
from rclpy.node import Node
from rclpy.qos import DurabilityPolicy, QoSProfile, ReliabilityPolicy
from std_msgs.msg import Int64MultiArray

from perf_math import summarize


class LatencyNode(Node):
    def __init__(self, mode):
        super().__init__(f"perf_{mode}")
        self.mode = mode
        self.finished = False
        self.sent = 0
        self.total = 20
        self.timeout_ns = 5_000_000_000
        self.pending = {}
        self.samples_ms = []
        self.ready_at_ns = time.monotonic_ns() + 2_000_000_000

        qos = QoSProfile(
            depth=10,
            reliability=ReliabilityPolicy.RELIABLE,
            durability=DurabilityPolicy.VOLATILE,
        )
        outgoing = "/perf/request" if mode == "probe" else "/perf/reply"
        incoming = "/perf/reply" if mode == "probe" else "/perf/request"
        self.publisher = self.create_publisher(Int64MultiArray, outgoing, qos)
        self.subscription = self.create_subscription(
            Int64MultiArray, incoming, self.on_message, qos
        )
        self.timer = None
        if mode == "probe":
            self.timer = self.create_timer(0.2, self.on_tick)

    def on_message(self, message):
        now_ns = time.monotonic_ns()
        if len(message.data) != 1:
            return

        if self.mode == "echo":
            self.publisher.publish(message)
            return

        sequence = message.data[0]
        start_ns = self.pending.pop(sequence, None)
        if start_ns is None:
            return

        elapsed_ns = now_ns - start_ns
        if elapsed_ns <= self.timeout_ns:
            self.samples_ms.append(elapsed_ns / 1_000_000.0)

    def on_tick(self):
        now_ns = time.monotonic_ns()
        for sequence, start_ns in list(self.pending.items()):
            if now_ns - start_ns > self.timeout_ns:
                del self.pending[sequence]

        if now_ns < self.ready_at_ns:
            return

        if self.sent < self.total:
            message = Int64MultiArray()
            message.data = [self.sent]
            self.pending[self.sent] = time.monotonic_ns()
            self.publisher.publish(message)
            self.sent += 1

        if self.sent == self.total and not self.pending:
            report = summarize(self.samples_ms, self.sent)
            report["samples_ms"] = self.samples_ms
            report["timeout_ms"] = self.timeout_ns / 1_000_000.0
            Path("latency_result.json").write_text(
                json.dumps(report, ensure_ascii=False, indent=2) + "\n",
                encoding="utf-8",
            )
            print("測定結果".replace("測定結果", "측정 결과를 latency_result.json에 저장했다."))
            self.finished = True
            self.timer.cancel()


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("mode", choices=["probe", "echo"])
    options, ros_args = parser.parse_known_args()

    rclpy.init(args=ros_args)
    node = None
    try:
        node = LatencyNode(options.mode)
        while rclpy.ok() and not node.finished:
            rclpy.spin_once(node, timeout_sec=0.1)
    except KeyboardInterrupt:
        pass
    finally:
        if node is not None:
            node.destroy_node()
        if rclpy.ok():
            rclpy.shutdown()


if __name__ == "__main__":
    main()

줄별 해설

summarize()의 첫 줄은 입력을 실수로 바꾸고 정렬한다. 표본 순서를 바꾸어도 평균과 표준편차는 같지만 백분위수에는 정렬이 필요하다. 요청 수보다 응답 수가 많거나 음수·무한대·비수인 지연이 들어오면 예외를 발생시켜 잘못된 자료를 조용히 통과시키지 않는다.

received는 제한 시간 안에 관측한 응답 수다. timed_out은 모든 요청의 관측을 끝냈다는 전제에서 요청 수와 응답 수의 차이로 계산한다. 아직 응답을 기다리는 중간 상태에 이 함수를 적용하면 대기 중인 요청까지 시간 초과로 세게 된다. ROS 프로그램은 대기 목록이 비었을 때만 이 함수를 호출한다.

math.ceil()로 백분위 순번을 올림하고 1을 빼서 목록 인덱스로 변환한다. 응답이 없으면 지연 통계는 None이며 JSON에는 null로 저장된다. 관측하지 못한 지연을 0 ms로 기록하면 가장 빠른 실행처럼 해석될 수 있다. 표본이 하나일 때 표준편차가 0이라는 사실도 안정성을 증명하지는 않는다.

보조 프로그램의 수신 시각 다섯 개에서는 간격 네 개가 나온다. 간격은 100, 100, 110, 90 ms이고 평균은 100 ms다. 이 예는 평균 주기가 유지되어도 도착 시점에는 흔들림이 존재함을 보여 준다. 마지막 출력의 48 ms는 별도의 단방향 시계 오차 예이며 앞의 왕복 표본을 반으로 나눈 결과가 아니다.

LatencyNode는 실행 인자에 따라 요청 또는 응답 역할을 맡는다. 두 역할은 같은 토픽 쌍과 통신 설정을 사용한다. 대기 요청은 일련번호를 키로 하는 pending 사전에 저장한다. 발행 직전의 단조 시각을 자기 프로세스에 보관하므로 다른 장치의 시계 값을 받을 필요가 없다.

on_message()는 콜백에 들어오자마자 시간을 기록한다. 응답 역할은 받은 번호를 그대로 돌려보낸다. 요청 역할은 사전에서 해당 번호를 꺼내 경과 시간을 계산한다. 이미 처리한 번호나 만료된 번호의 응답은 사전에 없으므로 무시한다. 같은 응답을 두 번 표본에 넣지 않는 구조다.

응답 콜백에서도 제한 시간을 검사한다. 타이머가 바빠서 만료 정리를 늦게 수행했더라도 5초를 넘긴 응답이 유효 표본으로 들어가지 않게 하기 위해서다. 만료 정리의 실행 시점은 타이머 지연만큼 늦어질 수 있지만, 유효 응답의 기준은 실제 경과 시간이다.

on_tick()은 만료 요청을 지우고, 초기 2초가 지난 뒤부터 명목상 0.2초마다 요청을 하나씩 보낸다. 발견 대기는 연결 완료를 보장하는 절차가 아니다. 상대 프로세스가 없으면 모든 요청이 시간 초과되고, 그 결과도 파일에 남는다. 스무 요청의 결과가 확정되면 통계를 기록하고 반복 실행을 종료한다.

파일 저장과 화면 출력은 측정 종료 시 한 번만 수행한다. 콜백마다 출력하면 터미널 처리 시간이 측정에 개입할 수 있다. parse_known_args()는 역할 인자 외의 인자를 ROS 초기화로 넘기므로 뒤에 보안 구획을 지정하는 인자를 붙일 수 있다. 한 측정에는 요청 프로세스 하나와 응답 프로세스 하나만 실행한다.

실행 결과

먼저 ROS 없이 두 파일의 구문을 검사하고 보조 프로그램을 실행한다. 구문 검사는 파일을 가져와 실행하지 않으므로 rclpy 설치가 없어도 가능하다. 정상적인 컴파일은 아무것도 출력하지 않는다. 다음 보조 프로그램의 출력은 고정된 입력으로 계산하므로 실행할 때마다 같다.

python3 -W error -m py_compile perf_math.py latency_probe.py
python3 perf_math.py

응답: 5/6, 시간 초과: 1 (16.67%)
왕복 지연: 평균 14.00 ms, p95 26.00 ms
왕복 지터(모표준편차): 6.32 ms
수신 간격: 평균 100.00 ms, 지터 7.07 ms
시계가 40 ms 앞선 수신자의 단방향 관측값: 48.00 ms

실제 ROS 측정은 Jazzy 환경이 설정된 터미널 두 개를 사용한다. 보안을 적용하기 전의 독립 실습에서 첫 터미널은 다음 명령으로 응답 노드를 실행한다. 응답 노드는 자체적으로 화면에 값을 출력하지 않고 계속 대기한다.

export ROS_DOMAIN_ID=42
python3 latency_probe.py echo

두 번째 터미널에서는 같은 도메인으로 요청 노드를 실행한다. 두 프로세스가 같은 디렉터리의 파일을 사용하도록 작업 디렉터리도 확인한다.

export ROS_DOMAIN_ID=42
python3 latency_probe.py probe

정상 종료 시 프로그램이 출력하는 문장은 다음과 같다. 실제 지연 수치는 환경에 따라 달라지므로 고정된 예상 수치로 제시하지 않는다. 측정값과 표본 목록은 요청 프로세스의 작업 디렉터리에 저장된다.

측정 결과를 latency_result.json에 저장했다.

파일을 확인할 때는 지연보다 먼저 sent, received, timed_out을 읽는다. 전부 시간 초과되어도 결과 저장 문장은 동일하다. 그 문장은 통신 성공을 뜻하지 않는다. 재실행하면 같은 파일을 덮어쓰므로 조건별 비교에서는 먼저 결과를 다른 이름으로 보관한다.

보안 실습에는 SROS2 명령행 도구와 보안을 지원하는 미들웨어가 필요하다. 두 프로세스가 종료된 상태에서, 새 실습 디렉터리를 기준으로 다음과 같이 개발용 자료를 만든다. 명령행 도구의 생성 로그는 버전에 따라 달라지므로 아래에는 입력할 명령만 제시한다.

ros2 security create_keystore duri_keys
ros2 security create_enclave duri_keys /duri/perf_probe
ros2 security create_enclave duri_keys /duri/perf_echo

각 터미널에서 다음 환경 변수를 설정한다. 두 터미널의 현재 디렉터리가 같아야 같은 저장소를 가리킨다. 다른 디렉터리라면 저장소의 실제 절대 경로를 지정한다.

export ROS_DOMAIN_ID=42
export ROS_SECURITY_KEYSTORE="$(pwd)/duri_keys"
export ROS_SECURITY_ENABLE=true
export ROS_SECURITY_STRATEGY=Enforce

첫 터미널과 두 번째 터미널에서 각각 다음 명령을 실행한다. 정상적인 요청 프로그램의 출력 문장은 앞과 같다.

python3 latency_probe.py echo --ros-args --enclave /duri/perf_echo

python3 latency_probe.py probe --ros-args --enclave /duri/perf_probe

구획 생성만으로 두 토픽에 한정된 최소 권한 정책이 완성되지는 않는다. 생성된 거버넌스와 권한 자료를 검토하고, 필요한 발행·구독만 허용하는 정책을 작성해 서명·배포해야 한다. 또한 보안 없는 참여자를 허용할지, 데이터 보호 방식을 어떻게 정할지도 거버넌스에서 확인해야 한다.

연결 성공 확인과 별도로, 존재하지 않는 구획으로 실행했을 때 초기화가 실패하는지 확인한다. 최소 권한 정책을 적용한 다음에는 허가되지 않은 발행과 구독이 성립하지 않는지도 점검한다. 초기화 실패의 오류 문구와 정책 거부 시점은 미들웨어마다 다를 수 있으므로 특정 문장 하나보다 종료 상태, 통신 결과, 진단 기록을 함께 본다.

실무에서 자주 틀리는 것

시스템 시각으로 경과 시간을 잰다

다음 코드는 시각 보정의 영향을 받는다. 날짜 기록에는 시스템 시각이 필요할 수 있지만 경과 시간은 단조 시계로 분리해서 잰다.

# 틀린 코드
start = time.time()
publisher.publish(message)
elapsed_ms = (time.time() - start) * 1000.0

# 고친 코드: 발행 호출에 걸린 시간만 측정한다.
start_ns = time.monotonic_ns()
publisher.publish(message)
publish_call_ms = (time.monotonic_ns() - start_ns) / 1_000_000.0

고친 코드도 수신 완료를 측정하지 않는다. 발행 호출 시간과 전달 지연은 다른 지표다. 왕복 지연이 목적이면 완성 코드처럼 응답 콜백에서 끝 시각을 기록한다.

응답이 없으면 평균을 0으로 채운다

0 ms는 관측 실패와 구별되어야 한다. 표본이 없는 상황을 빠른 응답처럼 표시하면 경보와 보고서의 판단이 뒤집힌다.

# 틀린 코드
mean_ms = fmean(samples_ms) if samples_ms else 0.0

# 고친 코드: 모든 요청의 결과가 확정된 뒤 호출한다.
report = summarize(samples_ms, sent)
mean_ms = report["rtt_mean_ms"]
timed_out = report["timed_out"]

시간 초과가 있는 실행과 없는 실행은 평균만으로 순위를 매기지 않는다. 상위 백분위수도 제한 시간 안에 응답한 표본에 대한 값이라는 조건을 붙인다.

도메인 번호를 접근 비밀로 취급한다

도메인을 바꾸는 것만으로 명령 토픽이 보호된다고 판단하는 설정은 부족하다. 운용 환경에서는 신원 자료와 정책을 함께 적용한다.

# 틀린 설정: 이것만으로 접근 통제가 된다고 가정한다.
export ROS_DOMAIN_ID=42

# 고친 설정: 준비한 보안 자료와 구획을 함께 사용한다.
export ROS_DOMAIN_ID=42
export ROS_SECURITY_KEYSTORE="$(pwd)/duri_keys"
export ROS_SECURITY_ENABLE=true
export ROS_SECURITY_STRATEGY=Enforce
python3 latency_probe.py probe --ros-args --enclave /duri/perf_probe

이 설정 역시 권한 정책 검토를 대신하지 않는다. 넓은 발행 권한을 가진 신원은 인증에 성공한 뒤 그 권한을 그대로 사용할 수 있다.

수신 콜백마다 출력을 추가한다

디버깅 출력은 처리 시간을 늘리고 다른 콜백을 늦출 수 있다. 측정 대상에 포함하려는 비용인지 먼저 결정한다.

# 틀린 코드: 측정 도중 표본마다 터미널 출력을 수행한다.
samples_ms.append(elapsed_ms)
print(f"응답 지연: {elapsed_ms:.3f} ms")

# 고친 코드: 콜백에서는 표본만 보관한다.
samples_ms.append(elapsed_ms)

# 모든 요청이 끝난 뒤 한 번 요약한다.
report = summarize(samples_ms, sent)

장시간 측정에서 표본을 계속 쌓으면 메모리 사용량도 증가한다. 완성 코드는 스무 요청으로 제한하지만, 장기 관측에서는 저장 한도와 구간별 집계를 별도로 설계한다.

한눈에 보기

두리의 성능과 보안을 점검할 때 함께 남길 정보

관심사	사용할 기준	함께 기록할 것
응답 지연	같은 단조 시계로 측정한 왕복 시간	시작·끝 경계, 메시지 크기, 발행 주기
시간 변동	왕복 표본의 모표준편차와 p95	표본 수, 백분위 계산법, 측정 구간
관측 실패	제한 시간 내 응답하지 않은 요청 수	제한 시간, 전체 요청 수, 유효 응답 수
통신 범위	일치하는 DDS 도메인 설정	프로세스 환경, 네트워크와 발견 조건
접근 통제	검증한 신원과 서명된 권한 정책	보안 구획, 미들웨어, 거부 동작 점검 결과

보안을 적용하기 전후의 성능도 같은 조건으로 비교한다. 초기 인증과 발견에 걸린 시간은 연결 이후의 왕복 지연과 분리한다. 암호화 비용이 얼마나 늘어나는지는 메시지 크기, 장치 성능, 구현에 따라 달라지므로 일정한 비율을 가정하지 않는다. 다음 장에서는 이렇게 정한 관측 기준과 보안 경계를 두리의 주행 스택을 묶는 기준으로 사용한다.

연습 문제

1. 왕복 지연이 6, 6, 6, 6, 36 ms이고 전체 요청이 다섯 번일 때 평균, 이 장의 방식으로 계산한 p95, 모표준편차를 구하라. 평균만 보고 판단할 때 놓치는 점도 설명하라.
2. 송신 장치가 기록한 시각은 1000 ms이고 수신 장치가 기록한 시각은 1055 ms다. 수신 시계가 송신 시계보다 40 ms 앞선다고 가정할 때 단방향 지연을 구하라. 시계 오차를 모른다면 어떤 값을 단정할 수 없는가.
3. ROS 없이 summarize([], sent=3)을 호출했을 때 시간 초과 수와 지연 통계가 무엇인지 설명하라. 이 함수가 대기 중인 요청까지 시간 초과로 잘못 세지 않으려면 언제 호출해야 하는가.
4. 두 프로세스의 도메인 번호가 같고 모두 유효한 인증서를 가졌다. 요청 프로세스에 /perf/reply 구독 권한이 없다면 무엇을 관찰할 수 있는가. 도메인 일치, 인증 성공, 권한 허용의 차이를 설명하라.

정답과 해설

1. 평균은 12 ms다. p95의 순번은 0.95 곱하기 5를 올림한 다섯 번째이므로 36 ms다. 평균과의 차이는 네 개의 -6과 한 개의 24이며, 제곱의 평균은 144이므로 모표준편차는 12 ms다. 대부분의 응답이 6 ms라는 사실과 한 응답이 36 ms라는 사실을 평균 하나로는 구별하기 어렵다.
2. 단순 차이는 55 ms이고, 시계가 앞선 40 ms를 빼면 단방향 지연은 15 ms다. 시계 차이를 모르면 55 ms 중 얼마가 실제 지연인지 단정할 수 없다. 왕복 측정은 같은 요청 장치의 시계를 사용하므로 이 시계 차이를 피할 수 있지만, 단방향 지연을 직접 제공하지는 않는다.
3. 시간 초과 수는 3이고 비율은 100%다. 평균, p95, 지터는 모두 None이다. 모든 요청이 유효 응답을 받았거나 제한 시간을 넘긴 뒤 호출해야 한다. 아직 기다리는 요청이 있다면 응답 수의 부족은 확정된 시간 초과가 아니다.
4. 정책에 따라 구독 엔드포인트 생성이나 통신 연결이 거부될 수 있다. 요청 노드가 계속 실행되어도 응답을 관측하지 못해 시간 초과가 생길 수 있다. 도메인 일치는 탐색 범위, 인증 성공은 신원 확인, 권한 허용은 해당 통신 동작의 승인에 해당한다. 시간 초과만으로 권한 문제라고 단정하지 말고 보안 진단 기록과 정책을 함께 확인한다.