Skip to content

Commit 3c3a48f

Browse files
authored
Merge pull request #590 from subspace/background-recommitment
Make recommitment not block other operations
2 parents f49009f + 2029fe4 commit 3c3a48f

File tree

1 file changed

+67
-54
lines changed

1 file changed

+67
-54
lines changed

crates/subspace-farmer/src/commitments.rs

Lines changed: 67 additions & 54 deletions
Original file line numberDiff line numberDiff line change
@@ -91,73 +91,78 @@ impl Commitments {
9191

9292
/// Create commitments for all pieces for a given salt
9393
pub fn create(&self, salt: Salt, plot: Plot) -> Result<(), CommitmentError> {
94-
let mut commitment_databases = self.inner.commitment_databases.lock();
95-
96-
let db_entry = match commitment_databases.create_db_entry(salt)? {
97-
Some(CreateDbEntryResult {
98-
db_entry,
99-
removed_entry_salt,
100-
}) => {
101-
if let Some(salt) = removed_entry_salt {
94+
{
95+
let mut commitment_databases = self.inner.commitment_databases.lock();
96+
97+
let db_entry = match commitment_databases.create_db_entry(salt)? {
98+
Some(CreateDbEntryResult {
99+
db_entry,
100+
removed_entry_salt,
101+
}) => {
102+
if let Some(salt) = removed_entry_salt {
103+
self.inner
104+
.handlers
105+
.status_change
106+
.call_simple(&CommitmentStatusChange::Removed { salt });
107+
}
102108
self.inner
103109
.handlers
104110
.status_change
105-
.call_simple(&CommitmentStatusChange::Removed { salt });
111+
.call_simple(&CommitmentStatusChange::Creating { salt });
112+
db_entry
106113
}
107-
self.inner
108-
.handlers
109-
.status_change
110-
.call_simple(&CommitmentStatusChange::Creating { salt });
111-
db_entry
112-
}
113-
None => {
114-
return Ok(());
115-
}
116-
};
117-
let (current, next) = commitment_databases.get_db_entries();
118-
self.inner.current.swap(current);
119-
self.inner.next.swap(next);
120-
121-
let mut db_guard = db_entry.lock();
122-
// Release lock to allow working with other databases, but hold lock for `db_entry.db` such
123-
// that nothing else can modify it.
124-
drop(commitment_databases);
125-
126-
let db_path = self.inner.base_directory.join(hex::encode(salt));
127-
128-
let db = {
129-
let db = DB::open_default(db_path).map_err(CommitmentError::CommitmentDb)?;
130-
let piece_count = plot.piece_count();
131-
for batch_start in (0..piece_count).step_by(BATCH_SIZE as usize) {
132-
let pieces_to_process = (batch_start + BATCH_SIZE).min(piece_count) - batch_start;
133-
// TODO: Read next batch while creating tags for the previous one for faster
134-
// recommitment.
135-
let pieces = plot
136-
.read_pieces(batch_start, pieces_to_process)
137-
.map_err(CommitmentError::Plot)?;
138-
139-
let tags: Vec<Tag> = pieces
140-
.par_chunks_exact(PIECE_SIZE)
141-
.map(|piece| create_tag(piece, salt))
142-
.collect();
114+
None => {
115+
return Ok(());
116+
}
117+
};
118+
let (current, next) = commitment_databases.get_db_entries();
119+
self.inner.current.swap(current);
120+
self.inner.next.swap(next);
121+
122+
let db_path = self.inner.base_directory.join(hex::encode(salt));
123+
db_entry.lock().replace(Arc::new(
124+
DB::open_default(db_path).map_err(CommitmentError::CommitmentDb)?,
125+
));
126+
}
143127

128+
let piece_count = plot.piece_count();
129+
for batch_start in (0..piece_count).step_by(BATCH_SIZE as usize) {
130+
let pieces_to_process = (batch_start + BATCH_SIZE).min(piece_count) - batch_start;
131+
// TODO: Read next batch while creating tags for the previous one for faster
132+
// recommitment.
133+
let pieces = plot
134+
.read_pieces(batch_start, pieces_to_process)
135+
.map_err(CommitmentError::Plot)?;
136+
137+
let tags: Vec<Tag> = pieces
138+
.par_chunks_exact(PIECE_SIZE)
139+
.map(|piece| create_tag(piece, salt))
140+
.collect();
141+
142+
let db_entry = match self.get_db_entry(salt) {
143+
Some(db_entry) => db_entry,
144+
None => {
145+
// Database was already removed, no need to continue
146+
break;
147+
}
148+
};
149+
150+
let db_guard = db_entry.lock();
151+
152+
if let Some(db) = db_guard.as_ref() {
144153
for (tag, offset) in tags.iter().zip(batch_start..) {
145154
db.put(tag, offset.to_le_bytes())
146155
.map_err(CommitmentError::CommitmentDb)?;
147156
}
157+
} else {
158+
// Database was already removed, no need to continue
159+
break;
148160
}
149-
150-
db
151-
};
152-
153-
db_guard.replace(Arc::new(db));
154-
// Drop guard because locks need to be taken in a specific order or else will result in a
155-
// deadlock
156-
drop(db_guard);
161+
}
157162

158163
let mut commitment_databases = self.inner.commitment_databases.lock();
159164

160-
// Check if database was already been removed
165+
// Check if database was already removed
161166
if commitment_databases
162167
.get_db_entry(&salt)
163168
.map(|db_entry| db_entry.lock().is_some())
@@ -180,6 +185,10 @@ impl Commitments {
180185
}
181186

182187
pub(crate) fn remove_pieces(&self, pieces: &[Piece]) -> Result<(), CommitmentError> {
188+
if pieces.is_empty() {
189+
return Ok(());
190+
}
191+
183192
for db_entry in self.get_db_entries() {
184193
let salt = db_entry.salt();
185194
let db_guard = db_entry.lock();
@@ -204,6 +213,10 @@ impl Commitments {
204213
F: Fn() -> Iter,
205214
Iter: Iterator<Item = (PieceOffset, &'iter [u8])>,
206215
{
216+
if pieces_with_offsets().next().is_none() {
217+
return Ok(());
218+
}
219+
207220
for db_entry in self.get_db_entries() {
208221
let salt = db_entry.salt();
209222
let db_guard = db_entry.lock();

0 commit comments

Comments
 (0)