From aa64eb4c7f7e1f2dda4fd19f20e0e586bdb62daf Mon Sep 17 00:00:00 2001
From: "andrey.bychkov" <andrey@signalmodelling.ru>
Date: Mon, 10 Aug 2026 15:48:40 +0300
Subject: [PATCH] =?UTF-8?q?fix(pibasetransfer):=20fix=20ABBA=20deadlock=20?=
 =?UTF-8?q?=E2=80=94=20enforce=20consistent=20mutex=20lock=20ordering?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

received() acquired mutex_header and mutex_session in different orders
depending on packet type:
  - pt_Data path:  mutex_header → mutex_session
  - pt_Start path: mutex_session → mutex_header

send_process() acquires mutex_session independently. When running
concurrently with received(), the reversed lock ordering creates a
classic ABBA deadlock scenario.

Fix: enforce consistent ordering (mutex_header → mutex_session) in
all code paths. Restructured pt_Start case and buildSession() to
always acquire mutex_header before mutex_session.
---
 libs/main/io_utils/pibasetransfer.cpp | 127 +++++++++++++-------------
 1 file changed, 63 insertions(+), 64 deletions(-)

diff --git a/libs/main/io_utils/pibasetransfer.cpp b/libs/main/io_utils/pibasetransfer.cpp
index 1829cd45..2f54fed9 100644
--- a/libs/main/io_utils/pibasetransfer.cpp
+++ b/libs/main/io_utils/pibasetransfer.cpp
@@ -133,6 +133,67 @@ void PIBaseTransfer::received(PIByteArray data) {
 		}
 		mutex_header.unlock();
 		break;
+	case pt_Start:
+		mutex_header.lock();
+		if (is_pause && (is_sending || is_receiving)) {
+			if (header.session_id == h.session_id) {
+				is_pause = false;
+				mutex_header.unlock();
+				resumed();
+				return;
+			}
+		}
+		if (is_sending && header.session_id != h.session_id) {
+			sendBreak(h.session_id);
+			mutex_header.unlock();
+			return;
+		}
+		if (is_receiving) {
+			if (header.session_id != h.session_id) {
+				piCoutObj << "restart receive"_tr("PIBaseTransfer");
+				mutex_header.unlock();
+				finish_receive(false, true);
+			} else {
+				header.id = 0;
+				sendReply(pt_ReplySuccess);
+				mutex_header.unlock();
+				return;
+			}
+		}
+		mutex_header.unlock();
+		if (data.size() == sizeof(StartRequest)) {
+			StartRequest sr;
+			data >> sr;
+			mutex_header.lock();
+			mutex_session.lock();
+			bytes_cur         = 0;
+			state_string      = "start request";
+			bytes_all         = sr.size;
+			header.session_id = h.session_id;
+			header.id         = 0;
+			packets_count     = 10;
+			session.clear();
+			replies.clear();
+			session.resize(sr.packets);
+			replies.resize(sr.packets + 1);
+			replies.fill(pt_Unknown);
+			pm_string.resize(replies.size(), '-');
+			diag.reset();
+			is_receiving = true;
+			break_       = false;
+			mutex_send.lock();
+			send_queue = 0;
+			mutex_send.unlock();
+			beginReceive();
+			receiveStarted();
+			state_string = "receiving";
+			replies[0]   = pt_ReplySuccess;
+			pm_string[0] = '#';
+			mutex_session.unlock();
+			sendReply(pt_ReplySuccess);
+			mutex_header.unlock();
+		}
+		break;
 	case pt_ReplySuccess:
 	case pt_ReplyInvalid:
 		mutex_header.lock();
@@ -199,68 +260,6 @@ void PIBaseTransfer::received(PIByteArray data) {
 			return;
 		}
 		break;
-	case pt_Start:
-		mutex_header.lock();
-		if (is_pause && (is_sending || is_receiving)) {
-			if (header.session_id == h.session_id) {
-				is_pause = false;
-				mutex_header.unlock();
-				resumed();
-				return;
-			}
-		}
-		if (is_sending && header.session_id != h.session_id) {
-			sendBreak(h.session_id);
-			mutex_header.unlock();
-			return;
-		}
-		if (is_receiving) {
-			if (header.session_id != h.session_id) {
-				piCoutObj << "restart receive"_tr("PIBaseTransfer");
-				mutex_header.unlock();
-				finish_receive(false, true);
-			} else {
-				header.id = 0;
-				sendReply(pt_ReplySuccess);
-				mutex_header.unlock();
-				return;
-			}
-		}
-		mutex_header.unlock();
-		if (data.size() == sizeof(StartRequest)) {
-			StartRequest sr;
-			data >> sr;
-			mutex_session.lock();
-			mutex_header.lock();
-			bytes_cur         = 0;
-			state_string      = "start request";
-			bytes_all         = sr.size;
-			header.session_id = h.session_id;
-			header.id         = 0;
-			packets_count     = 10;
-			session.clear();
-			replies.clear();
-			session.resize(sr.packets);
-			replies.resize(sr.packets + 1);
-			replies.fill(pt_Unknown);
-			pm_string.resize(replies.size(), '-');
-			diag.reset();
-			// piCoutObj << "receiveStarted()";
-			is_receiving = true;
-			break_       = false;
-			mutex_send.lock();
-			send_queue = 0;
-			mutex_send.unlock();
-			beginReceive();
-			receiveStarted();
-			state_string = "receiving";
-			replies[0]   = pt_ReplySuccess;
-			pm_string[0] = '#';
-			mutex_session.unlock();
-			sendReply(pt_ReplySuccess);
-			mutex_header.unlock();
-		}
-		break;
 	case pt_Pause:
 		mutex_header.lock();
 		if (header.session_id == h.session_id) {
@@ -424,8 +423,8 @@ int PIBaseTransfer::checkSession() {
 
 
 void PIBaseTransfer::buildSession(PIVector<Part> parts) {
-	mutex_session.lock();
 	mutex_header.lock();
+	mutex_session.lock();
 	state_string = "calculating parts ... ";
 	session.clear();
 	header.session_id = randomi();
@@ -478,8 +477,8 @@ void PIBaseTransfer::buildSession(PIVector<Part> parts) {
 		}
 	}
 	if (cur_size > min_size) session << lfi;
-	mutex_header.unlock();
 	mutex_session.unlock();
+	mutex_header.unlock();
 }