OpenRTM-aist » OpenRTM-aist (Python)

実装¶

Push型
以下のクラスを実装した。

InPortSHMProvider
インポートプロバイダクラス
InPortCorbaCdrProviderクラスを継承している。
InterfaceTypeは「shared_memory」に設定する。

InPortSHMConsumer
インポートコンシュマークラス
InPortCorbaCdrConsumerクラスを継承している。

FactoryInit関数でInPortSHMProviderInit関数とInPortSHMConsumerInit関数を呼び出すことでファクトリが登録され「shared_memory」のInterfaceTypeで通信可能になる。

通信までの手順を説明する。

InPortSHMProviderのコンストラクタ
self.shm_address = str(OpenRTM_aist.uuid1())
self._properties.append(OpenRTM_aist.NVUtil.newNV("dataport.shared_memory.address",self.shm_address))
UUIDにより共有メモリの空間名を生成し、dataport.shared_memory.addressというプロパティに保存する。

InPortSHMConsumerクラスのinit関数

self.shm_address = prop.getProperty("shared_memory.address")
if self.shm_address:
if self._shmem is None:
self._shmem = mmap.mmap(0, 256, self.shm_address, mmap.ACCESS_WRITE)

プロパティから取得した空間名から共有メモリを生成する。
init関数はInPortBaseクラスのcreateConsumer関数とInPortPushConnectorクラスのコンストラクタの2回呼び出されるが、createConsumer関数で呼び出された時には空間名のプロパティがないため共有メモリを生成するのはInPortPushConnectorクラスのコンストラクタとなっている。
現在は動作確認のためにデータサイズを256に固定しているが、送信するデータのサイズによって変更する必要がある。

InPortSHMConsumerクラスのput関数
self._shmem.seek(os.SEEK_SET)
self._shmem.write(data)
data_size = len(data)
mar_data_size = cdrMarshal(CORBA.TC_ushort, data_size)
return self.convertReturnCode(inportcdr.put(mar_data_size))

共有メモリにデータを書き込み、CORBAで符号化したデータサイズだけ送信している。
この方法とは別にデータサイズも共有メモリに書き込むという方法も考えられる。

InPortSHMProviderクラスのput関数
data_size = cdrUnmarshal(CORBA.TC_ushort, data)
self._shmem = mmap.mmap(0, data_size, self.shm_address, mmap.ACCESS_READ)
shm_data = self._shmem.read(data_size)
ret = self._connector.write(shm_data)
return self.convertReturn(ret, shm_data)

データサイズを復号化し、共有メモリからデータを取り出す。

Pull型
以下のクラスを実装した。

OutPortSHMProvider
アウトポートプロバイダクラス
OutPortCorbaCdrProviderクラスを継承している。
OutterfaceTypeは「shared_memory」に設定する。

OutPortSHMConsumer
アウトポートコンシュマークラス
OutPortCorbaCdrConsumerクラスを継承している。

Push型と同じくOutPortSHMProviderInit関数とOutPortSHMConsumerInit関数を呼び出すことでファクトリを登録する。

通信の手順を説明する。

OutPortSHMProviderクラスのコンストラクタ
self.shm_address = str(OpenRTM_aist.uuid1())
self._properties.append(OpenRTM_aist.NVUtil.newNV("dataport.shared_memory.address",self.shm_address))
self._shmem = mmap.mmap(0, 256, self.shm_address, mmap.ACCESS_WRITE)

空間名の生成、プロパティへの保存、共有メモリの生成を行う。

OutPortSHMConsumerクラスのinit関数
self.shm_address = prop.getProperty("shared_memory.address")

空間名を取得する。
現在のところ何故かOutPortPullConnectorのコンストラクタではOutPortConsumerオブジェクトのinit関数が呼ばれないため、実験のためにinit関数が呼ばれるようにしてある。

OutPortSHMProviderクラスのget関数
self._shmem.seek(os.SEEK_SET)
self._shmem.write(cdr[0])
data_size = len(cdr[0])
mar_data_size = cdrMarshal(CORBA.TC_ushort, data_size)
return self.convertReturn(ret, mar_data_size)

共有メモリにデータを書き込み符号化したデータサイズを送信する。

OutPortSHMConsumerクラスのget関数
data_size = cdrUnmarshal(CORBA.TC_ushort, cdr_data)
self._shmem = mmap.mmap(0, data_size, self.shm_address, mmap.ACCESS_READ)
shm_data = self._shmem.read(data_size)
data[0] = shm_data

データサイズを復号化して共有メモリから読み込んだデータを書き込む。

テスト¶

添付したテスト用コードでテストを行った。
テスト用コードは以下の動作を行う。

setUp関数

マネージャ初期化
self.manager = OpenRTM_aist.Manager.init(sys.argv)
self.manager.activateManager()

データポート生成
self._d_in = RTC.TimedLong(RTC.Time(0,0),0)
self._inIn = OpenRTM_aist.InPort("in", self._d_in)
prop = OpenRTM_aist.Properties()
self._inIn.init(prop)
self.inport_obj = self._inIn.getPortRef()
→インポートとアウトポートを生成

test_Push関数

データが正常に通信できているかの確認
self._d_out.data = 100
self._outOut.write()
ret = self._inIn.isNew()
self.assertTrue(ret)
data = self._inIn.read()
self.assertEqual(data.data, 100)

test_Pull関数

データが正常に通信できているかの確認

self._d_out.data = 100
self._outOut.write()
data = self._inIn.read()
self.assertEqual(data.data, 100)

共有メモリ通信のためのインターフェースとしてSharedMemory.idlを作成した。

interface SharedMemory
{
    void create_memory(in long memory_size, in string shm_address);
    void open_memory(in long memory_size, in string shm_address);
    void close_memory(in boolean unlink);
    void setInterface(in SharedMemory sm);
    PortStatus put();
    PortStatus get();
};

create_memory¶

初期化を行う。
Windowsではページングファイル上に領域を確保して、Linuxでは/dev/shm/以下にファイルを作成する。作成したファイルの内容をmmapにより仮想アドレスにマッピングする。

open_memory¶

既に確保済みの共有メモリ領域を仮想アドレスにマッピングする。

close_memory¶

アンマップ、共有領域の削除を行う。
unlinkをTrueにした場合は/dev/shm/以下に作成したファイルを削除する。

setInterface¶

通信先のCORBAインターフェースを設定する。
設定することによりcreate_memoryを呼び出したときに、通信先での共有メモリ領域のマッピングを自動的に行う。また設定したメモリのサイズをデータのサイズが上回った場合に、通信先の初期化も自動的に行う。

put¶

送信を知らせる。
putを呼び出す前に共有メモリにデータを書き込んでおくことにより、送信先でデータを読み込む事ができる。

get¶

送信を要求する。
getの処理内で共有メモリへのデータの書き込みを行う事により、送信元でデータを読み込む事ができる。

また共有メモリ操作クラスとしてSharedMemory.pyを作成した。
SharedMemoryクラスは上記のインターフェースによる操作ができるようになっている。

SharedMemoryクラスには以下の関数を定義してある。

string_to_MemorySize¶

1M、1k等のデータのサイズを表す文字列を数値に変換する

write¶

データを書き込む
データサイズを先頭8byteに保持させており、通信するデータはその後ろに書き込む。

read¶

データを読み込む。
先頭8byteからデータサイズを読み込み、取得したサイズだけデータを読み込む。

またInPortSHMProviderはInPortProvider、SharedMemoryを継承するように変更した。
OutPortSHMProviderはOutPortProvider、SharedMemoryを継承するように変更した。

以前はプロパティのdataport.shared_memory.addressで共有メモリ空間名を保持していたが、SharedMemoryインターフェースで渡せるようになったため削除した。

また共有メモリの初期の大きさはOutPort側でcreate_memory関数を呼び出すとInPort側のマッピングも自動的に行われるようになったため、rtc.confのport.outport.out.shem_default_sizeで設定するだけで済むようになった。具体的には以下のように指定する。

port.outport.out.shem_default_size: 10M
port.outport.out.shem_default_size: 100k
port.outport.out.shem_default_size: 300