실행기와 콜백 그룹 - 동시에 여러 일 하기

이 장에서 배우는 것

두리는 주행 중에도 배터리 상태를 읽고, 센서 점검 요청에 응답하고, 자신의 상태를 주기적으로 갱신해야 한다. 앞 장에서 통신 품질을 조정했다면 이제는 도착한 일을 언제, 어떤 순서로 실행할지 정해야 한다. 메시지가 잘 도착해도 긴 콜백이 실행 시간을 차지하면 다른 작업의 처리가 늦어진다.

이 장에서는 실행기(executor)와 콜백 그룹(callback group)을 조합해 서로 독립적인 작업이 진행되도록 만든다. 먼저 순수 Python으로 실행 규칙을 재현하고, 같은 원리를 ROS 2 Jazzy의 rclpy 프로그램에 적용한다.

• SingleThreadedExecutor와 MultiThreadedExecutor의 실행 차이를 설명한다.
• 상호배제 그룹과 재진입 그룹을 작업의 상태 공유 방식에 맞춰 선택한다.
• 기본 콜백 그룹 때문에 여러 스레드를 쓰고도 콜백이 직렬 실행되는 이유를 확인한다.
• 서비스 응답을 기다리다가 발생하는 교착을 분석하고 비동기 요청으로 고친다.

문제 상황

실외 시험 중 두리의 상태 표시가 간헐적으로 늦어진다. 상태 갱신 콜백은 0.05초마다 실행되도록 설정했지만, 센서 점검 서비스를 호출하면 다음 갱신까지 평소보다 긴 시간이 걸린다. 점검 콜백 안에는 장치 응답을 기다리는 작업이 있고, 시험 프로그램에서는 이 대기를 time.sleep(0.2)로 대신한다.

타이머 주기를 짧게 바꾸어도 문제가 해결되지는 않는다. 타이머는 실행할 일이 준비되었음을 알리는 장치다. 콜백을 실제로 실행할 자원이 비어 있어야 작업을 시작할 수 있다. 하나의 실행 스레드가 점검 콜백에서 기다리는 동안에는 준비된 상태 갱신 콜백도 차례를 기다린다.

작업자 스레드를 세 개로 늘렸는데도 결과가 비슷할 수 있다. 콜백을 만들 때 그룹을 지정하지 않았다면 같은 노드의 기본 그룹에 들어간다. 이 그룹은 상호배제 방식이므로 한 콜백이 끝나기 전에는 같은 그룹의 다른 콜백이 시작하지 못한다. 스레드 수와 콜백의 동시 실행 허용 여부를 함께 살펴야 하는 이유다.

실행기는 준비된 콜백을 실행한다

실행기는 자신에게 등록된 노드에서 처리할 일을 찾는다. 구독 메시지, 타이머 만료, 서비스 요청이나 응답 등이 처리할 일을 만든다. 실행기는 해당 작업이 준비되었고 콜백 그룹이 실행을 허용하는지 확인한 뒤 콜백을 실행한다. 노드를 만들기만 해서는 이러한 처리가 시작되지 않는다.

SingleThreadedExecutor는 spin을 호출한 스레드에서 콜백을 실행한다. 한 콜백이 반환하기 전에는 그 실행기가 다른 콜백을 실행하지 못한다. 상태 변경의 흐름을 따라가기 쉽고, 실행기 안의 일반적인 콜백끼리 동시에 같은 값을 고칠 가능성도 줄어든다. 다만 콜백 안의 긴 계산이나 대기는 뒤따르는 작업의 지연으로 이어진다.

MultiThreadedExecutor는 작업자 스레드 풀에서 콜백을 실행한다. 한 작업자가 장치 응답을 기다리는 동안 다른 작업자가 상태 갱신을 실행할 수 있다. 예제에서는 num_threads=3으로 작업자 수를 명시한다. 스레드 수를 생략했을 때의 환경별 선택에 의존하지 않고 실험 조건을 드러내기 위해서다.

여러 스레드가 있다고 해서 준비된 콜백이 모두 한꺼번에 실행되는 것은 아니다. 사용 가능한 작업자 수, 그룹의 허용 규칙, 운영체제의 스케줄링이 영향을 준다. 또한 타이머의 주기는 실행 완료 기한을 보장하지 않는다. 0.05초 주기는 매번 정확히 0.05초 간격으로 콜백이 시작된다는 뜻이 아니다.

Python에서는 동시성과 CPU 병렬 처리도 구분해야 한다. 일반적인 CPython 환경의 전역 인터프리터 잠금(GIL)은 여러 스레드가 Python 바이트코드를 동시에 실행하는 일을 제한한다. 따라서 순수 Python 계산을 여러 콜백으로 나누었다고 계산 시간이 스레드 수에 비례해 줄어들지는 않는다. 반면 입출력 대기나 time.sleep처럼 잠금을 놓는 작업은 다른 스레드의 진행 여지를 만든다.

실행기 선택은 콜백이 시간을 쓰는 방식에서 출발한다

실행기	콜백 실행 자원	검토할 상황	주의점
SingleThreadedExecutor	spin을 호출한 스레드	콜백이 짧고 실행 흐름이 단순할 때	긴 콜백이 다른 작업을 늦춘다
MultiThreadedExecutor	여러 작업자 스레드	독립적인 콜백이나 대기 작업이 있을 때	그룹이 허용해야 겹쳐 실행된다
두 방식의 공통점	준비된 작업을 처리한다	타이머·구독·서비스를 함께 운영할 때	주기와 처리 기한을 보장하지 않는다

콜백 그룹은 동시에 실행해도 되는 범위를 정한다

상호배제 콜백 그룹(MutuallyExclusiveCallbackGroup)은 같은 그룹의 콜백을 한 번에 하나만 실행하도록 제한한다. 타이머와 구독이 같은 상태를 함께 수정한다면 둘을 같은 상호배제 그룹에 두어 실행이 겹치지 않게 만들 수 있다. 다른 상호배제 그룹에 속한 콜백끼리는 작업자가 충분할 때 겹쳐 실행될 수 있다.

재진입 콜백 그룹(ReentrantCallbackGroup)은 그룹 자체에서 콜백의 동시 실행을 막지 않는다. 서로 다른 콜백뿐 아니라 같은 콜백의 여러 실행도 겹칠 수 있다. 예를 들어 타이머 콜백의 처리 시간이 주기보다 길면 이전 실행이 끝나기 전에 다음 실행이 시작될 수 있다. 호출별 데이터가 독립적이거나 공유 상태를 별도로 보호할 수 있을 때 선택한다.

그룹을 지정하지 않은 엔티티는 노드의 기본 그룹에 들어간다. rclpy 노드의 기본 그룹은 상호배제 그룹이다. 따라서 기본 그룹만 사용하는 노드의 콜백은 MultiThreadedExecutor에 등록해도 서로 겹쳐 실행되지 않는다. 다만 같은 실행기에 등록한 다른 노드의 콜백까지 모두 직렬화한다는 뜻은 아니다.

다음 그림은 준비된 두 작업에 실행 자원과 그룹 규칙을 적용한 개념도다. 점검 작업이 진행 중일 때 상태 갱신이 시작될 수 있는지가 핵심이다. 실제 실행기의 선택 순서를 예측하는 시간표는 아니다.

그룹 객체는 self.check_group처럼 노드 속성에 보관한다. 그 뒤 create_timer나 create_service의 callback_group 인수로 전달한다. 그룹을 만드는 일과 그 그룹에 엔티티를 배치하는 일은 별개다. 객체만 만들고 전달하지 않으면 실행 규칙이 달라지지 않는다.

클라이언트도 그룹을 가진다. 서비스 응답이 네트워크를 통해 도착하는 것과 실행기가 그 응답을 처리하여 요청의 완료 상태를 갱신하는 것은 구분해야 한다. 눈에 보이는 타이머 콜백만 분리하고 클라이언트의 응답 처리 경로를 놓치면 기다림이 풀리지 않을 수 있다.

그룹은 모든 스레드에 적용되는 메모리 잠금이 아니다. 같은 상호배제 그룹의 콜백은 서로 겹치지 않지만, 다른 그룹의 콜백이나 직접 만든 스레드는 같은 객체에 접근할 수 있다. 공유 상태의 보호 범위를 정할 때는 객체를 읽고 쓰는 모든 실행 경로를 확인해야 한다.

기다림이 순환하면 교착이 생긴다

교착(deadlock)은 작업들이 서로가 끝내야 가능한 일을 기다리면서 진행하지 못하는 상황이다. 두리의 타이머 콜백 A가 동기식 서비스 호출을 수행한다고 가정한다. A와 클라이언트는 같은 상호배제 그룹에 있다. 서비스 서버를 다른 프로세스에서 실행하더라도 클라이언트 쪽 응답 처리는 두리의 실행기가 맡는다.

A는 응답 처리가 완료되어야 반환한다. 그런데 응답을 처리하는 작업 B는 A가 그룹을 놓아야 실행될 수 있다. A는 B를 기다리고, B는 A가 반환하기를 기다린다. 남는 작업자 스레드가 있어도 그룹의 실행 제한을 통과하지 못하므로 스레드를 추가하는 것으로 해결되지 않는다.

다른 그룹으로 분리해도 SingleThreadedExecutor의 한 실행 스레드를 동기 대기로 붙잡으면 응답 처리 기회를 만들기 어렵다. 동기 대기를 유지하려면 그룹과 작업자 여유를 함께 검토해야 한다. 이 장의 완성 코드는 그 조건에 기대지 않고, 콜백이 요청을 보낸 뒤 바로 반환하도록 작성한다.

비동기 요청은 결과를 나중에 확인할 수 있는 Future 객체를 반환한다. 여기서는 짧은 타이머가 Future.done()을 확인하고 완료된 경우에만 결과를 읽는다. 완료되지 않았으면 곧바로 반환하므로 다른 작업이 실행될 수 있다. 요청 중인 Future를 하나만 보관하여 이전 요청이 끝나기 전에 다음 요청을 쌓지도 않는다.

rclpy의 Future.result()를 일반적인 블로킹 대기 함수로 생각해서는 안 된다. 완료 전 호출에 의존하지 말고 done()을 먼저 확인하거나 완료 콜백을 사용해야 한다. 또한 콜백 안에서 spin_until_future_complete를 중첩 호출하는 방식은 그룹과 실행기 문제를 감춘 채 남길 수 있다. 이 예제처럼 실행기 구동은 바깥에 두고, 콜백은 작업의 시작과 완료 확인을 나누는 편이 흐름을 이해하기 쉽다.

완성 코드

첫 프로그램은 표준 라이브러리만 사용하는 이산 시간 모형이다. 실제 스레드를 만들지 않고, 작업자 수와 그룹 규칙에 따라 작업의 시작 시점을 계산한다. 실행마다 결과가 같으므로 개념을 확인하기 좋다. 실제 ROS 실행기의 대기 구조나 공정성, 타이머 동작을 재현하는 프로그램은 아니다.

executor_model.py

from dataclasses import dataclass


@dataclass(frozen=True)
class Job:
    name: str
    group: str
    duration: int


def simulate(
    title: str,
    workers: int,
    shared_group: bool,
    reentrant: bool,
) -> None:
    pending = [
        Job("점검", "A", 3),
        Job("상태", "A" if shared_group else "B", 1),
    ]
    active: list[tuple[Job, int]] = []
    tick = 0
    print(f"[{title}]")

    while pending or active:
        for job, finish in active[:]:
            if finish == tick:
                print(f"t={tick} {job.name} 종료")
                active.remove((job, finish))

        busy_groups = {job.group for job, _ in active}
        for job in pending[:]:
            if len(active) >= workers:
                break
            if not reentrant and job.group in busy_groups:
                continue
            pending.remove(job)
            active.append((job, tick + job.duration))
            busy_groups.add(job.group)
            print(f"t={tick} {job.name} 시작")

        if not pending and not active:
            break
        tick += 1

    print(f"소요={tick}")


def main() -> None:
    simulate("작업자 1 / 그룹 분리", 1, False, False)
    simulate("작업자 2 / 같은 상호배제 그룹", 2, True, False)
    simulate("작업자 2 / 상호배제 그룹 분리", 2, False, False)
    simulate("작업자 2 / 같은 재진입 그룹", 2, True, True)


if __name__ == "__main__":
    main()

duri_executor.py

ROS 프로그램은 한 노드 안에 점검 서버와 클라이언트를 함께 둔다. 외부 장치 없이 요청·응답 흐름을 관찰하기 위한 배치다. 점검 서버는 0.2초를 기다리고 응답한다. 상태 갱신과 요청 진행 확인은 같은 상호배제 그룹에서 짧게 실행하며, 서버와 클라이언트는 각각 별도의 상호배제 그룹을 사용한다.

상태 갱신 횟수는 실행 환경에 따라 달라지므로 출력하지 않는다. 정상 응답을 세 번 받으면 정해진 완료 문장만 출력한다. 이 서비스는 실제 센서나 통신 장치를 검사하지 않으며, 배달 로봇의 점검 대기 시간을 모사한다.

import threading
import time

import rclpy
from rclpy.callback_groups import MutuallyExclusiveCallbackGroup
from rclpy.executors import (
    ExternalShutdownException,
    MultiThreadedExecutor,
)
from rclpy.node import Node
from std_srvs.srv import Trigger


class DuriExecutor(Node):
    def __init__(self) -> None:
        super().__init__("duri_executor")

        self.control_group = MutuallyExclusiveCallbackGroup()
        self.server_group = MutuallyExclusiveCallbackGroup()
        self.client_group = MutuallyExclusiveCallbackGroup()

        self.finished = threading.Event()
        self.pending = None
        self.completed = 0
        self.status_ticks = 0

        self.server = self.create_service(
            Trigger,
            "duri_check",
            self.check_device,
            callback_group=self.server_group,
        )
        self.client = self.create_client(
            Trigger,
            "duri_check",
            callback_group=self.client_group,
        )
        self.status_timer = self.create_timer(
            0.05,
            self.update_status,
            callback_group=self.control_group,
        )
        self.request_timer = self.create_timer(
            0.05,
            self.advance_request,
            callback_group=self.control_group,
        )

    def check_device(self, _request, response):
        time.sleep(0.2)
        response.success = True
        response.message = "점검 완료"
        return response

    def update_status(self) -> None:
        self.status_ticks += 1

    def advance_request(self) -> None:
        if self.finished.is_set():
            return

        if self.pending is not None:
            if not self.pending.done():
                return

            response = self.pending.result()
            if response is None or not response.success:
                raise RuntimeError("점검 응답 실패")

            self.pending = None
            self.completed += 1
            if self.completed == 3:
                self.request_timer.cancel()
                self.finished.set()
                return

        if self.client.service_is_ready():
            self.pending = self.client.call_async(Trigger.Request())


def main(args=None) -> None:
    rclpy.init(args=args)
    node = DuriExecutor()
    executor = MultiThreadedExecutor(num_threads=3)
    executor.add_node(node)

    try:
        while rclpy.ok() and not node.finished.is_set():
            executor.spin_once(timeout_sec=0.1)
        if node.finished.is_set():
            print("두리: 점검 3회 완료")
    except (KeyboardInterrupt, ExternalShutdownException):
        pass
    finally:
        executor.shutdown(wait_for_threads=True)
        node.destroy_node()
        if rclpy.ok():
            rclpy.shutdown()


if __name__ == "__main__":
    main()

줄별 해설

모형에서 실행 허용 여부를 계산하는 부분

Job의 name은 출력 이름이고 group은 그룹 식별자다. duration은 작업이 실행 자원을 점유하는 모형상의 시간이다. frozen=True로 지정했으므로 생성된 작업의 속성을 실행 도중 바꾸지 않는다. 시간 단위는 초가 아니라 정수 눈금이다.

pending에는 아직 시작하지 않은 두 작업을 넣는다. shared_group이 참이면 둘 다 A에 속하고, 거짓이면 상태 작업만 B에 속한다. active의 각 원소는 실행 중인 작업과 종료 시각이다. 작업이 끝나기 전에는 해당 작업자 자리를 계속 차지한다.

while 안의 첫 반복문은 현재 시각에 끝나는 작업을 제거한다. 그 다음 busy_groups를 계산하여 아직 실행 중인 작업의 그룹을 모은다. 종료 처리를 먼저 하므로 점검이 t=3에 끝나면 같은 그룹의 상태 작업도 t=3에 시작할 수 있다.

len(active)와 workers를 비교하는 줄은 작업자 수의 제한을 적용한다. 이어지는 그룹 검사는 상호배제 제한을 적용한다. reentrant가 참이면 그룹이 사용 중이어도 새 작업을 허용한다. 두 검사는 서로 독립적인 제한이라는 점이 중요하다.

pending[:]와 active[:]는 목록의 얕은 복사본이다. 복사본을 순회하면서 원래 목록에서 작업을 지우므로 순회 중 원소가 빠져 다음 원소를 건너뛰는 문제를 피한다. 같은 시각에 여러 작업이 가능하면 이 모형은 입력 목록 순서로 시작한다. 이 순서는 설명을 위해 정한 모형의 규칙이다.

ROS 노드의 작업을 나누는 부분

세 그룹을 self의 속성에 저장한 줄은 실행 정책을 노드 수명 동안 유지한다. control_group에는 상태 타이머와 요청 타이머가 들어간다. 둘 다 짧게 실행하고 요청 상태를 순차적으로 관리하므로 같은 상호배제 그룹에 둔다. server_group은 느린 점검을 담당하고 client_group은 응답 처리 경로를 분리한다.

finished는 작업자 콜백이 설정하고 바깥 실행 루프가 읽는 스레드 간 완료 신호다. 반면 pending과 completed는 요청 타이머 콜백만 변경한다. status_ticks도 상태 타이머만 변경한다. 불필요한 공유를 줄여 각 상태를 누가 수정하는지 드러낸다.

create_service와 create_client는 같은 서비스 이름과 Trigger 형식을 사용한다. Trigger.Request에는 이 예제에서 채울 입력 필드가 없다. 서버는 전달받은 응답 객체에 성공 여부와 메시지를 넣고 반환한다. time.sleep이 실행되는 동안 서버 그룹은 계속 사용 중이지만, 다른 그룹의 콜백은 남는 작업자를 사용할 수 있다.

advance_request의 첫 조건은 완료 후 추가 요청을 막는다. pending이 있으면 아직 끝나지 않은 요청인지 확인한다. 완료 전에는 바로 반환하고, 완료 후에만 result()를 호출한다. 요청에서 예외가 발생했다면 result()에서 예외가 전달될 수 있으므로 성공 횟수를 먼저 늘리지 않는다.

응답을 확인한 뒤 pending을 비우고 completed를 증가시킨다. 세 번째 성공에서는 요청 타이머를 취소하고 완료 신호를 설정한다. 그 전까지는 서비스가 준비된 경우 다음 요청을 보낸다. 따라서 같은 클라이언트에서 여러 점검 요청을 동시에 누적하지 않는다.

spin_once는 바깥의 한 스레드에서만 호출한다. MultiThreadedExecutor는 준비된 콜백을 작업자에게 맡긴다. timeout_sec=0.1은 바깥 루프가 완료 여부를 다시 확인할 기회를 주는 대기 설정이며, 콜백 실행 시간의 제한이 아니다. 종료 시에는 작업자 처리가 끝나기를 기다린 뒤 노드와 ROS 문맥을 정리한다.

실행 결과

파일은 UTF-8로 저장한다. 첫 프로그램은 Python 3.9 이상에서 실행할 수 있으며 ROS 설치가 필요 없다. Jazzy의 일반적인 Python 환경에서도 같은 결과가 나온다. 다음 컴파일 명령은 경고를 오류로 취급하여 두 파일의 문법을 검사한다. py_compile은 ROS 모듈을 가져와 실행하지 않으므로 ROS가 없는 환경에서도 이 검사는 가능하다.

python3 -W error -m py_compile executor_model.py duri_executor.py
python3 executor_model.py

컴파일이 성공하면 별도 출력이 없다. 모형의 예상 표준 출력은 다음과 같다.

[작업자 1 / 그룹 분리]
t=0 점검 시작
t=3 점검 종료
t=3 상태 시작
t=4 상태 종료
소요=4
[작업자 2 / 같은 상호배제 그룹]
t=0 점검 시작
t=3 점검 종료
t=3 상태 시작
t=4 상태 종료
소요=4
[작업자 2 / 상호배제 그룹 분리]
t=0 점검 시작
t=0 상태 시작
t=1 상태 종료
t=3 점검 종료
소요=3
[작업자 2 / 같은 재진입 그룹]
t=0 점검 시작
t=0 상태 시작
t=1 상태 종료
t=3 점검 종료
소요=3

앞의 두 경우는 제한의 원인이 다르지만 결과가 같다. 첫 경우에는 작업자가 하나여서, 둘째 경우에는 그룹이 같아서 상태 작업이 기다린다. 뒤의 두 경우는 작업자가 둘이고 그룹도 실행을 허용하므로 상태 작업이 먼저 끝난다.

ROS 프로그램은 Jazzy와 std_srvs를 사용할 수 있는 셸에서 실행한다. Linux에서는 설치 환경을, macOS에서는 준비한 Jazzy 작업공간의 환경을 먼저 불러온다. 설치 위치가 환경마다 다르므로 여기서는 환경 설정이 끝난 셸을 전제로 한다. 같은 서비스 이름을 가진 다른 서버가 실행 중이지 않은 상태에서 다음 명령을 사용한다.

python3 duri_executor.py

세 번의 요청이 정상 처리되었을 때 프로그램이 출력하는 문장은 다음과 같다. 실행을 중단하면 완료 문장은 출력하지 않는다.

두리: 점검 3회 완료

이 완료 문장만으로 상태 타이머의 지연이 줄었다고 판단해서는 안 된다. 출력은 요청·응답 흐름의 완료만 나타낸다. 콜백 시작 시각을 따로 기록하면 실행기와 그룹 조합에 따른 차이를 관찰할 수 있지만, 그 값은 모형처럼 고정되어 있지 않다. 또한 ROS가 없는 환경의 문법 검사는 실제 통신과 스케줄링 검증을 대신하지 않는다.

실무에서 자주 틀리는 것

실행기만 바꾸고 그룹을 지정하지 않는다

다음은 노드 초기화 내부의 잘못된 배치 예다. 두 타이머가 모두 기본 상호배제 그룹에 속한다. MultiThreadedExecutor를 사용하더라도 두 콜백은 겹쳐 실행되지 않는다.

self.slow_timer = self.create_timer(1.0, self.slow_check)
self.fast_timer = self.create_timer(0.05, self.update_status)

두 콜백이 공유 상태를 동시에 수정하지 않는지 확인한 다음 그룹을 분리한다. 느린 작업과 짧은 상태 갱신을 분리하려는 의도가 코드에 드러난다.

self.slow_group = MutuallyExclusiveCallbackGroup()
self.fast_group = MutuallyExclusiveCallbackGroup()

self.slow_timer = self.create_timer(
    1.0, self.slow_check, callback_group=self.slow_group
)
self.fast_timer = self.create_timer(
    0.05, self.update_status, callback_group=self.fast_group
)

타이머 안에서 같은 그룹의 서비스 완료를 기다린다

다음 코드는 타이머와 클라이언트가 같은 상호배제 그룹일 때 진행을 막을 수 있다. 서버가 응답을 보냈더라도 클라이언트 쪽 처리가 실행될 기회를 얻지 못한다.

def on_timer(self):
    response = self.client.call(Trigger.Request())
    self.last_ok = response.success

노드 초기화에서 self.pending = None을 설정하고, 다음처럼 요청과 완료 확인을 나눈다. 이 조각은 주기적으로 요청을 반복하는 형태다. 완성 프로그램은 여기에 성공 횟수와 종료 조건을 더한다.

def on_timer(self):
    if self.pending is None:
        if self.client.service_is_ready():
            self.pending = self.client.call_async(Trigger.Request())
        return

    if not self.pending.done():
        return

    response = self.pending.result()
    self.pending = None
    if response is not None:
        self.last_ok = response.success

비동기 호출 뒤에 긴 반복문으로 done()을 기다리면 콜백을 반환하지 않는다는 문제가 다시 생긴다. 핵심은 함수 이름의 async가 아니라, 실행기가 다른 작업을 처리하도록 콜백을 짧게 끝내는 것이다. 실제 시스템에서는 서비스가 준비되지 않거나 응답이 오래 걸릴 때의 기한과 오류 처리도 추가해야 한다.

재진입 그룹이면 공유 상태도 안전하다고 생각한다

다음은 같은 상태를 여러 콜백이 수정할 때의 잘못된 패턴이다. 값을 읽고 새 값을 쓰는 사이에 다른 실행이 끼어들면 한쪽의 갱신이 사라질 수 있다. GIL을 응용 프로그램의 상태 보호 규칙으로 사용해서는 안 된다.

def mark_checked(self):
    previous = self.checked_count
    self.checked_count = previous + 1

서로 겹칠 필요가 없는 콜백이라면 같은 상호배제 그룹에 배치하는 것이 간단하다. 여러 그룹에서 접근해야 한다면 threading.Lock으로 짧은 상태 변경 구간을 보호할 수 있다. 다음 초기화는 콜백 등록 전에 수행하며, 같은 값을 읽거나 수정하는 다른 경로도 동일한 잠금 규칙을 따라야 한다.

# __init__ 내부
self.state_lock = threading.Lock()
self.checked_count = 0

# 클래스의 메서드
def mark_checked(self):
    with self.state_lock:
        self.checked_count += 1

잠금을 잡은 채 서비스 응답을 기다리면 응답을 처리하는 콜백이 같은 잠금을 필요로 할 때 또 다른 대기 순환이 생길 수 있다. 잠금 구간에는 필요한 상태 변경만 넣는다. 그룹 분리는 실행 가능성을 정하고, 잠금은 공유 상태의 일관성을 지킨다는 역할 차이를 유지한다.

한눈에 보기

실행 자원과 그룹 규칙을 함께 확인하면 대기 원인을 구분할 수 있다

조건	동시 실행 가능성	두리에서의 의미
단일 작업자, 서로 다른 그룹	콜백을 하나씩 실행한다	느린 점검 중 상태 갱신이 기다린다
여러 작업자, 같은 상호배제 그룹	그룹 안에서는 하나씩 실행한다	기본 그룹만 쓰면 분리 효과가 없다
여러 작업자, 서로 다른 그룹	작업자가 남으면 겹칠 수 있다	점검 대기 중 상태를 갱신할 수 있다
여러 작업자, 같은 재진입 그룹	같은 콜백도 겹칠 수 있다	공유 상태와 중복 실행을 검토한다
콜백 안의 동기 응답 대기	응답 처리 경로에 따라 교착할 수 있다	비동기 요청 후 콜백을 반환한다

선택의 출발점은 스레드 수가 아니라 작업의 관계다. 어떤 콜백이 오래 기다리는지, 어떤 콜백끼리 상태를 공유하는지, 기다리는 결과를 누가 처리하는지 먼저 표시한다. 그 뒤 그룹을 나누고 필요한 작업자 수를 정한다.

API의 세부 동작은 Jazzy rclpy 실행기와 콜백 API 및 Jazzy 콜백 그룹 안내에서 확인할 수 있다. 다음 장에서는 이렇게 처리한 센서 데이터가 어느 좌표계를 기준으로 하는지 살펴본다.

연습 문제

1. 순수 Python 모형에서 점검의 duration을 5로 바꾼다. 네 경우의 전체 소요 시간과 상태 작업의 시작 시각을 실행 전에 예상한다.
2. 완성 ROS 코드의 실행기를 SingleThreadedExecutor로 바꾸면 같은 그룹의 동기 호출과 같은 교착이 발생하는지 설명한다. 점검이 진행되는 동안 상태 타이머에는 어떤 변화가 생기는지도 적는다.
3. 모형의 두 작업을 모두 A 그룹에 두고 작업자를 4개로 늘린다. 상호배제 설정과 재진입 설정의 차이를 설명한다. 재진입 설정을 실제 누적 카운터 콜백에 적용하려면 무엇을 확인해야 하는지도 적는다.
4. 완성 코드의 pending 검사 부분을 없애고 타이머가 실행될 때마다 call_async를 호출한다고 가정한다. 점검 서버의 처리보다 요청 생성이 빠를 때 생기는 문제와 이를 막는 방법을 설명한다.

정답과 해설

1. 작업자 1인 경우와 같은 상호배제 그룹인 경우에는 점검이 t=5에 끝난 뒤 상태 작업이 시작하므로 전체 소요는 6이다. 그룹을 분리한 경우와 같은 재진입 그룹인 경우에는 상태 작업이 t=0에 시작하고 t=1에 끝난다. 점검이 t=5에 끝나므로 전체 소요는 5다.
2. 요청 타이머가 비동기 요청을 보낸 뒤 반환하므로 설명한 대기 순환은 생기지 않는다. 서버 콜백이 끝난 후 클라이언트 응답 처리가 진행될 수 있다. 다만 점검 서버가 0.2초 동안 유일한 실행 스레드를 차지하므로 그동안 상태 타이머는 실행되지 못한다. 실행기 가져오기와 생성 코드를 함께 SingleThreadedExecutor로 바꾸어야 하며 num_threads 인수도 제거해야 한다.
3. 상호배제 그룹에서는 작업자가 네 개여도 두 작업이 직렬 실행된다. 재진입 그룹에서는 둘이 함께 시작할 수 있다. 실제 카운터 콜백에는 같은 콜백의 중복 실행, 여러 접근 경로, 읽기와 쓰기의 보호 규칙을 확인해야 한다. 겹칠 필요가 없다면 같은 상호배제 그룹을 유지하고, 겹쳐야 한다면 공유 상태를 짧은 잠금 구간으로 보호한다.
4. 서버 처리 속도보다 빠르게 요청을 만들면 미완료 요청이 누적되고 응답 지연과 자원 사용이 증가할 수 있다. self.pending에 마지막 Future만 계속 덮어쓰면 이전 요청의 완료를 추적하기도 어려워진다. 미완료 요청을 하나로 제한하거나 명시적인 동시 요청 한도를 둔다. 응답 지연에 대한 기한과 재시도 정책도 별도로 정해야 한다.