SHS/pip
4
3
Fork 0
Code Issues 16 Pull Requests 2 Actions Releases Wiki Activity

fix(pibasetransfer): fix ABBA deadlock — enforce consistent mutex lock ordering

Browse Source 
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.
This commit is contained in:
andrey
2026-08-10 15:48:40 +03:00
parent c3bdc5464e
commit aa64eb4c7f
1 changed files with 63 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libs/main/io_utils/pibasetransfer.cpp
+63 -64
View File
@@ -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();
}