QoS 이해 — 상황에 맞는 통신 품질 설계

1. QoS란 무엇인가

Topic 이름은 어디로, Message Type은 무엇을, QoS는 어떤 전달 조건으로 보낼지 정합니다.

Topic Name       /scan
Message Type     sensor_msgs/msg/LaserScan
QoS              BEST_EFFORT + VOLATILE + KEEP_LAST(5)

이름과 Type이 같아도 QoS가 호환되지 않으면 Publisher와 Subscription은 통신하지 않습니다. 또한 호환된다고 해서 원하는 성능이 자동 보장되는 것은 아닙니다. Queue가 너무 깊으면 오래된 Data가 쌓이고, Reliable 재전송은 손실망에서 지연을 키울 수 있습니다.

QoS는 “품질을 무조건 높이는 스위치”가 아니라 다음 요구의 Trade-off를 표현하는 계약입니다.

• 완전성: 하나도 놓치지 않아야 하는가?
• 최신성: 늦은 Data보다 최신 Data가 중요한가?
• Late Joiner: 나중에 켠 Node도 이전 값을 받아야 하는가?
• 자원: Memory와 Network 대역폭이 얼마나 있는가?
• 시간: 언제부터 Data가 쓸모없어지는가?
• 감시: Publisher 침묵을 얼마나 빨리 알아야 하는가?

2. 먼저 Data의 성격을 질문하라

QoS 이름부터 고르지 말고 다음 순서로 요구사항을 적습니다.

1. 한 Message를 놓치면 복구 가능한가?
2. 500ms 늦게 도착해도 가치가 있는가?
3. 새 Subscriber가 마지막 상태를 즉시 받아야 하는가?
4. Producer와 Consumer 속도 차이는 얼마인가?
5. 정상 발행 간격과 허용 지터는 얼마인가?
6. Publisher가 멈추면 어떤 안전 동작을 해야 하는가?

예를 들어 Camera 영상은 한 Frame 유실보다 Queue에 쌓인 2초 전 영상이 더 나쁠 수 있습니다. 반면 한 번 발행되는 Map은 늦게 켠 Planner도 받아야 합니다. 같은 “중요한 Data”여도 요구가 다르므로 Profile도 달라집니다.

3. Reliability: 재전송할 것인가

BEST_EFFORT

전송을 시도하지만 유실 Message의 재전송을 보장하지 않습니다. 고주기 Sensor처럼 다음 Data가 곧 오고 최신성이 중요한 경우에 적합합니다.

RELIABLE

Middleware가 ACK와 재전송을 사용해 전달 신뢰성을 높입니다. 이 의미는 Application 처리 완료나 영구 저장 보장이 아닙니다. Resource Limit, Process 종료와 Network 단절까지 없애는 것도 아닙니다.

“Sensor는 항상 BEST_EFFORT”, “Command는 항상 RELIABLE” 같은 절대 규칙은 없습니다. Command에 Watchdog가 있고 최신 명령만 의미 있다면 작은 Queue와 유효기간 관리가 더 중요할 수 있습니다. 실제 Driver와 Safety Architecture를 함께 봅니다.

4. Durability: 늦게 참가한 Node에게 과거 값을 줄 것인가

VOLATILE

Subscription이 연결된 이후의 Data만 받습니다. 연속 Sensor Stream과 속도 명령에 일반적입니다.

TRANSIENT_LOCAL

Publisher가 보관한 Sample을 Late Joiner에게 제공합니다. Map, Static Transform, 현재 Configuration처럼 마지막 값 자체가 계속 유효할 때 사용합니다.

Publisher: Map 1회 발행 ─────────────── 살아 있음
                              ↑
                   Subscriber 늦게 시작

VOLATILE        → 과거 Map을 받지 못함
TRANSIENT_LOCAL → 보관된 Map을 즉시 받음

주의할 점:

• 보관 책임은 일반적으로 Publisher 수명과 연관됩니다.
• TRANSIENT_LOCAL만 지정하고 Depth가 너무 작으면 필요한 Sample이 남지 않을 수 있습니다.
• 계속 변하는 Command에 과거 값을 재전달하면 Late Joiner가 오래된 동작을 실행할 수 있으므로 보통 VOLATILE을 사용합니다.

5. History와 Depth: Queue에 무엇을 남길 것인가

KEEP_LAST(N)

최근 N개 Sample을 보관합니다. 대부분의 실시간 Robot Topic에 적합합니다.

KEEP_ALL

Middleware Resource Limit까지 모든 Sample을 보관하려고 합니다. “무한 Memory”를 뜻하지 않으며 Consumer가 느리면 자원 압박이 커집니다.

Publisher:  1 2 3 4 5 6 7 8
Consumer:       느리게 처리

KEEP_LAST(1):             [8]
KEEP_LAST(3):         [6][7][8]
KEEP_ALL:       [1][2][3][4][5][6][7][8]  ← 자원 제한 주의

Depth를 키우면 Drop이 무조건 해결되는 것이 아닙니다. Consumer 처리율이 Producer보다 계속 낮으면 Queue는 결국 차거나 지연만 늘어납니다.

Backlog 증가율 ≈ Publish Rate - Processing Rate

Camera UI나 제어 입력은 KEEP_LAST(1)이 최신성을 지키는 경우가 많고, 짧은 Burst Event는 예상 Burst 크기만큼 Depth가 필요할 수 있습니다.

6. Deadline: 기대 발행 간격 감시

Deadline은 “이 시간 간격 안에 다음 Sample이 제공되어야 한다”는 시간 계약입니다. Publisher는 Offered Deadline, Subscriber는 Requested Deadline을 가집니다.

10Hz Sensor의 명목 주기 = 100ms
실측 간격: 96, 103, 111, 98, 130ms

Deadline을 100ms로 잡으면 정상 Scheduling Jitter에도 경보가 날 수 있습니다. 고정 배수를 맹목적으로 쓰기보다 실측 최대 지연, Startup, Network와 허용 감지 시간을 고려합니다.

Deadline 위반 Event는 감지 신호이지 자동 정지 기능이 아닙니다. Callback에서 상태를 갱신하고 Safety Controller가 정지 정책을 실행해야 합니다.

7. Lifespan: 오래된 Message의 유효기간

Lifespan은 발행 후 일정 시간이 지난 Sample을 만료시킵니다.

발행 ───── Network 지연 ───── 수신
t=0                            t=0.8s
Lifespan=0.2s → 이미 만료된 Sample

오래된 명령, Detection과 상태가 뒤늦게 사용되는 것을 줄이는 데 유용합니다. 그러나 모든 RMW 구현의 세부 지원과 동작은 확인해야 하며 Application은 Message Timestamp 기반 Age 검증과 Driver Watchdog도 유지해야 합니다.

age = self.get_clock().now() - rclpy.time.Time.from_msg(msg.header.stamp)
if age.nanoseconds > 200_000_000:
    self.get_logger().warning('200ms보다 오래된 Command 거절')
    return

Lifespan은 Header Stamp 의미를 대신하지 않습니다. Middleware 생성·전송 시간과 실제 Sensor 측정 시간은 다를 수 있습니다.

8. Liveliness와 Lease Duration

Liveliness는 Publisher가 살아 있음을 판단하는 정책입니다.

• AUTOMATIC: Middleware가 Participant의 생존을 자동 관리합니다.
• MANUAL_BY_TOPIC: Application이 해당 Publisher의 생존을 명시적으로 주장해야 합니다.
• Lease Duration: 생존 신호 없이 얼마나 기다릴지 정합니다.

Liveliness Lost는 Process가 완전히 종료된 경우뿐 아니라 Lease 조건을 충족하지 못한 경우에도 발생할 수 있습니다. 이를 단독 Emergency Stop 신호로 보지 말고 Deadline, Application Heartbeat, Command Watchdog와 함께 설계합니다.

9. 호환성은 Offered와 Requested의 계약이다

Publisher는 품질을 제공(Offered)하고 Subscription은 품질을 요청(Requested)합니다. 요청이 제공 수준을 넘으면 통신하지 않습니다.

Reliability 호환표

Durability 호환표

Deadline은 Publisher가 약속한 간격이 Subscriber가 요구한 간격보다 같거나 짧아야 합니다. Liveliness에도 Offered/Requested 호환 규칙이 있습니다.

History, Depth와 Lifespan은 중요한 동작 정책이지만 Reliability·Durability처럼 단순히 다르면 항상 Discovery 호환 실패가 되는 항목으로 취급하지 않습니다.

10. 가장 먼저 실행할 진단 명령

ros2 topic list -t
ros2 topic info /scan --verbose
ros2 topic hz /scan
ros2 topic bw /scan

info --verbose에서 각 Endpoint의 다음 항목을 비교합니다.

• Node 이름과 Namespace
• Topic Type
• Endpoint Type
• Reliability
• Durability
• History와 Depth
• Deadline·Lifespan·Liveliness

CLI 도움말에서 현재 배포판이 지원하는 QoS Option을 확인합니다.

ros2 topic echo --help
ros2 topic pub --help
ros2 bag record --help

11. CLI로 호환 실패 재현

Terminal A에서 BEST_EFFORT Publisher를 실행합니다.

ros2 topic pub /qos_demo std_msgs/msg/String \
  "{data: hello}" --rate 5 \
  --qos-reliability best_effort \
  --qos-durability volatile

Terminal B에서 RELIABLE을 요구합니다.

ros2 topic echo /qos_demo std_msgs/msg/String \
  --qos-reliability reliable \
  --qos-durability volatile

비호환 경고가 나타나거나 Data가 전달되지 않습니다. Subscription을 BEST_EFFORT로 맞춥니다.

ros2 topic echo /qos_demo std_msgs/msg/String \
  --qos-reliability best_effort \
  --qos-durability volatile

배포판별 Argument 위치와 명칭은 --help 결과를 따릅니다.

12. Late Joiner 실험

TRANSIENT_LOCAL Publisher가 한 번 발행한 값을 보관하도록 실행합니다.

ros2 topic pub --once /site_map_version std_msgs/msg/String \
  "{data: warehouse-v3}" \
  --qos-reliability reliable \
  --qos-durability transient_local \
  --qos-depth 1

Publisher Process가 유지되는 방식은 CLI 동작과 배포판에 따라 달라질 수 있습니다. 반복 저주기 발행으로 학습 효과를 확인할 수도 있습니다.

ros2 topic pub /site_map_version std_msgs/msg/String \
  "{data: warehouse-v3}" --rate 0.1 \
  --qos-reliability reliable \
  --qos-durability transient_local \
  --qos-depth 1

나중에 Subscriber를 시작합니다.

ros2 topic echo /site_map_version std_msgs/msg/String \
  --qos-reliability reliable \
  --qos-durability transient_local

VOLATILE과 비교해 첫 값 도착 시점을 관찰합니다.

13. Python QoS Publisher

# qos_lab_py/qos_publisher.py
import rclpy
from rclpy.node import Node
from rclpy.qos import (
    QoSDurabilityPolicy,
    QoSHistoryPolicy,
    QoSProfile,
    QoSReliabilityPolicy,
)
from std_msgs.msg import String


class QoSPublisher(Node):
    def __init__(self):
        super().__init__('qos_publisher')
        profile = QoSProfile(
            history=QoSHistoryPolicy.KEEP_LAST,
            depth=3,
            reliability=QoSReliabilityPolicy.BEST_EFFORT,
            durability=QoSDurabilityPolicy.VOLATILE,
        )
        self.publisher = self.create_publisher(
            String, 'qos_demo', profile)
        self.sequence = 0
        self.timer = self.create_timer(0.1, self.tick)

    def tick(self):
        message = String()
        message.data = f'sequence={self.sequence}'
        self.sequence += 1
        self.publisher.publish(message)


def main(args=None):
    rclpy.init(args=args)
    node = QoSPublisher()
    try:
        rclpy.spin(node)
    except KeyboardInterrupt:
        pass
    finally:
        node.destroy_node()
        rclpy.shutdown()


if __name__ == '__main__':
    main()

14. Python QoS Subscriber와 Sensor Profile

# qos_lab_py/scan_subscriber.py
import rclpy
from rclpy.node import Node
from rclpy.qos import qos_profile_sensor_data
from sensor_msgs.msg import LaserScan


class ScanSubscriber(Node):
    def __init__(self):
        super().__init__('scan_subscriber')
        self.subscription = self.create_subscription(
            LaserScan,
            '/scan',
            self.on_scan,
            qos_profile_sensor_data,
        )

    def on_scan(self, message):
        valid = [r for r in message.ranges
                 if message.range_min <= r <= message.range_max]
        if valid:
            self.get_logger().info(f'nearest={min(valid):.2f}m')


def main(args=None):
    rclpy.init(args=args)
    node = ScanSubscriber()
    try:
        rclpy.spin(node)
    except KeyboardInterrupt:
        pass
    finally:
        node.destroy_node()
        rclpy.shutdown()


if __name__ == '__main__':
    main()

qos_profile_sensor_data는 Sensor Data에 흔히 쓰는 Best Effort와 얕은 Queue를 제공합니다. 그러나 실제 Driver의 Offered QoS를 topic info --verbose로 확인하는 절차는 생략하지 않습니다.

15. Map과 지속 상태 Profile

from nav_msgs.msg import OccupancyGrid
from rclpy.qos import (
    QoSDurabilityPolicy,
    QoSHistoryPolicy,
    QoSProfile,
    QoSReliabilityPolicy,
)

map_qos = QoSProfile(
    reliability=QoSReliabilityPolicy.RELIABLE,
    durability=QoSDurabilityPolicy.TRANSIENT_LOCAL,
    history=QoSHistoryPolicy.KEEP_LAST,
    depth=1,
)

self.map_subscription = self.create_subscription(
    OccupancyGrid, '/map', self.on_map, map_qos)

적합한 Data:

• 마지막 Map
• Static Configuration Snapshot
• 현재 Route Definition
• Robot Description처럼 늦게 참가한 Node가 즉시 알아야 하는 상태

부적합한 Data:

• 과거 속도 명령
• 오래된 Emergency 해제 명령
• 매 Frame 새로 생성되는 Camera Stream

16. Deadline Event Callback

from rclpy.duration import Duration
from rclpy.event_handler import SubscriptionEventCallbacks
from rclpy.qos import QoSProfile, QoSReliabilityPolicy

sensor_qos = QoSProfile(
    depth=5,
    reliability=QoSReliabilityPolicy.BEST_EFFORT,
    deadline=Duration(seconds=0.2),
)

events = SubscriptionEventCallbacks(
    deadline=self.on_deadline_missed,
    incompatible_qos=self.on_incompatible_qos,
    liveliness=self.on_liveliness_changed,
    message_lost=self.on_message_lost,
)

self.subscription = self.create_subscription(
    LaserScan,
    '/scan',
    self.on_scan,
    sensor_qos,
    event_callbacks=events,
)

def on_deadline_missed(self, status):
    self.sensor_healthy = False
    self.get_logger().error(
        f'Scan Deadline 위반, total={status.total_count}')

def on_incompatible_qos(self, status):
    self.get_logger().error(
        f'QoS 비호환 Policy={status.last_policy_kind}')

def on_liveliness_changed(self, status):
    self.get_logger().warning(
        f'alive={status.alive_count}, not_alive={status.not_alive_count}')

def on_message_lost(self, status):
    self.get_logger().warning(
        f'Middleware Message Lost total={status.total_count}')

Event 지원 범위와 Status Field는 ROS 2 배포판 및 RMW에 따라 확인해야 합니다. message_lost가 End-to-End Application Drop 전체를 측정하는 만능 Counter는 아닙니다.

17. Deadline과 Application Watchdog 함께 사용

QoS Deadline만으로 Safety를 완성하지 않습니다. Header Stamp가 있는 Sensor는 Data Age도 검사합니다.

def on_scan(self, message):
    stamp = rclpy.time.Time.from_msg(message.header.stamp)
    age_sec = (self.get_clock().now() - stamp).nanoseconds / 1e9
    if age_sec > 0.25:
        self.get_logger().error(f'오래된 Scan: {age_sec:.3f}s')
        self.request_safe_stop()
        return
    self.last_valid_scan_time = self.get_clock().now()

별도 Timer가 Callback 침묵을 감시합니다.

def watchdog_tick(self):
    elapsed = self.get_clock().now() - self.last_valid_scan_time
    if elapsed.nanoseconds > 300_000_000:
        self.request_safe_stop()

QoS Deadline Event  → Middleware 계약 위반 감지
Header Age Check    → 측정 Data 신선도 검증
Application Watchdog→ 유효 Callback 침묵 감지
Driver Watchdog     → 최종 Actuator 명령 단절 시 정지

18. 상황별 Profile 선택표

다음은 시작점이지 모든 Robot의 절대 정답이 아닙니다.

19. /cmd_vel은 왜 하나의 정답이 없는가

속도 명령은 일반적으로 최신 값만 의미하고 이전 명령은 위험할 수 있습니다. 따라서 다음 항목을 함께 설계합니다.

• KEEP_LAST(1) 또는 작은 Depth
• VOLATILE
• Command Timestamp 또는 Sequence
• Source별 Timeout이 있는 Velocity Mux
• Driver의 Command Watchdog
• 명령 손실과 재전송 지연 중 어느 쪽이 더 위험한지 측정

Reliable은 단일 Packet 손실 복구에 유리하지만 Network가 혼잡할 때 오래된 명령 지연과 Back-pressure를 고려해야 합니다. Best Effort는 Drop을 허용하므로 충분한 발행률과 Watchdog가 필요합니다. Hardware Driver가 제공·요구하는 Profile을 먼저 확인합니다.

20. Wi-Fi와 원격 Robot

Wi-Fi에서 대용량 Reliable Camera Stream은 손실 Packet 재전송으로 Queue와 지연을 키워 작은 Control·State Packet까지 방해할 수 있습니다.

권장 실험 순서:

1. 유선 기준의 Hz·Bandwidth·Latency를 측정합니다.
2. Wi-Fi에서 같은 값을 측정합니다.
3. Camera를 Best Effort와 작은 Depth로 바꿉니다.
4. Resolution·Frame Rate·Compression을 조정합니다.
5. Control과 Telemetry를 대용량 Stream에서 논리·물리적으로 분리합니다.
6. Packet Loss, Jitter, CPU와 Queue Age를 기록합니다.

ros2 topic hz /camera/image_raw
ros2 topic bw /camera/image_raw
ros2 topic delay /camera/image_raw

topic delay 사용 가능 여부와 Message Header 요구는 배포판 도움말에서 확인합니다.

21. 느린 Subscriber와 Back-pressure

Callback이 100ms 걸리는데 Publisher가 30Hz라면 Consumer는 초당 10개밖에 처리하지 못합니다.

발행 30 msg/s - 처리 10 msg/s = 초당 20개 Backlog

해결책은 Depth를 무한히 키우는 것이 아닙니다.

• 불필요한 Data는 Publisher에서 줄입니다.
• Callback에서 무거운 처리를 Worker로 분리합니다.
• 최신 Data만 필요하면 Queue 1과 교체 정책을 씁니다.
• 처리율을 Profiling하고 알고리즘을 최적화합니다.
• 적절한 Best Effort와 Sampling을 검토합니다.
• Callback Group과 Executor가 다른 핵심 Callback을 굶기지 않게 합니다.

22. rosbag2 기록과 재생 QoS

rosbag2는 Topic Endpoint를 보고 QoS를 적응하려고 하지만 여러 Publisher가 서로 다른 Profile을 제공하거나 Late Joiner 의미를 보존해야 할 때 Override가 필요할 수 있습니다.

# qos_overrides.yaml
/scan:
  reliability: best_effort
  durability: volatile
  history: keep_last
  depth: 10

/map:
  reliability: reliable
  durability: transient_local
  history: keep_last
  depth: 1

ros2 bag record /scan /map \
  --qos-profile-overrides-path qos_overrides.yaml

ros2 bag play my_bag \
  --qos-profile-overrides-path qos_overrides.yaml

기록 전 topic info --verbose, 기록 후 ros2 bag info, 재생 중 Endpoint QoS와 실제 수신을 확인합니다. Bag에 Data가 있다는 사실만으로 재생 Subscriber와 호환되는 것은 아닙니다.

23. 여러 Publisher가 같은 Topic에 있을 때

같은 Topic·Type에 Publisher가 둘 이상이면 각각 다른 Offered QoS를 가질 수 있습니다. Subscription 하나가 모든 Publisher와 호환될 수도 있고 일부와만 호환될 수도 있습니다.

ros2 topic info /detections --verbose

예를 들어 한 Detection Publisher는 Reliable, 다른 Publisher는 Best Effort인데 Subscriber가 Reliable을 요구하면 Best Effort Publisher Data만 빠질 수 있습니다. “가끔 특정 Robot Data만 안 온다”는 증상이 생깁니다.

Source별 Topic 분리, 공통 Interface 계약 또는 Bridge에서 QoS 정규화를 검토합니다.

24. System Default와 Best Available

SYSTEM_DEFAULT는 실제 값을 Middleware 기본 설정에 맡깁니다. 이식성이 필요할 때는 배포 환경에서 실제 적용 QoS를 조사해야 합니다.

일부 Client Library와 배포판은 발견된 Endpoint에 맞추는 Best Available 계열 Profile을 제공합니다. 편리하지만 Application 요구를 대신 결정해 주지는 않습니다. 최소 보장과 Late Joiner 의미가 중요한 Topic은 명시적 Profile과 통합 Test가 더 분명합니다.

RMW 구현에 따라 일부 고급 QoS 정책 지원에 차이가 있을 수 있으므로 DDS Vendor나 Zenoh 계열 RMW를 변경하면 호환·Event·시간 정책 Test를 다시 수행합니다.

25. QoS 변경은 Interface 변경이다

QoS는 Code 내부 최적화가 아니라 다른 Node와의 통신 계약입니다. Publisher만 바꾸면 기존 Subscriber가 끊길 수 있습니다.

변경 절차:

1. 현재 Endpoint QoS와 Traffic을 기록합니다.
2. Data의 완전성·최신성·Late Joiner 요구를 문서화합니다.
3. Publisher와 모든 Subscriber의 호환표를 만듭니다.
4. Simulation과 손실 Network에서 Test합니다.
5. rosbag, Bridge, Visualization Tool도 확인합니다.
6. 단계적으로 배포하고 Drop·Latency·Deadline Event를 관찰합니다.
7. 이전 Profile로 되돌릴 기준을 정합니다.

26. 표준 진단 절차

1. Topic 이름과 Type 확인
        ↓
2. Publisher·Subscriber 존재 확인
        ↓
3. info --verbose로 Offered/Requested 비교
        ↓
4. CLI QoS를 명시해 최소 재현
        ↓
5. Hz·Bandwidth·Delay와 Queue Age 측정
        ↓
6. Event Callback·RMW Log·Network 조사

ros2 topic list -t
ros2 topic info /topic_name --verbose
ros2 topic echo /topic_name TYPE --qos-reliability best_effort
ros2 topic hz /topic_name
ros2 topic bw /topic_name
echo "$RMW_IMPLEMENTATION"
echo "$ROS_DOMAIN_ID"

QoS를 무작정 Reliable로 바꾸기 전에 현재 Publisher가 무엇을 제공하고 Application이 무엇을 요구하는지 비교합니다.

27. 증상별 해결 방법

28. 흔한 실수와 교정

1. Reliable을 무조건 더 좋은 QoS로 본다. 재전송 지연과 Back-pressure를 함께 봅니다.
2. Depth를 늘리면 손실이 해결된다고 생각한다. 지속적인 처리율 부족부터 해결합니다.
3. Publisher와 Subscriber Profile이 완전히 같아야 한다고 외운다. Offered/Requested 호환 규칙을 사용합니다.
4. History·Depth 불일치를 모두 연결 실패 원인으로 본다. 실제 호환 정책과 Queue 동작을 구분합니다.
5. TRANSIENT_LOCAL을 과거 명령에 사용한다. Late Joiner가 실행해도 안전한 상태 Data에 사용합니다.
6. Deadline Event가 Robot을 자동 정지한다고 믿는다. Application과 Driver Safety 동작을 연결합니다.
7. Lifespan만 믿고 Header Age를 검사하지 않는다. 측정 Timestamp 신선도를 별도로 검증합니다.
8. Sensor Profile을 모든 Topic에 사용한다. Data 의미별로 설계합니다.
9. CLI Echo가 되면 Code QoS도 맞다고 생각한다. Code Endpoint를 직접 조사합니다.
10. rosbag이 모든 QoS를 자동 해결한다고 믿는다. Record·Playback Override와 수신을 검증합니다.
11. RMW를 바꿔도 결과가 같다고 가정한다. 고급 정책과 Event Test를 반복합니다.
12. QoS만으로 안전 통신이 완성된다고 생각한다. Timestamp, Sequence, Watchdog와 독립 Safety 계층을 둡니다.

29. 실습 체크리스트

• [ ] topic info --verbose로 Endpoint별 QoS를 읽었다.
• [ ] Best Effort Publisher와 Reliable Subscriber 비호환을 재현했다.
• [ ] Reliability를 맞춰 다시 수신했다.
• [ ] VOLATILE과 TRANSIENT_LOCAL의 Late Joiner 차이를 확인했다.
• [ ] KEEP_LAST Depth 1과 깊은 Queue의 지연 차이를 관찰했다.
• [ ] Python QoSProfile로 Publisher·Subscriber를 작성했다.
• [ ] Sensor Data Profile로 /scan을 구독했다.
• [ ] Deadline과 Incompatible QoS Event Callback을 등록했다.
• [ ] Header Stamp 기반 Age Check를 구현했다.
• [ ] Application·Driver Watchdog 역할을 구분했다.
• [ ] Camera의 Hz·Bandwidth·Delay를 측정했다.
• [ ] Wi-Fi에서 Reliable·Best Effort를 비교했다.
• [ ] rosbag2 QoS Override로 기록·재생했다.
• [ ] 실제 Robot Topic별 QoS 계약표를 작성했다.
• [ ] QoS 변경 전후의 Drop·Latency·CPU를 기록했다.

30. 정리

QoS는 Topic Data의 전달 완전성, 최신성, 보관, 시간과 생존 감시를 설계하는 계약입니다. Reliability는 재전송, Durability는 Late Joiner, History와 Depth는 Queue, Deadline은 기대 간격, Lifespan은 유효기간, Liveliness는 Publisher 생존 의미를 다룹니다.

Profile은 Data 이름만 보고 정하지 않습니다. 한 Message 유실과 지연 중 무엇이 더 위험한지, 마지막 값을 늦게 참가한 Node가 받아야 하는지, Producer와 Consumer 처리율이 얼마인지 측정하여 선택합니다.

문제가 생기면 이름·Type 다음에 ros2 topic info --verbose로 Offered와 Requested QoS를 비교하십시오. 그 뒤 Hz, Bandwidth, Delay, Queue Age와 Event를 측정합니다. QoS는 Watchdog, Timestamp, Sequence와 독립 Safety Architecture를 보완하지만 대신하지 않습니다.

연습 문제

1. Topic 이름, Message Type과 QoS가 각각 정하는 계약을 설명하세요.
2. Camera Stream에서 Reliable보다 Best Effort가 유리할 수 있는 이유는 무엇인가요?
3. Reliable이 Application의 처리 완료를 보장하지 않는 이유는 무엇인가요?
4. VOLATILE과 TRANSIENT_LOCAL의 Late Joiner 동작 차이를 설명하세요.
5. KEEP_LAST Depth를 계속 늘렸을 때 지연과 Memory에 어떤 영향이 있나요?
6. Reliability와 Durability의 Offered/Requested 호환표를 설명하세요.
7. Deadline 호환 방향과 Deadline Event의 의미를 설명하세요.
8. Lifespan과 Message Header Stamp 기반 Age Check가 모두 필요한 이유는 무엇인가요?
9. Liveliness, Deadline, Application Watchdog와 Driver Watchdog의 역할 차이는 무엇인가요?
10. ros2 topic info /scan --verbose에서 반드시 비교할 항목을 쓰세요.
11. /cmd_vel QoS에 하나의 정답이 없는 이유와 함께 필요한 안전 장치를 설명하세요.
12. Wi-Fi에서 Reliable Camera가 제어 통신까지 느리게 만들 수 있는 과정을 설명하세요.
13. 느린 Subscriber 문제를 Depth 증가만으로 해결할 수 없는 이유는 무엇인가요?
14. rosbag2 기록·재생에서 QoS Override가 필요한 상황을 설명하세요.
15. Publisher는 보이지만 Callback이 호출되지 않을 때 Application부터 Network까지 진단 순서를 설명하세요.

참고 자료

• ROS 2 Quality of Service settings
• rclpy QoS API
• rclpy Event Handler API
• ROS 2 Topic statistics
• rosbag2 QoS Overrides
• ROS 2 Topic CLI