라이프사이클 노드 - 켜고 끄는 순서 관리

이 장에서 배우는 것

앞 장에서 오래 걸리는 작업을 요청하고 피드백을 받거나 취소하는 방법을 다뤘다. 그런데 작업을 요청하기 전에 확인할 조건이 있다. 두리의 센서가 준비되었는지, 주행에 필요한 설정이 적용되었는지, 명령을 받아도 되는 상태인지 알아야 한다. 프로세스가 실행 중이라는 사실만으로는 이 조건을 판단할 수 없다.

라이프사이클 노드(lifecycle node)는 준비와 실행, 정리와 종료를 명시적인 상태 전이로 나눈다. 이 장에서는 두리의 거리 센서와 주행 출력을 모형으로 삼아 관리형 노드를 만들고, 의존 관계에 맞춰 시스템을 기동한다. 실제 장치를 움직이는 대신 상태 문자열을 발행하므로 상태 전이 자체에 집중할 수 있다.

• 관리형 노드의 기본 상태와 전이 결과를 구분한다.
• 구성·활성화·비활성화·정리 콜백에 책임을 나누어 배치한다.
• 제공자를 먼저 활성화하고 소비자를 먼저 비활성화하는 순서를 설계한다.
• ROS 없이 실행하는 상태 기계로 기동 실패와 복구 순서를 확인한다.

문제 상황

두리의 전원을 켜면 거리 센서 프로세스와 주행 프로세스가 함께 시작된다고 하자. 거리 센서는 연결을 열고 설정값을 읽는 데 시간이 걸린다. 주행 프로세스는 먼저 시작되었지만 아직 유효한 거리 정보를 받지 못했다. 이때 주행 명령을 허용하면 필요한 입력이 없는 상태에서 동작하게 된다.

노드 시작 사이에 일정한 대기 시간을 넣는 방법은 환경에 따라 흔들린다. 장치 연결이 평소보다 늦으면 대기가 부족하고, 연결이 빠르면 불필요하게 기다린다. 필요한 것은 몇 초가 지났다는 정보가 아니라 준비 단계가 성공했다는 응답이다.

종료도 순서가 있다. 주행 쪽이 거리 센서의 결과를 사용하고 있다면 주행 출력을 먼저 멈추고 센서를 내려야 한다. 센서를 먼저 끄면 주행 쪽이 마지막 값을 계속 사용하는 구간이 생길 수 있다. 다시 구성하려는 경우에도 실행을 멈춘 뒤 자원을 정리해야 한다.

예제에서는 센서 역할의 노드를 /range_gate, 주행 역할의 노드를 /drive_gate라고 부른다. 두 노드는 같은 프로그램으로 실행하지만 관리 순서는 다르게 적용한다. 실제 센서 데이터 전달과 구동기 제어는 구현하지 않는다. 특히 두 노드 사이의 의존 관계는 ROS가 이름을 보고 알아내는 것이 아니라 운영자가 정한 관계다.

상태는 프로세스 생존 여부와 다르다

관리형 노드에는 네 가지 기본 상태가 있다. 생성 직후에는 미구성 상태이고, 구성에 성공하면 비활성 상태가 된다. 활성화에 성공해야 정상 작업을 수행하는 활성 상태로 들어간다. 종료 전이가 끝나면 최종 상태가 된다. 상태 이름은 명령행 출력과 대응하도록 코드 표기를 함께 기억한다.

관리형 노드의 기본 상태와 예제의 동작

상태	의미	예제의 자원	상태 발행
unconfigured	구성 전 또는 정리 후	없음	하지 않음
inactive	구성되었지만 작업 중지	발행자와 타이머 보유	하지 않음
active	정상 작업 허용	구성한 자원 사용	주기적으로 수행
finalized	라이프사이클 종료	종료 콜백에서 정리	하지 않음

최종 상태는 운영체제 프로세스가 사라졌다는 뜻이 아니다. 종료 전이가 성공해도 실행기를 돌리는 코드가 남아 있으면 프로세스는 계속 존재할 수 있다. 반대로 프로세스를 강제로 끝낸다고 종료 콜백이 실행되는 것도 아니다. 정상적인 상태 전이와 프로세스 종료는 별도의 절차로 다뤄야 한다.

구성과 활성화를 분리하면 비용이 큰 준비를 마친 채 동작만 멈출 수 있다. 예를 들어 센서 연결과 내부 버퍼는 유지하면서 외부로 결과를 내보내지 않을 수 있다. 비활성화 후 활성화하는 것은 정리 후 다시 구성하는 것과 비용과 의미가 다르다.

기본 상태 사이에는 configuring, activating 같은 전이 상태가 있다. 요청을 받으면 대응 콜백이 실행되고, 콜백의 반환값에 따라 다음 상태가 결정된다. 구성 콜백이 실행되고 있다는 사실과 구성이 완료되었다는 사실을 구분해야 한다.

구성 콜백의 반환값이 이후 상태에 미치는 영향

반환값	의미	이후 흐름
SUCCESS	구성 완료	inactive로 이동
FAILURE	예상 가능한 구성 거부	unconfigured로 복귀
ERROR	오류 처리 필요	on_error() 실행 경로로 이동

이 표의 복귀 상태는 구성 전이에 대한 설명이다. 모든 콜백에서 FAILURE를 반환하면 미구성 상태가 되는 것은 아니다. 예를 들어 활성화 실패는 비활성 상태로 돌아간다. 전이마다 출발 상태와 결과 상태를 따로 살펴야 한다.

오류 처리 콜백이 성공을 반환하면 미구성 상태로 돌아갈 수 있다. 따라서 성공을 반환할 때에는 남은 자원을 정리하고 다시 구성할 수 있는 조건을 실제로 복구해야 한다. 오류 메시지를 남기는 것만으로 복구가 끝난 것은 아니다.

콜백에 자원과 작업의 책임을 나눈다

생성자에는 노드 이름과 파라미터 선언, 아직 자원이 없음을 나타내는 필드를 둔다. 외부 장치를 열거나 작업을 시작하는 코드를 모두 생성자에 넣으면 관리자가 구성 요청을 보내기도 전에 준비와 동작이 섞인다. 관리형 노드로 바꾼 효과가 줄어드는 배치다.

on_configure()에서는 파라미터를 검사하고 작업에 필요한 자원을 만든다. 예제는 발행자와 타이머를 만들되 타이머를 즉시 취소한다. on_activate()에서는 발행자를 활성화한 뒤 타이머를 재개한다. on_deactivate()에서는 타이머를 먼저 멈추고 발행자를 비활성화한다.

on_cleanup()은 구성을 해제하여 미구성 상태로 돌아갈 준비를 한다. 발행자와 타이머를 없애고 필드를 다시 None으로 바꾼다. 여러 번 호출되어도 이미 없는 자원을 다시 파괴하지 않도록 정리 함수를 작성하면 부분 구성 실패에도 대응하기 쉽다.

일반 발행자와 라이프사이클 발행자는 다르다. 후자는 활성 여부를 관리하지만, 노드 안의 모든 동작이 자동으로 제어되는 것은 아니다. 일반 구독 콜백이나 타이머, 별도로 만든 작업 스레드까지 상태 변화만으로 멈추지는 않는다. 예제에서 타이머 취소와 실행 허용 플래그를 함께 사용하는 이유다.

활성화와 비활성화 콜백을 재정의할 때에는 부모 구현을 호출한다. 부모 구현이 관리 대상에 등록된 발행자에 상태 변화를 전달하기 때문이다. 직접 성공만 반환하면 노드 상태는 활성으로 바뀌더라도 라이프사이클 발행자는 비활성 상태에 남을 수 있다.

예제는 기본 실행 방식에서 콜백을 직렬로 처리한다. 타이머를 만들고 취소하는 구성 콜백이 끝나기 전에 다른 작업 콜백이 끼어들지 않는 조건이다. 콜백을 병렬로 실행하도록 바꾸면 이미 실행 중인 작업과 자원 해제 사이의 동기화도 별도로 설계해야 한다.

시스템 기동은 의존 관계를 따라간다

두리에서는 거리 센서가 제공자이고 주행 쪽이 소비자다. 예제의 기동 정책은 센서 구성, 주행 구성, 센서 활성화, 주행 활성화 순서다. 모든 노드가 준비 자원을 확보한 뒤 실제 작업을 시작하도록 구성 단계와 활성화 단계를 묶는다.

이 순서는 주행 노드의 구성 자체에는 활성 센서가 필요 없다는 가정을 포함한다. 어떤 노드는 구성 과정에서 제공자의 응답을 요구할 수 있다. 그런 시스템에서는 제공자를 먼저 활성화한 뒤 소비자를 구성하는 순서가 필요하다. 이름이나 실행 파일 순서가 아니라 실제 의존 조건을 기준으로 결정한다.

또한 센서의 활성화 성공이 첫 번째 유효 측정값의 도착을 뜻하지는 않는다. 활성화는 해당 노드가 정의한 시작 절차가 성공했다는 뜻이다. 실제 주행을 허용하려면 측정값의 유효성과 최신성 같은 준비 조건을 추가로 검사해야 한다. 이번 예제의 활성 상태는 문자열 발행 준비만 보장한다.

launch로 여러 프로세스를 시작해도 구성과 활성화 전이가 저절로 수행되지는 않는다. 시작한 관리형 노드에 전이 요청을 보내는 주체가 필요하다. 개발 중에는 명령행을 사용하고, 운영 시스템에서는 별도의 관리자가 전이 서비스 응답과 현재 상태를 확인하며 진행하도록 만들 수 있다.

상태 변경 서비스의 응답을 받지 못했을 때도 주의한다. 시간 초과는 실패가 확정되었다는 뜻이 아니다. 노드에서는 전이가 완료되었지만 응답만 늦어졌을 수 있다. 실제 관리자는 현재 상태를 다시 조회한 뒤 재시도나 복구를 결정해야 한다. 아래 보조 예제는 이런 통신 불확실성을 제외하고 순서와 실패 처리만 재현한다.

완성 코드

두 파일을 같은 디렉터리에 저장한다. 첫 파일은 ROS 2 Jazzy의 rclpy와 std_msgs가 필요하다. 두 번째 파일은 Python 표준 기능만 사용하므로 ROS가 없는 macOS와 Linux에서도 실행할 수 있다. ROS 프로그램은 해당 운영체제에 Jazzy와 Python 바인딩이 준비된 환경을 전제로 한다.

duri_lifecycle.py

import math

import rclpy
from rclpy.lifecycle import LifecycleNode, State, TransitionCallbackReturn
from std_msgs.msg import String


class DuriGate(LifecycleNode):
    def __init__(self):
        super().__init__("duri_gate")
        self.declare_parameter("period", 1.0)
        self.publisher = None
        self.timer = None
        self.enabled = False

    def on_configure(self, state: State) -> TransitionCallbackReturn:
        period = self.get_parameter("period").value
        if not isinstance(period, float):
            return TransitionCallbackReturn.FAILURE
        if not math.isfinite(period) or period <= 0.0:
            return TransitionCallbackReturn.FAILURE

        try:
            self.publisher = self.create_lifecycle_publisher(
                String, "~/status", 10
            )
            self.timer = self.create_timer(period, self.publish_status)
            self.timer.cancel()
        except Exception:
            self.release_resources()
            return TransitionCallbackReturn.ERROR
        return TransitionCallbackReturn.SUCCESS

    def on_activate(self, state: State) -> TransitionCallbackReturn:
        if self.publisher is None or self.timer is None:
            return TransitionCallbackReturn.FAILURE
        result = super().on_activate(state)
        if result == TransitionCallbackReturn.SUCCESS:
            self.enabled = True
            self.timer.reset()
        return result

    def on_deactivate(self, state: State) -> TransitionCallbackReturn:
        self.enabled = False
        if self.timer is not None:
            self.timer.cancel()
        return super().on_deactivate(state)

    def on_cleanup(self, state: State) -> TransitionCallbackReturn:
        self.release_resources()
        return TransitionCallbackReturn.SUCCESS

    def on_shutdown(self, state: State) -> TransitionCallbackReturn:
        self.release_resources()
        return TransitionCallbackReturn.SUCCESS

    def on_error(self, state: State) -> TransitionCallbackReturn:
        self.release_resources()
        return TransitionCallbackReturn.SUCCESS

    def publish_status(self):
        if not self.enabled or self.publisher is None:
            return
        message = String()
        message.data = f"{self.get_name()}:ready"
        self.publisher.publish(message)

    def release_resources(self):
        self.enabled = False
        if self.timer is not None:
            self.timer.cancel()
            self.destroy_timer(self.timer)
            self.timer = None
        if self.publisher is not None:
            self.destroy_publisher(self.publisher)
            self.publisher = None

    def close(self):
        self.release_resources()


def main():
    rclpy.init()
    node = DuriGate()
    try:
        rclpy.spin(node)
    except (KeyboardInterrupt, rclpy.executors.ExternalShutdownException):
        pass
    finally:
        node.close()
        node.destroy_node()
        rclpy.try_shutdown()


if __name__ == "__main__":
    main()

lifecycle_model.py

from dataclasses import dataclass


@dataclass
class Unit:
    name: str
    state: str = "unconfigured"
    fail_activate: bool = False

    def move(self, event: str) -> None:
        transitions = {
            ("unconfigured", "configure"): "inactive",
            ("inactive", "activate"): "active",
            ("active", "deactivate"): "inactive",
            ("inactive", "cleanup"): "unconfigured",
        }
        target = transitions.get((self.state, event))
        if target is None:
            raise RuntimeError(
                f"{self.name}: {self.state}에서 {event} 불가"
            )
        if event == "activate" and self.fail_activate:
            raise RuntimeError(f"{self.name}: 활성화 실패")
        self.state = target
        print(f"{self.name}: {event} -> {self.state}")


def bring_down(units: list[Unit]) -> None:
    for unit in reversed(units):
        if unit.state == "active":
            unit.move("deactivate")
    for unit in reversed(units):
        if unit.state == "inactive":
            unit.move("cleanup")


def bring_up(units: list[Unit]) -> bool:
    try:
        for unit in units:
            unit.move("configure")
        for index, unit in enumerate(units):
            if any(item.state != "active" for item in units[:index]):
                raise RuntimeError(f"{unit.name}: 제공자 준비 안 됨")
            unit.move("activate")
    except RuntimeError as error:
        print(f"기동 중단: {error}")
        bring_down(units)
        return False
    return True


def show_states(units: list[Unit]) -> None:
    summary = ", ".join(f"{unit.name}={unit.state}" for unit in units)
    print(f"최종 상태: {summary}")


def main():
    print("[정상 기동]")
    normal = [Unit("range_gate"), Unit("drive_gate")]
    assert bring_up(normal)
    assert all(unit.state == "active" for unit in normal)
    print("[정상 정리]")
    bring_down(normal)
    assert all(unit.state == "unconfigured" for unit in normal)
    show_states(normal)

    print("[활성화 실패]")
    failing = [
        Unit("range_gate"),
        Unit("drive_gate", fail_activate=True),
    ]
    assert not bring_up(failing)
    assert all(unit.state == "unconfigured" for unit in failing)
    show_states(failing)

    print("[잘못된 전이]")
    try:
        Unit("drive_gate").move("activate")
    except RuntimeError as error:
        print(f"거부: {error}")
    else:
        raise AssertionError("잘못된 전이가 허용됨")


if __name__ == "__main__":
    main()

보조 예제는 세 가지 기본 상태와 네 가지 전이만 구현한다. ROS의 전이 상태, 서비스 통신, 오류 처리 상태, 최종 상태 전체를 대체하는 구현은 아니다. 실패 시 활성화 대상이 비활성 상태에 남는 동작과, 성공한 선행 노드를 거꾸로 내리는 정책을 읽기 쉽게 드러내는 모형이다.

줄별 해설

노드 생성과 구성

super().__init__("duri_gate")는 관리형 노드를 생성한다. 실행할 때 이름을 재매핑하므로 한 파일로 두 역할을 띄울 수 있다. period는 부동소수점 파라미터로 선언한다. 명령행에서도 1.0처럼 같은 자료형의 값을 전달한다.

세 개의 필드는 자원의 존재와 작업 허용 여부를 분리한다. publisher와 timer가 존재해도 enabled가 거짓이면 발행 작업을 수행하지 않는다. 비활성 상태에서 구성 자원을 유지하는 구조를 코드에 표현한 것이다.

구성 콜백의 앞부분은 주기가 유한한 양수인지 확인한다. 검사에 실패하면 자원을 만들기 전에 FAILURE를 반환한다. NaN과 무한대도 제외하므로 단순히 양수인지 비교하는 것보다 타이머 입력 조건이 분명하다.

create_lifecycle_publisher()의 토픽 이름 ~/status는 노드 이름에 따라 확장된다. 센서 역할에서는 /range_gate/status가 된다. 뒤의 숫자는 발행자 생성에 필요한 이력 깊이이며, 여기서는 통신 정책을 바꾸는 것이 목적이 아니다.

자원을 만드는 도중 예외가 발생하면 이미 만든 자원을 정리한 뒤 ERROR를 반환한다. 이어지는 오류 처리 콜백도 같은 정리 함수를 사용한다. 예제에서는 자원 해제 자체가 정상적으로 끝난다는 조건에서 미구성 상태로 복귀한다.

작업 시작과 중지

on_activate()의 첫 검사는 필요한 두 자원이 있는지 확인한다. 이어서 부모 활성화 구현의 반환값을 검사한다. 성공일 때만 플래그를 켜고 reset()으로 타이머를 다시 시작하므로 발행자가 준비되기 전에 주기 작업을 허용하지 않는다.

on_deactivate()는 반대 순서로 처리한다. 작업 허용을 먼저 끄고 타이머를 취소한 뒤 부모 구현으로 발행자를 비활성화한다. 실제 주행 노드라면 이 위치에서 구동기 정지 요청과 정지 확인 같은 별도의 동작이 필요하다. 타이머 취소만으로 물리적인 로봇 정지를 증명할 수는 없다.

publish_status()는 허용 상태를 다시 확인한 뒤 문자열을 만든다. 타이머 제어가 일차적인 실행 제어이고, 플래그는 콜백 본문에 들어왔을 때의 추가 조건이다. 이 플래그가 노드의 공식 라이프사이클 상태를 대체하는 것은 아니다.

정리와 프로세스 종료

release_resources()는 타이머부터 없앤다. 발행자를 먼저 파괴하면 작업이 남아 있는 구조에서 존재하지 않는 발행자를 사용하기 쉽기 때문이다. 파괴 후 필드를 None으로 바꾸어 이후 정리 호출이 같은 자원을 다시 다루지 않게 한다.

정리·종료·오류 콜백은 모두 이 함수로 현재 자원을 제거한다. 예제에는 별도로 전파해야 할 사용자 정의 관리 객체가 없으므로 해당 콜백에서 정리 완료를 직접 반환한다. 관리 객체를 추가한다면 그 객체의 상태 전파와 해제 책임도 함께 검토해야 한다.

finally의 close()는 키보드 인터럽트처럼 라이프사이클 종료 요청을 거치지 않은 경로에서도 자원을 정리하기 위한 것이다. 여기서 정리한다고 공식 상태가 최종 상태로 바뀌지는 않는다. 프로세스 수준의 해제와 상태 전이를 의도적으로 구분한다.

순서 모형과 실패 복구

보조 예제의 transitions는 현재 상태와 요청의 조합을 다음 상태로 대응시킨다. 표에 없는 조합은 상태를 바꾸기 전에 거부한다. 활성화 실패도 대입문보다 먼저 발생하므로 실패한 노드는 비활성 상태를 유지한다.

bring_up()은 모든 노드를 구성하고 순서대로 활성화한다. 앞선 노드가 활성인지 검사하는 부분은 이 예제의 선형 의존 관계를 나타낸다. 모든 시스템의 노드가 앞선 노드 전체에 의존한다는 뜻은 아니다.

bring_down()은 먼저 활성 노드를 역순으로 비활성화하고, 그다음 비활성 노드를 역순으로 정리한다. 주행 활성화가 실패하면 센서만 비활성화하고 두 노드를 정리한다. 이미 비활성인 주행 노드에 불필요한 비활성화 요청을 보내지 않는다. 이 모형은 복구 전이가 성공한다고 가정한다.

실행 결과

먼저 두 파일의 문법을 검사한다. 이 명령은 ROS 모듈을 가져오지 않으므로 ROS가 없어도 사용할 수 있다. 정상적으로 끝나면 출력이 없다. 컴파일 성공은 API 호출과 실제 상태 전이의 성공까지 검증한 결과는 아니다.

python3 -W error -m py_compile duri_lifecycle.py lifecycle_model.py

순수 Python 보조 예제는 다음과 같이 실행한다. 아래 출력은 코드의 정상 기동, 활성화 실패, 잘못된 전이 처리 순서에 대응하는 예상 결과다. 여기서는 도구를 실행하지 않았으므로 실제 실행 검증 결과로 제시하는 것은 아니다.

python3 lifecycle_model.py
[정상 기동]
range_gate: configure -> inactive
drive_gate: configure -> inactive
range_gate: activate -> active
drive_gate: activate -> active
[정상 정리]
drive_gate: deactivate -> inactive
range_gate: deactivate -> inactive
drive_gate: cleanup -> unconfigured
range_gate: cleanup -> unconfigured
최종 상태: range_gate=unconfigured, drive_gate=unconfigured
[활성화 실패]
range_gate: configure -> inactive
drive_gate: configure -> inactive
range_gate: activate -> active
기동 중단: drive_gate: 활성화 실패
range_gate: deactivate -> inactive
drive_gate: cleanup -> unconfigured
range_gate: cleanup -> unconfigured
최종 상태: range_gate=unconfigured, drive_gate=unconfigured
[잘못된 전이]
거부: drive_gate: unconfigured에서 activate 불가

ROS 실행은 Jazzy 환경을 불러온 터미널에서 진행한다. 각 터미널은 같은 ROS 도메인을 사용해야 한다. 첫 번째 터미널에서 센서 역할을 실행한다. 프로그램은 자체 시작 메시지를 출력하지 않고 요청을 기다린다.

python3 duri_lifecycle.py --ros-args -r __node:=range_gate

두 번째 터미널에서는 주행 역할을 실행한다.

python3 duri_lifecycle.py --ros-args -r __node:=drive_gate

세 번째 터미널에서 초기 상태를 확인한다. 노드가 검색된 뒤의 정상 응답은 다음과 같다.

ros2 lifecycle get /range_gate
unconfigured [1]

다음 요청은 앞 요청이 성공한 것을 확인한 뒤 한 줄씩 실행한다. 이 명령 목록 자체에 실패 복구 기능이 있는 것은 아니다. 응답이 실패이거나 노드를 찾을 수 없으면 다음 활성화 요청으로 넘어가지 않는다.

ros2 lifecycle set /range_gate configure
Transitioning successful
ros2 lifecycle set /drive_gate configure
Transitioning successful
ros2 lifecycle set /range_gate activate
Transitioning successful
ros2 lifecycle set /drive_gate activate
Transitioning successful
ros2 lifecycle get /drive_gate
active [3]

주행 역할의 상태 토픽을 한 번 읽는다. 기본 주기는 1초이므로 활성 상태에서 다음 발행 시점까지 기다릴 수 있다. 메시지 내용에는 시각이나 난수를 넣지 않았다.

ros2 topic echo /drive_gate/status --once
data: drive_gate:ready
---

두 노드를 다시 구성할 수 있는 상태로 내릴 때에는 다음 순서를 적용한다. 비활성화 이후에는 새로운 상태 메시지를 발행하지 않는다.

ros2 lifecycle set /drive_gate deactivate
Transitioning successful
ros2 lifecycle set /range_gate deactivate
Transitioning successful
ros2 lifecycle set /drive_gate cleanup
Transitioning successful
ros2 lifecycle set /range_gate cleanup
Transitioning successful
ros2 lifecycle get /drive_gate
unconfigured [1]

재구성 대신 라이프사이클을 끝내려면 이 상태에서 종료 전이를 요청한다. 아래 명령 후에도 두 Python 프로세스는 실행기에서 대기한다. 각 실행 터미널에서 Ctrl+C를 눌러 프로세스를 끝낸다.

ros2 lifecycle set /drive_gate shutdown
Transitioning successful
ros2 lifecycle set /range_gate shutdown
Transitioning successful
ros2 lifecycle get /drive_gate
finalized [4]

실무에서 자주 틀리는 것

활성 상태로 바꾸면서 발행자는 활성화하지 않는다

다음 코드는 활성화 요청에 성공을 반환하지만 관리 대상 발행자에 전이를 전달하지 않는다. 노드 상태만 보고 발행도 시작되었다고 판단하면 원인을 찾기 어렵다.

def on_activate(self, state):
    self.timer.reset()
    return TransitionCallbackReturn.SUCCESS

부모 구현의 결과를 확인한 뒤 주기 작업을 허용하도록 고친다.

def on_activate(self, state):
    result = super().on_activate(state)
    if result == TransitionCallbackReturn.SUCCESS:
        self.enabled = True
        self.timer.reset()
    return result

비활성 상태이면 모든 콜백이 멈춘다고 가정한다

일반 타이머는 라이프사이클 상태만으로 멈추지 않는다. 아래처럼 구성 과정에서 타이머를 실행한 채 두면 비활성 상태에서도 콜백이 호출된다. 발행이 막혀 있더라도 콜백 안의 다른 부수 효과는 실행될 수 있다.

self.timer = self.create_timer(period, self.publish_status)
return TransitionCallbackReturn.SUCCESS

구성에서는 타이머를 취소하고, 활성화에서 재개한다. 구독 콜백에도 필요한 작업 허용 조건을 별도로 둔다.

self.timer = self.create_timer(period, self.publish_status)
self.timer.cancel()
return TransitionCallbackReturn.SUCCESS

실행 중인 노드에 바로 정리를 요청한다

활성 상태에서 아래 요청을 보내면 허용된 정리 전이가 아니므로 자원 해제로 이어지지 않는다. 정리 콜백이 실행된다고 가정해서도 안 된다.

ros2 lifecycle set /drive_gate cleanup

먼저 비활성화 성공을 확인하고 정리를 요청한다. 활성 상태에서 종료 전이를 요청할 수는 있지만, 다시 구성할 목적이라면 비활성화와 정리를 사용한다.

ros2 lifecycle set /drive_gate deactivate
ros2 lifecycle set /drive_gate cleanup

실패하면 모든 노드에 같은 복구 요청을 보낸다

주행 활성화에 실패한 경우 센서는 활성이고 주행은 비활성일 수 있다. 아래 코드는 상태를 무시하며 제공자부터 내리기도 한다.

for unit in units:
    unit.move("deactivate")

보조 예제처럼 역순으로 상태를 검사한다. 실제 ROS 관리자에서는 메모리에 기록한 이전 성공 여부만 믿지 말고 상태 조회 결과도 사용한다. 비활성화에 실패했다면 제공자를 계속 내려도 되는지도 별도로 판단해야 한다.

for unit in reversed(units):
    if unit.state == "active":
        unit.move("deactivate")

한눈에 보기

두리의 라이프사이클 설계에서 확인할 결정

단계	노드의 책임	관리자의 확인
생성	파라미터 선언과 초기 필드 설정	노드와 관리 서비스 검색
구성	입력 검사와 자원 확보	성공 응답과 비활성 상태
활성화	발행자와 작업 활성화	제공자 준비 후 소비자 진행
비활성화	작업 중지와 발행자 비활성화	소비자부터 중지
정리	구성 자원 해제	미구성 상태 복귀
종료	남은 자원 해제	최종 상태와 프로세스 종료 구분

라이프사이클은 노드가 무엇을 준비했고 언제 작업을 허용하는지 외부에서 확인할 수 있게 만든다. 여러 노드의 전이를 하나의 원자적인 작업으로 묶지는 않는다. 중간 실패, 응답 지연, 재시작에 대한 정책은 시스템 관리자의 책임으로 남는다.

상태 전이 정의는 ROS 2 관리형 노드 설계에서, Python API는 Jazzy의 rclpy 라이프사이클 참조에서 확인할 수 있다. 두 문서는 상태와 API의 사실 확인에 사용하는 자료다. 다음 장에서는 노드의 실행 위치와 프로세스 내 통신을 다룬다.

연습 문제

1. 두 노드가 모두 활성 상태일 때 주행 쪽의 주기만 다시 구성하려 한다. 센서는 유지하면서 주행 노드를 다시 활성 상태로 만드는 전이 순서를 적는다. 파라미터는 어느 구간에서 변경할지도 설명한다.
2. 보조 예제에서 센서 역할의 fail_activate를 참으로 바꾼다. 어느 노드에 비활성화가 요청되는지, 정리는 어떤 순서로 수행되는지 실행 전에 예측한다.
3. 제공자인 센서가 활성 상태인데 아직 유효한 측정값이 없다. 관리자가 곧바로 주행을 활성화해도 되는지 판단하고 추가로 확인할 조건 두 가지를 제안한다.
4. 주행 활성화 서비스 호출이 시간 초과되었다. 관리자가 같은 요청을 즉시 반복하는 방식의 문제를 설명하고, 상태 조회 결과가 활성 또는 비활성일 때의 후속 동작을 각각 제안한다.

정답과 해설

1. 주행 노드에 deactivate, cleanup, configure, activate 순서로 요청한다. 주기는 정리가 끝난 뒤 다음 구성 전에 부동소수점 값으로 변경한다. 예제는 구성 콜백에서 주기를 읽으므로 활성 상태에서 파라미터만 바꾸어도 기존 타이머 주기가 갱신되지는 않는다. 각 전이의 성공을 확인하며 센서는 활성 상태로 유지한다.
2. 두 노드의 구성은 성공하지만 센서 활성화에서 중단된다. 센서와 주행이 모두 비활성이므로 어느 노드에도 비활성화를 요청하지 않는다. 정리는 역순인 주행, 센서 순서로 수행한다. 최종적으로 두 노드 모두 미구성 상태가 된다.
3. 이 조건만으로 주행을 허용하기에는 정보가 부족하다. 측정값이 유효 범위에 있는지와 마지막 측정 시각이 허용된 최신성 기준 안에 있는지를 확인할 수 있다. 센서 활성화 성공은 노드의 시작 절차가 성공했다는 뜻이며, 실제 입력 데이터의 준비 조건은 추가로 정의해야 한다.
4. 첫 요청이 이미 성공했다면 반복 요청은 활성 상태에서 다시 활성화하려는 잘못된 전이가 된다. 먼저 현재 상태를 조회한다. 활성이라면 추가 준비 조건을 확인하고 다음 단계로 진행할 수 있다. 비활성이라면 실패 원인과 재시도 정책을 확인한 뒤 제한적으로 다시 요청한다. 전이 중이거나 상태를 조회할 수 없다면 소비자 기동을 진행하지 않고 결과를 더 확인한다.