reivilibre/yama
1
0
Fork 0
Code Issues 27 Pull Requests Releases 16 Wiki Activity

Actually commit when flushing bloblog pointers

Browse Source
This commit is contained in:
Olivier 'reivilibre' 2021-11-21 12:11:51 +00:00
parent 8eeafa7626
commit a37acf3e74
1 changed files with 25 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

46
yama/src/pile/local_sqlitebloblogs.rs
View File

@ -26,7 +26,7 @@ use std::{fs, thread};
use anyhow::{bail, Context};
use byteorder::{BigEndian, ReadBytesExt, WriteBytesExt};
use log::warn;
use log::{info, warn};
use nix::unistd::sync;
use rusqlite::{params, Error, ErrorCode};
use rusqlite::{Connection, OptionalExtension};
@ -383,29 +383,32 @@ impl SqliteBloblogPile {
pointers_buffered: &mut Vec<(ChunkId, BloblogPointer)>,
) -> anyhow::Result<()> {
let mut inner = this.inner.lock().unwrap();
let txn = inner.connection.transaction()?;
let mut stmt = txn.prepare(
"INSERT OR FAIL INTO chunks (chunk_id, bloblog, offset) VALUES (?1, ?2, ?3)",
)?;
for (chunk_id, pointer) in pointers_buffered.drain(..) {
match stmt.execute(params![
&chunk_id[..],
pointer.bloblog,
pointer.offset as i64
]) {
Err(Error::SqliteFailure(e, str))
if e.code == ErrorCode::ConstraintViolation =>
{
warn!(
"(ignoring) SQLite constraint violation on insertion... {:?}",
str
);
}
other => {
other?;
let mut txn = inner.connection.transaction()?;
{
let mut stmt = txn.prepare(
"INSERT OR FAIL INTO chunks (chunk_id, bloblog, offset) VALUES (?1, ?2, ?3)",
)?;
for (chunk_id, pointer) in pointers_buffered.drain(..) {
match stmt.execute(params![
&chunk_id[..],
pointer.bloblog,
pointer.offset as i64
]) {
Err(Error::SqliteFailure(e, str))
if e.code == ErrorCode::ConstraintViolation =>
{
warn!(
"(ignoring) SQLite constraint violation on insertion... {:?}",
str
);
}
other => {
other?;
}
}
}
}
txn.commit()?;
Ok(())
}
@ -459,6 +462,7 @@ impl SqliteBloblogPile {
self.return_writing_bloblog(bloblog_id, bloglog_mutex)?;
}
info!("Flushing pointers (storage pipeline shutdown).");
flush_pointers(self, &mut pointers_buffered)?;
// we MUST have flushed ALL the pointers by now.