Python 가상 발행·구독

발행과 수신 처리를 다른 순간으로 만듭니다

센서가 값을 만들자마자 화면 함수를 직접 호출하면 화면 처리 시간이 센서 처리 경로에 섞입니다. 이번 레슨은 발행할 때 메시지를 큐에 넣고, 수신 처리 단계에서 꺼내는 단일 스레드 버스를 구현합니다. 비동기 시스템의 모든 특성을 구현하는 것은 아니지만 생산과 소비 사이에 대기 데이터가 생기는 상황을 재현할 수 있습니다. 성공 기준은 FIFO 순서, 큐 용량, 소유 복사, 화면 상태가 같은 테스트에서 일관되게 지켜지는지입니다.

실습의 topic_runtime.py에는 Bus와 StatusNode가 있습니다. Bus.publish는 이름과 타입을 비교한 뒤 validate로 필드를 검사하여 큐에 넣습니다. Bus.drain은 비어 있지 않은 동안 가장 오래된 항목을 꺼내 콜백에 전달합니다. StatusNode.on_message가 최근 상태를 저장하고 render가 화면 문자열을 만듭니다. 센서 역할은 StatusNode 객체를 직접 수정하지 않습니다. 전달은 버스 경계를 통과합니다.

deque의 양 끝을 의도에 맞게 씁니다

FIFO는 먼저 들어온 항목을 먼저 꺼내는 순서입니다. 오른쪽 append로 넣고 왼쪽 popleft로 꺼내면 발행 순서가 유지됩니다. starter는 appendleft로 넣고 popleft로 꺼내므로 가장 최근 항목부터 전달합니다. 항목 하나만 넣는 테스트에서는 문제가 보이지 않지만 0,1,2를 넣으면 순서가 역전됩니다. TODO를 고칠 때 함수 이름만 암기하기보다 어느 끝에 넣고 어느 끝에서 빼는지 그림으로 확인합니다.

큐 깊이는 이번 모델에서 구독 대기 항목 수의 상한입니다. deque(maxlen=4)에 append를 계속하면 다섯 번째부터 왼쪽의 오래된 항목이 제거됩니다. 0부터 5까지 여섯 건을 넣고 나중에 처리하면 2,3,4,5가 남습니다. appendleft 결함은 순서만 바꾸는 것이 아니라 제거되는 방향까지 바꿔 최신 네 건 유지 계약도 깨뜨립니다. FIFO 검사와 용량 검사 두 개가 함께 실패하는 이유를 연결해 읽습니다.

제거된 항목을 카운터로 드러냅니다

maxlen은 메모리를 제한하지만 제거된 횟수를 자동으로 업무 로그에 남기지는 않습니다. publish는 append 전에 len(queue)가 maxlen과 같은지 확인하고 dropped를 증가시킵니다. 깊이 4의 정상 상태에서 다섯 번째를 넣을 때 1, 여섯 번째를 넣을 때 2가 됩니다. len이 깊이와 같다는 사실만으로 이미 손실됐다고 계산하면 네 번째 정상 입력부터 손실로 잘못 기록하므로 증가 시점을 주의합니다.

published는 publish를 호출한 횟수이며 이름과 타입 불일치 호출도 포함합니다. received는 콜백이 성공적으로 반환한 항목 수입니다. dropped는 용량 때문에 제거한 수입니다. 세 값은 의미가 달라 무수신의 위치를 찾는 데 도움이 됩니다. 이번 실습은 정상 콜백만 대상으로 합니다. 콜백 예외가 나면 꺼낸 항목의 재시도와 실패 카운터를 별도 설계해야 하며 현재 버스가 그 보장을 제공한다고 설명하지 않습니다.

값 복사는 노드 간 소유권 계약입니다

발행자가 같은 딕셔너리를 재사용하면서 range_m를 바꾸면 큐에 저장한 과거 메시지도 바뀔 수 있습니다. 큐가 객체 참조만 보유하기 때문입니다. validate는 deepcopy로 검증된 값을 복사해 반환합니다. 발행 후 생산자가 원본 거리를 7로 바꾸더라도 이미 대기 중인 메시지에는 1이 남아야 합니다. 테스트는 이 변화가 버스 밖에서 발생해도 전달 내용이 안정적인지 확인합니다.

StatusNode도 최근 메시지를 복사해 저장합니다. 현재 필드는 대부분 단순값이지만 추가 필드에 중첩 구조가 들어갈 수 있어 deepcopy를 사용합니다. 이 복사가 네트워크 직렬화를 구현하는 것은 아닙니다. 목적은 PC 모델의 수정 가능한 객체 공유를 차단하는 데 있습니다. 복사 비용은 실제 성능 설계에서 고려할 사항이며 이번 작은 fixture에서는 일관성부터 검증합니다.

20건 정상 수신과 과부하를 구별합니다

정상 데모는 정수 tick 20개를 돌면서 매 발행 뒤 바로 drain합니다. 큐가 다음 발행 전에 비워지므로 깊이 4라도 20건 모두 수신하고 dropped는 0입니다. 큐 깊이가 4이므로 총 네 건만 받을 수 있다는 해석은 틀립니다. 깊이는 누적 수신량이 아니라 동시에 대기하는 항목 수입니다. 버퍼와 총 처리량의 차이를 실행 기록으로 설명합니다.

과부하 데모는 drain 없이 여섯 건을 넣은 다음 한 번에 비웁니다. 정상과 같은 publish 함수를 사용하지만 소비 시점이 달라 손실 두 건이 생깁니다. 깊이를 늘리면 일시적인 대기를 더 수용할 수 있지만 소비자가 계속 느리면 처리 능력 부족은 남습니다. 모든 관측 저장이 요구되는 기록 경로와 최신 상태 표시만 필요한 화면 경로에 같은 제거 정책이 맞는지 목적별로 판단합니다.

상태 화면은 수신 메시지의 의미를 보존합니다

최근 메시지가 없으면 render는 NO_DATA를 반환합니다. 정상 메시지면 seq, stamp, 소수점 세 자리 거리, status를 표시합니다. sample_valid가 false면 거리 위치에 INVALID를 넣습니다. 숫자가 유한하다는 검사와 표시 가능 여부의 판단을 한 조건으로 합치지 않습니다. 결함 행은 정상적인 전달 대상으로 남기면서 관측 사용을 차단합니다.

독립 레슨 데모는 매 tick마다 새 메시지를 만들어 마지막 seq=19, stamp=0.950입니다. 모듈 미션은 실제 C 행을 재발행하므로 마지막 측정 seq=2, stamp=0.200으로 남습니다. 두 출력이 다른 것은 오류가 아니라 입력 방식이 다르기 때문입니다. 단계 실행 결과를 비교할 때 어떤 fixture를 사용했는지 함께 읽어야 합니다. 발행 횟수만 보고 측정 개수까지 같다고 결론 내리지 않습니다.

실패 출력에서 기능을 좁힙니다

처음 bash check.sh를 실행하면 test_fifo와 test_bounded 등이 실패할 수 있습니다. AssertionError에 출력된 실제 목록과 기대 목록을 비교해 방향을 확인합니다. ImportError나 SyntaxError는 의도한 TODO 실패가 아니라 실행 환경이나 편집 오류입니다. 테스트 기대 순서를 역순으로 바꾸어 통과시키지 않습니다. 요구사항은 발행 순서를 유지하고 최신 네 건을 보존하는 것입니다.

수정 후 같은 명령에서 12개 검사가 OK인지 확인합니다. test_copy는 원본 수정, test_depth_one은 깊이 1, test_bad_fields는 불리언 번호와 비유한 거리 등을 확인합니다. 이런 검사가 남아 있어야 FIFO 수정이 입력 계약을 약화시키지 않았다는 근거가 됩니다. python3 demo.py의 카운터와 목록까지 관찰하고 어떤 처리 일정이 손실을 만들었는지 짧은 설명을 제출합니다.

실행 모델의 한계를 정확히 인계합니다

Bus는 네트워크 발견, 여러 프로세스 간 전송, 직렬화, 재전송, ROS 실행기의 콜백 스케줄을 구현하지 않습니다. 한 프로세스에서 함수를 호출하는 실습입니다. 여기서 20건을 받았다는 결과는 지정 fixture와 소비 일정의 증거입니다. 실제 장치의 20Hz나 네트워크 손실 없는 운전을 증명하지 않습니다. 실물로 옮기는 단계에서는 주기와 지연, 끝점 설정을 별도로 측정합니다.

코드 리뷰에는 큐의 양 끝 정책, 소유 복사 이유, 세 카운터의 정의를 포함합니다. 더 읽기는 실제 발행자·구독자 API를 연결하는 참고 자료입니다. 다음 레슨은 이 버스에 이름 오타와 중단, 느린 소비를 주입해 서로 다른 증상을 구분합니다. 이번에는 먼저 정상 전달 경로의 계약을 안정적으로 만들고 그 경로를 기준으로 실패를 비교할 수 있게 준비합니다.