leyoda/rqbit
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Debugged one more deadlock (or it was the same one)

Browse source
This commit is contained in:
Igor Katson 2023-11-19 18:15:18 +00:00
parent 19c3fd70ff
commit 2b842024c0
No known key found for this signature in database
GPG key ID: B4EC22B66D61A3F5
4 changed files with 1072 additions and 23 deletions
Download patch file Download diff file Expand all files Collapse all files

51
crates/librqbit/src/torrent_state.rs
View file

@ -1,6 +1,11 @@
// The main logic of rqbit is here - connecting to peers, reading and writing messages
// to them, tracking peer state etc.
// NOTE: deadlock notice:
// peers and stateLocked are behind 2 different locks.
// if you lock them in different order, this may deadlock.
// so always lock the peers one first, and unlock it before stateLocked is locked.
use std::{
collections::HashMap,
fs::File,
@ -427,6 +432,7 @@ impl TorrentState {
}
fn reserve_next_needed_piece(&self, peer_handle: PeerHandle) -> Option<ValidPieceIndex> {
// TODO: locking one inside the other in different order results in deadlocks.
self.peers
.with_live_mut(peer_handle, |live| {
if live.i_am_choked {
@ -1053,8 +1059,6 @@ impl PeerHandler {
}
};
let mut g = self.state.locked.write();
self.state
.peers
.with_live_mut(handle, |h| {
@ -1079,30 +1083,31 @@ impl PeerHandler {
})
.context("peer not found")??;
let full_piece_download_time = match g.chunks.mark_chunk_downloaded(&piece) {
Some(ChunkMarkingResult::Completed) => {
debug!("piece={} done, will write and checksum", piece.index,);
// This will prevent others from stealing it.
g.remove_inflight_piece(chunk_info.piece_index)
.map(|t| t.started.elapsed())
}
Some(ChunkMarkingResult::PreviouslyCompleted) => {
// TODO: we might need to send cancellations here.
debug!("piece={} was done by someone else, ignoring", piece.index,);
return Ok(());
}
Some(ChunkMarkingResult::NotCompleted) => None,
None => {
anyhow::bail!(
"bogus data received: {:?}, cannot map this to a chunk, dropping peer",
piece
);
let full_piece_download_time = {
let mut g = self.state.locked.write();
match g.chunks.mark_chunk_downloaded(&piece) {
Some(ChunkMarkingResult::Completed) => {
debug!("piece={} done, will write and checksum", piece.index,);
// This will prevent others from stealing it.
g.remove_inflight_piece(chunk_info.piece_index)
.map(|t| t.started.elapsed())
}
Some(ChunkMarkingResult::PreviouslyCompleted) => {
// TODO: we might need to send cancellations here.
debug!("piece={} was done by someone else, ignoring", piece.index,);
return Ok(());
}
Some(ChunkMarkingResult::NotCompleted) => None,
None => {
anyhow::bail!(
"bogus data received: {:?}, cannot map this to a chunk, dropping peer",
piece
);
}
}
};
// to prevent deadlocks.
drop(g);
self.spawner
.spawn_block_in_place(move || {
let index = piece.index;