학습 목표
- Topic, Service, Action과 Parameter 중 요구사항에 맞는 Interface를 선택할 수 있다.
- Service 이름·Type·Server·Client·Request·Response의 관계를 설명할 수 있다.
- CLI로 Service 목록, Type과 구조를 조사하고 직접 호출할 수 있다.
- Python Service Server와 비동기 Client를 작성·Build·실행할 수 있다.
- Custom
.srv를 Interface Package에서 생성하고 다른 Package에서 사용할 수 있다. - 값 검증, Timeout, 통신 실패와 업무 거절을 구분해 처리할 수 있다.
- 동기 호출과 Callback Group에서 발생하는 Deadlock을 설명하고 예방할 수 있다.
- 실제 Robot의 Mode 변경, Sensor Zeroing과 Controller 관리에 Service를 안전하게 적용할 수 있다.
1. Service란 무엇인가
ROS 2 Service는 Client가 Request를 보내면 Server가 계산하고 Response를 돌려주는 원격 절차 호출입니다. Topic이 시간에 따라 계속 흐르는 Stream에 적합하다면 Service는 필요할 때 한 번 묻고 짧은 결과를 확인하는 작업에 적합합니다.
Client Server
│ │
│ Request(mode='auto') │
├──────────────────────────────>│
│ 검증·처리
│ Response(success=True) │
│<──────────────────────────────┤
│ │
하나의 Service 이름에는 Server를 하나만 두어야 합니다. 같은 이름에 여러 Server가 존재하면 어느 Server가 Request를 받을지 정의되지 않습니다. 반면 여러 Client가 같은 Server를 사용할 수 있습니다. ROS 2 공식 문서도 Service를 빠르게 반환해야 하는 RPC로 설명하며, 오래 걸리거나 중간 취소가 필요한 작업에는 Action을 권장합니다. ROS 2 Services
“Request 하나에 Response 하나”라는 논리 구조와 “동시에 Client 하나만 가능”하다는 말은 다릅니다. 여러 Client가 Request를 보낼 수 있지만 Server의 실제 동시 처리 여부는 Executor와 Callback Group 구성에 따라 달라집니다.
2. Topic·Service·Action·Parameter 선택
| 요구사항 | 권장 Interface | 이유 |
|---|---|---|
| LiDAR·Camera·Odometry Stream | Topic | 주기적으로 계속 생성되고 여러 Consumer가 받음 |
| Motor Driver Mode 변경 | Service | 짧은 요청 후 수락·거절 결과가 필요함 |
| 목표 지점까지 Navigation | Action | 오래 걸리고 Feedback·Cancel이 필요함 |
| 최대 속도 설정값 유지 | Parameter | 값의 저장·조회·초기 설정이 중요함 |
| 지도 저장 요청 | Service 또는 Action | 구현 시간이 짧으면 Service, 오래 걸리고 취소가 필요하면 Action |
| Emergency Stop | 독립 Safety 계층과 상태·명령 Interface | 일반 Service 응답만 Safety 기능으로 믿으면 안 됨 |
Service의 선택 기준은 특정 숫자 하나가 아니라 다음 질문입니다.
- 결과를 기다려야 하는가?
- 정상 상황에서 빠르게 끝나는가?
- 진행률이 필요한가?
- 중간 취소가 필요한가?
- 값이 장기간 유지되어야 하는가?
계속 흐르는 Data인가? ── Yes → Topic
│ No
지속 설정값인가? ─────── Yes → Parameter
│ No
오래 걸리거나 Cancel·Feedback 필요? ── Yes → Action
│ No
짧게 끝나고 Response 필요? ─────────── Yes → Service
3. Service를 구성하는 여섯 요소
Service Name /set_mode
Service Type lab_interfaces/srv/SetMode
Server Request를 검증하고 Response를 생성
Client Request를 만들고 호출
Request mode, speed_limit
Response accepted, message, previous_mode
Client와 Server가 연결되려면 최종 Service 이름과 Type이 같아야 하며 ROS Graph에서 서로 발견할 수 있어야 합니다. Service 이름은 Topic처럼 Namespace와 Remapping의 영향을 받지만 Topic과는 별도 이름 영역입니다.
Service의 Transport가 연결됐다고 업무 성공이 보장되는 것은 아닙니다.
Transport 성공: Request가 Server에 도착하고 Response를 받음
Business 성공: Server가 Request 내용을 수락하고 작업을 수행함
따라서 Custom Response에는 accepted 또는 success와 사람이 이해할 message를 두는 것이 실용적입니다. 단, 모든 표준 Service가 이 Field를 갖는 것은 아니며 Interface 요구사항에 맞춰 설계해야 합니다.
4. CLI로 Service 관찰하기
먼저 Demo Node를 실행합니다.
ros2 run demo_nodes_py add_two_ints_server
다른 Terminal에서 조사합니다.
# Service 이름 목록
ros2 service list
# 이름과 Type 함께 보기
ros2 service list -t
# 특정 Service Type 확인
ros2 service type /add_two_ints
# 같은 Type을 사용하는 Service 찾기
ros2 service find example_interfaces/srv/AddTwoInts
# Request와 Response 구조 확인
ros2 interface show example_interfaces/srv/AddTwoInts
# YAML Prototype 확인
ros2 interface proto example_interfaces/srv/AddTwoInts
출력 구조는 다음과 같습니다.
int64 a
int64 b
---
int64 sum
직접 호출합니다.
ros2 service call /add_two_ints \
example_interfaces/srv/AddTwoInts \
"{a: 7, b: 35}"
예상 Response:
sum: 42
CLI Option은 배포판에 따라 달라질 수 있으므로 설치 환경에서 확인합니다.
ros2 service --help
ros2 service call --help
ros2 interface show --help
ROS 2 Node가 자동 제공하는 Parameter 관련 Service가 목록에 많이 나타나는 것은 정상입니다. 목록을 지우려 하지 말고 이름과 Type으로 필요한 Service를 찾습니다.
5. 첫 Python Server 만들기
Package를 만듭니다.
mkdir -p ~/ros2_ws/src
cd ~/ros2_ws/src
ros2 pkg create --build-type ament_python service_lab \
--dependencies rclpy example_interfaces
service_lab/service_lab/add_server.py:
import rclpy
from rclpy.node import Node
from example_interfaces.srv import AddTwoInts
class AddServer(Node):
def __init__(self):
super().__init__('add_server')
self.service = self.create_service(
AddTwoInts,
'add_two_ints',
self.on_add,
)
self.get_logger().info('add_two_ints service ready')
def on_add(self, request, response):
response.sum = request.a + request.b
self.get_logger().info(
f'request: {request.a} + {request.b} = {response.sum}'
)
return response
def main(args=None):
rclpy.init(args=args)
node = AddServer()
try:
rclpy.spin(node)
finally:
node.destroy_node()
rclpy.shutdown()
if __name__ == '__main__':
main()
setup.py의 console_scripts에 등록합니다.
'add_server = service_lab.add_server:main',
Build하고 실행합니다.
cd ~/ros2_ws
colcon build --packages-select service_lab --symlink-install
source install/setup.bash
ros2 run service_lab add_server
다른 Terminal에서 CLI로 Server부터 검증합니다.
source ~/ros2_ws/install/setup.bash
ros2 service call /add_two_ints \
example_interfaces/srv/AddTwoInts \
"{a: 10, b: -3}"
6. 첫 비동기 Python Client 만들기
service_lab/service_lab/add_client.py:
import argparse
import rclpy
from rclpy.node import Node
from example_interfaces.srv import AddTwoInts
class AddClient(Node):
def __init__(self):
super().__init__('add_client')
self.client = self.create_client(AddTwoInts, 'add_two_ints')
def send(self, a: int, b: int):
request = AddTwoInts.Request()
request.a = a
request.b = b
return self.client.call_async(request)
def main(args=None):
parser = argparse.ArgumentParser()
parser.add_argument('a', type=int)
parser.add_argument('b', type=int)
known, ros_args = parser.parse_known_args(args=args)
rclpy.init(args=ros_args)
node = AddClient()
try:
if not node.client.wait_for_service(timeout_sec=3.0):
node.get_logger().error('service not available within 3 seconds')
return
future = node.send(known.a, known.b)
rclpy.spin_until_future_complete(node, future, timeout_sec=2.0)
if not future.done():
future.cancel()
node.get_logger().error('response timeout')
return
error = future.exception()
if error is not None:
node.get_logger().error(f'call failed: {error!r}')
return
response = future.result()
node.get_logger().info(f'result: {response.sum}')
finally:
node.destroy_node()
rclpy.shutdown()
if __name__ == '__main__':
main()
setup.py에 추가합니다.
'add_client = service_lab.add_client:main',
실행합니다.
cd ~/ros2_ws
colcon build --packages-select service_lab --symlink-install
source install/setup.bash
# Terminal 1
ros2 run service_lab add_server
# Terminal 2
ros2 run service_lab add_client -- 12 30
call_async()는 즉시 Future를 반환합니다. Future가 완료됐다는 것은 Response를 받았거나 오류로 종료됐다는 의미이므로 done(), exception(), result()를 구분해 확인합니다. 공식 Python Tutorial도 비동기 호출과 Callback 밖의 spin_until_future_complete()를 사용합니다. Python service and client
7. Custom SetMode.srv 설계
실제 Robot Mode를 바꾸는 Service를 만듭니다.
cd ~/ros2_ws/src
ros2 pkg create --build-type ament_cmake lab_interfaces
mkdir -p lab_interfaces/srv
lab_interfaces/srv/SetMode.srv:
string MODE_IDLE=idle
string MODE_MANUAL=manual
string MODE_AUTO=auto
string mode
float32 speed_limit # m/s, 0.0 이상 Hardware 한계 이하
---
bool accepted
string message
string previous_mode
float32 applied_speed_limit
설계 의도는 다음과 같습니다.
- 문자열 상수로 허용 Mode를 공유합니다.
- Server가 Request 값을 다시 검증합니다.
- Transport 성공과 Mode 전환 수락을
accepted로 구분합니다. - 실제 적용된 제한값을 Response로 돌려 Client 추측을 없앱니다.
- 이전 Mode를 반환해 Log와 운영 UI에서 상태 변화를 설명합니다.
이 Service는 상태 전환 “요청”을 처리합니다. 현재 상태를 지속적으로 알리는 기능은 별도 상태 Topic을 함께 사용하는 편이 좋습니다. 새로 참가한 Node가 과거 Service Response를 자동으로 알 수는 없습니다.
8. Custom Interface Build
lab_interfaces/CMakeLists.txt:
cmake_minimum_required(VERSION 3.8)
project(lab_interfaces)
find_package(ament_cmake REQUIRED)
find_package(rosidl_default_generators REQUIRED)
rosidl_generate_interfaces(${PROJECT_NAME}
"srv/SetMode.srv"
)
ament_export_dependencies(rosidl_default_runtime)
ament_package()
lab_interfaces/package.xml 핵심 항목:
<buildtool_depend>ament_cmake</buildtool_depend>
<build_depend>rosidl_default_generators</build_depend>
<exec_depend>rosidl_default_runtime</exec_depend>
<member_of_group>rosidl_interface_packages</member_of_group>
Build와 확인:
cd ~/ros2_ws
rosdep install --from-paths src --ignore-src -r -y
colcon build --packages-select lab_interfaces
source install/setup.bash
ros2 interface show lab_interfaces/srv/SetMode
ros2 interface proto lab_interfaces/srv/SetMode
Service Application Package에도 의존성을 추가합니다.
cd ~/ros2_ws/src/service_lab
# package.xml에 lab_interfaces exec_depend를 추가하거나
# 생성 시 --dependencies에 포함하는 방식을 사용한다.
package.xml:
<exec_depend>lab_interfaces</exec_depend>
9. 검증형 Mode Server
Server Callback은 Request를 검증하고 내부 목표 상태를 원자적으로 바꾼 뒤 빠르게 Response를 돌려줍니다. 실제 Motor 출력은 별도 Control Loop가 담당합니다.
service_lab/service_lab/mode_server.py:
import math
import rclpy
from rclpy.node import Node
from std_msgs.msg import String
from lab_interfaces.srv import SetMode
HARDWARE_MAX_SPEED = 1.5
VALID_MODES = {
SetMode.Request.MODE_IDLE,
SetMode.Request.MODE_MANUAL,
SetMode.Request.MODE_AUTO,
}
class ModeServer(Node):
def __init__(self):
super().__init__('mode_server')
self.mode = SetMode.Request.MODE_IDLE
self.speed_limit = 0.0
self.service = self.create_service(
SetMode, 'set_mode', self.on_set_mode
)
self.state_publisher = self.create_publisher(
String, 'mode_state', 10
)
self.timer = self.create_timer(0.5, self.publish_state)
self.get_logger().info('set_mode service ready')
def reject(self, response, reason):
response.accepted = False
response.message = reason
response.previous_mode = self.mode
response.applied_speed_limit = self.speed_limit
self.get_logger().warning(f'rejected: {reason}')
return response
def on_set_mode(self, request, response):
if request.mode not in VALID_MODES:
return self.reject(
response,
f'unknown mode {request.mode!r}; valid={sorted(VALID_MODES)}',
)
if not math.isfinite(request.speed_limit):
return self.reject(response, 'speed_limit must be finite')
if not 0.0 <= request.speed_limit <= HARDWARE_MAX_SPEED:
return self.reject(
response,
f'speed_limit must be 0.0..{HARDWARE_MAX_SPEED:.1f} m/s',
)
if request.mode == SetMode.Request.MODE_IDLE:
applied_limit = 0.0
else:
applied_limit = request.speed_limit
previous_mode = self.mode
self.mode = request.mode
self.speed_limit = applied_limit
response.accepted = True
response.previous_mode = previous_mode
response.applied_speed_limit = applied_limit
response.message = (
f'{previous_mode} -> {self.mode}, '
f'limit={applied_limit:.2f} m/s'
)
self.get_logger().info(response.message)
return response
def publish_state(self):
msg = String()
msg.data = f'{self.mode}:{self.speed_limit:.2f}'
self.state_publisher.publish(msg)
def main(args=None):
rclpy.init(args=args)
node = ModeServer()
try:
rclpy.spin(node)
finally:
node.destroy_node()
rclpy.shutdown()
실제 Robot에서는 Mode 변경 전에 E-stop 상태, Controller 상태, Sensor Health, Battery와 속도 0 여부 같은 전제조건을 확인합니다. Service Response는 “요청을 수락했다”는 Application 결과이지 Safety 인증이 아닙니다.
10. CLI로 정상·거절 경로 시험
cd ~/ros2_ws
colcon build --packages-select lab_interfaces service_lab --symlink-install
source install/setup.bash
ros2 run service_lab mode_server
정상 Request:
ros2 service call /set_mode lab_interfaces/srv/SetMode \
"{mode: 'auto', speed_limit: 0.6}"
Idle 전환:
ros2 service call /set_mode lab_interfaces/srv/SetMode \
"{mode: 'idle', speed_limit: 0.0}"
잘못된 Mode와 속도:
ros2 service call /set_mode lab_interfaces/srv/SetMode \
"{mode: 'fly', speed_limit: 8.0}"
ros2 service call /set_mode lab_interfaces/srv/SetMode \
"{mode: 'auto', speed_limit: -0.2}"
상태 Topic 확인:
ros2 topic echo /mode_state
정상 경로만 시험하면 검증 Code가 실제로 작동하는지 알 수 없습니다. 최소값, 최대값, 범위 밖, 빈 문자열, NaN 가능성과 반복 호출을 시험합니다.
11. Callback 안에서 비동기 호출하기
Timer나 Subscription Callback에서 다른 Service를 호출해야 한다면 응답을 기다리며 Callback을 막지 말고 Future 완료 처리를 분리합니다.
import rclpy
from rclpy.node import Node
from std_msgs.msg import Bool
from lab_interfaces.srv import SetMode
class SafetyModeClient(Node):
def __init__(self):
super().__init__('safety_mode_client')
self.client = self.create_client(SetMode, 'set_mode')
self.pending_future = None
self.subscription = self.create_subscription(
Bool, 'obstacle_stop', self.on_obstacle, 10
)
def on_obstacle(self, msg):
if not msg.data:
return
if self.pending_future is not None and not self.pending_future.done():
self.get_logger().warning('idle request already pending')
return
if not self.client.service_is_ready():
self.get_logger().error('set_mode service is not ready')
return
request = SetMode.Request()
request.mode = SetMode.Request.MODE_IDLE
request.speed_limit = 0.0
self.pending_future = self.client.call_async(request)
self.pending_future.add_done_callback(self.on_response)
def on_response(self, future):
try:
response = future.result()
except Exception as error:
self.get_logger().error(f'service call failed: {error!r}')
else:
if response.accepted:
self.get_logger().info(response.message)
else:
self.get_logger().error(f'request rejected: {response.message}')
finally:
self.pending_future = None
여기서 중복 Pending Request를 막는 이유는 Sensor Callback이 빠르게 반복될 때 동일 Service 호출이 폭주하는 것을 방지하기 위해서입니다. 다만 장애물 정지는 Service 왕복만 의존하지 않고 독립적인 즉시 정지 경로를 갖춰야 합니다.
12. 동기 호출과 Deadlock
Python의 client.call()은 Response가 올 때까지 현재 Thread를 막습니다. 같은 Executor Thread가 Response 완료 Callback도 실행해야 하는 상황에서 Callback 내부 동기 호출을 하면 다음 순환 대기가 생깁니다.
Timer Callback
└─ client.call()이 Response를 기다림
└─ Response 완료 처리는 Executor가 해야 함
└─ Executor는 현재 Timer Callback이 끝나길 기다림
└─ Deadlock
# 나쁜 예: Callback 안의 동기 호출
def timer_callback(self):
request = SetMode.Request()
response = self.client.call(request) # 멈출 수 있음
ROS 2 공식 가이드는 Python 동기 Service 호출을 권장하지 않으며, Deadlock이 발생해도 Warning이나 Exception이 없을 수 있다고 설명합니다. Synchronous vs. asynchronous clients
안전한 기본 원칙:
- 기본적으로
call_async()를 사용합니다. spin_until_future_complete()는 이미 실행 중인 ROS Callback 안에서 사용하지 않습니다.- Callback 안에서는
add_done_callback()또는 상태 Machine으로 완료를 처리합니다. - Callback은
sleep, Network I/O와 긴 File I/O로 막지 않습니다.
13. MultiThreadedExecutor와 Callback Group
Multi-threaded Executor를 선택했다고 모든 Callback이 자동 병렬 실행되는 것은 아닙니다. 기본 Callback Group은 Mutually Exclusive이므로 같은 Group의 Callback은 동시에 실행되지 않습니다.
from rclpy.callback_groups import MutuallyExclusiveCallbackGroup
from rclpy.executors import MultiThreadedExecutor
self.control_group = MutuallyExclusiveCallbackGroup()
self.service_group = MutuallyExclusiveCallbackGroup()
self.timer = self.create_timer(
0.02,
self.control_loop,
callback_group=self.control_group,
)
self.service = self.create_service(
SetMode,
'set_mode',
self.on_set_mode,
callback_group=self.service_group,
)
# main
executor = MultiThreadedExecutor(num_threads=2)
executor.add_node(node)
executor.spin()
이 구조는 Service Callback과 Control Timer가 다른 Group에서 실행될 수 있게 하지만 공유 상태에 대한 동기화 책임이 생깁니다. mode와 speed_limit을 여러 Thread가 읽고 쓴다면 Lock, Immutable Snapshot 또는 Message Passing 방식으로 Race Condition을 방지해야 합니다.
Callback Group의 구성이 잘못되면 MultiThreadedExecutor도 사실상 Single-threaded처럼 동작하거나 동기 호출에서 Deadlock이 발생할 수 있습니다. Using Callback Groups
14. Timeout·재시도·멱등성
Service Client는 두 가지 대기 시간을 구분해야 합니다.
- Discovery Timeout: Server가 Graph에 나타날 때까지 기다리는 시간
- Response Timeout: Request 전송 후 Response를 기다리는 시간
Timeout이 발생하면 Server가 Request를 처리하지 않았다고 단정할 수 없습니다. Response만 유실되거나 늦었을 수 있습니다. 그래서 무조건 재시도하면 같은 작업이 두 번 실행될 수 있습니다.
Client ── Request(save map) ──> Server
Client <── Response 유실 ────── Server는 이미 저장 완료
Client ── 같은 Request 재시도 > 두 번째 저장 또는 충돌
재시도 가능한 작업은 가능한 한 멱등적으로 설계합니다. 같은 Request를 여러 번 실행해도 최종 상태가 같아야 합니다.
set_mode(auto)는 대체로 멱등적으로 만들 수 있습니다.increment_counter()는 멱등적이지 않습니다.save_map(name, overwrite=False)는 중복을 명확히 거절할 수 있습니다.- 중요한 작업은
request_id를 받아 Server가 처리 이력을 제한 시간 동안 기억하는 방식을 검토합니다.
재시도 정책에는 최대 횟수, Backoff, 재시도 가능한 오류와 운영 Log가 필요합니다. Server의 명시적 거절은 같은 Request를 반복해도 해결되지 않으므로 통신 Timeout과 구분합니다.
15. Namespace와 Remapping
상대 Service 이름을 사용하면 같은 Node를 여러 Robot에 재사용하기 쉽습니다.
# Robot 1 Server
ros2 run service_lab mode_server --ros-args -r __ns:=/robot_01
# Robot 2 Server
ros2 run service_lab mode_server --ros-args -r __ns:=/robot_02
ros2 service list -t | grep set_mode
# /robot_01/set_mode
# /robot_02/set_mode
Client도 Namespace를 맞춥니다.
ros2 run service_lab mode_client --ros-args -r __ns:=/robot_01
Service 이름만 Remap할 수도 있습니다.
ros2 run service_lab mode_server --ros-args \
-r set_mode:=control/set_mode
Source Code에 /robot_01/set_mode 같은 절대 이름을 고정하면 Multi-robot 재사용과 Launch 구성이 어려워집니다.
16. 실제 Robot 적용 사례
16.1 운전 Mode 전환
Operator UI ── /set_mode Request ──> Mode Manager
│
Preconditions 검사
Controller 상태 전환
│
Operator UI <── accepted·reason ─────────┘
Mode Manager ── /mode_state Topic ──> 전체 System
Service는 전환 요청과 즉시 판단 결과를 전달하고 Topic은 현재 Mode를 지속적으로 알립니다. E-stop, Motor Enable과 안전 상태는 위험 분석에 따른 별도 계층으로 구성합니다.
16.2 IMU Bias Zeroing
Robot이 완전히 정지한 상태에서 짧게 내부 Accumulator를 Reset하는 작업은 Service 후보입니다. 하지만 수 초 동안 Sample을 모아 진행률을 보여주거나 움직임이 감지되면 취소해야 한다면 Action이 더 적절합니다.
16.3 Camera Exposure 설정
장기간 유지되고 조회해야 하는 Exposure 값은 Parameter가 자연스럽습니다. 반면 “현재 설정을 Sensor Hardware에 다시 적용하고 성공 여부를 확인”하는 단발 작업은 Service가 될 수 있습니다.
16.4 ros2_control Controller 관리
Controller 목록 조회, Load·Configure·Switch처럼 짧고 결과 확인이 필요한 관리 작업은 Service를 사용합니다. 실제 Controller 전환 중 Hardware 안전 조건, Strictness와 Timeout은 해당 Package의 Interface 문서를 확인합니다.
16.5 지도 저장
작고 빠른 지도 저장은 Service로 처리할 수 있지만 Storage가 느리거나 Cloud Upload, 진행률·취소가 필요하면 Action 또는 Background Job 구조가 적합합니다. “보통 빠르다”가 아니라 Worst-case를 측정해 결정합니다.
17. Service 진단 순서
1단계: Server Process와 Log
ros2 node list
ros2 node info /mode_server
2단계: 이름과 Type
ros2 service list -t
ros2 service type /set_mode
ros2 interface show lab_interfaces/srv/SetMode
3단계: CLI 직접 호출
ros2 service call /set_mode lab_interfaces/srv/SetMode \
"{mode: 'idle', speed_limit: 0.0}"
4단계: Namespace·Domain·Network
echo "$ROS_DOMAIN_ID"
ros2 daemon stop
ros2 daemon start
5단계: Callback 실행 시간과 예외
Server Callback 시작·종료와 처리 시간을 Log로 남기고 Block I/O, Nested Service Call과 Exception을 확인합니다.
| 증상 | 확인할 항목 |
|---|---|
| Service가 목록에 없음 | Server 실행, 최종 Namespace, Domain, Discovery |
| 목록에는 있지만 Client가 못 찾음 | 이름·Type·Overlay Source 차이 |
| CLI는 되지만 Program Client 실패 | Request 생성, Future 처리, Timeout Logic |
| Server Log에 Request가 없음 | 다른 이름·Domain·Network 또는 Type |
| Request Log 후 Response 없음 | 긴 Callback, Exception, Deadlock |
| 응답은 왔지만 작업 실패 | accepted와 이유, Preconditions, Business Rule |
| 반복 호출 때만 이상 | Shared State Race, 중복 Request, 멱등성 문제 |
Topic처럼 무조건
echo한다고 생각하지 마십시오. 최신 ROS 2에는 Service Introspection 기능이 있지만 배포판·구현과 활성화 여부가 관련되므로, 기본 진단은 CLI 직접 호출과 구조화된 Server Log부터 시작하는 것이 이식성이 좋습니다.
18. 흔한 실수와 교정
- 같은 이름에 Server를 두 개 실행한다. Server Owner를 하나로 정합니다.
- 오래 걸리는 Robot 동작을 Service로 만든다. Feedback·Cancel이 필요하면 Action을 사용합니다.
- Callback 안에서
call()또는spin_until_future_complete()로 기다린다. 비동기 Future 완료 처리를 분리합니다. wait_for_service()를 무한 반복한다. 운영 정책에 맞는 Discovery Timeout과 실패 경로를 둡니다.- Timeout이면 작업이 실행되지 않았다고 믿고 즉시 재시도한다. 멱등성과 중복 처리를 설계합니다.
- Server에서 Request 범위를 검증하지 않는다. 모든 외부 입력을 불신하고 NaN까지 검사합니다.
- Transport 성공과 업무 성공을 섞는다. Response의
accepted와 이유를 확인합니다. - Mode 현재 상태를 Service Response로만 알린다. 별도 상태 Topic을 제공합니다.
- MultiThreadedExecutor만 켜면 병렬이라고 생각한다. Callback Group과 공유 상태를 함께 설계합니다.
- Custom
.srv변경 후 일부 Package만 Build·Source한다. 모든 Producer·Consumer와 Overlay를 맞춥니다. - 대용량 Image를 Request에 넣는다. Stream은 Topic으로 전달하고 Service에는 짧은 참조와 작업 조건을 둡니다.
- Emergency Stop을 일반 Service 왕복 하나로 구현한다. 독립 Safety Architecture와 상태 확인을 설계합니다.
19. 실습 체크리스트
- [ ] Demo AddTwoInts Server를 CLI로 호출했다.
- [ ]
service list·type·find와interface show·proto를 사용했다. - [ ] Python Server와 비동기 Client를 Build·실행했다.
- [ ]
SetMode.srv를 Interface Package에서 생성했다. - [ ] 정상값과 범위 밖 Request를 모두 시험했다.
- [ ] Server가 Transport 성공과 Request 거절을 구분해 응답한다.
- [ ] Discovery와 Response Timeout을 각각 처리한다.
- [ ] Callback 내부 Service 호출은
call_async()로 분리했다. - [ ] 중복 Pending Request를 제한했다.
- [ ] 재시도 대상 작업의 멱등성을 검토했다.
- [ ] Namespace 두 개에서 같은 Server Code를 실행했다.
- [ ] Mode 상태를 별도 Topic으로 확인했다.
- [ ] 실제 구동 전 Simulation과 Motor Disable 상태에서 시험했다.
20. 정리
Service는 짧은 Request에 계산 결과나 수락 여부가 필요한 ROS 2 RPC입니다. 잘 만든 Service는 Request가 완결되어 있고, Server가 입력을 검증하며, Response가 업무 성공과 실패 이유를 명확히 표현합니다.
Client는 call_async()를 기본으로 사용하고 Discovery Timeout, Response Timeout, Future Exception과 Server 거절을 구분해야 합니다. Callback 내부에서 동기 대기를 만들지 말고, Multi-threading을 사용한다면 Callback Group과 공유 상태까지 함께 설계하십시오.
실제 Robot에서는 Service 하나만 보지 않습니다. 전환 요청은 Service, 지속 상태는 Topic, 장시간 작업은 Action, 유지 설정은 Parameter, 생명·장비 안전은 독립 Safety 계층이 담당하도록 전체 구조를 나눠야 합니다.
- Service서비스
- Client의 Request를 Server가 처리하고 하나의 Response를 반환하는 ROS 2 원격 절차 호출 Interface입니다.
- Service Server서비스 서버
- 특정 Service 이름과 Type의 Request를 받아 계산하고 Response를 생성하는 ROS Entity입니다.
- Service Client서비스 클라이언트
- Service Server를 발견하고 Type에 맞는 Request를 보내 Response를 받는 ROS Entity입니다.
- Request요청
- Client가 Server에 수행할 작업과 입력값을 전달하는 Service Interface의 앞쪽 Message입니다.
- Response응답
- Server가 처리 결과와 출력값을 Client에 돌려주는 Service Interface의 뒤쪽 Message입니다.
- RPC원격 절차 호출
- 다른 Process나 Computer의 기능을 Request와 Response로 호출하는 통신 방식입니다.
- Future비동기 결과 객체
- 아직 완료되지 않은 Service 호출의 완료 상태, 예외와 결과를 나중에 확인하는 객체입니다.
- Asynchronous Call비동기 호출
- 호출 Thread를 Response 대기로 막지 않고 Future를 즉시 반환하는 Service 호출 방식입니다.
- Synchronous Call동기 호출
- Response가 도착할 때까지 호출 Thread를 Block하는 방식으로 Callback 안에서는 Deadlock 위험이 큽니다.
- Deadlock교착 상태
- Callback과 Executor처럼 둘 이상의 실행 흐름이 서로의 완료를 기다려 영원히 진행하지 못하는 상태입니다.
- Discovery Timeout발견 제한시간
- Client가 일치하는 Service Server가 Graph에 나타나기를 기다리는 최대 시간입니다.
- Response Timeout응답 제한시간
- Request를 보낸 뒤 Response 완료를 기다리는 최대 시간입니다.
- Business Rejection업무 규칙 거절
- 통신은 성공했지만 Server가 값 범위나 상태 전제조건 때문에 Request 수행을 수락하지 않은 결과입니다.
- Idempotency멱등성
- 같은 Request를 여러 번 수행해도 한 번 수행한 것과 최종 상태가 같은 성질입니다.
- Retry재시도
- 일시적 통신 실패 가능성을 고려해 횟수와 간격을 제한하여 Request를 다시 보내는 정책입니다.
- Backoff재시도 대기 증가
- 연속 실패 시 재시도 간격을 늘려 Server와 Network 부하의 폭주를 줄이는 방식입니다.
- Callback Group콜백 그룹
- Executor가 관련 Callback의 상호 배타 또는 재진입 병렬 실행을 결정하는 Group입니다.
- Mutually Exclusive상호 배타 실행
- 한 Callback이 실행되는 동안 같은 Group의 다른 Callback이 동시에 실행되지 않도록 하는 규칙입니다.
- Reentrant재진입 가능 실행
- 같은 Callback의 여러 Instance를 포함해 같은 Group의 Callback이 병렬 실행될 수 있게 하는 규칙입니다.
- Precondition사전 조건
- Mode 전환이나 Hardware 작업을 수락하기 전에 반드시 만족해야 하는 Robot 상태와 안전 조건입니다.
연습 문제
- Service가 Topic보다 적합한 요구사항 세 가지를 쓰세요.
- 같은 Service 이름에 Server를 두 개 두면 안 되는 이유는 무엇인가요?
- Transport 성공과 업무 성공의 차이를
SetMode예로 설명하세요. ros2 service list,type,find,interface show,call의 역할을 각각 설명하세요..srv의---위와 아래에는 무엇이 정의되나요?call_async()가 반환하는 Future에서done,exception,result를 구분해야 하는 이유는 무엇인가요?- Discovery Timeout과 Response Timeout의 차이를 설명하세요.
- Callback 안에서 동기
call()을 사용할 때 Deadlock이 생기는 과정을 설명하세요. - MultiThreadedExecutor를 사용해도 Callback이 병렬로 실행되지 않을 수 있는 이유는 무엇인가요?
- Server가
speed_limit에서 범위뿐 아니라 NaN·Infinity도 검사해야 하는 이유는 무엇인가요? - Mode 변경 Service와 Mode 상태 Topic을 함께 사용해야 하는 이유는 무엇인가요?
- Response Timeout 후 같은 Request를 즉시 재시도할 때 중복 실행이 가능한 이유는 무엇인가요?
- 멱등적인 Service와 멱등적이지 않은 Service의 예를 하나씩 쓰세요.
- IMU Calibration을 Service 대신 Action으로 만들어야 하는 조건은 무엇인가요?
- 실제 Robot에서
/set_mode가 응답하지 않을 때 이름·Type부터 Callback까지 진단하는 순서를 설명하세요.
COMMUNITY
강의 댓글
질문과 학습 경험을 함께 나눠보세요.댓글을 불러오는 중입니다.