はじめに
前回は、ABC467C - Adjacent Sums (easy)をShrike-Liteへ素朴に実装しました。
この実装では、RP2040からFPGAへ1byteずつデータを送り、そのたびにMicroPythonのSPI送受信関数を呼び出しています。
SPIクロックは1MHzなので、1byteを転送するために必要な理論時間は次の通りです。
8bit ÷ 1MHz = 8µs
ところが、これまでの実測では、1byteの送受信におよそ150～200µsかかっていました。
1byteあたり約150µsとしても、2秒間で転送できるデータ量は約13,300byte、つまり約13,300組の(A, B)を送るのが限界です。
つまり現在のSPI転送方式が制限時間内に解答を得るボトルネックになっています。
そこで今回はRP2040とForgeFPGA間のSPI転送の仕組みを見直し、より高速にデータの受け渡しができるようにしてみます。
SPI MasterはRP2040で、そのMicroPythonプログラムから通信を制御しています。
今回はFPGA側を前回の素朴実装のまま固定し、RP2040側のプログラムだけを変更しながら、SPI転送時間の内訳を調べます。
最終的には、現在のSPI通信のチューニングを通じて、ABC467CのACを目指します。
今回確認すること
最初に、次の仮説を立てました。
1byteごとに送受信バッファを作成し、CSを操作し、SPI関数を呼び出しているため、SPI通信そのものよりもMicroPython側の処理時間が支配的になっているのではないか。
この仮説を確認するため、次の順序で試験します。
- 現在の1byte送受信時間を再測定する
- 改善①: 送受信バッファを再利用する
- 改善②: CSをLowのまま保持する
- 改善③: 複数byteを一度のSPI関数呼び出しで転送する
- 改善④:
write()とwrite_readinto()を比較する - 転送データパッケージごとのCS操作の影響を確認する
- 改善⑤: SPIクロックを変更する
- 実際のABC467Cデータを送り、正しく処理できるか確認する
- 素朴実装のまま、2秒で処理できるNの上限を再探索する
FPGA側は、前回作成したabc467c_naive.binをそのまま使用します。
試験環境
試験条件は現状の実装環境をそのまま転用します。
RP2040 CPUクロック : 125MHz
SPI : SPI0
SCK : GP2
MOSI : GP3
MISO : GP0
CS : GP1
FPGA RESET : GP14
SPI Mode : Mode 0
初期SPIクロック : 1MHz
FPGA bitstream : abc467c_naive.bin
FPGA側のVerilogコードは変更しません。
なぜ1byteごとの送受信になっていたのか
今回チューニングを行う1byte送受信関数は、元々がShrike公式のstack_processorサンプルを参考にしたものです。
公式サンプルでは、FPGAへ1byteの命令を送るたびに、次の処理を行います。
送信用bytesを作る
受信用bytearrayを作る
CSをLowにする
1byteをwrite_readinto()で送受信する
CSをHighに戻す
stack_processorは、Push、Pop、演算などの命令を1byteずつ送り、結果を確認するサンプルです。
この用途では、1回の通信処理が短く、コードも理解しやすいため、1byteごとに送受信する実装は自然です。公式サンプルでは、大量データを高速転送することではなく、SPI経由でFPGAへ命令を送り、動作を確認することを重視しているようです。
連載では、この分かりやすい構成をもとにAtCoder用SPIテンプレートを作り、その後の問題でも同じ1byte送受信関数を使い続けました。
初期の問題では転送するbyte数が少なかったため、大きな問題にはなりませんでした。
しかし、ABC467CではNに比例して多数のbyteを送るため、公式サンプルから引き継いだ「1byteごとに箱を作り、CSを切り替え、SPI関数を呼ぶ」という構成の固定費が、処理時間の大部分を占めるようになりました。
今回は、バッファの扱いやCS操作、SPI関数の呼び出し方を見直し、大量データの転送効率アップに挑戦します。MicroPythonのmachine.SPIを使用する基本構造はそのまま利用し、PIOやC SDKを使った独自実装までは行いません。
参考：
現在の実装で転送時間を測る
ここから行う転送速度の試験では、ABC467Cの入力データではなく、FPGAの状態を変化させないNOPコマンドを繰り返し送信します。
今回測定したいのは、RP2040側で送受信バッファを準備し、SPI関数を呼び出して、データを転送するまでの時間です。ABC467C用の送信データを作る処理や、FPGA側の問題固有処理が測定結果へ混ざらないように、単純なNOPの転送時間を比較します。
まず、前回まで使用していた1byte送受信関数を何も変更せず、そのまま測定します。
def spi_exchange_1byte(value):
tx = bytes([value])
rx = bytearray(1)
cs.value(0)
spi.write_readinto(tx, rx)
cs.value(1)
return rx[0]
200回を個別に測定し、さらに2,000回をまとめて測定しました。
CONFIG CPU_FREQ_HZ=125000000 SPI_BAUDRATE=1000000 THEORY_US_PER_BYTE=8
SINGLE COUNT=200 MIN_US=161 AVG_US=170.3 MAX_US=203 THEORY_US=8 AVG_EXTRA_US=162.3 CHECKSUM=0x00
BATCH COUNT=2000 TOTAL_US=295892 AVG_US=147.9 THEORY_TOTAL_US=16000 EXTRA_TOTAL_US=279892 AVG_EXTRA_US=139.9 CHECKSUM=0x00
個別測定では、各回のticks_us()呼び出しも含まれるため、平均170.3µsになりました。
時間測定をループ全体の前後だけで行った場合は、1byteあたり147.9µsでした。
理論転送時間 8.0µs/byte
実測時間 147.9µs/byte
理論値との差 139.9µs/byte
現行方式では、全体の約95%がSPI線上の転送以外に使われています。
2秒間に転送できるbyte数を単純換算すると、次のようになります。
2,000,000µs ÷ 147.9µs/byte
≒ 13,500byte
現在の通信形式では、(A, B)の1組をほぼ1byteで送っているため、約13,500組分に相当します。これは、前回テストした素朴な実装の実行結果とほぼ一致しています。
現在地：2秒間に転送できる量は約13,500byte
なお、以降に示す「2秒間の転送量」は、測定した1byteあたりの時間から求めた単純換算です。AtCoder入力の読み込みや転送データの作成、FPGAのRESETなどの時間は含みません。
改善ポイントを探す
実測時間が理論値の約18倍になっている原因を探すため、先ほどの関数をもう一度見てみます。
def spi_exchange_1byte(value):
tx = bytes([value]) # 候補1
rx = bytearray(1) # 候補1
cs.value(0) # 候補2
spi.write_readinto(tx, rx) # 候補3・4
cs.value(1) # 候補2
return rx[0]
1byteを送受信するたびに、次の処理を繰り返しています。
- 送信用と受信用のバッファを新規作成する
- CSをLow／Highへ切り替える
-
write_readinto()を1byteごとに呼び出す
SPI線上で8bitを送受信する時間は8µsですが、その前後にあるMicroPython側の処理に固定的な時間がかかっている可能性があります。
さらに、ABC467Cの入力データを送信している間はMISOから返る値を使用していません。受信値が不要であれば、同時送受信を行うwrite_readinto()ではなく、送信だけを行うwrite()のほうが速い可能性もあります。
そこで、次の4つをボトルネック候補として順に確認します。
- 送受信バッファの毎回作成
- 1byteごとのCS操作
- 1byteごとのSPI関数呼び出し
- 受信値が不要な場面での
write_readinto()使用
ボトルネック候補1：送受信バッファの毎回作成
最初に、次の2行へ注目します。
tx = bytes([value])
rx = bytearray(1)
この2行だけでも、3種類のPythonオブジェクトが登場します。
valueが0xA5の場合は、概念的に次の順序で処理されます。
value = 0xA5
↓
[value] bytes関数の引数用list作成
↓ bytesへ変換
b'\xA5' 送信バッファ用bytes作成
bytearray(b'\x00') 受信バッファ用bytearray作成
それぞれの役割は次の通りです。
| 種類 | 特徴 | 今回の役割 |
|---|---|---|
list |
Pythonの値を並べる汎用的な入れ物 |
bytesを作るための一時的な材料 |
bytes |
上書き変更できないbyte列 | SPIの送信バッファ |
bytearray |
内容を変更できるbyte列 | SPIの受信バッファ |
bytes([value])では、まず[value]というlistを作り、その中の整数が0～255の範囲にあることを確認し、新しく確保したbytesへ1byteを書き込みます。
bytearray(1)では、受信用のbyte列を新しく確保し、最初の値を0で初期化します。
SPI送受信が終わってrx[0]を取り出すと、これらのオブジェクトは不要になります。次の1byteでは、また新しいlist、bytes、bytearrayを作り直します。
1byte目 : 新規list作成 → 新規bytes作成 → 新規bytearray作成 → SPI送受信
2byte目 : 新規list作成 → 新規bytes作成 → 新規bytearray作成 → SPI送受信
3byte目 : 新規list作成 → 新規bytes作成 → 新規bytearray作成 → SPI送受信
そこで、送受信バッファを関数の外で一度だけ確保し、毎回再利用します。
tx_buffer = bytearray(1)
rx_buffer = bytearray(1)
def spi_exchange_reuse_buffer(value):
tx_buffer[0] = value
cs.value(0)
spi.write_readinto(tx_buffer, rx_buffer)
cs.value(1)
return rx_buffer[0]
結果は次の通りです。
CREATE_BUFFER COUNT=2000 TOTAL_US=293392 AVG_US=146.7 THEORY_TOTAL_US=16000 EXTRA_TOTAL_US=277392 AVG_EXTRA_US=138.7 CHECKSUM=0x00
REUSE_BUFFER COUNT=2000 TOTAL_US=113130 AVG_US=56.6 THEORY_TOTAL_US=16000 EXTRA_TOTAL_US=97130 AVG_EXTRA_US=48.6 CHECKSUM=0x00
COMPARISON CREATE_BUFFER_AVG_US=146.7 REUSE_BUFFER_AVG_US=56.6 DIFF_US=90.1 IMPROVEMENT_PERCENT=61.4
送受信バッファを再利用しただけで、1byteあたりの時間は146.7µsから56.6µsへ短縮されました。
短縮時間 90.1µs/byte
短縮率 61.4%
毎回の一時list、bytes、bytearrayの生成と初期化が、非常に大きな負担になっていたことが分かります。
再利用版では、送信・受信バッファを最初に一度だけ作り、送信時には既存のtx_buffer[0]を書き換えます。
変更前 : 箱を毎回作り直してから1byte送る
変更後 : 既存の箱の中身だけ書き換えて1byte送る
2秒間の転送量へ換算すると、次のようになります。
変更前 146.7µs/byte → 約13,600byte/2秒
変更後 56.6µs/byte → 約35,300byte/2秒
現在地：2秒間の転送量が約13,600byteから約35,300byteへ増加
最初の候補は、明確なボトルネックでした。
ボトルネック候補2：1byteごとのCS操作
次に、CSのLow／High操作へ注目します。
CSは、SPI Slaveを選択する信号です。
複数のSPI SlaveがSPI Masterに接続されている場合、SPI MasterがどのSPI Slaveと通信するかを制御するための信号線ですが、Shrike-Lite上で接続しているSPI Slaveは、基板上のForgeFPGA 1台だけです。
SPI Slaveの切り替えが無いならば、CSを必ず1byteごとにHighへ戻す必要はなさそうです。
また、今回のForgeFPGA側SPI Slave実装は、CSがLowの間も8bit受信するごとに1byteのコマンドとして処理できます。
そのため、Master側のRP2040は1byteごとにCSをHighへ戻さず、複数byteを連続して送信しても問題ないはずです。
元の関数では、1byteごとに次の操作を行っています。
CS Low
1byte送受信
CS High
これを以下のようなシーケンスに変更します。
CS Low
1byte送受信
1byte送受信
1byte送受信
...
CS High
cs.value(0)やcs.value(1)では、すでに出力ピンとして設定されているGPIOの値だけを切り替えます。GPIOを毎回初期設定し直しているわけではありません。
ただし、MicroPythonからこの処理を呼び出すたびに、Pythonのメソッド呼び出し、引数の受け渡し、C側のGPIO操作を経て、再びMicroPythonの処理へ戻ります。GPIOの切り替え自体は軽くても、この一連の呼び出しを1byteごとに2回繰り返すと、その固定費が積み重なります。
そこで、write_readinto()を1byteごとに呼び出す仕組みはそのままに、CSを操作する位置だけを変更しました。
これまでの実装では、1byteを送受信する関数の中でCSをLow／Highへ切り替えていました。
def spi_exchange_cs_each(value):
tx_buffer[0] = value
cs.value(0)
spi.write_readinto(tx_buffer, rx_buffer)
cs.value(1)
return rx_buffer[0]
変更後は、CS操作を送受信ループの外へ移動します。
cs.value(0)
for value in test_data:
tx_buffer[0] = value
spi.write_readinto(tx_buffer, rx_buffer)
cs.value(1)
write_readinto()は従来どおり1byteごとに呼び出しますが、2,000byte全体を転送する間、CSはLowのまま保持されます。
比較した条件は次の二つです。
CS_EACH : 1byteごとにCSをLow／High
CS_HELD : 2,000byte全体でCSをLow／Highするのは1回だけ
送受信バッファは、どちらも事前に確保して再利用します。
結果は次の通りです。
CS_EACH COUNT=2000 TOTAL_US=115274 AVG_US=57.6 AVG_EXTRA_US=49.6 CHECKSUM=0x00
CS_HELD COUNT=2000 TOTAL_US=57506 AVG_US=28.8 AVG_EXTRA_US=20.8 CHECKSUM=0x00
COMPARISON CS_EACH_AVG_US=57.6 CS_HELD_AVG_US=28.8 DIFF_US=28.9 IMPROVEMENT_PERCENT=50.1
CSをLowのまま保持すると、1byteあたり57.6µsから28.8µsへ短縮されました。
2秒間の転送量へ換算すると、次のようになります。
変更前 57.6µs/byte → 約34,700byte/2秒
変更後 28.8µs/byte → 約69,400byte/2秒
現在地：2秒間の転送量が約34,700byteから約69,400byteへ増加
1byteごとに行っていたCS操作も、大きなボトルネックになっていました。
ボトルネック候補3：1byteごとのSPI関数呼び出し
次に、write_readinto()を1byteごとに呼び出している点を見直します。
これまでは、送信用と受信用のバッファをそれぞれ1byteで作成していました。
tx = bytearray(1)
rx = bytearray(1)
spi.write_readinto(tx, rx)
今回のSPI設定では8bitを1単位としてデータを送受信しますが、write_readinto()へ渡すバッファの長さが1byteに決まっているわけではありません。
例えば、256byteの送受信バッファを渡すと、1回のwrite_readinto()で256byteが順番に転送されます。
tx = bytearray(256)
rx = bytearray(256)
spi.write_readinto(tx, rx)
FPGA側では、これまでどおり8bit受信するたびに1byteのコマンドとして処理します。
RP2040側
256byteのバッファを1回のwrite_readinto()へ渡す
↓
SPI線上
1byteずつ順番に転送される
↓
FPGA側
受信したbyteを1byteずつ処理する
つまり、通信データの形式やFPGA側の処理を変更せず、RP2040側のSPI関数呼び出し回数だけを減らせます。
そこで、2,048byteの転送データを用意し、write_readinto()で転送する1回あたりのbyte数（=送受信バッファサイズ）を変更しながら、転送時間を測定してみました。
1byte × 2048回
2byte × 1024回
4byte × 512回
8byte × 256回
16byte × 128回
32byte × 64回
64byte × 32回
128byte × 16回
256byte × 8回
送受信バッファは事前に作成し、CSは2,048byte全体の転送が終わるまでLowに保持します。
また、この試験では転送サイズを切り替えられるように、送受信処理を関数へ分けず、測定ループ内で直接実行しています。そのため、1byte条件でも前の試験の28.8µs/byteより短い21.19µs/byteになっています。
比較するのは、同じ測定プログラム内で転送サイズだけを変更した結果です。
実測結果は次の通りです。
| 1回の転送サイズ | SPI関数呼び出し回数 | 合計時間 | 1byteあたり | 2秒間の単純換算 |
|---|---|---|---|---|
| 1byte | 2,048回 | 43,406µs | 21.19µs | 約94,000byte |
| 2byte | 1,024回 | 31,489µs | 15.38µs | 約130,000byte |
| 4byte | 512回 | 25,464µs | 12.43µs | 約161,000byte |
| 8byte | 256回 | 22,482µs | 10.98µs | 約182,000byte |
| 16byte | 128回 | 20,983µs | 10.25µs | 約195,000byte |
| 32byte | 64回 | 20,504µs | 10.01µs | 約200,000byte |
| 64byte | 32回 | 20,005µs | 9.77µs | 約205,000byte |
| 128byte | 16回 | 19,743µs | 9.64µs | 約207,000byte |
| 256byte | 8回 | 19,622µs | 9.58µs | 約209,000byte |
転送する総byte数とSPIクロックは同じですが、256byte単位では、SPI関数の呼び出し回数が2,048回から8回へ減ります。
1byte単位
write_readinto()を2,048回呼ぶ
2,048byteを転送する
256byte単位
write_readinto()を8回呼ぶ
2,048byteを転送する
その結果、1byteあたりの転送時間は21.19µsから9.58µsへ短縮されました。
1byte単位 21.19µs/byte → 約94,000byte/2秒
256byte単位 9.58µs/byte → 約209,000byte/2秒
現在地：2秒間に転送できる量は約209,000byte
最初の実装で得た約13,500byteと比較すると、SPIクロックが1MHzのままでも約15倍の転送量になりました。
長い転送ではDMAの利用を試みる
複数byte転送が速くなる理由は、MicroPython関数の固定費を複数byteへ分散できることだけではないようです。
今回使用しているMicroPython v1.26のRP2向けSPI実装では、転送長が32byte以上の場合、利用可能なDMAチャンネルを確保し、SPI送受信にDMAを使用しようとします。
DMA（Direct Memory Access）とは、CPUの代わりにメモリと周辺回路の間でデータを転送する仕組みです。
通常の転送では、CPUがSPIの送受信FIFOへデータを出し入れします。
メモリ ↔ CPU ↔ SPI
DMAを使用すると、このデータ移動をDMAコントローラへまとめて任せられます。
メモリ ↔ DMA ↔ SPI
これにより、CPUが1byteずつ送受信バッファを操作する必要がなくなります。
今回のwrite_readinto()は、DMA転送の完了を待ってからMicroPythonへ戻ります。そのため、DMA転送中に別のPython処理を実行できるわけではありませんが、RP2040のコアが転送するbyteを一つずつFIFOへ出し入れするための処理時間は短縮されるはずです。
DMA利用に関する、RP2向けMicroPython v1.26の実装は、次のようになっています。
転送長が32byte未満
→ CPUによる通常のblocking転送
転送長が32byte以上
→ 送受信用DMAチャンネルの確保を試す
├─ 2チャンネル確保できた：DMA転送
└─ 確保できなかった ：通常のblocking転送
したがって、今回の32byte、64byte、256byte転送では、DMAが利用されている可能性が高いと考えられます。
ただし、今回の試験ではDMAを有効・無効にして比較していません。そのため、どの程度がDMA単独の効果なのかは分かりません。
複数byte転送の高速化は、主に次の効果が重なった結果として扱います。
write_readinto()の呼び出し回数を削減
CS操作の回数を削減
固定費を複数byteへ分散
長い転送ではMicroPython内部でDMAの利用を試みる
参考：
ボトルネック候補4：送信だけならwrite()のほうが速いのか
ここまでの試験では、SPI送受信にwrite_readinto()を使用してきました。
しかし、ABC467Cの入力データを送っている間は、MISOから返る値を使用していません。
受信データが不要であれば、同時送受信を行うwrite_readinto()ではなく、送信だけを行うwrite()へ変更することで、さらに固定費を減らせる可能性があります。
そこで、同じ2,048byteをwrite_readinto()とwrite()で転送し、処理時間を比較しました。
| パッケージサイズ | write_readinto() |
write() |
差 |
|---|---|---|---|
| 1byte | 21.19µs/byte | 20.74µs/byte | 2.1% |
| 8byte | 10.97µs/byte | 11.02µs/byte | -0.4% |
| 32byte | 10.01µs/byte | 9.98µs/byte | 0.3% |
| 64byte | 9.77µs/byte | 9.75µs/byte | 0.2% |
| 256byte | 9.58µs/byte | 9.58µs/byte | ほぼ0% |
パッケージサイズが大きくなると差はほぼなくなりました。
SPIでは送信と受信が同時に進むため、MISOの値を使用しなくても、物理的な転送時間は変わりません。
候補4は、今回の構成では大きなボトルネックではありませんでした。速度面ではwrite()へ変更する必要はなさそうです。今後も、必要に応じてMISOを確認できるwrite_readinto()を使用します。
転送データパッケージごとのCS操作を確認する
ここまでの試験では、2,048byte全体を転送し終わるまで、CSをLowのまま保持していました。
しかし、実際のABC467Cでは、入力データを256byteなどの一定サイズの送信バッファへ分けて転送します。
例えば、1,024byteのデータを256byteのバッファで送る場合は、write_readinto()を4回呼び出します。
256byte転送
256byte転送
256byte転送
256byte転送
このとき、全体の転送が終わるまでCSをLowのまま保持する方法と、各write_readinto()の前後でCSをLow／Highへ切り替える方法の二つが考えられます。
パッケージごとにCSを切り替えると、FPGA側から見た転送の区切りが明確になります。一方、そのたびにMicroPythonからCSを操作する固定費が加わります。
今回のFPGA側SPI Slaveは、CSをLowのまま保持しても連続して受信できますが、今後の実装のために、一定サイズごとにCSのLow/High制御を行っても、大きな性能低下が起きないかを確認しておきます。
この記事では、1回のwrite_readinto()で連続して送る複数byteのまとまりを、転送データパッケージと呼ぶことにします。
同じ2,048byteを、次の二つの条件で転送しました。
HELD : 2,048byte全体でCSを1回だけLow／Highする
EACH : 各転送データパッケージでCSをLow／Highする
結果は次の通りです。
| パッケージサイズ | パッケージ数 | HELD | EACH | CS追加時間／パッケージ |
|---|---|---|---|---|
| 1byte | 2,048回 | 21.18µs/byte | 33.51µs/byte | 12.33µs |
| 2byte | 1,024回 | 15.38µs/byte | 21.54µs/byte | 12.34µs |
| 4byte | 512回 | 12.43µs/byte | 15.51µs/byte | 12.30µs |
| 8byte | 256回 | 10.97µs/byte | 12.51µs/byte | 12.32µs |
| 16byte | 128回 | 10.25µs/byte | 11.01µs/byte | 12.24µs |
| 32byte | 64回 | 10.01µs/byte | 10.39µs/byte | 12.14µs |
| 64byte | 32回 | 9.77µs/byte | 9.95µs/byte | 11.94µs |
| 128byte | 16回 | 9.64µs/byte | 9.73µs/byte | 11.38µs |
| 256byte | 8回 | 9.59µs/byte | 9.62µs/byte | 9.00µs |
CS操作による追加時間は、次の式で求めました。
CS追加時間／パッケージ
=（EACHの総転送時間 − HELDの総転送時間）
÷ パッケージ数
例えば、4byteパッケージでは次の結果になります。
HELD_TOTAL_US = 25,461µs
EACH_TOTAL_US = 31,761µs
パッケージ数 = 512回
(31,761 − 25,461) ÷ 512
≒ 12.30µs／パッケージ
1～64byteの範囲では、パッケージごとのCS操作に、およそ12µsの固定費が加わりました。
この固定費はパッケージサイズによらずほぼ一定なので、パッケージを大きくするほど1byteあたりへの影響は小さくなります。
実測では、CSを保持した場合の短縮率は、1byteパッケージでは36.8%でしたが、32byteでは3.7%、64byteでは1.9%でした。
128byteと256byteはパッケージ数が少ないため、測定値のばらつきが1パッケージあたりの計算結果へ大きく反映された可能性があります。
以上から、32byte以上、特に64byte以上の転送データパッケージであれば、パッケージごとにCSを切り替えても、速度への影響は限定的だと判断できます。
固定費を減らしてからSPIクロックを上げる
ここまでの改善により、SPIクロックを1MHzのままでも、2秒間に転送できるデータ量は約209,000byteまで増えました。
ABC467Cの最大制約はN = 200,000です。現在の通信形式では(A, B)の1組をほぼ1byteで送るため、単純な転送量だけを見れば、最大制約へ届く水準です。
しかし、実際に問題を解くには、入力データをFPGAへ転送するだけではないため、この転送性能では、最大制約の約200,000組に対する余裕がほとんどありません。
そこで、改善を行った現在の構成で、さらにSPIクロックを上げたときに転送速度がどこまで伸びるか確認します。
まずは、通信内容が正しく届いているかどうかは考えず、500kHzから8MHzまでSPIクロックを変更して、2,048byteデータの転送時間を測定します。
ここでは速度向上の見通しを立てることを優先し、通信エラーの有無は後で改めて確認します。
試験したSPIクロックは次の通りです。
500kHz
1MHz
2MHz
4MHz
8MHz
各クロックで、1回のSPI関数呼び出しあたりの転送バッファサイズを変えて、転送時間を測定しました。
なお、以下のSPIクロックはSPI()へ指定した値であり、理論時間も指定値をもとに計算しています。
1byte単位ではクロックを上げても効果が小さい
1byte単位の結果は次の通りです。
| SPIクロック指定値 | 理論時間 | 実測時間 | 理論値との差 |
|---|---|---|---|
| 500kHz | 16.000µs/byte | 29.150µs/byte | 13.150µs |
| 1MHz | 8.000µs/byte | 21.240µs/byte | 13.240µs |
| 2MHz | 4.000µs/byte | 17.188µs/byte | 13.188µs |
| 4MHz | 2.000µs/byte | 15.157µs/byte | 13.157µs |
| 8MHz | 1.000µs/byte | 14.186µs/byte | 13.186µs |
SPIクロックを変更しても、理論値との差は約13.2µsでほぼ一定です。
これは、write_readinto()を1回呼び出すたびに、およそ13µsの固定コストが発生していることを示しています。
1MHzから8MHzへSPIクロックを8倍にしても、1byte単位の転送時間は21.24µsから14.19µsへ短縮されるだけです。
MicroPython側の固定コストが支配的なままなので、SPIクロックを上げても十分な効果が得られません。
256byte単位ではクロック向上の効果が出る
256byte単位の結果は次の通りです。
| SPIクロック指定値 | 理論時間 | 実測時間 | 実効速度 |
|---|---|---|---|
| 500kHz | 16.000µs/byte | 19.076µs/byte | 0.419Mbps |
| 1MHz | 8.000µs/byte | 9.583µs/byte | 0.835Mbps |
| 2MHz | 4.000µs/byte | 4.835µs/byte | 1.655Mbps |
| 4MHz | 2.000µs/byte | 2.459µs/byte | 3.253Mbps |
| 8MHz | 1.000µs/byte | 1.275µs/byte | 6.273Mbps |
1MHzから8MHzへ変更すると、9.583µs/byteから1.275µs/byteへ短縮されました。
どうやら、複数byteをまとめて送り、SPI関数の固定コストを小さくして初めて、SPIクロック向上の効果が得られるようです。
この時点で最も速かった8MHz・256byte単位の1.275µs/byteを、2秒間の転送量へ単純換算すると次のようになります。
2,000,000µs ÷ 1.275µs/byte
≒ 1,568,000byte
速度だけを見た現在地：2秒間の転送量は約1,568,000byte
最初の実装で得た約13,500byteと比較すると、単純な転送性能は約116倍です。
ただし、この測定では、FPGAへ送ったデータが正しく受信されているかを確認していません。
理論値1µs/byteへ近づいた理由
8MHzでは、1byteをSPI線上で転送する理論時間は1µsです。
今回、送受信バッファサイズを256byteにした場合の実測値は1.275µs/byteでした。
理論転送時間 1.000µs/byte
実測時間 1.275µs/byte
理論値との差 0.275µs/byte
1byteずつ送信していた元の実装では、バッファの生成、CS操作、write_readinto()の呼び出しといった処理を、1byteごとに繰り返していました。
改善版では、送受信バッファを再利用し、CS操作とSPI関数の呼び出しを256byteごとにまとめています。さらに、長い転送ではMicroPython内部でDMAの利用も試みられます。
この結果、256byteを1回のSPI関数呼び出しで転送することで、呼び出しやCS操作に伴う固定費の1byteあたりの負担を小さくできました。
ただし、これはDMAだけの効果ではありません。バッファの再利用、CS操作回数の削減、SPI関数呼び出し回数の削減、DMA利用の可能性、SPIクロックの8MHz化を積み重ねた結果です。
速度が出ても正しく受信できるとは限らない
速度測定では、8MHz・256byte単位で、2秒間に約1,568,000byteを転送できる見通しが得られました。
ここまでの速度試験では、前述の通り、FPGAの状態を変化させないNOPを送信して転送時間を測っていました。この段階では、すべてのbitとbyteが正しく受信されたことまでは確認していません。
速度が出ても、通信内容が壊れていてはABC467Cを解くことはできません。
そこで、前回のABC467C素朴実装へ実際の入力データを送り、答えが正しく返るか確認しました。
試験したSPIクロックは次の通りです。
1MHz
2MHz
4MHz
8MHz
パッケージサイズは次の5種類です。
1byte
16byte
32byte
64byte
256byte
公式サンプル、小さな境界ケース、N=257、N=1024のパターンデータを含む7ケースを使用しました。
1～4MHzでは全ケースPASS
1MHz、2MHz、4MHzでは、すべてのパッケージサイズで全ケースPASSしました。
| SPIクロック | パッケージサイズ | PASS | FAIL |
|---|---|---|---|
| 1MHz | 1 / 16 / 32 / 64 / 256byte | 35 | 0 |
| 2MHz | 1 / 16 / 32 / 64 / 256byte | 35 | 0 |
| 4MHz | 1 / 16 / 32 / 64 / 256byte | 35 | 0 |
前回の素朴実装のFPGAでも、最大256byteを一度のwrite_readinto()で送り、受信したbyteを順番に処理できることを確認できました。
8MHzでは一部のケースがFAIL
8MHzでは、すべてのパッケージサイズで失敗が発生しました。
PACKAGE_SIZE=1 PASS=5 FAIL=2
PACKAGE_SIZE=16 PASS=5 FAIL=2
PACKAGE_SIZE=32 PASS=3 FAIL=4
PACKAGE_SIZE=64 PASS=3 FAIL=4
PACKAGE_SIZE=256 PASS=3 FAIL=4
失敗例は次の通りです。
EXPECT=4 RESULT=6
EXPECT=512 RESULT=768
EXPECT=128 RESULT=192
EXPECT=1 RESULT=0
VALID=1は返っているため、FPGAが完全に応答しなくなっているわけではありません。
一部のbitまたはbyteを誤って取り込んだ結果、計算結果が壊れているようです。
パッケージサイズ1byteでも失敗しているため、パッケージが大きすぎることが原因ではありません。
現在のSPI Slave実装では、8MHz付近で受信タイミングの余裕が不足している可能性があります。
現時点で、実データを使って正常動作を確認できた最高のSPIクロックは4MHzです。
4MHz・256byte単位の実測値は2.459µs/byteでした。これを2秒間の転送量へ単純換算すると、次のようになります。
2,000,000µs ÷ 2.459µs/byte
≒ 813,000byte
正常動作を確認できた現在地：2秒間の転送量は約813,000byte
それでも最初の実装で得た約13,500byteと比較すると、単純な転送性能は約60倍になりました。
4MHzのマージンを確認する
数ケースがPASSしただけでは、4MHzで安定して動作しているとは言い切れません。
そこで、採用候補の4MHzについて、長い入力を繰り返し送るマージン試験を行いました。
試験条件は次の通りです。
SPIクロック 4MHz
N 4096
入力ストリーム長 4100byte
パッケージサイズ 64byte / 256byte
入力パターン 2種類
各条件の反復回数 500回
入力パターンには、次の2種類を用意しました。
-
quartet：SEND_PAIRのDATA下位2bitに4種類の組み合わせが繰り返し現れるパターン -
mixed：添字をXORして生成し、結果が2048になるパターン
試験結果は次の通りです。
| パッケージサイズ | パターン | PASS | FAIL | 平均時間 |
|---|---|---|---|---|
| 64byte | quartet | 500 | 0 | 12,005.5µs |
| 64byte | mixed | 500 | 0 | 11,966.4µs |
| 256byte | quartet | 500 | 0 | 10,569.2µs |
| 256byte | mixed | 500 | 0 | 10,568.9µs |
PHASE_SUMMARY PHASE=STABILITY PASS=2000 FAIL=0 TOTAL=2000 PASS
4条件で合計2,000回を実行し、すべてPASSしました。
1回あたり4,100byteを送っているため、FPGAへ送った入力データの合計は次の通りです。
4,100byte × 2,000回 = 8,200,000byte
約8.2MBの実データをエラーなしで処理できたため、今回の構成では4MHzを安定して使用できると判断します。
4MHzより上の境界も確認する
4MHzが失敗境界のすぐ手前ではないことを確認するため、次のSPIクロック指定値でも試験しました。
4.0MHz
4.5MHz
5.0MHz
6.0MHz
7.0MHz
8.0MHz
パッケージサイズは64byteと256byte、入力パターンはquartetとmixedを使用し、それぞれ50回ずつ実行しました。
4MHzから7MHzまでの指定では、すべての試験がPASSしました。
8MHzではquartetはPASSしましたが、mixedで多数のFAILが発生しました。
| SPIクロック指定 | パッケージサイズ | パターン | PASS | FAIL |
|---|---|---|---|---|
| 8MHz | 64byte | quartet | 50 | 0 |
| 8MHz | 64byte | mixed | 11 | 39 |
| 8MHz | 256byte | quartet | 50 | 0 |
| 8MHz | 256byte | mixed | 13 | 37 |
quartetの期待値は0なので、一部の入力を誤って受信しても結果に表れにくい可能性があります。
一方、mixedでは期待値2048に対して、0、1024、3072などが返りました。8MHzでは、入力データを安定して受信できていないと判断できます。
また、処理時間を見ると、4.0MHz指定と4.5MHz指定、6.0MHz指定と7.0MHz指定では、ほぼ同じ値になりました。RP2040のSPIクロックは、指定した周波数そのものではなく、利用可能な分周比に応じた実クロックになります。また、今回の処理時間にはMicroPython側の固定費も含まれるため、指定値の差が処理時間へそのまま現れなかったと考えられます。
さて、今回の目的は最高クロックを確定させることではなく、4MHzが安定して使用できる条件であるかを判断することです。
これまでの結果から、今後は次の条件を使用します。
RP2040 CPUクロック : 125MHz
SPIクロック : 4MHz
実際のABC467C処理時間
4MHzで、最も長いN=1024のケースにかかった時間は次の通りです。
| パッケージサイズ | 処理時間 |
|---|---|
| 1byte | 58,599µs |
| 16byte | 6,381µs |
| 32byte | 4,798µs |
| 64byte | 3,906µs |
| 256byte | 3,228µs |
1byteずつ送った場合は58.6msでしたが、256byteずつ送ると3.2msまで短縮されました。
約18倍高速化
FPGA側の問題固有回路を変更せず、RP2040側の転送単位とCS操作を見直しただけで、これだけ大きな差が生じました。
素朴実装のまま処理可能なNを再探索する
ここまでの試験では、FPGA側の素朴実装も、MOSIとMISOのbyte構成も変更していません。
そこで最後に、前回使用した処理上限探索用のMicroPythonプログラムを、次の条件へ変更しました。
FPGA bitstream abc467c_naive.bin
FPGA側の処理 素朴実装のまま
MOSI/MISO形式 変更なし
RP2040 CPUクロック 125MHz
SPIクロック 4MHz
転送パッケージ 最大256byte
NとA_1を表す5byteは一度に送信し、以降のSEND_PAIRは256byte単位で送信します。
ベンチマークではAとBをすべて0としています。同じSEND_PAIRを格納した256byteのバッファを繰り返し再利用するため、Nが大きくなっても巨大な配列を確保する必要はありません。
また、現在の通信形式ではNを18bitで送信しています。そのため、探索上限をAtCoderの最大制約200,000ではなく、18bitで表せる最大値まで広げました。
2^18 - 1 = 262,143
まず、公式サンプルなどに加えて、256byteの境界をまたぐN=257のケースを確認しました。
NAME=official_sample_1 N=3 PACKAGES=1 ... EXPECT=1 RESULT=1 PASS TIME_US=245
NAME=official_sample_2 N=2 PACKAGES=1 ... EXPECT=0 RESULT=0 PASS TIME_US=233
NAME=official_sample_3 N=10 PACKAGES=1 ... EXPECT=4 RESULT=4 PASS TIME_US=247
NAME=all_zero N=5 PACKAGES=1 ... EXPECT=0 RESULT=0 PASS TIME_US=232
NAME=two_elements_mismatch N=2 PACKAGES=1 ... EXPECT=1 RESULT=1 PASS TIME_US=228
NAME=package_boundary_n257 N=257 PACKAGES=2 ... EXPECT=128 RESULT=128 PASS TIME_US=903
FUNCTION_SUMMARY PASS=6 FAIL=0 TOTAL=6
機能テストはすべてPASSしました。
続いて、指数探索で2秒を超えるNを探します。
EXPAND N=1024 PACKAGES=5 TIME_US=2773 PASS
EXPAND N=2048 PACKAGES=9 TIME_US=5309 PASS
EXPAND N=4096 PACKAGES=17 TIME_US=10384 PASS
EXPAND N=8192 PACKAGES=33 TIME_US=20548 PASS
EXPAND N=16384 PACKAGES=65 TIME_US=40860 PASS
EXPAND N=32768 PACKAGES=129 TIME_US=81474 PASS
EXPAND N=65536 PACKAGES=257 TIME_US=162804 PASS
EXPAND N=131072 PACKAGES=513 TIME_US=325187 PASS
EXPAND N=262143 PACKAGES=1025 TIME_US=650233 PASS
18bitで表せる最大のN=262,143でも、処理時間は約0.65秒でした。
SEARCH_BOUNDARY PASS_N=262143 FAIL_N=NONE LIMIT_REASON=18BIT_PROTOCOL_MAX
BENCHMARK_ESTIMATE TIME_LIMIT_US=2000000 AT_LEAST_N=262143 PROTOCOL_MAX_REACHED=1
2秒を超える境界を探すために指数探索と二分探索を用意しましたが、二分探索へ進む前に、時間制限ではなく現在の通信仕様の上限へ到達しました。
N=262,143では、入力と答え受信を合わせて約262,150byteを転送します。4MHz SPIにおける物理転送時間は1byteあたり2µsなので、SPI線上だけでも約524ms必要です。
実測値は約650msであり、1byteあたりでは次の値になります。
650,233µs ÷ 262,150byte ≒ 2.48µs/byte
以前の転送試験で得た、4MHz・256byte単位の約2.46µs/byteともほぼ一致しています。
NOPを送った単純な転送試験だけでなく、ABC467Cのコマンドを解釈しながら処理する実際のNaive版でも、理論値2µs/byteに対して約2.48µs/byteで動作しました。
ABC467Cの最大制約N=200,000について単純比例で見積もると、SPI送信とFPGAの処理、答え受信は約0.50秒です。
ただし、このベンチマークでは次の時間を測定範囲に含めていません。
- AtCoder入力の読み込み
- AとBから送信データを作る処理
- 転送バッファの準備
- GC
- FPGAのRESET
したがって、約0.50秒はプログラム全体の所要時間ではなく、準備済みのデータをFPGAへ送り、答えを受け取る部分の目安です。
それでも、FPGA側を変更せず、RP2040側の転送方法とSPIクロックを見直しただけで、AtCoderの最大制約を大きく超えるNまで余裕を持って転送できるようになりました。
これでACがとれそうですね。
ここまでの結果
今回の改善版では、MicroPythonからSPI関数を呼び出す回数を減らすため、転送データパッケージを最大256byteとしました。最後のパッケージだけは、残っているデータの長さに合わせて送信します。
このパッケージサイズはRP2040側の転送単位であり、FPGA側の並列演算幅ではありません。今回のFPGAは、256byteを受信している間も従来どおり1byteずつ処理します。
この構成を、今後のSPI通信で使用する基本形とします。ただし、常に256byte単位で送るわけではなく、問題で扱うデータ幅や通信手順に応じて変更します。
試験の結果、最終的に次の構成を採用しました。
SPIクロック 4MHz
転送データパッケージ 最大256byte
送受信バッファ 事前に確保して再利用
CS操作 転送データパッケージごと
SPI関数 write_readinto()
最初の1byte送受信方式と比較すると、転送速度は次のように高速化できました。
| 段階 | 実測時間 | 2秒間の単純換算 |
|---|---|---|
| 前回の実装 | 約147.9µs/byte | 約13,500byte |
| バッファ再利用 | 約56.6µs/byte | 約35,300byte |
| CSをLowのまま保持 | 約28.8µs/byte | 約69,400byte |
| 256byte単位・1MHz | 約9.58µs/byte | 約209,000byte |
| 256byte単位・4MHz（NOP転送試験・CS保持） | 約2.46µs/byte | 約813,000byte |
write()とwrite_readinto()の差は、転送データパッケージを大きくするとほぼ見られませんでした。そのため、MISOから答えを受信する処理と共通化しやすいwrite_readinto()を引き続き使用します。
MicroPythonスクリプト
以下が、4MHz・最大256byteの転送データパッケージを使って、機能テストと処理上限探索を行った最終版のMicroPythonプログラムです。
ファイル名はabc467c_spi_call_improved_test_estimate.pyとしました。
from machine import Pin, SPI
import gc
import time
import shrike
# ===== 共通部分：bitstream名とShrike-Liteのピン設定 =====
BITSTREAM = "abc467c_naive.bin"
SCK = 2
CS = 1
MOSI = 3
MISO = 0
FPGA_RESET = 14
# ===== SPI転送設定 =====
SPI_BAUDRATE = 4_000_000
PACKAGE_SIZE = 256
# ===== ベンチマーク設定 =====
TIME_LIMIT_US = 2_000_000
MIN_N = 2
# 現在の通信形式では、Nを18bitで送信する。
# そのため、FPGA側を変更せずに扱える最大値は2^18-1。
PROTOCOL_MAX_N = (1 << 18) - 1
MAX_N = PROTOCOL_MAX_N
INITIAL_N = 1_024
ESTIMATE_UNIT = 100
# ===== 共通部分：FPGAへのbitstream書き込みとリセット =====
shrike.reset()
shrike.flash(BITSTREAM)
reset_pin = Pin(FPGA_RESET, Pin.OUT, value=1)
reset_pin.value(0)
time.sleep_ms(100)
reset_pin.value(1)
time.sleep_ms(100)
# ===== 共通部分：SPI Masterの初期化 =====
cs = Pin(CS, Pin.OUT, value=1)
spi = SPI(
0,
baudrate=SPI_BAUDRATE,
polarity=0,
phase=0,
bits=8,
firstbit=SPI.MSB,
sck=Pin(SCK),
mosi=Pin(MOSI),
miso=Pin(MISO)
)
# ===== ABC467C固有処理 =====
NOP = 0b000
SEND_N_17_15 = 0b001
SEND_N_14_10 = 0b010
SEND_N_9_5 = 0b011
SEND_N_4_0 = 0b100
SEND_A1 = 0b101
SEND_PAIR = 0b110
RESET = 0b111
def make_command(command, data=0):
return (command << 5) | (data & 0x1F)
NOP_BYTE = make_command(NOP)
RESET_BYTE = make_command(RESET)
PAIR_ZERO = make_command(SEND_PAIR, 0)
# ===== 再利用するSPI送受信バッファ =====
# RESETと答え受信に使用する1byteバッファ
single_tx = bytearray(1)
single_rx = bytearray(1)
# NとA_1を送る5byteヘッダ
header_tx = bytearray(5)
header_rx = bytearray(5)
# ベンチマーク用のPAIR_ZEROパッケージ
pair_zero_tx = bytearray(PACKAGE_SIZE)
pair_zero_rx = bytearray(PACKAGE_SIZE)
for i in range(PACKAGE_SIZE):
pair_zero_tx[i] = PAIR_ZERO
# ===== SPI送受信 =====
def spi_transfer(tx_buffer, rx_buffer):
# 複数byteを一度のwrite_readinto()で転送する。
cs.value(0)
spi.write_readinto(tx_buffer, rx_buffer)
cs.value(1)
def spi_exchange_1byte(value):
# RESETと答え受信では、再利用する1byteバッファを使用する。
single_tx[0] = value
spi_transfer(single_tx, single_rx)
return single_rx[0]
# ===== ABC467C固有の通信処理 =====
def set_header(n, a_first):
if n < MIN_N or n > PROTOCOL_MAX_N:
raise ValueError("N is outside the 18bit protocol range")
header_tx[0] = make_command(
SEND_N_17_15,
(n >> 15) & 0x07
)
header_tx[1] = make_command(
SEND_N_14_10,
(n >> 10) & 0x1F
)
header_tx[2] = make_command(
SEND_N_9_5,
(n >> 5) & 0x1F
)
header_tx[3] = make_command(
SEND_N_4_0,
n & 0x1F
)
header_tx[4] = make_command(
SEND_A1,
a_first
)
def reset_problem():
# RESETの処理結果を次のNOPでSPI送信側へ反映させる。
spi_exchange_1byte(RESET_BYTE)
spi_exchange_1byte(NOP_BYTE)
def receive_answer():
rx_hi = spi_exchange_1byte(NOP_BYTE)
rx_mid = spi_exchange_1byte(NOP_BYTE)
rx_lo = spi_exchange_1byte(NOP_BYTE)
valid = (rx_hi >> 7) & 0x01
answer = (
((rx_hi & 0x03) << 16)
| (rx_mid << 8)
| rx_lo
)
return valid, answer, (rx_hi, rx_mid, rx_lo)
# ===== 機能テスト用の入力stream作成 =====
def build_input_stream(a_values, b_values):
n = len(a_values)
if n < MIN_N:
raise ValueError("N must be at least 2")
if n > PROTOCOL_MAX_N:
raise ValueError("N exceeds the 18bit protocol range")
if len(b_values) != n - 1:
raise ValueError("len(B) must be N - 1")
# N送信4byte、A_1送信1byte、PAIR送信N-1byte
tx_stream = bytearray(n + 4)
index = 0
tx_stream[index] = make_command(
SEND_N_17_15,
(n >> 15) & 0x07
)
index += 1
tx_stream[index] = make_command(
SEND_N_14_10,
(n >> 10) & 0x1F
)
index += 1
tx_stream[index] = make_command(
SEND_N_9_5,
(n >> 5) & 0x1F
)
index += 1
tx_stream[index] = make_command(
SEND_N_4_0,
n & 0x1F
)
index += 1
tx_stream[index] = make_command(
SEND_A1,
a_values[0]
)
index += 1
for i in range(n - 1):
# DATA bit1=A_(i+1)、bit0=B_i
pair_data = (
((a_values[i + 1] & 0x01) << 1)
| (b_values[i] & 0x01)
)
tx_stream[index] = make_command(
SEND_PAIR,
pair_data
)
index += 1
return tx_stream
def make_packages(tx_stream):
# パッケージ生成は測定前に行う。
packages = []
start = 0
stream_length = len(tx_stream)
while start < stream_length:
end = start + PACKAGE_SIZE
if end > stream_length:
end = stream_length
tx_package = tx_stream[start:end]
rx_package = bytearray(len(tx_package))
packages.append((tx_package, rx_package))
start = end
return packages
def send_packages(packages):
for tx_package, rx_package in packages:
spi_transfer(tx_package, rx_package)
def run_test_case(name, a_values, b_values, expected):
n = len(a_values)
# 入力streamとパッケージは測定前に作る。
tx_stream = build_input_stream(
a_values,
b_values
)
packages = make_packages(tx_stream)
reset_problem()
# GCの実行時間はTIME_USに含めない。
gc.collect()
start_us = time.ticks_us()
send_packages(packages)
valid, result, rx_bytes = receive_answer()
elapsed_us = time.ticks_diff(
time.ticks_us(),
start_us
)
passed = valid == 1 and result == expected
status = "PASS" if passed else "FAIL"
print(
"NAME={} N={} PACKAGES={} RX=[0x{:02X},0x{:02X},0x{:02X}] "
"VALID={} EXPECT={} RESULT={} {} TIME_US={}".format(
name,
n,
len(packages),
rx_bytes[0],
rx_bytes[1],
rx_bytes[2],
valid,
expected,
result,
status,
elapsed_us
)
)
return passed
# ===== 参照用のABC467C計算 =====
def solve_reference(a_values, b_values):
value0 = 0
value1 = 1
cost0 = 1 if value0 != a_values[0] else 0
cost1 = 1 if value1 != a_values[0] else 0
for i in range(len(b_values)):
value0 ^= b_values[i]
value1 ^= b_values[i]
if value0 != a_values[i + 1]:
cost0 += 1
if value1 != a_values[i + 1]:
cost1 += 1
return cost0 if cost0 < cost1 else cost1
def make_boundary_test_case(n):
# 256byte境界をまたぐ機能テストを作る。
a_values = [0] * n
b_values = [0] * (n - 1)
for i in range(n):
a_values[i] = (
i
^ (i >> 2)
^ (i >> 5)
) & 0x01
for i in range(n - 1):
b_values[i] = (
(i * 3)
^ (i >> 1)
^ 1
) & 0x01
expected = solve_reference(
a_values,
b_values
)
return (
"package_boundary_n{}".format(n),
a_values,
b_values,
expected
)
# ===== ベンチマーク用のゼロ入力転送 =====
def prepare_zero_case(n):
# 測定中にバッファを生成しないよう、
# ヘッダと最終パッケージを測定前に準備する。
set_header(n, 0)
pair_count = n - 1
full_package_count = pair_count // PACKAGE_SIZE
tail_count = pair_count % PACKAGE_SIZE
if tail_count == 0:
tail_tx = None
tail_rx = None
else:
tail_tx = bytearray(tail_count)
tail_rx = bytearray(tail_count)
for i in range(tail_count):
tail_tx[i] = PAIR_ZERO
return (
full_package_count,
tail_count,
tail_tx,
tail_rx
)
def send_zero_case(
full_package_count,
tail_count,
tail_tx,
tail_rx
):
# NとA_1を5byteまとめて送信する。
spi_transfer(header_tx, header_rx)
# PAIR_ZEROを256byte単位で繰り返し送信する。
for _ in range(full_package_count):
spi_transfer(pair_zero_tx, pair_zero_rx)
# 最後の端数だけ短いパッケージで送信する。
if tail_count != 0:
spi_transfer(tail_tx, tail_rx)
def measure_zero_case(n, label="SEARCH"):
(
full_package_count,
tail_count,
tail_tx,
tail_rx
) = prepare_zero_case(n)
reset_problem()
# 各測定の直前にGCを実行する。
# バッファ生成時間とGC時間はTIME_USに含めない。
gc.collect()
start_us = time.ticks_us()
send_zero_case(
full_package_count,
tail_count,
tail_tx,
tail_rx
)
valid, answer, _ = receive_answer()
elapsed_us = time.ticks_diff(
time.ticks_us(),
start_us
)
correct = valid == 1 and answer == 0
within_limit = (
correct
and elapsed_us <= TIME_LIMIT_US
)
package_count = (
1
+ full_package_count
+ (1 if tail_count != 0 else 0)
)
print(
"{} N={} PACKAGES={} TIME_US={} VALID={} ANSWER={} {}".format(
label,
n,
package_count,
elapsed_us,
valid,
answer,
"PASS" if within_limit else "FAIL"
)
)
return within_limit, elapsed_us, correct
# ===== 指数探索と二分探索 =====
def find_upper_bound():
# まず指数探索で2秒を超える最初のNを探す。
# 現在の18bit通信形式で扱える最大値まで探索する。
n = INITIAL_N
last_pass = MIN_N
while True:
if n > MAX_N:
n = MAX_N
passed, _, correct = measure_zero_case(
n,
"EXPAND"
)
if not correct:
raise RuntimeError(
"FPGA reply error during benchmark"
)
if not passed:
return last_pass, n
last_pass = n
if n == MAX_N:
# 2秒境界へ到達する前に18bit上限へ到達した。
return MAX_N, MAX_N
n *= 2
def binary_search_limit(low_pass, high_fail):
if low_pass == MAX_N:
return MAX_N, None
low = low_pass
high = high_fail
# 実測値にはばらつきがあるため、厳密な最大値ではなく
# おおよその2秒境界を得る目的で二分探索する。
while high - low > 1:
mid = (low + high) // 2
passed, _, correct = measure_zero_case(
mid,
"BINARY"
)
if not correct:
raise RuntimeError(
"FPGA reply error during benchmark"
)
if passed:
low = mid
else:
high = mid
return low, high
def round_to_estimate_unit(value):
estimated = (
(value + (ESTIMATE_UNIT // 2))
// ESTIMATE_UNIT
) * ESTIMATE_UNIT
if estimated < MIN_N:
return MIN_N
if estimated > MAX_N:
return MAX_N
return estimated
# ===== 機能テスト =====
TEST_CASES = [
(
"official_sample_1",
[1, 1, 1],
[1, 1],
1
),
(
"official_sample_2",
[1, 1],
[0],
0
),
(
"official_sample_3",
[0, 0, 0, 1, 1, 0, 1, 0, 1, 0],
[0, 1, 0, 1, 0, 1, 0, 1, 0],
4
),
(
"all_zero",
[0, 0, 0, 0, 0],
[0, 0, 0, 0],
0
),
(
"two_elements_mismatch",
[0, 1],
[0],
1
),
make_boundary_test_case(257),
]
print(
"CONFIG "
"SPI_BAUDRATE={} "
"PACKAGE_SIZE={} "
"TIME_LIMIT_US={} "
"MAX_N={} "
"PROTOCOL_BITS=18".format(
SPI_BAUDRATE,
PACKAGE_SIZE,
TIME_LIMIT_US,
MAX_N
)
)
pass_count = 0
for name, a_values, b_values, expected in TEST_CASES:
if run_test_case(
name,
a_values,
b_values,
expected
):
pass_count += 1
fail_count = len(TEST_CASES) - pass_count
print(
"FUNCTION_SUMMARY PASS={} FAIL={} TOTAL={}".format(
pass_count,
fail_count,
len(TEST_CASES)
)
)
if fail_count != 0:
raise RuntimeError("Functional test failed")
# ===== 2秒前後となるNの推定 =====
# MicroPythonの実行時間にはばらつきがあるため、
# 厳密な最大値ではなく、おおよその目安として扱う。
low_pass, high_fail = find_upper_bound()
raw_pass_n, raw_fail_n = binary_search_limit(
low_pass,
high_fail
)
if raw_fail_n is None:
print(
"SEARCH_BOUNDARY "
"PASS_N={} "
"FAIL_N=NONE "
"LIMIT_REASON=18BIT_PROTOCOL_MAX".format(
raw_pass_n
)
)
print(
"BENCHMARK_ESTIMATE "
"TIME_LIMIT_US={} "
"AT_LEAST_N={} "
"PROTOCOL_MAX_REACHED=1".format(
TIME_LIMIT_US,
raw_pass_n
)
)
else:
estimated_n = round_to_estimate_unit(
raw_pass_n
)
print(
"SEARCH_BOUNDARY PASS_N={} FAIL_N={}".format(
raw_pass_n,
raw_fail_n
)
)
print(
"BENCHMARK_ESTIMATE "
"ESTIMATED_N_AROUND_2S={} "
"ROUND_UNIT={} "
"PROTOCOL_MAX_REACHED=0".format(
estimated_n,
ESTIMATE_UNIT
)
)
まとめ
今回は、1MHz SPIで1byteの転送に約150µsかかっていた原因を調べました。
主な原因は、送受信バッファの生成、CS操作、write_readinto()の呼び出しを、1byteごとに繰り返していたことでした。
そこで、送受信バッファを再利用し、複数byteをまとめて転送するように変更したうえで、SPIクロックを4MHzへ上げました。
結果として、1byteあたりの転送時間は約2.5µsまで短縮されました。最初の約150µs/byteと比較すると、約60倍の高速化です。
この構成で処理上限を探索したところ、現在の18bit通信形式で扱える最大のN = 262,143を約0.65秒で処理できました。
RP2040側のSPI転送方法を改善するだけで、ABC467Cの最大制約N = 200,000を、FPGA側の素朴実装のまま処理できる見込みが立ちました。
次回
次回は、1組2bitの(A_{i+1}, B_i)を4組ずつ1byteへ詰める通信形式と、それを処理するFPGA側のPrefix XOR回路を実装します。
通信量を減らしてさらなる解答時間短縮を目指し、ABC467Cの仕上げとします。
お楽しみに。
前回：
Shrike-LiteでAtCoder問題を解く(14)：ABC467C - Adjacent Sums (easy)（素朴実装編）
次回：
Shrike-LiteでAtCoder問題を解く(16)：ABC467C - Adjacent Sums (easy)（4組パック＋Prefix XOR編）