4. QoS(Quality of Service)
- 데이터 통신 옵션
- ROS1 --> 자체 프로토콜 TCPROS
- ROS2 --> TCP(신뢰성 중심), UDP(속도 중심) 방식을 선택적으로 사용 가능
- ROS2에서 DDS의 QOS를 도입
- 퍼블리셔 또는 서브스크라이버 선언 시 QoS를 매개 변수로 지정하여 원하는 통신 방식 설정
4.1 QoS의 종류
qos_profile = QoSProfile([option]=QoS[option]Policy.[setting])
History: 정해진 크기만큼 데이터를 보관하는 기능
qos_profile = QoSProfile(history=QoSHistoryPolicy.KEEP_LAST, depth=10)
| option | explanation |
|---|---|
| KEEP_LAST | 정해진 메시지 큐 크기(depth)만큼의 데이터를 보관 |
| KEEP_ALL | 모든 데이터를 보관 |
Reliability: TCP 통신 방식으로, 데이터 손실을 방지하여 신뢰도를 우선(Reliable)로 설정하거나, UDP 통신 방식과 같이 통신 속도 최우선(best effort)로 설정
qos_profile = QoSProfile(reliability=QoReliabilityPolicy.BEST_EFFORT)
| option | explanation |
|---|---|
| BEST_EFFORT | 데이터 송신에 집중, 전송 속도를 중시하며 네트워크 상태에 따라 유실이 발생 |
| RELIABLE | 데이터 수신에 집중, 신뢰성을 중시하며 유실 발생시 재전송을 통해 수신을 보장 |
Durability: 데이터 수신하는 서브스크라이버가 생성되거 전, 데이터의 사용 유무를 설정
qos_profile = QoSProfile(duarability=QoSDurabilityPolicy.TRANSIENT_LOCAL)
| option | explanation |
|---|---|
| TRANSIENT_LOCAL | 서브스크립션이 생성되기 전의 데이터도 보관(publisher에만 적용 가능) |
| VOCATILE | 서브스크립션이 생성되기 전의 데이터는 무효 |
Deadline: 정해진 주기 내 데이터의 발신 및 수신이 없는 경우 이벤트 함수 실행
qos_profile = QoSProfile(depth=10, deadline=Duration(0.1))
| option | explanation |
|---|---|
| deadline_duration | deadline을 확인하는 주기 |
Lifespan: 정해진 주기 내 수신되는 데이터에만 유효 판정, 이외 데이터는 삭제
qos_profile = QoSProfile(lifespan=Duration(0.01))
| option | explanation |
|---|---|
| lifespan_duration | lifespan을 확인하는 주기 |
Liveliness: 정해진 주기 내 노드 또는 토픽의 생사를 확인
qos_profile = QoSProfile(liveliness=AUTOMATIC, liveliness_lease_duration=Duration(1.0))
| option | explanation |
|---|---|
| liveliness | 자동 또는 매뉴얼로 확인할지를 지정하는 옵션(AUTOMATIC, MANUAL_BY_ALONE, MANUAL_BY_TOPIC) 중 선택 |
| lease_duration | Liveliness을 확인하는 주기 |
4.1.1 RMW QoS Profile
- qos profile은 사전 정의된 모듈을 import하여 사용가능
from rclpy.qos import
(qos_profile_default, # 기본 qos 설정
qos_profile_sensor_data, # 센서 데이터 스트림에 적합한 qos
qos_profile_services_default,
qos_profile_system_default,
QoSProfile)
class Node_class(Node):
def __init__(self):
super().__init__('node_name')
self.publisher_ = self.create_publisher(<msg type>, 'msg_name', qos_profile) - qos profile은 개인 커스텀 가능, 실제 개발시 더 많이 사용됨
from rclpy.qos import
QoSProfile,
QoSReliabilityPolicy,
QoSHistoryPolicy,
QoSDurabilityPolicy
qos_profile = QoSProfile(
reliability=QoSReliabilityPolicy.RELIABLE,
history=QoSHistoryPolicy.KEEP_LAST,
depth=10,
durability=QoSDurabilityPolicy.VOLATILE
)5. Qos Programming
5.1 topic
- topic의 기본 qos설정은 RMW QoS profile과 동일
qos_profile = QoSProfile(
reliability=QoSReliabilityPolicy.RELIABLE,
history=QoSHistoryPolicy.KEEP_LAST,
depth=10,
durability=QoSDurabilityPolicy.VOLATILE
)5.2 service
- service의 경우 기본 QoS를 주로 사용
qos_profile: QosProfile = qos_profile_services_default
RMW_QOS_POLICY_LIVELINESS_SYSTEM_DEFUALT = 0
RMW_QOS_DEADLINE_DEFAULT {0, 0}
RMW_QOS_LIFESPAN_DEFAULT {0, 0}
RMW_QOS_LIVELINESS_LEASE_DURATION_DEFAULT {0, 0}action
-
액션 토픽 : qos_profile_services_default를 기본 설정
-
피드백 퍼블리셔: QoSProfile, rmw_qos_profile_default를 초기값으로 사용
-
액션 상태 퍼블리셔: qos_profile_action_status_default를 기본값으로 사용
-
파이썬의 경우 goal_service_qos_profile, result_service_qos_profile, cancel_service_qos_profile,feedback_pub_qos_profile,status_pub_qos_profile에 대한 기본 설정을 사용
6. 실습
py_pubsub/src/publisher_member_function.py을 수정하여 확인
import rclpy
from rclpy.node import Node
from std_msgs.msg import String
from rclpy.qos import QoSProfile, QoSReliabilityPolicy, QoSHistoryPolicy, QoSDurabilityPolicy
qos_profile = QoSProfile(
reliability=QoSReliabilityPolicy.RELIABLE,
history=QoSHistoryPolicy.KEEP_LAST,
depth=10,
durability=QoSDurabilityPolicy.VOLATILE
)
class MinimalPublisher(Node): # rclpy의 node로부터 상속되는 클래스의 선언
def __init__(self): # 생성자를 정의
super().__init__('minimal_publisher') # 부모 클래스 호출
self.publisher_ = self.create_publisher(String, 'topic', qos_profile=qos_profile) # string 메시지 형식을 사용하여 topic 토픽에 발행하는 발행자 생성
timer_period = 0.5 # seconds, 발행주기 설정
self.timer = self.create_timer(timer_period, self.timer_callback) # 주기적으로 timer_callback() 함수 호출
self.i = 0 # 카운트 변수 초기화
def timer_callback(self):
msg = String() # string 메세지 객체 생성
msg.data = 'Hello World: %d' % self.i # 메시지 데이터 설정
self.publisher_.publish(msg) # publisher를 사용하여 메시지를 'topic' 토픽에 발생
self.get_logger().info('Publishing: "%s"' % msg.data) # 로그에 발행한 메시지 출력
self.i += 1 # 카운트 증가
def main(args=None):
rclpy.init(args=args) # ros2 초기화
minimal_publisher = MinimalPublisher() # minimalpublisher 클래스 인스턴스화
rclpy.spin(minimal_publisher) # 노드실행 및 이벤트 루프 시작
minimal_publisher.destroy_node()
rclpy.shutdown() # ros2 종료
if __name__ == '__main__':
main()6.1 history
- KEEP_LAST
- KEEP_ALL
6.2 Reliable
$ sudo tc qdisc add dev lo root netem loss 45%
$ ros2 run demo_nodes_cpp listener_best_effort
$ ros2 run demo_nodes_cpp talker
6.3 Durability
- TRANSIENT_LOCAL: subcription이 생성되기 전 데이터도 보관(publisher에만 적용가능)
- VOLATILE: subcription이 생성되기 전 데이터는 사용 x
6.4 Deadline
- 정해진 주기 내 데이터의 발신 및 수신이 없는 경우 EventCallback 함수를 실행하는 QoS옵션
ros2 run quality_of_service_demo_py deadline 700 --publish-for 3000 --pause-for 1000 - 700: deadline의 시간 700밀리초
- publish-for 3000: 발행기간을 설정하느 옵션, 3초(3000밀리초)
- pause-for 1000: 1초동안 발행 멈춤
6.5 Lifespan
- 정해진 주기 내 수신되는 데이터만 유효 판정, 이외 데이터는 삭제하는 QoS 옵션
ros2 run quality_of_service_demo_py lifespan 1000 --publish-count 10 --subscribe-after 3000- lifespan 1000: 발행한 메시지의 유효시간 1초
- publish-count 10: 발행할 메시지의 수 지정, 10개
- subscriber-after 3000: 메시지를 3초 후에 구독을 시작
6.6 Liveliness
- 정해진 주기 내 노드 또는 토픽의 생사를 확인하는 QoS 옵션
- AUTOMATIC, MANUAL_BY_NODE, MANUAL_BY_TOPIC 중 택 1하여 자동 또는 매뉴얼로 확인할 지 결정
liveliness_lease_duartion = Duration(seconds=parsed_args.liveliness_lease_duration/1000.0)
liveliness_policy = POLICY_MAP[parsed_args.policy]ros2 run quality_of_service_demo_py liveliness 1000 --kill-publisher-after 2000 --policy AUTOMATIC- liveliness 1000: 1초마다 발행자가 자신이 활성 상태임을 신호
- kill-publisher-after 2000: 2초 후에 발행자 강제 중단
- Automatic: 자동으로 활성 상태를 신호하는 정책
qos_profile = QosProfile(
depth=10,
liveliness=liveliess_policy,
liveliness_lease_duartion = liveliness_lease_duration)
공부하는 거북이