ROS 2 実践コース｜#A2 詰まりを観察して直す（executor・launch の引数・テスト）

延長コース A の 2 週目です。前の週（#A1）は「型」という契約を自分で決めました。この週は、1 つのノードの中で処理が詰まったときに、外から何が見えるかを観察します。

コアの #11 では、scan_avoid が /scan を受けるたびに計算していました。あのとき、もし計算が長引いたら何が起きるか——という話を、ここで実際に作って確かめます。

この週の問い：1 つのノードの中で、ある処理が 3 秒かかるようになったら、同じノードの中のほかの処理（1 秒ごとのタイマーなど）はどうなるでしょうか。答えは A3-2 で自分の端末に出た回数で確かめ、図は A3-5 の結果のところに載せています。

この記事の位置

この記事は延長コース Aの 2 回分（A3・A4）です。授業で受ける人は、記録を実習レポート（Word）の「回 A3」「回 A4」のシートに分けて書きます。スクリーンショットは、シートの中の「画面：」の欄（実験の表のすぐ下）に貼ります。延長コースのレポートは、コア（第 1〜12 回）とは別の 1 本にします。ファイルは Google Classroom で配られ、コースの最後に Google Classroom で 1 回だけ出します。期限は最後の実習から 1 週間後の 23:59 です。ひとりで学ぶ人も、予想と結果を同じ形でノートに残しておくと、あとで見返せます。

回 A3　executor とコールバックグループ

この回は、わざと詰まるノードを作り、詰まりを外から見つけて、executor とコールバックグループで直します。6 つの実験をします。使うノードは 1 つで、少しずつ書き換えていきます。（結果の画面では、版どうしを並べて見比べるために、slow_node_sleep のように版ごとにノードの名前を変えて動かしています）

# わざと詰まらせるノード（この形だけ覚えておく）
import time
import rclpy
from rclpy.node import Node
from std_msgs.msg import String


class SlowNode(Node):
    def __init__(self):
        super().__init__("slow_node")
        self.count = 0
        self.received = 0
        self.create_timer(1.0, self.on_timer)              # 1 秒ごと
        self.create_subscription(String, "/chatter", self.on_msg, 10)

    def on_timer(self):
        self.count += 1
        self.get_logger().info(f"timer {self.count}（受けた件数 {self.received}）")

    def on_msg(self, msg):
        # time.sleep(3.0)                                  # A3-2 で、この行の先頭の # を外す
        self.received += 1


def main():
    rclpy.init()
    rclpy.spin(SlowNode())                                 # 既定は 1 スレッド

on_timer は 1 秒ごとに呼ばれ、タイマーの回数と、それまでに受けた件数をログに出します。on_msg は /chatter が 1 件届くたびに呼ばれ、件数を 1 つ増やします。time.sleep(3.0) の行は、いまは先頭に # があるので動きません（A3-2 で使うので、import time は先に書いてあります）。

A3-1　正常なときの回数を数える

ここでは、詰まらせる前の、ふつうの動き方を数えておきます。

time.sleep(3.0) はコメントのまま動かし、10 秒間でタイマーが何回出たか、受けた件数がいくつ増えたかを数えます。同時に、別の端末から /chatter を流しておきます（talker は 1 秒に 1 件送ります）。

# 端末 1：メッセージを流す
ros2 run sakigake_samples talker

# 端末 2：自分のノード
ros2 run <自分のパッケージ> slow_node

<自分のパッケージ> は、ノードを入れたパッケージの名前に置き換えます。止めるときは Ctrl+C です。

A3-2　壊す実験：購読の処理を 3 秒かける

ここでは、購読の処理だけを遅くしたとき、タイマーがどうなるかを確かめます。

on_msg の time.sleep(3.0) の行の先頭の # を外し、ビルドし直してから同じことをします。予想を先に書きます。タイマーは止まるでしょうか、遅れるだけでしょうか。10 秒間に何回出ると思うか、数字で書いてください。

手がかり：コードの最後の行のコメントを見てください。rclpy.spin は、届いたコールバックを何本のスレッドで呼んでいるでしょうか。1 本なら、on_msg が 3 秒眠っている間、ほかのコールバックを呼べるでしょうか。

結果の画面（予想を書いてから開く）

A3-3　送り側と受け側の Hz を比べる

ここでは、ノードの中の詰まりが、ノードの外からどう見えるかを Hz で確かめます。

ros2 topic hz は、トピックに 1 秒あたり何回メッセージが出ているかを測るコマンドです。ただし、測れるのはトピックに出ているものだけです。そこで、処理が 1 件終わるたびに /processed というトピックに件数を出すように、A3-2 のノードに 3 か所足します。

# A3-3：処理が 1 件終わるたびに、/processed に件数を出す
from std_msgs.msg import Int32                             # import に足す

    # __init__ の最後に足す
        self.pub_done = self.create_publisher(Int32, "/processed", 10)

    # on_msg をこうする
    def on_msg(self, msg):
        time.sleep(3.0)
        self.received += 1
        self.pub_done.publish(Int32(data=self.received))   # 1 件終わったら 1 回出す

Int32 は整数 1 つだけを入れる型です。Int32(data=self.received) で、いまの件数を入れたメッセージを作って出します。ビルドし直したら、4 つの端末で次のように動かします。

# 端末 1：メッセージを流す
ros2 run sakigake_samples talker

# 端末 2：自分のノード（/processed を足した版）
ros2 run <自分のパッケージ> slow_node

# 端末 3：送られている周期
ros2 topic hz /chatter

# 端末 4：自分のノードが処理できている周期
ros2 topic hz /processed

予想：/chatter と /processed の Hz は、それぞれいくつになるでしょうか。手がかりは、talker が 1 秒に何件送っているかと、on_msg が 1 件に何秒かかるかです。

結果の画面（予想を書いてから開く）

送り側の Hz は変わっていないのに、処理できた件数だけが減っています。/processed の Hz は、1 件に 3 秒かかるので、およそ 1 ÷ 3 ＝ 0.33 になるはずです。「遅い」はノードの中の話ですが、外からは Hz の食い違いとして見えます。これが、この回のいちばん大事なところです。

A3-4　executor を複数スレッドにする

ここでは、コールバックを呼ぶスレッドを増やして、詰まりが直るかを確かめます。

main を次のように書き換えます（/processed を足した行は、残しても外してもかまいません）。MultiThreadedExecutor は、コールバックを複数のスレッドで呼べる executor（コールバックを呼ぶ係）です。

# A3-4：executor を複数スレッドにする
from rclpy.executors import MultiThreadedExecutor

def main():
    rclpy.init()
    node = SlowNode()
    executor = MultiThreadedExecutor()
    executor.add_node(node)
    executor.spin()

予想：これで直るでしょうか。直らないなら、何が足りないのでしょうか。

手がかり：複数のスレッドがあっても、「同時に走ってよい」と決めたものしか同時には走りません。タイマーと購読は、何も指定しないと、どのコールバックグループに入るでしょうか。

A3-5　コールバックグループを 2 つに分ける

ここでは、タイマーと購読を別のグループに入れて、別々のスレッドで走れるようにします。

# A3-5：コールバックグループを 2 つに分ける（main は A3-4 のまま）
from rclpy.callback_groups import MutuallyExclusiveCallbackGroup

class SlowNode(Node):
    def __init__(self):
        super().__init__("slow_node")
        self.count = 0
        self.received = 0
        self.g_timer = MutuallyExclusiveCallbackGroup()
        self.g_sub = MutuallyExclusiveCallbackGroup()
        self.create_timer(1.0, self.on_timer, callback_group=self.g_timer)
        self.create_subscription(String, "/chatter", self.on_msg, 10,
                                 callback_group=self.g_sub)

MutuallyExclusiveCallbackGroup は「このグループの中は 1 つずつ走る」グループです。タイマー用と購読用に 1 つずつ作り、callback_group= でそれぞれを入れます。main は A3-4 の MultiThreadedExecutor のままにして、もう一度 10 秒間の回数を数えます。

予想：タイマーの回数と、受けた件数は、それぞれ 10 秒間でいくつになるでしょうか。

結果の画面（予想を書いてから開く）

タイマーは直りましたが、受けた件数は 3 秒に 1 つのままです。on_msg が 1 件に 3 秒かかることは変わっていないので、購読の側の詰まりは残っています。

A3-6　同じ処理が重なって走るようにする

ここでは、同じコールバックが重なって走れるようにして、何が困るかを確かめます。

ReentrantCallbackGroup は「同じグループの中でも、同時に走ってよい」グループです。タイマーも購読も、この 1 つのグループに入れます。on_msg は、重なったことが見えるように、入ったときと出るときにログを出し、件数を「読んでおいて、3 秒後に 1 を足して書き戻す」形に変えます。

# A3-6：1 つの ReentrantCallbackGroup に、タイマーも購読も入れる（main は A3-4 のまま）
from rclpy.callback_groups import ReentrantCallbackGroup

class SlowNode(Node):
    def __init__(self):
        super().__init__("slow_node")
        self.count = 0
        self.received = 0
        self.g = ReentrantCallbackGroup()
        self.create_timer(1.0, self.on_timer, callback_group=self.g)
        self.create_subscription(String, "/chatter", self.on_msg, 10,
                                 callback_group=self.g)

    def on_timer(self):
        self.count += 1
        self.get_logger().info(f"timer {self.count}（受けた件数 {self.received}）")

    def on_msg(self, msg):
        n = self.received                                  # 件数を読んでおく
        self.get_logger().info(f"  入る（いまの件数 {n}）")
        time.sleep(3.0)                                    # この 3 秒の間に、別の on_msg が入りうる
        self.received = n + 1                              # 読んでおいた値に 1 を足して書き戻す
        self.get_logger().info(f"  出る（書いた件数 {self.received}）")

予想：/chatter は 1 秒に 1 件届きます。10 秒たつと、受けた件数はいくつ増えるでしょうか。出力がどう混ざるかを見て、重なると何が困るかを書いてください。

手がかり：on_msg は件数を読んでから 3 秒待って書き戻します。その 3 秒の間に次の on_msg が入ってきたら、その処理はどの値を読むでしょうか。

結果の画面（予想を書いてから開く）

同時に 3 つの on_msg が走っていて、どれも「同じ件数を読み、3 秒後にその値に 1 を足して書き戻す」ので、ほかの処理が足した 1 を上書きしてしまいます。これを取りこぼしといいます。ノードの中の変数を 2 つ以上の処理が同時に書き換えると、エラーは出ないまま値が壊れます。

この回で撮っておく画面

動画は撮りません。そのかわり、下の画面を撮って残します。授業で受ける人は、実習レポートの「回 A3」の、それぞれの「画面：」の欄に貼ります。ひとりで学ぶ人も、同じ画面を残しておくと、あとで値を見直せます。最初の 1 枚は、実習を始める前に端末で TZ=Asia/Tokyo date と打ち、日付と時刻が出た画面です。授業の日に実習したことの記録になります。講義用の環境の時計は世界標準時（UTC）なので、TZ=Asia/Tokyo を付けて日本時間で出します。

• TZ=Asia/Tokyo date の画面（回の頭）
• 正常なときのタイマーの出力（実験 A3-1 の欄）
• sleep を入れたときの出力（実験 A3-2 の欄）
• ros2 topic hz /chatter と、処理できている周期（/processed）を並べた画面（実験 A3-3 の欄）
• MultiThreadedExecutor にしただけのときの出力（実験 A3-4 の欄）
• コールバックグループを 2 つに分けたときの出力（実験 A3-5 の欄）
• 重なって走ったときの出力（実験 A3-6 の欄）

本質の問い（回 A3）

• 1 つのノードの中で処理が詰まったとき、外から見て何が変わるか。自分の Hz の値で説明しなさい。

回 A4　launch の引数とテスト

この回は、#A1 のサービスと周期を変えられるノードを 1 つの launch で起動し、引数が効いているかを外から確かめ、テストが「落ちるべきときに落ちる」かまで確かめます。

1. 設計シート：#A1 で作ったサービスと、周期を変えられるノードを 1 つの launch で起動します。引数（周期・トピック名・出し先）を決めます。シートの欄は自分で埋めます。
2. AI に実装させる：launch ファイルと、テストを作らせます。テストは、ノードの計算部分を確かめる pytest にします。
3. 検証する：引数を変えて起動し、ros2 param get と ros2 topic hz で反映を確かめます（A4-1・A4-2）。そのあと colcon test を走らせ、結果を読みます（A4-3）。
4. わざと落とす：テストが落ちるように 1 行だけ変えて、落ちた表示を記録します（A4-4）。
5. つながりを見る：rqt_graph で、launch で起動したノードとトピックのつながりを見ます（A4-5）。

設計シートの例

ここでは、launch とテストの設計シートを、例と欄ごとの説明で確かめます。

launch ファイル：my_task パッケージの task.launch.py
起動するノード：my_node（my_task の中）と status_server（#A1 で作ったサービス）

引数 period：型 float、単位 s（秒）、既定 0.5
  範囲：0.05〜2.0。範囲の外の値が来たら、止めずに端の値に収める
出すトピック：/my_topic（std_msgs/msg/String）。この例では引数にせず固定
ログの出し先：端末（output="screen"）

確かめ方：
  1. period:=0.2 で起動し、ros2 param get /my_node period が 0.2 か
  2. ros2 topic hz /my_topic が 1 ÷ 0.2 ＝ 5 に近いか
テストで確かめること（pytest）：
  範囲の中はそのまま／短すぎる値は下限に／長すぎる値は上限に／
  ちょうど境界の値はそのまま／周期から 1 秒あたりの回数が出る

AI に頼むときは、たとえば次のように書いて、自分の設計シートを貼ります。

次の設計シートのとおりに、ament_python のパッケージ my_task を作ってください。
・ノード my_node は、パラメータ period の周期で /my_topic に String を出します。
・period を範囲に収める計算は、ノードとは別のファイル my_task/rate.py に関数として分けてください。
  ノードを起動しなくても、この関数だけをテストできるようにするためです。
・launch ファイル launch/task.launch.py で、引数 period を受け取り、my_node と status_server を起動してください。
・test/ の下に、設計シートの「テストで確かめること」を 1 つずつ確かめる pytest を作ってください。

（ここに自分の設計シートを貼る）

このあとのコマンドに出てくる名前は、例の設計シートの名前です。自分の設計シートで別の名前にしたときは、次のように置き換えます。

A4-1　launch の引数を変えて起動する

ここでは、コードを書き換えずに、launch の引数だけで周期を変えて起動します。

ros2 launch のあとに period:=0.2 と書くと、引数 period に 0.2 を渡して起動します。起動したら、別の端末で ros2 param get（ノードのパラメータの値を出す）と ros2 topic hz を打ちます。ros2 topic hz は Ctrl+C で止めます。

# 引数を変えて起動する
ros2 launch my_task task.launch.py period:=0.2

# 反映されているかを、外から確かめる（別の端末で）
ros2 param get /my_node period
ros2 topic hz /my_topic

A4-2　引数が効いているかを外から確かめる

ここでは、渡した引数がノードまで届き、周期として効いているかを 2 つの値で確かめます。

予想：period:=0.2 のとき、ros2 param get と ros2 topic hz はそれぞれいくつを示すでしょうか。手がかりは、周期［s］と、1 秒あたりの回数［Hz］の関係です。

続けて、範囲の外の period:=0.01 でも起動し直して、同じ 2 つを確かめます。予想：ros2 param get は 0.01 と 0.05 のどちらを示すでしょうか。ros2 topic hz はどうでしょうか。手がかりは、設計シートの「範囲の外の値は端に収める」を、ノードの中の計算でやっているのか、パラメータの値そのものを書き換えているのか、です。

結果の画面（予想を書いてから開く）

period:=0.01 のとき、例のノードは範囲に収める計算をノードの中でしているので、ros2 param get は渡したままの 0.01 を示し、ros2 topic hz は下限 0.05 秒に当たる 20 に近い値になります。パラメータの値と、実際に使われている値が違うことがある、ということです。自分のノードでどうなったかは、自分の画面で確かめてください。

A4-3　テストを走らせる

ここでは、AI が作ったテストの中身を読んでから走らせ、全部通ることを確かめます。

例のテストは、範囲に収める計算（rate.py）を、ノードを起動せずに確かめます。

# my_task/rate.py（周期の決め方だけを取り出したもの）
MIN_PERIOD = 0.05      # これより短くはしない［s］
MAX_PERIOD = 2.0       # これより長くもしない［s］


def clamp_period(period):
    """設定された周期を、この講座で扱う範囲に収める。"""
    if period < MIN_PERIOD:
        return MIN_PERIOD                 # ← 下限に収める 1 行（A4-4 でここを変える）
    if period > MAX_PERIOD:
        return MAX_PERIOD
    return period


def period_to_hz(period):
    """周期［s］から、1 秒あたりの回数を出す。"""
    return 1.0 / clamp_period(period)

# test/test_rate.py
from my_task.rate import MAX_PERIOD, MIN_PERIOD, clamp_period, period_to_hz


def test_範囲の中はそのまま():
    assert clamp_period(0.5) == 0.5


def test_短すぎる値は下限に収まる():
    assert clamp_period(0.001) == MIN_PERIOD


def test_長すぎる値は上限に収まる():
    assert clamp_period(99.0) == MAX_PERIOD


def test_境界はそのまま通る():
    assert clamp_period(MIN_PERIOD) == MIN_PERIOD
    assert clamp_period(MAX_PERIOD) == MAX_PERIOD


def test_周期から回数が出る():
    assert period_to_hz(0.5) == 2.0
    assert period_to_hz(0.001) == 1.0 / MIN_PERIOD

test_ で始まる関数が 1 つのテストです。assert のあとに書いた式が正しくなければ、そのテストは落ちます。5 つのテストは、設計シートの「テストで確かめること」の 5 つに 1 つずつ対応しています。読んだら、次のコマンドで走らせます。colcon test がテストを走らせ、colcon test-result --verbose が結果をまとめて表示します。

cd ~/ros2_ws
colcon build --packages-select my_task
colcon test --packages-select my_task
colcon test-result --verbose

A4-4　わざと落とす

ここでは、テストが「落ちるべきときに落ちる」かを、コードを 1 行壊して確かめます。

rate.py の、下限に収める 1 行（return MIN_PERIOD）を return period に変えます。短すぎる値が来ても、そのまま返してしまうようになります。

# 下限に収める 1 行を、わざと「そのまま返す」に変える
#   変える前：        return MIN_PERIOD
#   変えたあと：      return period
# 変えたら、上と同じ 4 行（build → test → test-result）をもう一度打つ

予想：5 つのテストのうち、いくつが落ちるでしょうか。どのテストが落ちるでしょうか。手がかりは、テストの中で 0.05 より短い値を渡しているのがどれか、です。

わざと落とすのは、いじわるではありません。テストは「通ったこと」より「落ちるべきときに落ちること」のほうが大事で、それを確かめずに信じると、動いていないものを動いていると思い込むからです。記録したら、変えた 1 行を元に戻し、もう一度テストが全部通ることを確かめます。

結果の画面（予想を書いてから開く）

落ちるのは、0.001 を渡している「短すぎる値は下限に収まる」と「周期から回数が出る」の 2 つです。「境界はそのまま通る」は、0.05 ちょうどを渡しているので、壊した行を通らず、落ちません。

A4-5　rqt_graph でつながりを見る

ここでは、launch で起動したものが、ノードとトピックとしてどうつながっているかを図で確かめます。

rqt_graph は、いま動いているノードとトピックのつながりを図にするツールです。

# 端末 1：launch を動かしたままにしておく
ros2 launch my_task task.launch.py period:=0.2

# 端末 2：つながりを図で見る
rqt_graph

rqt_graph が開いたら、左上の選び方を「Nodes/Topics (active)」にします。/my_topic にはまだ受ける側がいないので、そのままでは図から隠されます。上の段の「Hide:」の並びにある「Leaf topics」のチェックを外し、左上の更新ボタン（丸い矢印）を押します。予想：/my_node と /my_topic はどうつながって見えるでしょうか。status_server の /get_status は、線として出るでしょうか。手がかりは、rqt_graph が描くのは「トピック」のつながりだ、ということです。

余った時間に（発展）

lifecycle ノードというものがあります。「つなぐ前に設定する」「止めてから片付ける」という状態を持つノードです。ここでは、ROS 2 に入っている見本のノード lifecycle_talker を使って、状態を 1 段ずつ進めてみます。

# 端末 1：lifecycle ノードの見本を動かす
ros2 run lifecycle lifecycle_talker

# 端末 2：状態を見て、1 段ずつ進める
ros2 lifecycle nodes
ros2 lifecycle get /lc_talker
ros2 lifecycle set /lc_talker configure
ros2 lifecycle get /lc_talker
ros2 lifecycle set /lc_talker activate

ros2 lifecycle nodes は状態を持つノードの一覧、ros2 lifecycle get はいまの状態、ros2 lifecycle set は次の状態へ進める指示です。見本のノードの名前は /lc_talker です。予想：configure しただけで、端末 1 の lifecycle_talker は送り始めるでしょうか。activate まで進めると、どう変わるでしょうか。端末 1 のログを見比べてください。

状態を持たせると、起動の順番に左右されにくくなります。コアの #03 で「起動の順番を変えても、あとからつながる」ことを確かめましたが、設定が要るノードでは、それだけでは足りない場合がある、という話です。

この回で撮っておく画面

動画は撮りません。そのかわり、下の画面を撮って残します。授業で受ける人は、実習レポートの「回 A4」の、それぞれの「画面：」の欄に貼ります。ひとりで学ぶ人も、同じ画面を残しておくと、あとで値を見直せます。最初の 1 枚は、実習を始める前に端末で TZ=Asia/Tokyo date と打ち、日付と時刻が出た画面です。授業の日に実習したことの記録になります。講義用の環境の時計は世界標準時（UTC）なので、TZ=Asia/Tokyo を付けて日本時間で出します。

• TZ=Asia/Tokyo date の画面（回の頭）
• launch の引数を変えて起動した端末（実験 A4-1 の欄）
• ros2 param get の結果（実験 A4-2 の欄）
• colcon test-result --verbose（成功）（実験 A4-3 の欄）
• わざと落としたときのテスト結果（実験 A4-4 の欄）
• rqt_graph（実験 A4-5 の欄）

つまずきやすい点

まとめ

これで延長コース A は終わりです。ここで作った自作サービスは、延長コース D（LLM とロボット）で「AI に呼ばせる道具」としてそのまま使います。