@geolab/serverのデータ読み出し高速化
データ読み取り速度が2025-09-06現在だと100msが関の山
データ読み出しのたびにDB書き込みをawaitしているせい
またdata loggerとserverが別々にデータを読みだしていて非効率かつ整合性がとれない
とくにserverは/api/readなどの呼び出しごとに全く違うタイミングで得たデータを流している
方針
ModbusServiceではraw valueだけ配信し、configを使った計算は各自で行う
configによる計算がどれだけ重くなっても、読み出し速度に影響を与えないようにする
ModbusServiceからのデータ読み出しをブロッキングしない
ModbusServiceからのデータ読み出しは、すべて単一のストリームから読みだす
実装段階
だいぶ大規模なrefactoringになるため、少しずつ変更 & lintとtestが通ることを確認しながら進める
1. readAllData()にConfigを渡す引数追加
代わりにreadAllData()内でthis.configServiceを使わない
readAllData()の使用箇所を修正し、テストも合わせる
2. readAllData()からraw vauleだけ返すようにする
readAllData(): Promise<ModbusRegisterData>となる
1.で入れたConfigの引数は削る
readAllData()の使用箇所を修正し、テストも合わせる
3. readAllData()を削除しget readable(): ReadableStream<ModbusRegisterData>を作る
constructor内でReadableStream<ModbusRegisterData>を作成し読み取り始める
複数のインスタンスが同時にデータを読みだせるようにする
データ読み取りのブロッキングも解消できる
エラーでReadableStreamが止まった時どうする？
再生成の仕組み必要か？
内部キューに溜まりすぎたらプログラムが止まる？
わからない
背圧制御入れるべきか？
readAllData()を使っていたコードはreadableを使うコードに書き換える
テストも追随させる
DataLoggerServiceでは、DB書き込みをWritableStreamにしてpipeToでつなぐ
間にDBに書き込むデータ形式に変換するTransformStreamを挟む
serialportの呼び出しはnpm@serialport/bindings-interfaceで抽象化されている
これはread,writeなどが単独かつ非同期関数となっていて、callback地獄がない
callbackをpromiseでwrapする必要もなく、コードが簡単
とくにSerialPortはDuplexでwrapされていて、無駄な層が増えてオーバーヘッドになる
しかもcpp実装の場合はread/writeなどとCppコードがほぼ１対１対応になっていて、オーバーヘッドがかなり小さい
そこで、SerialPortではなくBindingInterfaceを直接呼び出す設計にModbusClientを買えたい
現在のModbusSamplerとModbusServiceもModbusClientに吸収させた感じ
特徴
一般のmodbus通信への対応を削り、今回のmodbusボードの仕様にあわせた関数のみ実装する
データ読み出しの間の処理を大幅に削る
下の設計ではcontroller.enqueueとsleepを挟んでいる程度
テストはconstructorに@serialport/bindings-mockで定義されたものを渡せばいい
本番環境では@serialport/bindings-cppを使う
code:ts
export type ModbusRawData = [
inputs: number[];
outputs: number[];
timestamp: number;
];
export class ModbusClient implements AsyncDisposable {
constructor(binding: BindingInterface, options) {
// ...
}
/** 再接続も兼ねる */
async open() {
// ...
}
/**
* push sourceで得たアナログ入出力のstream
*
* DataLoggerServiceなどで使う
*/
get readable(): ReadableStream<ModbusRawData> {
// 単一のstreamからデータを読み出す
this.#readable ??= new ReadableStream({
async start(controller) {
while (this.#port.isOpen) {
// 実際にはノイズや0x03,0x04以外のレスポンスの分別などが必要
const data = await getRawData(binding);
this.#latestRawData = data;
controller.enqueue(data);
await sleep(interval); // たとえば20ms, 0msにすれば理論上最高速度でデータを呼び出せる
}
// 再接続を待機する処理を書く
// readableは切断中はエラーにせず、streamを開いたままにする
}
});
return this.#readable;
}
// openやcloseなどのイベントを扱うstreamかeventが別途必要かも
/**
* pull sourceで最新のデータだけ得る
* 変数名思いつかなかった。うまいの考えて
*
* /api/readなどで使う
*/
get readable2(): ReadableStream<ModbusRawData> {
return new ReadableStream({
// 要求されたときだけデータを取得する
pull(controller) {
controller.enqueue(this.#latestRawData);
}
});
}
/**
* すべてのアナログ出力値を書き出す
*
* 一部の値だけ変えたいときは、関数を引数に渡す
*/
async write(outputs: number[] | (prevOutputs: number[]) => number[]): Promise<void> {
// ...
}
if (!this.#port.isOpen) return;
// this.readable
}
}
// binding.writeで0x03,0x04を連続で書き込み、入出力データをすべて取り出す
// どのポートまで存在するかはシリアル接続開始時に調べておく
const getRawData = async (binding: BindingInterface): Promise<ModbusRawData> => {
// ...
}
あくまでAPI設計のイメージなので、詳細設計はもっと詰める必要ある。エラー処理や既存テスト・コードの書き換えも必要。
そこらへんを詰めた上で、この方針で実装してほしい。
もし方針自体に問題があったり、実装が困難な場合は伝えて。