非同期処理を行うライブラリの複数回呼び出しにおける排他制御

前提・実現したいこと

次のライブラリを利用してModusTCP通信を行うプログラムを作成中です。
Modbus Master TCP
こちらがマスターとして複数の離れたアドレスから値を取得し、組み合わせて利用する形になります。

kotlin
1 //イメージ
2 val future: CompletableFuture<ReadHoldingRegistersResponse> =
3    modbusMaster.sendRequest(ReadHoldingRegistersRequest(0, 5), 1)
4 val future2: CompletableFuture<ReadHoldingRegistersResponse> =
5    modbusMaster.sendRequest(ReadHoldingRegistersRequest(1000, 5), 1)
6 //両方の取得完了を待って処理
7 CompletableFuture.allOf(future,future2).whenComplete { response, ex -> 
8    if(ex==null) {
9      hoge = Hoge(future.get().registers, future2.get().registers)
10    }
11  }

発生している問題・エラーメッセージ

GitHubリポジトリに記載のサンプルを参考に値の取得はできたのですが、
複数のレジストリから値を取得しようと複数sendRequestを呼び出すと応答がなくなり、タイムアウトします。
sendRequest呼び出し(非同期処理発行)->処理完了待ち->sendRequest呼び出し(非同期処理発行)->処理完了待ち
の順で処理した場合は正常に取得でき、
sendRequest呼び出し(1)->sendRequest呼び出し(2)->(1)処理完了待ち->(2)処理完了待ち
の場合はタイムアウトとなります。
おそらくはsendRequestで使用している機器へのコネクションなどの単一リソースの専有あたりが問題になっているのではないかと思うのですが、解決策が思いつかず手詰まりとなっております。

該当のソースコード

CompletableFuture.allOfで完了を待つのをやめて上記
sendRequest呼び出し(1)->sendRequest呼び出し(2)->(1)処理完了待ち->(2)処理完了待ちの順で
確認するようにしたものが以下のコードです。

kotlin
1val modbusConfig: ModbusTcpMasterConfig = ModbusTcpMasterConfig.Builder("127.0.0.1")
2    .setPort(502)
3    .setTimeout(Duration.ofMillis(120000))
4    .build()
5  val modbusMaster = ModbusTcpMaster(modbusConfig)
6
7  modbusMaster.connect()
8
9  val future: CompletableFuture<ReadHoldingRegistersResponse> =
10    modbusMaster.sendRequest(ReadHoldingRegistersRequest(0, 5), 1) //1回目呼び出し
11  val future2: CompletableFuture<ReadHoldingRegistersResponse> =
12    modbusMaster.sendRequest(ReadHoldingRegistersRequest(1000, 5), 1) //2回目呼び出し
13  future.whenCompleteAsync { response: ReadHoldingRegistersResponse?, ex: Throwable? ->
14    if (ex == null) {
15      println("Response: " + ByteBufUtil.hexDump(response.registers))
16      ReferenceCountUtil.release(response)
17    } else {
18      println("Completed exceptionally, message={}" + (ex?.message ?: "") + ex)
19      ex?.printStackTrace()
20    }
21  }.join() //1回目処理完了待ち
22  future2.whenCompleteAsync { response: ReadHoldingRegistersResponse?, ex: Throwable? ->
23    if (ex == null) {
24      println("Response: " + ByteBufUtil.hexDump(response.registers))
25      ReferenceCountUtil.release(response)
26    } else {
27      println("Completed exceptionally, message={}" + (ex?.message ?: "") + ex)
28      ex?.printStackTrace()
29    }
30  }.join() //2回目処理完了待ち

試したこと

上記コードでは同様にタイムアウトが発生し、future2の変数宣言(2回目のsendRequest呼び出し)を1回目の処理の後ろに持ってきた場合は正常に値の取得が可能でした。
前述の通りsendRequestで使用している機器へのコネクションなどの単一リソースの専有あたりが問題になっているのではないかと思い、
Lock.withLockブロックによる排他制御をかけてみたのですがsendRequestの呼び出し側(非同期処理の発行)の制御となり、内部の制御を行う方法がわかりませんでした。

補足情報（FW/ツールのバージョンなど）

JDK:11
Kotlin:1.5
ライブラリ
modbus-master-tcp:1.2.0

dodox86

2021/10/13 19:33

同期でも非同期でも、その複数のModbusリクエストを送信しようとしているスレーブの機器は、127.0.0.1:502 の同じTCP接続先の1つだけなのでしょうか？　そうであると、結局TCPの接続先/Modbusのスレーブは1つなのですからライブラリとか関係なく、そのスレーブの機器にとってリクエストを受信してレスポンスを返す前に非同期で送られてきた2つ目以降のリクエストを受信してしまう(あるいは受信しようとする)のであるから、機器によっては混乱をきたして期待するように動かない気がしますが。非同期で正しく動くケースは、複数のスレーブがマスターにとってそれぞれ別のリモートのTCPのアドレス:ポートである場合のみ、な気がしますがそんなことはないですか。

kurogo

2021/10/13 23:09

おっしゃるとおりで、質問にも書きましたがリクエスト発行->レスポンス処理完了待機->次のリクエストを発行の順で処理した場合は正常に値を取得できます。なので、根本の原因としては同一コネクションで複数のリクエストが連続して投げられたことだと思いますが、これだとライブラリが非同期に作られている意味がないためレスポンス待ち中は次のリクエストを待機するような制御が組めないかと思い質問を立てました。

dodox86

2021/10/13 23:22 編集

ライブラリは、Modbusを意識したものではありますが、同一の、切断されないTCPコネクションで同一のスレーブに対して複数の同時の非同期リクエストを想定したものではないように思います。

kurogo

2021/10/16 04:53

ライブラリがCompletableFutureを返すように実装されていたので非同期で値を集めれるかと思いましたが、レスポンス取得完了待ちまでをメソッドに押し込めて一つづつ取得するしかなさそうですね。