ROS2 - Qos, DDS(10)

이준혁·2025년 1월 1일

ROS2

목록 보기
14/14

제가 작성하는 이 글은 제가 공부하기 위해 작성하고 제 학교 후배들에게 조금이라도 ros에 대해서 접하게 하고자 작성하게 되었습니다. 그러니 만약 틀린내용이나 그러니 제가 부족한 부분이 있을수 있으니 주의해주면서 봐주시면 감사하겠습니다 -Happy Lee-


이번에는 다양한 개념을 알아보기 위해 작성하는 블로그이다. 너무 깊지 않을수 있으니 궁금한게 있으면 더 찾아보면서 공부하는게 좋을것이다.

목차

  1. RCLPY
  2. ROS2BAG
  3. CLI
  4. Intra-process communication
  5. Lifecycle
    6. Qos
    7. DDS

QOS

QOS란 뭘까?

Qos는 쉽게 말해 데이터 통신 옵션입니다.
기존 ros1에서는 TCPROS로 tcp 통신을 진행하였는데 ROS2는 TCP 방식과 UDP 방식으로 선택적으로 사용이 가능하다.
TCP : 신뢰성 중심
UDP : 속도 중심
그래서 Qos의 도입으로 데이터 전송시 실시간성 설정 관련 부분, 대역폭 옵션, 데이터 지속성, 중복성 등을 바꿀수 있습니다.

QOS 항목

  • Reliability: ROS2에서는 TCP 통신 방식으로, 데이터 손실을 방지함으로써 신뢰도를 우선(reliable)으로 설정하거나, UDP 통신 방식과 같이 통신 속도 최우선으로 설정
  • History: 정해진 크기만큼 데이터를 보관하는 기능
  • Durability: 데이터 수신하는 서브스크라이버가 생성되기 전, 데이터의 사용유무를설정
  • Deadline: 정해진 주기 내 데이터의 발신 및 수신이 없는 경우 이벤트 함수실행
  • Lifespan: 정해진 주기 내 수신되는 데이터에만 유효판정, 이외의 데이터는 삭제
  • Liveliness: 정해진 주기 내 노드 또는 토픽의 생사를 확인

Qos history

데이터를 몇 개나 보관할지 결정하는 QoS 옵션

qos_profile = QoSProfile(history=QoSHistoryPolicy.KEEP_LAST, depth=10)

Qos Reliability

신뢰성 또는 속도 우선설정

qos_profile = QoSProfile(reliability=QoSReliabilityPolicy.BEST_EFFORT)

▪ TCP처럼손실을방지하면서신뢰도를우선시: RELIABLE
▪ UDP처럼손실을감안하고통신속도를우선시: BEST_EFFORT

Qos Durability

데이터 수신하는 서브스크르라이버가 생성되기 전, 데이터의 사용 유무를 설정

qos_profile = QoSProfile(durability=QoSDurabilityPolicy.TRANSIENT_LOCAL)

Qos Deadline

정해진 주기 내 데이터의 발신 및 수신이 없는 경우 이벤트 함수 실행

qos_profile = QoSProfile(depth=10, deadline = Duration(0.1))

Qos Lifespan

정해진 주기 내 수신되는 데이터에만 유효 판정되어 이외 데이터는 삭제됨

qos_profile = QoSProfile(lifespan = Duration(0.01))

Qos Liveliness

정해진 주기 내 노드 또는 토픽의 생사를 확인하는 것

qos_profile = QoSProfile(liveliness=LivelinessPolicy.AUTOMATIC,liveliness_lease_duration=Duration(seconds=1.0))

그래서 이런식으로 Qos를 설정하여 세밀하게 사용 가능하다.

import rclpy
from rclpy.node import Node
from std_msgs.msg import String
from rclpy.qos import QoSDurabilityPolicy, QoSHistoryPolicy, QoSLivelinessPolicy, QoSReliabilityPolicy
from rclpy.qos import QoSProfile, Duration

class FullQoSProfilePublisher(Node):
    def __init__(self):
        super().__init__('full_qos_profile_publisher')
        
        # QoS 프로필 설정
        qos_profile = QoSProfile(
            history=QoSHistoryPolicy.KEEP_LAST,
            depth=10,
            reliability=QoSReliabilityPolicy.RELIABLE,
            durability=QoSDurabilityPolicy.TRANSIENT_LOCAL,
            liveliness=QoSLivelinessPolicy.AUTOMATIC,
            liveliness_lease_duration=Duration(seconds=1.0),
            deadline=Duration(seconds=1.5),
            lifespan=Duration(seconds=5),
            avoid_ros_namespace_conventions=False
        )
        
        # 퍼블리셔 생성, QoS 프로필 적용
        self.publisher = self.create_publisher(String, 'full_qos_chatter', qos_profile)ㅊ

DDS

ROS2에서는 DDS 프로토콜을 사용하여 데이터를 분산시키는 방법을 활용하고 있습니다. DDS는 OMG(Object Management Group)에 의해 표준화된 데이터를 분산 서비스로, 실시간 시스템에서의 효율적인 데이터 교환을 목표로 합니다.


ROS1은 TCPROS를 사용하여 통신을 구현했습니다. ROS1에서 널리 사용된 브로커 기반의 MQTT는 TCP를 사용하여 브로커를 통해 메시지를 중개했습니다.

이와 달리 ROS2에서는 DDS를 통해 UDP를 사용하여 데이터 교환의 지연 시간을 줄이고, QoS를 통해 더 세밀한 서비스 품질 관리를 제공합니다.


▪ TCP처럼손실을방지하면서신뢰도를우선시: RELIABLE
▪ UDP처럼손실을감안하고통신속도를우선시: BEST_EFFORT

profile
#자기공부 #틀린것도많음 #자기개발 여러분 인생이 힘들다 하더라도 그것을 깨는 순간 큰 희열감으로 옵니다~

0개의 댓글