GRIN/grin-node
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'refs/heads/lmdb_update' into grim

Browse Source
This commit is contained in:
ardocrat
2026-04-16 14:54:33 +03:00
parent 42b928a42f 02cce56741
commit 0c3e18a913
16 changed files with 621 additions and 535 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+182 -104
View File
@@ -186,18 +186,21 @@ version = "1.6.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8c3c1a368f70d6cf7302d78f8f7093da241fb8e8807c05cc9e51a125895a6d5b"
[[package]]
name = "bincode"
version = "1.3.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b1f45e9417d87227c7a56d22e471c6206462cba514c7590c09aff4cf6d1ddcad"
dependencies = [
"serde",
]
[[package]]
name = "bit-vec"
version = "0.6.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "349f9b6a179ed607305526ca489b34ad0a41aed5f7980fa90eb03160b69598fb"
[[package]]
name = "bitflags"
version = "0.9.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4efd02e230a02e18f92fc2735f44597385ed02ad8f831e7c1c1156ee5e1ab3a5"
[[package]]
name = "bitflags"
version = "1.3.2"
@@ -206,9 +209,12 @@ checksum = "bef38d45163c2f1dde094a7dfd33ccf595c92905c8f8f4fdc18d06fb1037718a"
[[package]]
name = "bitflags"
version = "2.6.0"
version = "2.11.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b048fb63fd8b5923fc5aa7b340d8e156aec7ec02f0c78fa8a6ddc2613f6f71de"
checksum = "c4512299f36f043ab09a583e57bceb5a5aab7a73db1805848e8fef3c9e8c78b3"
dependencies = [
"serde_core",
]
[[package]]
name = "blake2-rfc"
@@ -273,10 +279,11 @@ dependencies = [
[[package]]
name = "cc"
version = "1.2.16"
version = "1.2.60"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "be714c154be609ec7f5dad223a33bf1482fff90472de28f7362806e6d4832b8c"
checksum = "43c5703da9466b66a946814e1adf53ea2c90f10063b86290cc9eb67ce3478a20"
dependencies = [
"find-msvc-tools",
"jobserver",
"libc",
"shlex",
@@ -422,6 +429,15 @@ dependencies = [
"crossbeam-utils",
]
[[package]]
name = "crossbeam-queue"
version = "0.3.12"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0f58bbc28f91df819d0aa2a2c00cd19754769c2fad90579b3592b1c9ba7a3115"
dependencies = [
"crossbeam-utils",
]
[[package]]
name = "crossbeam-utils"
version = "0.8.20"
@@ -471,8 +487,6 @@ dependencies = [
"lazy_static",
"libc",
"log",
"maplit",
"pancurses",
"signal-hook",
"unicode-segmentation",
"unicode-width",
@@ -626,6 +640,15 @@ dependencies = [
"syn 2.0.87",
]
[[package]]
name = "doxygen-rs"
version = "0.4.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "415b6ec780d34dcf624666747194393603d0373b7141eef01d12ee58881507d9"
dependencies = [
"phf",
]
[[package]]
name = "easy-jsonrpc-mw"
version = "0.5.4"
@@ -748,6 +771,12 @@ dependencies = [
"windows-sys 0.59.0",
]
[[package]]
name = "find-msvc-tools"
version = "0.1.9"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582"
[[package]]
name = "flate2"
version = "1.0.32"
@@ -884,12 +913,6 @@ dependencies = [
"slab",
]
[[package]]
name = "gcc"
version = "0.3.55"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8f5f3913fa0bfe7ee1fd8248b6b9f42a5af4b9d65ec2dd2c3c26132b950ecfc2"
[[package]]
name = "generic-array"
version = "0.14.7"
@@ -923,7 +946,7 @@ version = "0.20.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2deb07a133b1520dc1a5690e9bd08950108873d7ed5de38dcc74d3b5ebffa110"
dependencies = [
"bitflags 2.6.0",
"bitflags 2.11.1",
"libc",
"libgit2-sys",
"log",
@@ -1054,7 +1077,7 @@ dependencies = [
"serde",
"serde_derive",
"serde_json",
"siphasher",
"siphasher 0.3.11",
"thiserror",
"zeroize",
]
@@ -1153,7 +1176,6 @@ dependencies = [
"http",
"hyper",
"hyper-rustls",
"lmdb-zero",
"log",
"rand 0.6.5",
"rustls",
@@ -1176,8 +1198,8 @@ dependencies = [
"filetime",
"grin_core",
"grin_util",
"heed",
"libc",
"lmdb-zero",
"log",
"memmap",
"rand 0.6.5",
@@ -1242,6 +1264,44 @@ dependencies = [
"unicode-segmentation",
]
[[package]]
name = "heed"
version = "0.22.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ad82d6598ccf1dac15c8b758a1bd282b755b6776be600429176757190a1b0202"
dependencies = [
"bitflags 2.11.1",
"byteorder",
"heed-traits",
"heed-types",
"libc",
"lmdb-master-sys",
"once_cell",
"page_size",
"serde",
"synchronoise",
"url",
]
[[package]]
name = "heed-traits"
version = "0.20.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "eb3130048d404c57ce5a1ac61a903696e8fcde7e8c2991e9fcfc1f27c3ef74ff"
[[package]]
name = "heed-types"
version = "0.21.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "13c255bdf46e07fb840d120a36dcc81f385140d7191c76a7391672675c01a55d"
dependencies = [
"bincode",
"byteorder",
"heed-traits",
"serde",
"serde_json",
]
[[package]]
name = "hermit-abi"
version = "0.1.19"
@@ -1559,9 +1619,9 @@ checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe"
[[package]]
name = "libc"
version = "0.2.158"
version = "0.2.185"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d8adc4bb1803a324070e64a98ae98f38934d91957a99cfb3a43dcbc01bc56439"
checksum = "52ff2c0fe9bc6cb6b14a0592c2ff4fa9ceb83eea9db979b0487cd054946a2b8f"
[[package]]
name = "libgit2-sys"
@@ -1575,23 +1635,13 @@ dependencies = [
"pkg-config",
]
[[package]]
name = "liblmdb-sys"
version = "0.2.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "feed38a3a580f60bf61aaa067b0ff4123395966839adeaf67258a9e50c4d2e49"
dependencies = [
"gcc",
"libc",
]
[[package]]
name = "libredox"
version = "0.1.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c0ff37bd590ca25063e35af745c343cb7a0271906fb7b37e4813e8f79f00268d"
dependencies = [
"bitflags 2.6.0",
"bitflags 2.11.1",
"libc",
"redox_syscall 0.5.3",
]
@@ -1627,15 +1677,14 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6373607a59f0be73a39b6fe456b8192fcc3585f602af20751600e974dd455e77"
[[package]]
name = "lmdb-zero"
version = "0.4.4"
name = "lmdb-master-sys"
version = "0.2.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "13416eee745b087c22934f35f1f24da22da41ba2a5ce197143d168ce055cc58d"
checksum = "aaeb9bd22e73bd1babffff614994b341e9b2008de7bb73bf1f7e9154f1978f8b"
dependencies = [
"bitflags 0.9.1",
"cc",
"doxygen-rs",
"libc",
"liblmdb-sys",
"supercow",
]
[[package]]
@@ -1710,12 +1759,6 @@ dependencies = [
"linked-hash-map",
]
[[package]]
name = "maplit"
version = "1.0.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3e2e65a1a2e43cfcb47a895c4c8b10d1f4a61097f9f254f183aee60cad9c651d"
[[package]]
name = "memchr"
version = "2.7.4"
@@ -1762,24 +1805,13 @@ dependencies = [
"windows-sys 0.52.0",
]
[[package]]
name = "ncurses"
version = "5.101.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5e2c5d34d72657dc4b638a1c25d40aae81e4f1c699062f72f467237920752032"
dependencies = [
"cc",
"libc",
"pkg-config",
]
[[package]]
name = "nix"
version = "0.29.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "71e2746dc3a24dd78b3cfcb7be93368c6de9963d30f43a6a73998a9cf4b17b46"
dependencies = [
"bitflags 2.6.0",
"bitflags 2.11.1",
"cfg-if 1.0.0",
"cfg_aliases",
"libc",
@@ -1934,9 +1966,9 @@ dependencies = [
[[package]]
name = "once_cell"
version = "1.19.0"
version = "1.21.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3fdb12b2476b595f9358c5161aa467c2438859caa136dec86c26fdd2efe17b92"
checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50"
[[package]]
name = "opaque-debug"
@@ -1969,16 +2001,13 @@ dependencies = [
]
[[package]]
name = "pancurses"
version = "0.17.0"
name = "page_size"
version = "0.6.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0352975c36cbacb9ee99bfb709b9db818bed43af57751797f8633649759d13db"
checksum = "30d5b2194ed13191c1999ae0704b7839fb18384fa22e49b57eeaa97d79ce40da"
dependencies = [
"libc",
"log",
"ncurses",
"pdcurses-sys",
"winreg",
"winapi",
]
[[package]]
@@ -2052,22 +2081,54 @@ dependencies = [
"sha2",
]
[[package]]
name = "pdcurses-sys"
version = "0.7.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "084dd22796ff60f1225d4eb6329f33afaf4c85419d51d440ab6b8c6f4529166b"
dependencies = [
"cc",
"libc",
]
[[package]]
name = "percent-encoding"
version = "2.3.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220"
[[package]]
name = "phf"
version = "0.11.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1fd6780a80ae0c52cc120a26a1a42c1ae51b247a253e4e06113d23d2c2edd078"
dependencies = [
"phf_macros",
"phf_shared",
]
[[package]]
name = "phf_generator"
version = "0.11.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3c80231409c20246a13fddb31776fb942c38553c51e871f8cbd687a4cfb5843d"
dependencies = [
"phf_shared",
"rand 0.8.5",
]
[[package]]
name = "phf_macros"
version = "0.11.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f84ac04429c13a7ff43785d75ad27569f2951ce0ffd30a3321230db2fc727216"
dependencies = [
"phf_generator",
"phf_shared",
"proc-macro2 1.0.86",
"quote 1.0.36",
"syn 2.0.87",
]
[[package]]
name = "phf_shared"
version = "0.11.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "67eabc2ef2a60eb7faa00097bd1ffdb5bd28e62bf39990626a582201b7a754e5"
dependencies = [
"siphasher 1.0.2",
]
[[package]]
name = "pin-project-lite"
version = "0.2.14"
@@ -2334,7 +2395,7 @@ version = "0.5.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2a908a6e00f1fdd0dfd9c0eb08ce85126f6d8bbda50017e74bc4a4b7d4a926a4"
dependencies = [
"bitflags 2.6.0",
"bitflags 2.11.1",
]
[[package]]
@@ -2429,7 +2490,7 @@ version = "0.38.34"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "70dc5ec042f7a43c4a73241207cecc9873a06d45debb38b329f8541d85c2730f"
dependencies = [
"bitflags 2.6.0",
"bitflags 2.11.1",
"errno",
"libc",
"linux-raw-sys",
@@ -2521,7 +2582,7 @@ version = "2.11.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "897b2245f0b511c87893af39b033e5ca9cce68824c4d7e7630b5a1d339658d02"
dependencies = [
"bitflags 2.6.0",
"bitflags 2.11.1",
"core-foundation",
"core-foundation-sys",
"libc",
@@ -2540,10 +2601,11 @@ dependencies = [
[[package]]
name = "serde"
version = "1.0.208"
version = "1.0.228"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cff085d2cb684faa248efb494c39b68e522822ac0de72ccf08109abde717cfb2"
checksum = "9a8e94ea7f378bd32cbbd37198a4a91436180c5bb472411e48b5ec2e2124ae9e"
dependencies = [
"serde_core",
"serde_derive",
]
@@ -2558,10 +2620,19 @@ dependencies = [
]
[[package]]
name = "serde_derive"
version = "1.0.208"
name = "serde_core"
version = "1.0.228"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "24008e81ff7613ed8e5ba0cfaf24e2c2f1e5b8a0495711e44fcd4882fca62bcf"
checksum = "41d385c7d4ca58e59fc732af25c3983b67ac852c1a25000afe1175de458b67ad"
dependencies = [
"serde_derive",
]
[[package]]
name = "serde_derive"
version = "1.0.228"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d540f220d3187173da220f885ab66608367b6574e925011a9353e4badda91d79"
dependencies = [
"proc-macro2 1.0.86",
"quote 1.0.36",
@@ -2570,14 +2641,15 @@ dependencies = [
[[package]]
name = "serde_json"
version = "1.0.125"
version = "1.0.149"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "83c8e735a073ccf5be70aa8066aa984eaf2fa000db6c8d0100ae605b366d31ed"
checksum = "83fc039473c5595ace860d8c4fafa220ff474b3fc6bfdb4293327f1a37e94d86"
dependencies = [
"itoa",
"memchr",
"ryu",
"serde",
"serde_core",
"zmij",
]
[[package]]
@@ -2637,6 +2709,12 @@ version = "0.3.11"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "38b58827f4464d87d377d175e90bf58eb00fd8716ff0a62f80356b5e61555d0d"
[[package]]
name = "siphasher"
version = "1.0.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b2aa850e253778c88a04c3d7323b043aeda9d3e30d5971937c1855769763678e"
[[package]]
name = "slab"
version = "0.4.9"
@@ -2692,12 +2770,6 @@ version = "2.6.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292"
[[package]]
name = "supercow"
version = "0.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "171758edb47aa306a78dfa4ab9aeb5167405bd4e3dc2b64e88f6a84bbe98bd63"
[[package]]
name = "syn"
version = "0.15.44"
@@ -2731,6 +2803,15 @@ dependencies = [
"unicode-ident",
]
[[package]]
name = "synchronoise"
version = "1.0.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3dbc01390fc626ce8d1cffe3376ded2b72a11bb70e1c75f404a210e4daa4def2"
dependencies = [
"crossbeam-queue",
]
[[package]]
name = "synstructure"
version = "0.13.2"
@@ -3283,15 +3364,6 @@ version = "0.52.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec"
[[package]]
name = "winreg"
version = "0.5.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a27a759395c1195c4cc5cda607ef6f8f6498f64e78f7900f5de0a127a424704a"
dependencies = [
"winapi",
]
[[package]]
name = "writeable"
version = "0.6.2"
@@ -3438,3 +3510,9 @@ dependencies = [
"crc32fast",
"thiserror",
]
[[package]]
name = "zmij"
version = "1.0.21"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b8848ee67ecc8aedbaf3e4122217aff892639231befc6a1b58d29fff4c2cabaa"
chain/src/chain.rs
+25 -24
View File
@@ -150,8 +150,8 @@ pub struct Chain {
store: Arc<store::ChainStore>,
adapter: Arc<dyn ChainAdapter + Send + Sync>,
orphans: Arc<OrphanBlockPool>,
txhashset: Arc<RwLock<txhashset::TxHashSet>>,
header_pmmr: Arc<RwLock<txhashset::PMMRHandle<BlockHeader>>>,
txhashset: Arc<RwLock<TxHashSet>>,
header_pmmr: Arc<RwLock<PMMRHandle<BlockHeader>>>,
pibd_segmenter: Arc<RwLock<Option<Segmenter>>>,
pibd_desegmenter: Arc<RwLock<Option<Desegmenter>>>,
// POW verification function
@@ -189,9 +189,9 @@ impl Chain {
// Initialize the output_pos index based on UTXO set
// and NRD kernel_pos index based recent kernel history.
{
let batch = store.batch()?;
txhashset.init_output_pos_index(&header_pmmr, &batch)?;
txhashset.init_recent_kernel_pos_index(&header_pmmr, &batch)?;
let mut batch = store.batch()?;
txhashset.init_output_pos_index(&header_pmmr, &mut batch)?;
txhashset.init_recent_kernel_pos_index(&header_pmmr, &mut batch)?;
batch.commit()?;
}
@@ -275,7 +275,7 @@ impl Chain {
pub fn reset_chain_head_to_genesis(&self) -> Result<(), Error> {
let mut header_pmmr = self.header_pmmr.write();
let mut txhashset = self.txhashset.write();
let batch = self.store.batch()?;
let mut batch = self.store.batch()?;
// Change head back to genesis
{
@@ -314,7 +314,7 @@ impl Chain {
/// Reset PIBD head
pub fn reset_pibd_head(&self) -> Result<(), Error> {
let batch = self.store.batch()?;
let mut batch = self.store.batch()?;
batch.save_pibd_head(&self.genesis().into())?;
Ok(())
}
@@ -530,9 +530,9 @@ impl Chain {
pub fn new_ctx<'a>(
&self,
opts: Options,
batch: store::Batch<'a>,
header_pmmr: &'a mut txhashset::PMMRHandle<BlockHeader>,
txhashset: &'a mut txhashset::TxHashSet,
batch: Batch<'a>,
header_pmmr: &'a mut PMMRHandle<BlockHeader>,
txhashset: &'a mut TxHashSet,
) -> Result<pipe::BlockContext<'a>, Error> {
let denylist = self.denylist.read().clone();
Ok(pipe::BlockContext {
@@ -832,7 +832,7 @@ impl Chain {
&self,
header: &BlockHeader,
ext: &mut ExtensionPair,
batch: &Batch,
batch: &mut Batch,
) -> Result<BlockHeader, Error> {
let denylist = self.denylist.read().clone();
pipe::rewind_and_apply_fork(header, ext, batch, &|header| {
@@ -846,7 +846,7 @@ impl Chain {
&self,
header: &BlockHeader,
ext: &mut HeaderExtension,
batch: &Batch,
batch: &mut Batch,
) -> Result<(), Error> {
let denylist = self.denylist.read().clone();
pipe::rewind_and_apply_header_fork(header, ext, batch, &|header| {
@@ -1015,7 +1015,7 @@ impl Chain {
fn validate_kernel_history(
&self,
header: &BlockHeader,
txhashset: &txhashset::TxHashSet,
txhashset: &TxHashSet,
) -> Result<(), Error> {
debug!("validate_kernel_history: rewinding and validating kernel history (readonly)");
@@ -1151,11 +1151,11 @@ impl Chain {
self.validate_kernel_history(&header, &txhashset)?;
let header_pmmr = self.header_pmmr.read();
let batch = self.store.batch()?;
let mut batch = self.store.batch()?;
txhashset.verify_kernel_pos_index(
&self.genesis.header,
&header_pmmr,
&batch,
&mut batch,
None,
None,
)?;
@@ -1213,10 +1213,10 @@ impl Chain {
}
// Rebuild our output_pos index in the db based on fresh UTXO set.
txhashset.init_output_pos_index(&header_pmmr, &batch)?;
txhashset.init_output_pos_index(&header_pmmr, &mut batch)?;
// Rebuild our NRD kernel_pos index based on recent kernel history.
txhashset.init_recent_kernel_pos_index(&header_pmmr, &batch)?;
txhashset.init_recent_kernel_pos_index(&header_pmmr, &mut batch)?;
// Commit all the changes to the db.
batch.commit()?;
@@ -1257,9 +1257,9 @@ impl Chain {
/// *Only* runs if we are not in archive mode.
fn remove_historical_blocks(
&self,
header_pmmr: &txhashset::PMMRHandle<BlockHeader>,
header_pmmr: &PMMRHandle<BlockHeader>,
archive_header: BlockHeader,
batch: &store::Batch<'_>,
batch: &mut Batch<'_>,
) -> Result<(), Error> {
if self.archive_mode() {
return Ok(());
@@ -1345,7 +1345,7 @@ impl Chain {
// Take a write lock on the txhashet and start a new writeable db batch.
let header_pmmr = self.header_pmmr.read();
let mut txhashset = self.txhashset.write();
let batch = self.store.batch()?;
let mut batch = self.store.batch()?;
// Compact the txhashset itself (rewriting the pruned backend files).
{
@@ -1361,15 +1361,15 @@ impl Chain {
// If we are not in archival mode remove historical blocks from the db.
if !self.archive_mode() {
self.remove_historical_blocks(&header_pmmr, archive_header, &batch)?;
self.remove_historical_blocks(&header_pmmr, archive_header, &mut batch)?;
}
// Make sure our output_pos index is consistent with the UTXO set.
txhashset.init_output_pos_index(&header_pmmr, &batch)?;
txhashset.init_output_pos_index(&header_pmmr, &mut batch)?;
// TODO - Why is this part of chain compaction?
// Rebuild our NRD kernel_pos index based on recent kernel history.
txhashset.init_recent_kernel_pos_index(&header_pmmr, &batch)?;
txhashset.init_recent_kernel_pos_index(&header_pmmr, &mut batch)?;
// Commit all the above db changes.
batch.commit()?;
@@ -1439,7 +1439,8 @@ impl Chain {
0
} else {
self.get_header_by_height(start_block_height - 1)?
.output_mmr_size + 1
.output_mmr_size
+ 1
};
let end_mmr_size = self.get_header_by_height(end_block_height)?.output_mmr_size;
Ok((start_mmr_size, end_mmr_size))
chain/src/linked_list.rs
+27 -12
View File
@@ -108,7 +108,7 @@ pub trait ListIndex {
/// Push a pos onto the list for the specified commitment.
fn push_pos(
&self,
batch: &Batch<'_>,
batch: &mut Batch<'_>,
commit: Commitment,
new_pos: <Self::Entry as ListIndexEntry>::Pos,
) -> Result<(), Error>;
@@ -116,7 +116,7 @@ pub trait ListIndex {
/// Pop a pos off the list for the specified commitment.
fn pop_pos(
&self,
batch: &Batch<'_>,
batch: &mut Batch<'_>,
commit: Commitment,
) -> Result<Option<<Self::Entry as ListIndexEntry>::Pos>, Error>;
}
@@ -124,7 +124,12 @@ pub trait ListIndex {
/// Supports "rewind" given the provided commit and a pos to rewind back to.
pub trait RewindableListIndex {
/// Rewind the index for the given commitment to the specified position.
fn rewind(&self, batch: &Batch<'_>, commit: Commitment, rewind_pos: u64) -> Result<(), Error>;
fn rewind(
&self,
batch: &mut Batch<'_>,
commit: Commitment,
rewind_pos: u64,
) -> Result<(), Error>;
}
/// A pruneable list index supports pruning of old data from the index lists.
@@ -133,15 +138,20 @@ pub trait RewindableListIndex {
pub trait PruneableListIndex: ListIndex {
/// Clear all data from the index.
/// Used when rebuilding the index.
fn clear(&self, batch: &Batch<'_>) -> Result<(), Error>;
fn clear(&self, batch: &mut Batch<'_>) -> Result<(), Error>;
/// Prune old data.
fn prune(&self, batch: &Batch<'_>, commit: Commitment, cutoff_pos: u64) -> Result<(), Error>;
fn prune(
&self,
batch: &mut Batch<'_>,
commit: Commitment,
cutoff_pos: u64,
) -> Result<(), Error>;
/// Pop a pos off the back of the list (used for pruning old data).
fn pop_pos_back(
&self,
batch: &Batch<'_>,
batch: &mut Batch<'_>,
commit: Commitment,
) -> Result<Option<<Self::Entry as ListIndexEntry>::Pos>, Error>;
}
@@ -255,7 +265,7 @@ where
}
}
fn push_pos(&self, batch: &Batch<'_>, commit: Commitment, new_pos: T) -> Result<(), Error> {
fn push_pos(&self, batch: &mut Batch<'_>, commit: Commitment, new_pos: T) -> Result<(), Error> {
match self.get_list(batch, commit)? {
None => {
let list = ListWrapper::Single { pos: new_pos };
@@ -327,7 +337,7 @@ where
/// Pop the head of the list.
/// Returns the output_pos.
/// Returns None if list was empty.
fn pop_pos(&self, batch: &Batch<'_>, commit: Commitment) -> Result<Option<T>, Error> {
fn pop_pos(&self, batch: &mut Batch<'_>, commit: Commitment) -> Result<Option<T>, Error> {
match self.get_list(batch, commit)? {
None => Ok(None),
Some(ListWrapper::Single { pos }) => {
@@ -373,7 +383,12 @@ where
/// List index that supports rewind.
impl<T: PosEntry> RewindableListIndex for MultiIndex<T> {
fn rewind(&self, batch: &Batch<'_>, commit: Commitment, rewind_pos: u64) -> Result<(), Error> {
fn rewind(
&self,
batch: &mut Batch<'_>,
commit: Commitment,
rewind_pos: u64,
) -> Result<(), Error> {
while self
.peek_pos(batch, commit)?
.map(|x| x.pos() > rewind_pos)
@@ -386,7 +401,7 @@ impl<T: PosEntry> RewindableListIndex for MultiIndex<T> {
}
impl<T: PosEntry> PruneableListIndex for MultiIndex<T> {
fn clear(&self, batch: &Batch<'_>) -> Result<(), Error> {
fn clear(&self, batch: &mut Batch<'_>) -> Result<(), Error> {
let mut list_count = 0;
let mut entry_count = 0;
let prefix = to_key(self.list_prefix, "");
@@ -409,7 +424,7 @@ impl<T: PosEntry> PruneableListIndex for MultiIndex<T> {
/// Pruning will be more performant than full rebuild but not yet necessary.
fn prune(
&self,
_batch: &Batch<'_>,
_batch: &mut Batch<'_>,
_commit: Commitment,
_cutoff_pos: u64,
) -> Result<(), Error> {
@@ -420,7 +435,7 @@ impl<T: PosEntry> PruneableListIndex for MultiIndex<T> {
/// Pop off the back/tail of the linked list.
/// Used when pruning old data.
fn pop_pos_back(&self, batch: &Batch<'_>, commit: Commitment) -> Result<Option<T>, Error> {
fn pop_pos_back(&self, batch: &mut Batch<'_>, commit: Commitment) -> Result<Option<T>, Error> {
match self.get_list(batch, commit)? {
None => Ok(None),
Some(ListWrapper::Single { pos }) => {
chain/src/pipe.rs
+18 -17
View File
@@ -161,11 +161,11 @@ pub fn process_block(
// Note we do this in the outer batch, not the child batch from the extension
// as we only commit the child batch if the extension increases total work.
// We want to save the block to the db regardless.
add_block(b, &ctx.batch)?;
add_block(b, &mut ctx.batch)?;
// If we have no "tail" then set it now.
if ctx.batch.tail().is_err() {
update_body_tail(&b.header, &ctx.batch)?;
update_body_tail(&b.header, &mut ctx.batch)?;
}
if has_more_work(&b.header, &head) {
@@ -198,13 +198,13 @@ pub fn process_block_headers(
// Note: This batch may later be committed even if the MMR itself is rollbacked.
for header in headers {
validate_header(header, ctx)?;
add_block_header(header, &ctx.batch)?;
add_block_header(header, &mut ctx.batch)?;
}
let ctx_specific_validation = &ctx.header_allowed;
// Now apply this entire chunk of headers to the header MMR.
txhashset::header_extending(&mut ctx.header_pmmr, &mut ctx.batch, |ext, batch| {
txhashset::header_extending(&mut ctx.header_pmmr, &mut ctx.batch, |ext, mut batch| {
rewind_and_apply_header_fork(&last_header, ext, batch, ctx_specific_validation)?;
// If previous sync_head is not on the "current" chain then
@@ -216,7 +216,7 @@ pub fn process_block_headers(
// Note the outer batch may still be committed to db assuming no errors occur in the extension.
if has_more_work(last_header, &head) {
let header_head = last_header.into();
update_header_head(&header_head, &batch)?;
update_header_head(&header_head, &mut batch)?;
} else {
ext.force_rollback();
};
@@ -275,7 +275,7 @@ pub fn process_block_header(header: &BlockHeader, ctx: &mut BlockContext<'_>) ->
})?;
// Add this new block header to the db.
add_block_header(header, &ctx.batch)?;
add_block_header(header, &mut ctx.batch)?;
if has_more_work(header, &header_head) {
update_header_head(&Tip::from_header(header), &mut ctx.batch)?;
@@ -478,7 +478,7 @@ fn verify_coinbase_maturity(
/// Verify kernel sums across the full utxo and kernel sets based on block_sums
/// of previous block accounting for the inputs|outputs|kernels of the new block.
/// Saves the new block_sums to the db via the current batch if successful.
fn verify_block_sums(b: &Block, batch: &store::Batch<'_>) -> Result<(), Error> {
fn verify_block_sums(b: &Block, batch: &mut store::Batch<'_>) -> Result<(), Error> {
// Retrieve the block_sums for the previous block.
let block_sums = batch.get_block_sums(&b.header.prev_hash)?;
@@ -509,7 +509,7 @@ fn verify_block_sums(b: &Block, batch: &store::Batch<'_>) -> Result<(), Error> {
fn apply_block_to_txhashset(
block: &Block,
ext: &mut txhashset::ExtensionPair<'_>,
batch: &store::Batch<'_>,
batch: &mut store::Batch<'_>,
) -> Result<(), Error> {
ext.extension
.apply_block(block, ext.header_extension, batch)?;
@@ -520,13 +520,13 @@ fn apply_block_to_txhashset(
/// Officially adds the block to our chain (possibly on a losing fork).
/// Header must be added separately (assume this has been done previously).
fn add_block(b: &Block, batch: &store::Batch<'_>) -> Result<(), Error> {
fn add_block(b: &Block, batch: &mut store::Batch<'_>) -> Result<(), Error> {
batch.save_block(b)?;
Ok(())
}
/// Update the block chain tail so we can know the exact tail of full blocks in this node
fn update_body_tail(bh: &BlockHeader, batch: &store::Batch<'_>) -> Result<(), Error> {
fn update_body_tail(bh: &BlockHeader, batch: &mut store::Batch<'_>) -> Result<(), Error> {
let tip = Tip::from_header(bh);
batch
.save_body_tail(&tip)
@@ -536,27 +536,28 @@ fn update_body_tail(bh: &BlockHeader, batch: &store::Batch<'_>) -> Result<(), Er
}
/// Officially adds the block header to our header chain.
fn add_block_header(bh: &BlockHeader, batch: &store::Batch<'_>) -> Result<(), Error> {
fn add_block_header(bh: &BlockHeader, batch: &mut store::Batch<'_>) -> Result<(), Error> {
batch
.save_block_header(bh)
.map_err(|e| Error::StoreErr(e, "pipe save header".to_owned()))?;
Ok(())
}
fn update_header_head(head: &Tip, batch: &store::Batch<'_>) -> Result<(), Error> {
fn update_header_head(head: &Tip, batch: &mut store::Batch<'_>) -> Result<(), Error> {
batch
.save_header_head(&head)
.map_err(|e| Error::StoreErr(e, "pipe save header head".to_owned()))?;
debug!(
trace!(
"header head updated to {} at {}",
head.last_block_h, head.height
head.last_block_h,
head.height
);
Ok(())
}
fn update_head(head: &Tip, batch: &store::Batch<'_>) -> Result<(), Error> {
fn update_head(head: &Tip, batch: &mut store::Batch<'_>) -> Result<(), Error> {
batch
.save_body_head(&head)
.map_err(|e| Error::StoreErr(e, "pipe save body".to_owned()))?;
@@ -575,7 +576,7 @@ fn has_more_work(header: &BlockHeader, head: &Tip) -> bool {
pub fn rewind_and_apply_header_fork(
header: &BlockHeader,
ext: &mut txhashset::HeaderExtension<'_>,
batch: &store::Batch<'_>,
batch: &mut store::Batch<'_>,
ctx_specific_validation: &dyn Fn(&BlockHeader) -> Result<(), Error>,
) -> Result<(), Error> {
let mut fork_hashes = vec![];
@@ -616,7 +617,7 @@ pub fn rewind_and_apply_header_fork(
pub fn rewind_and_apply_fork(
header: &BlockHeader,
ext: &mut txhashset::ExtensionPair<'_>,
batch: &store::Batch<'_>,
batch: &mut store::Batch<'_>,
ctx_specific_validation: &dyn Fn(&BlockHeader) -> Result<(), Error>,
) -> Result<BlockHeader, Error> {
let extension = &mut ext.extension;
chain/src/store.rs
+23 -17
View File
@@ -201,22 +201,22 @@ impl<'a> Batch<'a> {
}
/// Save body head to db.
pub fn save_body_head(&self, t: &Tip) -> Result<(), Error> {
pub fn save_body_head(&mut self, t: &Tip) -> Result<(), Error> {
self.db.put_ser(&[HEAD_PREFIX], t)
}
/// Save body "tail" to db.
pub fn save_body_tail(&self, t: &Tip) -> Result<(), Error> {
pub fn save_body_tail(&mut self, t: &Tip) -> Result<(), Error> {
self.db.put_ser(&[TAIL_PREFIX], t)
}
/// Save header head to db.
pub fn save_header_head(&self, t: &Tip) -> Result<(), Error> {
pub fn save_header_head(&mut self, t: &Tip) -> Result<(), Error> {
self.db.put_ser(&[HEADER_HEAD_PREFIX], t)
}
/// Save PIBD head to db.
pub fn save_pibd_head(&self, t: &Tip) -> Result<(), Error> {
pub fn save_pibd_head(&mut self, t: &Tip) -> Result<(), Error> {
self.db.put_ser(&[PIBD_HEAD_PREFIX], t)
}
@@ -234,7 +234,7 @@ impl<'a> Batch<'a> {
/// Save the block to the db.
/// Note: the block header is not saved to the db here, assumes this has already been done.
pub fn save_block(&self, b: &Block) -> Result<(), Error> {
pub fn save_block(&mut self, b: &Block) -> Result<(), Error> {
debug!(
"save_block: {} at {} ({} -> v{})",
b.header.hash(),
@@ -248,20 +248,20 @@ impl<'a> Batch<'a> {
/// We maintain a "spent" index for each full block to allow the output_pos
/// to be easily reverted during rewind.
pub fn save_spent_index(&self, h: &Hash, spent: &[CommitPos]) -> Result<(), Error> {
pub fn save_spent_index(&mut self, h: &Hash, spent: &[CommitPos]) -> Result<(), Error> {
self.db
.put_ser(&to_key(BLOCK_SPENT_PREFIX, h)[..], &spent.to_vec())?;
Ok(())
}
/// Low level function to delete directly by raw key.
pub fn delete(&self, key: &[u8]) -> Result<(), Error> {
pub fn delete(&mut self, key: &[u8]) -> Result<(), Error> {
self.db.delete(key)
}
/// Delete a full block. Does not delete any record associated with a block
/// header.
pub fn delete_block(&self, bh: &Hash) -> Result<(), Error> {
pub fn delete_block(&mut self, bh: &Hash) -> Result<(), Error> {
self.db.delete(&to_key(BLOCK_PREFIX, bh)[..])?;
// Best effort at deleting associated data for this block.
@@ -275,7 +275,7 @@ impl<'a> Batch<'a> {
}
/// Save block header to db.
pub fn save_block_header(&self, header: &BlockHeader) -> Result<(), Error> {
pub fn save_block_header(&mut self, header: &BlockHeader) -> Result<(), Error> {
let hash = header.hash();
// Store the header itself indexed by hash.
@@ -286,13 +286,17 @@ impl<'a> Batch<'a> {
}
/// Save output_pos and block height to index.
pub fn save_output_pos_height(&self, commit: &Commitment, pos: CommitPos) -> Result<(), Error> {
pub fn save_output_pos_height(
&mut self,
commit: &Commitment,
pos: CommitPos,
) -> Result<(), Error> {
self.db
.put_ser(&to_key(OUTPUT_POS_PREFIX, commit)[..], &pos)
}
/// Delete the output_pos index entry for a spent output.
pub fn delete_output_pos_height(&self, commit: &Commitment) -> Result<(), Error> {
pub fn delete_output_pos_height(&mut self, commit: &Commitment) -> Result<(), Error> {
self.db.delete(&to_key(OUTPUT_POS_PREFIX, commit))
}
@@ -305,7 +309,9 @@ impl<'a> Batch<'a> {
}
/// Iterator over the output_pos index.
pub fn output_pos_iter(&self) -> Result<impl Iterator<Item = (Vec<u8>, CommitPos)>, Error> {
pub fn output_pos_iter(
&self,
) -> Result<impl Iterator<Item = (Vec<u8>, CommitPos)> + 'a, Error> {
let key = to_key(OUTPUT_POS_PREFIX, "");
let protocol_version = self.db.protocol_version();
self.db.iter(&key, move |k, mut v| {
@@ -366,12 +372,12 @@ impl<'a> Batch<'a> {
}
/// Delete the block spent index.
fn delete_spent_index(&self, bh: &Hash) -> Result<(), Error> {
fn delete_spent_index(&mut self, bh: &Hash) -> Result<(), Error> {
self.db.delete(&to_key(BLOCK_SPENT_PREFIX, bh))
}
/// Save block_sums for the block.
pub fn save_block_sums(&self, h: &Hash, sums: BlockSums) -> Result<(), Error> {
pub fn save_block_sums(&mut self, h: &Hash, sums: BlockSums) -> Result<(), Error> {
self.db.put_ser(&to_key(BLOCK_SUMS_PREFIX, h)[..], &sums)
}
@@ -383,7 +389,7 @@ impl<'a> Batch<'a> {
}
/// Delete the block_sums for the block.
fn delete_block_sums(&self, bh: &Hash) -> Result<(), Error> {
fn delete_block_sums(&mut self, bh: &Hash) -> Result<(), Error> {
self.db.delete(&to_key(BLOCK_SUMS_PREFIX, bh))
}
@@ -423,7 +429,7 @@ impl<'a> Batch<'a> {
/// Iterator over all full blocks in the db.
/// Uses default db serialization strategy via db protocol version.
pub fn blocks_iter(&self) -> Result<impl Iterator<Item = Block>, Error> {
pub fn blocks_iter(&self) -> Result<impl Iterator<Item = Block> + 'a, Error> {
let key = to_key(BLOCK_PREFIX, "");
let protocol_version = self.db.protocol_version();
self.db.iter(&key, move |_, mut v| {
@@ -434,7 +440,7 @@ impl<'a> Batch<'a> {
/// Iterator over raw data for full blocks in the db.
/// Used during block migration (we need flexibility around deserialization).
pub fn blocks_raw_iter(&self) -> Result<impl Iterator<Item = (Vec<u8>, Vec<u8>)>, Error> {
pub fn blocks_raw_iter(&self) -> Result<impl Iterator<Item = (Vec<u8>, Vec<u8>)> + 'a, Error> {
let key = to_key(BLOCK_PREFIX, "");
self.db.iter(&key, |k, v| Ok((k.to_vec(), v.to_vec())))
}
chain/src/txhashset/desegmenter.rs
+7 -7
View File
@@ -189,7 +189,7 @@ impl Desegmenter {
// TODO: Unwraps
let tip = Tip::from_header(&h);
let batch = self.store.batch()?;
let mut batch = self.store.batch()?;
batch.save_pibd_head(&tip)?;
batch.commit()?;
@@ -283,11 +283,11 @@ impl Desegmenter {
{
let header_pmmr = self.header_pmmr.read();
let txhashset = self.txhashset.read();
let batch = self.store.batch()?;
let mut batch = self.store.batch()?;
txhashset.verify_kernel_pos_index(
&self.genesis,
&header_pmmr,
&batch,
&mut batch,
Some(status.clone()),
Some(stop_state.clone()),
)?;
@@ -308,9 +308,9 @@ impl Desegmenter {
&mut header_pmmr,
&mut txhashset,
&mut batch,
|ext, batch| {
|ext, mut batch| {
let extension = &mut ext.extension;
extension.rewind(&self.archive_header, batch)?;
extension.rewind(&self.archive_header, &mut batch)?;
// Validate the extension, generating the utxo_sum and kernel_sum.
// Full validation, including rangeproofs and kernel signature verification.
@@ -359,10 +359,10 @@ impl Desegmenter {
}
// Rebuild our output_pos index in the db based on fresh UTXO set.
txhashset.init_output_pos_index(&header_pmmr, &batch)?;
txhashset.init_output_pos_index(&header_pmmr, &mut batch)?;
// Rebuild our NRD kernel_pos index based on recent kernel history.
txhashset.init_recent_kernel_pos_index(&header_pmmr, &batch)?;
txhashset.init_recent_kernel_pos_index(&header_pmmr, &mut batch)?;
// Commit all the changes to the db.
batch.commit()?;
chain/src/txhashset/txhashset.rs
+30 -22
View File
@@ -538,7 +538,7 @@ impl TxHashSet {
pub fn init_recent_kernel_pos_index(
&self,
header_pmmr: &PMMRHandle<BlockHeader>,
batch: &Batch<'_>,
batch: &mut Batch<'_>,
) -> Result<(), Error> {
let head = batch.head()?;
let cutoff = head.height.saturating_sub(WEEK_HEIGHT * 2);
@@ -552,7 +552,7 @@ impl TxHashSet {
&self,
from_header: &BlockHeader,
header_pmmr: &PMMRHandle<BlockHeader>,
batch: &Batch<'_>,
batch: &mut Batch<'_>,
status: Option<Arc<SyncState>>,
stop_state: Option<Arc<StopState>>,
) -> Result<(), Error> {
@@ -635,7 +635,7 @@ impl TxHashSet {
pub fn init_output_pos_index(
&self,
header_pmmr: &PMMRHandle<BlockHeader>,
batch: &Batch<'_>,
batch: &mut Batch<'_>,
) -> Result<(), Error> {
let now = Instant::now();
@@ -733,10 +733,10 @@ pub fn extending_readonly<F, T>(
inner: F,
) -> Result<T, Error>
where
F: FnOnce(&mut ExtensionPair<'_>, &Batch<'_>) -> Result<T, Error>,
F: FnOnce(&mut ExtensionPair<'_>, &mut Batch<'_>) -> Result<T, Error>,
{
let commit_index = trees.commit_index.clone();
let batch = commit_index.batch()?;
let mut batch = commit_index.batch()?;
trace!("Starting new txhashset (readonly) extension.");
@@ -751,7 +751,7 @@ where
header_extension: &mut header_extension,
extension: &mut extension,
};
inner(&mut extension_pair, &batch)
inner(&mut extension_pair, &mut batch)
};
trace!("Rollbacking txhashset (readonly) extension.");
@@ -830,7 +830,7 @@ pub fn extending<'a, F, T>(
inner: F,
) -> Result<T, Error>
where
F: FnOnce(&mut ExtensionPair<'_>, &Batch<'_>) -> Result<T, Error>,
F: FnOnce(&mut ExtensionPair<'_>, &mut Batch<'_>) -> Result<T, Error>,
{
let sizes: (u64, u64, u64);
let res: Result<T, Error>;
@@ -842,7 +842,7 @@ where
// create a child transaction so if the state is rolled back by itself, all
// index saving can be undone
let child_batch = batch.child()?;
let mut child_batch = batch.child()?;
{
trace!("Starting new txhashset extension.");
@@ -853,7 +853,7 @@ where
header_extension: &mut header_extension,
extension: &mut extension,
};
res = inner(&mut extension_pair, &child_batch);
res = inner(&mut extension_pair, &mut child_batch);
rollback = extension_pair.extension.rollback;
sizes = extension_pair.extension.sizes();
@@ -901,15 +901,15 @@ where
/// Start a new readonly header MMR extension.
/// This MMR can be extended individually beyond the other (output, rangeproof and kernel) MMRs
/// to allow headers to be validated before we receive the full block data.
pub fn header_extending_readonly<'a, F, T>(
handle: &'a mut PMMRHandle<BlockHeader>,
pub fn header_extending_readonly<F, T>(
handle: &mut PMMRHandle<BlockHeader>,
store: &ChainStore,
inner: F,
) -> Result<T, Error>
where
F: FnOnce(&mut HeaderExtension<'_>, &Batch<'_>) -> Result<T, Error>,
F: FnOnce(&mut HeaderExtension<'_>, &mut Batch<'_>) -> Result<T, Error>,
{
let batch = store.batch()?;
let mut batch = store.batch()?;
let head = match handle.head_hash() {
Ok(hash) => {
@@ -921,7 +921,7 @@ where
let pmmr = PMMR::at(&mut handle.backend, handle.size);
let mut extension = HeaderExtension::new(pmmr, head);
let res = inner(&mut extension, &batch);
let res = inner(&mut extension, &mut batch);
handle.backend.discard();
@@ -937,7 +937,7 @@ pub fn header_extending<'a, F, T>(
inner: F,
) -> Result<T, Error>
where
F: FnOnce(&mut HeaderExtension<'_>, &Batch<'_>) -> Result<T, Error>,
F: FnOnce(&mut HeaderExtension<'_>, &mut Batch<'_>) -> Result<T, Error>,
{
let size: u64;
let res: Result<T, Error>;
@@ -945,7 +945,7 @@ where
// create a child transaction so if the state is rolled back by itself, all
// index saving can be undone
let child_batch = batch.child()?;
let mut child_batch = batch.child()?;
let head = match handle.head_hash() {
Ok(hash) => {
@@ -958,7 +958,7 @@ where
{
let pmmr = PMMR::at(&mut handle.backend, handle.size);
let mut extension = HeaderExtension::new(pmmr, head);
res = inner(&mut extension, &child_batch);
res = inner(&mut extension, &mut child_batch);
rollback = extension.rollback;
size = extension.size();
@@ -1233,7 +1233,7 @@ impl<'a> Extension<'a> {
&mut self,
b: &Block,
header_ext: &HeaderExtension<'_>,
batch: &Batch<'_>,
batch: &mut Batch<'_>,
) -> Result<(), Error> {
let mut affected_pos = vec![];
@@ -1499,7 +1499,7 @@ impl<'a> Extension<'a> {
&mut self,
kernels: &[TxKernel],
height: u64,
batch: &Batch<'_>,
batch: &mut Batch<'_>,
) -> Result<(), Error> {
for kernel in kernels {
let pos = self.apply_kernel(kernel)?;
@@ -1579,7 +1579,7 @@ impl<'a> Extension<'a> {
/// Rewinds the MMRs to the provided block, rewinding to the last output pos
/// and last kernel pos of that block. If `updated_bitmap` is supplied, the
/// bitmap accumulator will be replaced with its contents
pub fn rewind(&mut self, header: &BlockHeader, batch: &Batch<'_>) -> Result<(), Error> {
pub fn rewind(&mut self, header: &BlockHeader, batch: &mut Batch) -> Result<(), Error> {
debug!(
"Rewind extension to {} at {} from {} at {}",
header.hash(),
@@ -1622,7 +1622,11 @@ impl<'a> Extension<'a> {
// Rewind the MMRs and the output_pos index.
// Returns a vec of "affected_pos" so we can apply the necessary updates to the bitmap
// accumulator in a single pass for all rewound blocks.
fn rewind_single_block(&mut self, block: &Block, batch: &Batch<'_>) -> Result<Vec<u64>, Error> {
fn rewind_single_block(
&mut self,
block: &Block,
batch: &mut Batch<'_>,
) -> Result<Vec<u64>, Error> {
let header = &block.header;
let prev_header = batch.get_previous_header(&header)?;
@@ -2206,7 +2210,11 @@ fn input_pos_to_rewind(
}
/// If NRD enabled then enforce NRD relative height rules.
fn apply_kernel_rules(kernel: &TxKernel, pos: CommitPos, batch: &Batch<'_>) -> Result<(), Error> {
fn apply_kernel_rules(
kernel: &TxKernel,
pos: CommitPos,
batch: &mut Batch<'_>,
) -> Result<(), Error> {
if !global::is_nrd_enabled() {
return Ok(());
}
chain/tests/store_indices.rs
+1 -1
View File
@@ -50,7 +50,7 @@ fn test_store_indices() {
{
// Start a new batch and delete the block.
let store = chain.store();
let batch = store.batch().unwrap();
let mut batch = store.batch().unwrap();
assert!(batch.delete_block(&block_hash).is_ok());
// Block is deleted within this batch.
chain/tests/store_kernel_pos_index.rs
+53 -53
View File
@@ -39,14 +39,14 @@ fn test_store_kernel_idx() {
let commit = Commitment::from_vec(vec![]);
let store = ChainStore::new(chain_dir).unwrap();
let batch = store.batch().unwrap();
let mut batch = store.batch().unwrap();
let index = store::nrd_recent_kernel_index();
assert_eq!(index.peek_pos(&batch, commit), Ok(None));
assert_eq!(index.get_list(&batch, commit), Ok(None));
assert_eq!(
index.push_pos(&batch, commit, CommitPos { pos: 1, height: 1 }),
index.push_pos(&mut batch, commit, CommitPos { pos: 1, height: 1 }),
Ok(()),
);
@@ -63,7 +63,7 @@ fn test_store_kernel_idx() {
);
assert_eq!(
index.push_pos(&batch, commit, CommitPos { pos: 2, height: 2 }),
index.push_pos(&mut batch, commit, CommitPos { pos: 2, height: 2 }),
Ok(()),
);
@@ -79,17 +79,17 @@ fn test_store_kernel_idx() {
// Pos must always increase.
assert_eq!(
index.push_pos(&batch, commit, CommitPos { pos: 1, height: 1 }),
index.push_pos(&mut batch, commit, CommitPos { pos: 1, height: 1 }),
Err(Error::OtherErr("pos must be increasing".into())),
);
assert_eq!(
index.push_pos(&batch, commit, CommitPos { pos: 2, height: 2 }),
index.push_pos(&mut batch, commit, CommitPos { pos: 2, height: 2 }),
Err(Error::OtherErr("pos must be increasing".into())),
);
assert_eq!(
index.push_pos(&batch, commit, CommitPos { pos: 3, height: 3 }),
index.push_pos(&mut batch, commit, CommitPos { pos: 3, height: 3 }),
Ok(()),
);
@@ -104,7 +104,7 @@ fn test_store_kernel_idx() {
);
assert_eq!(
index.pop_pos(&batch, commit),
index.pop_pos(&mut batch, commit),
Ok(Some(CommitPos { pos: 3, height: 3 })),
);
@@ -119,7 +119,7 @@ fn test_store_kernel_idx() {
);
assert_eq!(
index.push_pos(&batch, commit, CommitPos { pos: 3, height: 3 }),
index.push_pos(&mut batch, commit, CommitPos { pos: 3, height: 3 }),
Ok(()),
);
@@ -134,7 +134,7 @@ fn test_store_kernel_idx() {
);
assert_eq!(
index.pop_pos(&batch, commit),
index.pop_pos(&mut batch, commit),
Ok(Some(CommitPos { pos: 3, height: 3 })),
);
@@ -149,7 +149,7 @@ fn test_store_kernel_idx() {
);
assert_eq!(
index.pop_pos(&batch, commit),
index.pop_pos(&mut batch, commit),
Ok(Some(CommitPos { pos: 2, height: 2 })),
);
@@ -166,7 +166,7 @@ fn test_store_kernel_idx() {
);
assert_eq!(
index.pop_pos(&batch, commit),
index.pop_pos(&mut batch, commit),
Ok(Some(CommitPos { pos: 1, height: 1 })),
);
@@ -186,14 +186,14 @@ fn test_store_kernel_idx_pop_back() {
let commit = Commitment::from_vec(vec![]);
let store = ChainStore::new(chain_dir).unwrap();
let batch = store.batch().unwrap();
let mut batch = store.batch().unwrap();
let index = store::nrd_recent_kernel_index();
assert_eq!(index.peek_pos(&batch, commit), Ok(None));
assert_eq!(index.get_list(&batch, commit), Ok(None));
assert_eq!(
index.push_pos(&batch, commit, CommitPos { pos: 1, height: 1 }),
index.push_pos(&mut batch, commit, CommitPos { pos: 1, height: 1 }),
Ok(()),
);
@@ -210,7 +210,7 @@ fn test_store_kernel_idx_pop_back() {
);
assert_eq!(
index.pop_pos_back(&batch, commit),
index.pop_pos_back(&mut batch, commit),
Ok(Some(CommitPos { pos: 1, height: 1 })),
);
@@ -218,32 +218,32 @@ fn test_store_kernel_idx_pop_back() {
assert_eq!(index.get_list(&batch, commit), Ok(None));
assert_eq!(
index.push_pos(&batch, commit, CommitPos { pos: 1, height: 1 }),
index.push_pos(&mut batch, commit, CommitPos { pos: 1, height: 1 }),
Ok(()),
);
assert_eq!(
index.push_pos(&batch, commit, CommitPos { pos: 2, height: 2 }),
index.push_pos(&mut batch, commit, CommitPos { pos: 2, height: 2 }),
Ok(()),
);
assert_eq!(
index.push_pos(&batch, commit, CommitPos { pos: 3, height: 3 }),
index.push_pos(&mut batch, commit, CommitPos { pos: 3, height: 3 }),
Ok(()),
);
assert_eq!(
index.peek_pos(&batch, commit),
index.peek_pos(&mut batch, commit),
Ok(Some(CommitPos { pos: 3, height: 3 })),
);
assert_eq!(
index.get_list(&batch, commit),
index.get_list(&mut batch, commit),
Ok(Some(ListWrapper::Multi { head: 3, tail: 1 })),
);
assert_eq!(
index.pop_pos_back(&batch, commit),
index.pop_pos_back(&mut batch, commit),
Ok(Some(CommitPos { pos: 1, height: 1 })),
);
@@ -258,7 +258,7 @@ fn test_store_kernel_idx_pop_back() {
);
assert_eq!(
index.pop_pos_back(&batch, commit),
index.pop_pos_back(&mut batch, commit),
Ok(Some(CommitPos { pos: 2, height: 2 })),
);
@@ -275,7 +275,7 @@ fn test_store_kernel_idx_pop_back() {
);
assert_eq!(
index.pop_pos_back(&batch, commit),
index.pop_pos_back(&mut batch, commit),
Ok(Some(CommitPos { pos: 3, height: 3 })),
);
@@ -294,21 +294,21 @@ fn test_store_kernel_idx_rewind() {
let commit = Commitment::from_vec(vec![]);
let store = ChainStore::new(chain_dir).unwrap();
let batch = store.batch().unwrap();
let mut batch = store.batch().unwrap();
let index = store::nrd_recent_kernel_index();
assert_eq!(
index.push_pos(&batch, commit, CommitPos { pos: 1, height: 1 }),
index.push_pos(&mut batch, commit, CommitPos { pos: 1, height: 1 }),
Ok(()),
);
assert_eq!(
index.push_pos(&batch, commit, CommitPos { pos: 2, height: 2 }),
index.push_pos(&mut batch, commit, CommitPos { pos: 2, height: 2 }),
Ok(()),
);
assert_eq!(
index.push_pos(&batch, commit, CommitPos { pos: 3, height: 3 }),
index.push_pos(&mut batch, commit, CommitPos { pos: 3, height: 3 }),
Ok(()),
);
@@ -317,7 +317,7 @@ fn test_store_kernel_idx_rewind() {
Ok(Some(ListWrapper::Multi { head: 3, tail: 1 })),
);
assert_eq!(index.rewind(&batch, commit, 1), Ok(()),);
assert_eq!(index.rewind(&mut batch, commit, 1), Ok(()),);
assert_eq!(
index.get_list(&batch, commit),
@@ -327,7 +327,7 @@ fn test_store_kernel_idx_rewind() {
);
// Check we can safely noop rewind.
assert_eq!(index.rewind(&batch, commit, 2), Ok(()),);
assert_eq!(index.rewind(&mut batch, commit, 2), Ok(()),);
assert_eq!(
index.get_list(&batch, commit),
@@ -336,7 +336,7 @@ fn test_store_kernel_idx_rewind() {
})),
);
assert_eq!(index.rewind(&batch, commit, 1), Ok(()),);
assert_eq!(index.rewind(&mut batch, commit, 1), Ok(()),);
assert_eq!(
index.get_list(&batch, commit),
@@ -346,42 +346,42 @@ fn test_store_kernel_idx_rewind() {
);
// Check we can rewind back to 0.
assert_eq!(index.rewind(&batch, commit, 0), Ok(()),);
assert_eq!(index.rewind(&mut batch, commit, 0), Ok(()),);
assert_eq!(index.get_list(&batch, commit), Ok(None),);
assert_eq!(index.rewind(&batch, commit, 0), Ok(()),);
assert_eq!(index.rewind(&mut batch, commit, 0), Ok(()),);
// Now check we can rewind past the end of a list safely.
assert_eq!(
index.push_pos(&batch, commit, CommitPos { pos: 1, height: 1 }),
index.push_pos(&mut batch, commit, CommitPos { pos: 1, height: 1 }),
Ok(()),
);
assert_eq!(
index.push_pos(&batch, commit, CommitPos { pos: 2, height: 2 }),
index.push_pos(&mut batch, commit, CommitPos { pos: 2, height: 2 }),
Ok(()),
);
assert_eq!(
index.push_pos(&batch, commit, CommitPos { pos: 3, height: 3 }),
index.push_pos(&mut batch, commit, CommitPos { pos: 3, height: 3 }),
Ok(()),
);
assert_eq!(
index.pop_pos_back(&batch, commit),
index.pop_pos_back(&mut batch, commit),
Ok(Some(CommitPos { pos: 1, height: 1 })),
);
assert_eq!(
index.get_list(&batch, commit),
index.get_list(&mut batch, commit),
Ok(Some(ListWrapper::Multi { head: 3, tail: 2 })),
);
assert_eq!(index.rewind(&batch, commit, 1), Ok(()),);
assert_eq!(index.rewind(&mut batch, commit, 1), Ok(()),);
assert_eq!(index.get_list(&batch, commit), Ok(None),);
assert_eq!(index.get_list(&mut batch, commit), Ok(None),);
clean_output_dir(chain_dir);
}
@@ -396,14 +396,14 @@ fn test_store_kernel_idx_multiple_commits() {
let commit2 = Commitment::from_vec(vec![1]);
let store = ChainStore::new(chain_dir).unwrap();
let batch = store.batch().unwrap();
let mut batch = store.batch().unwrap();
let index = store::nrd_recent_kernel_index();
assert_eq!(index.get_list(&batch, commit), Ok(None));
assert_eq!(index.get_list(&batch, commit2), Ok(None));
assert_eq!(
index.push_pos(&batch, commit, CommitPos { pos: 1, height: 1 }),
index.push_pos(&mut batch, commit, CommitPos { pos: 1, height: 1 }),
Ok(()),
);
@@ -417,7 +417,7 @@ fn test_store_kernel_idx_multiple_commits() {
assert_eq!(index.get_list(&batch, commit2), Ok(None));
assert_eq!(
index.push_pos(&batch, commit2, CommitPos { pos: 2, height: 2 }),
index.push_pos(&mut batch, commit2, CommitPos { pos: 2, height: 2 }),
Ok(()),
);
@@ -436,7 +436,7 @@ fn test_store_kernel_idx_multiple_commits() {
);
assert_eq!(
index.push_pos(&batch, commit, CommitPos { pos: 3, height: 3 }),
index.push_pos(&mut batch, commit, CommitPos { pos: 3, height: 3 }),
Ok(()),
);
@@ -453,7 +453,7 @@ fn test_store_kernel_idx_multiple_commits() {
);
assert_eq!(
index.pop_pos(&batch, commit),
index.pop_pos(&mut batch, commit),
Ok(Some(CommitPos { pos: 3, height: 3 })),
);
@@ -488,18 +488,18 @@ fn test_store_kernel_idx_clear() -> Result<(), Error> {
// Add a couple of single entries to the index and commit the batch.
{
let batch = store.batch()?;
let mut batch = store.batch()?;
assert_eq!(index.peek_pos(&batch, commit), Ok(None));
assert_eq!(index.get_list(&batch, commit), Ok(None));
assert_eq!(
index.push_pos(&batch, commit, CommitPos { pos: 1, height: 1 }),
index.push_pos(&mut batch, commit, CommitPos { pos: 1, height: 1 }),
Ok(()),
);
assert_eq!(
index.push_pos(
&batch,
&mut batch,
commit2,
CommitPos {
pos: 10,
@@ -544,8 +544,8 @@ fn test_store_kernel_idx_clear() -> Result<(), Error> {
// Clear the index and confirm everything was deleted as expected.
{
let batch = store.batch()?;
assert_eq!(index.clear(&batch), Ok(()));
let mut batch = store.batch()?;
assert_eq!(index.clear(&mut batch), Ok(()));
assert_eq!(index.peek_pos(&batch, commit), Ok(None));
assert_eq!(index.get_list(&batch, commit), Ok(None));
assert_eq!(index.peek_pos(&batch, commit2), Ok(None));
@@ -555,13 +555,13 @@ fn test_store_kernel_idx_clear() -> Result<(), Error> {
// Add multiple entries to the index, commit the batch.
{
let batch = store.batch()?;
let mut batch = store.batch()?;
assert_eq!(
index.push_pos(&batch, commit, CommitPos { pos: 1, height: 1 }),
index.push_pos(&mut batch, commit, CommitPos { pos: 1, height: 1 }),
Ok(()),
);
assert_eq!(
index.push_pos(&batch, commit, CommitPos { pos: 2, height: 2 }),
index.push_pos(&mut batch, commit, CommitPos { pos: 2, height: 2 }),
Ok(()),
);
assert_eq!(
@@ -577,8 +577,8 @@ fn test_store_kernel_idx_clear() -> Result<(), Error> {
// Clear the index and confirm everything was deleted as expected.
{
let batch = store.batch()?;
assert_eq!(index.clear(&batch), Ok(()));
let mut batch = store.batch()?;
assert_eq!(index.clear(&mut batch), Ok(()));
assert_eq!(index.peek_pos(&batch, commit), Ok(None));
assert_eq!(index.get_list(&batch, commit), Ok(None));
batch.commit()?;
chain/tests/test_block_known.rs
+1 -1
View File
@@ -61,7 +61,7 @@ fn check_known() {
{
let chain = init_chain(chain_dir, genesis.clone());
let store = chain.store();
let batch = store.batch().unwrap();
let mut batch = store.batch().unwrap();
let head_header = chain.head_header().unwrap();
let prev = batch.get_previous_header(&head_header).unwrap();
batch.save_body_head(&Tip::from_header(&prev)).unwrap();
p2p/src/peers.rs
+35 -25
View File
@@ -275,22 +275,30 @@ impl Peers {
}
}
/// Iterator over all peers we know about (stored in our db).
pub fn peer_data_iter(&self) -> Result<impl Iterator<Item = PeerData>, Error> {
self.store.peers_iter().map_err(From::from)
}
/// Convenience for reading all peer data from the db.
pub fn all_peer_data(&self) -> Vec<PeerData> {
self.peer_data_iter()
.map(|peers| peers.collect())
.unwrap_or(vec![])
match self.store.iter_batch() {
Ok(batch) => match batch.peers_iter() {
Ok(iter) => iter.collect(),
Err(e) => {
error!("failed to get all peer data: {:?}", e);
vec![]
}
},
Err(e) => {
error!("failed to get all peer data: {:?}", e);
vec![]
}
}
}
/// Find peers in store (not necessarily connected) and return their data
pub fn find_peers(&self, state: State, cap: Capabilities, count: usize) -> Vec<PeerData> {
match self.store.find_peers(state, cap, count) {
Ok(peers) => peers,
match self.store.iter_batch() {
Ok(batch) => batch.find_peers(state, cap, count).unwrap_or_else(|e| {
error!("failed to find peers: {:?}", e);
vec![]
}),
Err(e) => {
error!("failed to find peers: {:?}", e);
vec![]
@@ -490,24 +498,26 @@ impl Peers {
let now = Utc::now();
// Delete defunct peers from storage
let _ = self.store.delete_peers(|peer| {
let diff = now - Utc.timestamp_opt(peer.last_connected, 0).unwrap();
if let Ok(batch) = self.store.iter_batch() {
let _ = batch.delete_peers(|peer| {
let diff = now - Utc.timestamp_opt(peer.last_connected, 0).unwrap();
let should_remove = peer.flags == State::Defunct
&& diff > Duration::seconds(global::PEER_EXPIRATION_REMOVE_TIME);
let should_remove = peer.flags == State::Defunct
&& diff > Duration::seconds(global::PEER_EXPIRATION_REMOVE_TIME);
if should_remove {
debug!(
"removing peer {:?}: last connected {} days {} hours {} minutes ago.",
peer.addr,
diff.num_days(),
diff.num_hours(),
diff.num_minutes()
);
}
if should_remove {
debug!(
"removing peer {:?}: last connected {} days {} hours {} minutes ago.",
peer.addr,
diff.num_days(),
diff.num_hours(),
diff.num_minutes()
);
}
should_remove
});
should_remove
});
}
}
}
p2p/src/store.rs
+51 -49
View File
@@ -119,19 +119,19 @@ impl PeerStore {
/// Instantiates a new peer store under the provided root path.
pub fn new(db_root: &str) -> Result<PeerStore, Error> {
let db = grin_store::Store::new(db_root, Some(DB_NAME), Some(STORE_SUBPATH), None)?;
Ok(PeerStore { db: db })
Ok(PeerStore { db })
}
pub fn save_peer(&self, p: &PeerData) -> Result<(), Error> {
debug!("save_peer: {:?} marked {:?}", p.addr, p.flags);
let batch = self.db.batch()?;
let mut batch = self.db.batch()?;
batch.put_ser(&peer_key(p.addr)[..], p)?;
batch.commit()
}
pub fn save_peers(&self, p: Vec<PeerData>) -> Result<(), Error> {
let batch = self.db.batch()?;
let mut batch = self.db.batch()?;
for pd in p {
debug!("save_peers: {:?} marked {:?}", pd.addr, pd.flags);
batch.put_ser(&peer_key(pd.addr)[..], &pd)?;
@@ -149,50 +149,11 @@ impl PeerStore {
self.db.exists(&peer_key(peer_addr)[..])
}
/// TODO - allow below added to avoid github issue reports
#[allow(dead_code)]
pub fn delete_peer(&self, peer_addr: PeerAddr) -> Result<(), Error> {
let batch = self.db.batch()?;
batch.delete(&peer_key(peer_addr)[..])?;
batch.commit()
}
/// Find some peers in our local db.
pub fn find_peers(
&self,
state: State,
cap: Capabilities,
count: usize,
) -> Result<Vec<PeerData>, Error> {
let peers = self
.peers_iter()?
.filter(|p| p.flags == state && p.capabilities.contains(cap))
.choose_multiple(&mut thread_rng(), count);
Ok(peers)
}
/// Iterator over all known peers.
pub fn peers_iter(&self) -> Result<impl Iterator<Item = PeerData>, Error> {
let key = to_key(PEER_PREFIX, "");
let protocol_version = self.db.protocol_version();
self.db.iter(&key, move |_, mut v| {
ser::deserialize(&mut v, protocol_version, DeserializationMode::default())
.map_err(From::from)
})
}
/// List all known peers
/// Used for /v1/peers/all api endpoint
pub fn all_peers(&self) -> Result<Vec<PeerData>, Error> {
let peers: Vec<PeerData> = self.peers_iter()?.collect();
Ok(peers)
}
/// Convenience method to load a peer data, update its status and save it
/// back. If new state is Banned its last banned time will be updated too.
/// If new state is Defunct last connection attempt will be updated too.
pub fn update_state(&self, peer_addr: PeerAddr, new_state: State) -> Result<(), Error> {
let batch = self.db.batch()?;
let mut batch = self.db.batch()?;
let mut peer = option_to_not_found(
batch.get_ser::<PeerData>(&peer_key(peer_addr)[..], None),
@@ -209,8 +170,52 @@ impl PeerStore {
batch.commit()
}
/// Builds a new iterator batch to be used with this store.
pub fn iter_batch(&self) -> Result<PeersIterBatch<'_>, Error> {
Ok(PeersIterBatch {
db: self.db.batch()?,
})
}
}
pub struct PeersIterBatch<'a> {
db: grin_store::Batch<'a>,
}
impl<'a> PeersIterBatch<'a> {
/// Iterator over all known peers.
pub fn peers_iter(&self) -> Result<impl Iterator<Item = PeerData> + 'a, Error> {
let key = to_key(PEER_PREFIX, "");
let protocol_version = self.db.protocol_version();
self.db.iter(&key, move |_, mut v| {
ser::deserialize(&mut v, protocol_version, DeserializationMode::default())
.map_err(From::from)
})
}
/// Find some peers in our local db.
pub fn find_peers(
&self,
state: State,
cap: Capabilities,
count: usize,
) -> Result<Vec<PeerData>, Error> {
let peers = self
.peers_iter()?
.filter(|p| p.flags == state && p.capabilities.contains(cap))
.choose_multiple(&mut thread_rng(), count);
Ok(peers)
}
/// List all known peers
/// Used for /v1/peers/all api endpoint
pub fn all_peers(&self) -> Result<Vec<PeerData>, Error> {
let peers: Vec<PeerData> = self.peers_iter()?.collect();
Ok(peers)
}
/// Deletes peers from the storage that satisfy some condition `predicate`
pub fn delete_peers<F>(&self, predicate: F) -> Result<(), Error>
pub fn delete_peers<F>(mut self, predicate: F) -> Result<(), Error>
where
F: Fn(&PeerData) -> bool,
{
@@ -224,13 +229,10 @@ impl PeerStore {
// Delete peers in single batch
if !to_remove.is_empty() {
let batch = self.db.batch()?;
for peer in to_remove {
batch.delete(&peer_key(peer.addr)[..])?;
self.db.delete(&peer_key(peer.addr)[..])?;
}
batch.commit()?;
self.db.commit()?;
}
Ok(())
servers/Cargo.toml
-1
View File
@@ -15,7 +15,6 @@ hyper-rustls = "0.23"
fs2 = "0.4"
futures = "0.3"
http = "0.2"
lmdb-zero = "0.4.4"
rand = "0.6"
serde = "1"
log = "0.4"
store/Cargo.toml
+1 -1
View File
@@ -13,7 +13,7 @@ edition = "2018"
byteorder = "1"
croaring = "1.0.1"
libc = "0.2"
lmdb-zero = "0.4.4"
heed = "0.22.1"
memmap = "0.7"
tempfile = "3.1"
serde = "1"
store/src/lmdb.rs
+163 -197
View File
@@ -14,19 +14,19 @@
//! Storage of core types using LMDB.
use std::fs;
use heed::types::Bytes;
use heed::{Database, Env, EnvOpenOptions, RoTxn, RwTxn, WithoutTls};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use lmdb_zero as lmdb;
use lmdb_zero::traits::CreateCursor;
use lmdb_zero::LmdbResultExt;
use std::time::Duration;
use std::{fs, thread};
use crate::grin_core::global;
use crate::grin_core::ser::{self, DeserializationMode, ProtocolVersion};
use crate::util::RwLock;
/// number of bytes to grow the database by when needed
pub const ALLOC_CHUNK_SIZE_DEFAULT: usize = 134_217_728; //128 MB
pub const ALLOC_CHUNK_SIZE_DEFAULT: usize = 134_217_728 / 32; //128 MB
/// And for test mode, to avoid too much disk allocation on windows
pub const ALLOC_CHUNK_SIZE_DEFAULT_TEST: usize = 1_048_576; //1 MB
const RESIZE_PERCENT: f32 = 0.9;
@@ -42,7 +42,7 @@ pub enum Error {
NotFoundErr(String),
/// Wraps an error originating from LMDB
#[error("LMDB error: {0}")]
LmdbErr(lmdb::error::Error),
LmdbErr(String),
/// Wraps a serialization error for Writeable or Readable
#[error("Serialization Error: {0}")]
SerErr(ser::Error),
@@ -54,9 +54,9 @@ pub enum Error {
OtherErr(String),
}
impl From<lmdb::error::Error> for Error {
fn from(e: lmdb::error::Error) -> Error {
Error::LmdbErr(e)
impl From<heed::Error> for Error {
fn from(e: heed::Error) -> Error {
Error::LmdbErr(e.to_string())
}
}
@@ -83,11 +83,12 @@ const DEFAULT_DB_VERSION: ProtocolVersion = ProtocolVersion(3);
/// LMDB-backed store facilitating data access and serialization. All writes
/// are done through a Batch abstraction providing atomicity.
pub struct Store {
env: Arc<lmdb::Environment>,
db: Arc<RwLock<Option<Arc<lmdb::Database<'static>>>>>,
env: Arc<Env<WithoutTls>>,
db: Arc<RwLock<Option<Arc<Database<Bytes, Bytes>>>>>,
name: String,
version: ProtocolVersion,
alloc_chunk_size: usize,
resizing: Arc<AtomicBool>,
}
impl Store {
@@ -117,53 +118,55 @@ impl Store {
))
})?;
let mut env_builder = lmdb::EnvBuilder::new()?;
env_builder.set_maxdbs(8)?;
if let Some(max_readers) = max_readers {
env_builder.set_maxreaders(max_readers)?;
}
let alloc_chunk_size = match global::is_production_mode() {
true => ALLOC_CHUNK_SIZE_DEFAULT,
false => ALLOC_CHUNK_SIZE_DEFAULT_TEST,
};
let env = unsafe { env_builder.open(&full_path, lmdb::open::NOTLS, 0o600)? };
debug!("DB Mapsize for {} is {}", full_path, env.info()?.mapsize);
let res = Store {
let env = unsafe {
let mut options = EnvOpenOptions::new().read_txn_without_tls();
let mut env_options = options.map_size(alloc_chunk_size).max_dbs(8);
if let Some(max_readers) = max_readers {
env_options = env_options.max_readers(max_readers);
}
env_options.open(&full_path)?
};
let (resize, new_size) = Self::needs_resize(Arc::new(env.clone()), alloc_chunk_size);
if resize {
unsafe {
env.resize(new_size)?;
};
}
let s = Store {
env: Arc::new(env),
db: Arc::new(RwLock::new(None)),
name: db_name,
version: DEFAULT_DB_VERSION,
alloc_chunk_size,
resizing: Default::default(),
};
{
let mut w = res.db.write();
*w = Some(Arc::new(lmdb::Database::open(
res.env.clone(),
Some(&res.name),
&lmdb::DatabaseOptions::new(lmdb::db::CREATE),
)?));
let mut write = s.env.write_txn()?;
let db = s.env.create_database(&mut write, Some(&s.name))?;
write.commit()?;
let mut w_db = s.db.write();
*w_db = Some(Arc::new(db));
}
Ok(res)
Ok(s)
// debug!("DB Mapsize for {} is {}", full_path, env.info().map_size);
}
/// Construct a new store using a specific protocol version.
/// Permits access to the db with legacy protocol versions for db migrations.
pub fn with_version(&self, version: ProtocolVersion) -> Store {
let alloc_chunk_size = match global::is_production_mode() {
true => ALLOC_CHUNK_SIZE_DEFAULT,
false => ALLOC_CHUNK_SIZE_DEFAULT_TEST,
};
Store {
env: self.env.clone(),
db: self.db.clone(),
name: self.name.clone(),
version,
alloc_chunk_size,
alloc_chunk_size: self.alloc_chunk_size,
resizing: self.resizing.clone(),
}
}
@@ -172,97 +175,58 @@ impl Store {
self.version
}
/// Opens the database environment
pub fn open(&self) -> Result<(), Error> {
let mut w = self.db.write();
*w = Some(Arc::new(lmdb::Database::open(
self.env.clone(),
Some(&self.name),
&lmdb::DatabaseOptions::new(lmdb::db::CREATE),
)?));
Ok(())
}
/// Determines whether the environment needs a resize based on a simple percentage threshold
pub fn needs_resize(&self) -> Result<bool, Error> {
let env_info = self.env.info()?;
let stat = self.env.stat()?;
let size_used = stat.psize as usize * env_info.last_pgno;
trace!("DB map size: {}", env_info.mapsize);
/// Determines whether the environment needs a resize based on a simple percentage threshold.
pub fn needs_resize(env: Arc<Env<WithoutTls>>, alloc_chunk_size: usize) -> (bool, usize) {
let env_info = env.info();
let stat = env.stat();
let size_used = stat.page_size as usize * env_info.last_page_number;
trace!("DB map size: {}", env_info.map_size);
trace!("Space used: {}", size_used);
trace!("Space remaining: {}", env_info.mapsize - size_used);
trace!("Space remaining: {}", env_info.map_size - size_used);
let resize_percent = RESIZE_PERCENT;
trace!(
"Percent used: {:.*} Percent threshold: {:.*}",
4,
size_used as f64 / env_info.mapsize as f64,
size_used as f64 / env_info.map_size as f64,
4,
resize_percent
);
if size_used as f32 / env_info.mapsize as f32 > resize_percent
|| env_info.mapsize < self.alloc_chunk_size
let resize = if size_used as f32 / env_info.map_size as f32 > resize_percent
|| env_info.map_size < alloc_chunk_size
{
trace!("Resize threshold met (percent-based)");
Ok(true)
true
} else {
trace!("Resize threshold not met (percent-based)");
Ok(false)
}
}
/// Increments the database size by as many ALLOC_CHUNK_SIZES
/// to give a minimum threshold of free space
pub fn do_resize(&self) -> Result<(), Error> {
let env_info = self.env.info()?;
let stat = self.env.stat()?;
let size_used = stat.psize as usize * env_info.last_pgno;
let new_mapsize = if env_info.mapsize < self.alloc_chunk_size {
self.alloc_chunk_size
} else {
let mut tot = env_info.mapsize;
while size_used as f32 / tot as f32 > RESIZE_MIN_TARGET_PERCENT {
tot += self.alloc_chunk_size;
}
tot
false
};
// close
let mut w = self.db.write();
*w = None;
let new_size = if resize {
if env_info.map_size < alloc_chunk_size {
alloc_chunk_size
} else {
let mut tot = env_info.map_size;
while size_used as f32 / tot as f32 > RESIZE_MIN_TARGET_PERCENT {
tot += alloc_chunk_size;
}
tot
}
} else {
env_info.map_size
};
unsafe {
self.env.set_mapsize(new_mapsize)?;
}
*w = Some(Arc::new(lmdb::Database::open(
self.env.clone(),
Some(&self.name),
&lmdb::DatabaseOptions::new(lmdb::db::CREATE),
)?));
info!(
"Resized database from {} to {}",
env_info.mapsize, new_mapsize
);
Ok(())
(resize, new_size)
}
/// Gets a value from the db, provided its key.
/// Deserializes the retrieved data using the provided function.
pub fn get_with<F, T>(
&self,
key: &[u8],
access: &lmdb::ConstAccessor<'_>,
db: &lmdb::Database<'_>,
deserialize: F,
) -> Result<Option<T>, Error>
fn get_with<F, T>(&self, key: &[u8], read: &RoTxn, deserialize: F) -> Result<Option<T>, Error>
where
F: Fn(&[u8], &[u8]) -> Result<T, Error>,
{
let res: Option<&[u8]> = access.get(db, key).to_opt()?;
let db = self.db.read();
let res: Option<&[u8]> = db.as_ref().unwrap().get(read, key)?;
match res {
None => Ok(None),
Some(res) => deserialize(key, res).map(Some),
@@ -276,81 +240,82 @@ impl Store {
key: &[u8],
deser_mode: Option<DeserializationMode>,
) -> Result<Option<T>, Error> {
let lock = self.db.read();
let db = lock
.as_ref()
.ok_or_else(|| Error::NotFoundErr("chain db is None".to_string()))?;
let txn = lmdb::ReadTransaction::new(self.env.clone())?;
let access = txn.access();
let d = match deser_mode {
Some(d) => d,
_ => DeserializationMode::default(),
};
self.get_with(key, &access, &db, |_, mut data| {
let read = self.env.read_txn()?;
self.get_with(key, &read, |_, mut data| {
ser::deserialize(&mut data, self.protocol_version(), d).map_err(From::from)
})
}
/// Whether the provided key exists
/// Whether the provided key exists.
pub fn exists(&self, key: &[u8]) -> Result<bool, Error> {
let lock = self.db.read();
let db = lock
.as_ref()
.ok_or_else(|| Error::NotFoundErr("chain db is None".to_string()))?;
let txn = lmdb::ReadTransaction::new(self.env.clone())?;
let access = txn.access();
let res: Option<&lmdb::Ignore> = access.get(db, key).to_opt()?;
let db = self.db.read();
let read = self.env.read_txn()?;
let res = db.as_ref().unwrap().get(&read, key)?;
Ok(res.is_some())
}
/// Produces an iterator from the provided key prefix.
pub fn iter<F, T>(&self, prefix: &[u8], deserialize: F) -> Result<PrefixIterator<F, T>, Error>
pub fn iter<F, T>(
&'_ self,
prefix: &[u8],
deserialize: F,
) -> Result<PrefixIterator<'_, F, T>, Error>
where
F: Fn(&[u8], &[u8]) -> Result<T, Error>,
{
let lock = self.db.read();
let db = lock
.as_ref()
.ok_or_else(|| Error::NotFoundErr("chain db is None".to_string()))?;
let tx = Arc::new(lmdb::ReadTransaction::new(self.env.clone())?);
let cursor = Arc::new(tx.cursor(db.clone())?);
Ok(PrefixIterator::new(tx, cursor, prefix, deserialize))
let db = Arc::new(self.db.read().clone());
let read = self.env.read_txn()?;
Ok(PrefixIterator::new(db.clone(), read, prefix, deserialize))
}
/// Builds a new batch to be used with this store.
pub fn batch(&self) -> Result<Batch<'_>, Error> {
// check if the db needs resizing before returning the batch
if self.needs_resize()? {
self.do_resize()?;
while self.resizing.load(Ordering::SeqCst) {
debug!("Wait resizing {} DB", self.name);
thread::sleep(Duration::from_millis(100));
}
let tx = lmdb::WriteTransaction::new(self.env.clone())?;
Ok(Batch { store: self, tx })
let (resize, new_size) = Self::needs_resize(self.env.clone(), self.alloc_chunk_size);
if resize {
self.resizing.store(true, Ordering::SeqCst);
debug!("Start resizing {} DB", self.name);
unsafe {
thread::sleep(Duration::from_millis(3000));
self.env.resize(new_size)?;
}
self.resizing.store(false, Ordering::SeqCst);
debug!("End resizing {} DB", self.name);
}
Ok(Batch::new(self)?)
}
}
/// Batch to write multiple Writeables to db in an atomic manner.
pub struct Batch<'a> {
store: &'a Store,
tx: lmdb::WriteTransaction<'a>,
write: RwTxn<'a>,
}
impl<'a> Batch<'a> {
/// Writes a single key/value pair to the db
pub fn put(&self, key: &[u8], value: &[u8]) -> Result<(), Error> {
let lock = self.store.db.read();
let db = lock
.as_ref()
.ok_or_else(|| Error::NotFoundErr("chain db is None".to_string()))?;
self.tx
.access()
.put(db, key, value, lmdb::put::Flags::empty())?;
/// Creates a new batch for provided db.
pub fn new(store: &'a Store) -> Result<Batch<'a>, Error> {
let write = store.env.write_txn()?;
Ok(Batch { store, write })
}
/// Writes a single key/value pair to the db.
pub fn put(&mut self, key: &[u8], value: &[u8]) -> Result<(), Error> {
let db = self.store.db.read();
db.as_ref().unwrap().put(&mut self.write, key, value)?;
Ok(())
}
/// Writes a single key and its `Writeable` value to the db.
/// Encapsulates serialization using the (default) version configured on the store instance.
pub fn put_ser<W: ser::Writeable>(&self, key: &[u8], value: &W) -> Result<(), Error> {
pub fn put_ser<W: ser::Writeable>(&mut self, key: &[u8], value: &W) -> Result<(), Error> {
self.put_ser_with_version(key, value, self.store.protocol_version())
}
@@ -362,7 +327,7 @@ impl<'a> Batch<'a> {
/// Writes a single key and its `Writeable` value to the db.
/// Encapsulates serialization using the specified protocol version.
pub fn put_ser_with_version<W: ser::Writeable>(
&self,
&mut self,
key: &[u8],
value: &W,
version: ProtocolVersion,
@@ -376,33 +341,29 @@ impl<'a> Batch<'a> {
/// Low-level access for retrieving data by key.
/// Takes a function for flexible deserialization.
pub fn get_with<F, T>(&self, key: &[u8], deserialize: F) -> Result<Option<T>, Error>
fn get_with<F, T>(&self, key: &[u8], deserialize: F) -> Result<Option<T>, Error>
where
F: Fn(&[u8], &[u8]) -> Result<T, Error>,
{
let access = self.tx.access();
let lock = self.store.db.read();
let db = lock
.as_ref()
.ok_or_else(|| Error::NotFoundErr("chain db is None".to_string()))?;
self.store.get_with(key, &access, &db, deserialize)
let read = self.write.nested_read_txn()?;
self.store.get_with(key, &read, deserialize)
}
/// Whether the provided key exists.
/// This is in the context of the current write transaction.
pub fn exists(&self, key: &[u8]) -> Result<bool, Error> {
let access = self.tx.access();
let lock = self.store.db.read();
let db = lock
.as_ref()
.ok_or_else(|| Error::NotFoundErr("chain db is None".to_string()))?;
let res: Option<&lmdb::Ignore> = access.get(db, key).to_opt()?;
let db = self.store.db.read();
let read = self.write.nested_read_txn()?;
let res = db.as_ref().unwrap().get(&read, key)?;
Ok(res.is_some())
}
/// Produces an iterator from the provided key prefix.
pub fn iter<F, T>(&self, prefix: &[u8], deserialize: F) -> Result<PrefixIterator<F, T>, Error>
pub fn iter<F, T>(
&self,
prefix: &[u8],
deserialize: F,
) -> Result<PrefixIterator<'a, F, T>, Error>
where
F: Fn(&[u8], &[u8]) -> Result<T, Error>,
{
@@ -427,85 +388,90 @@ impl<'a> Batch<'a> {
})
}
/// Deletes a key/value pair from the db
pub fn delete(&self, key: &[u8]) -> Result<(), Error> {
let lock = self.store.db.read();
let db = lock
.as_ref()
.ok_or_else(|| Error::NotFoundErr("chain db is None".to_string()))?;
self.tx.access().del_key(db, key)?;
/// Deletes a key/value pair from the db.
pub fn delete(&mut self, key: &[u8]) -> Result<(), Error> {
let db = self.store.db.read();
db.as_ref().unwrap().delete(&mut self.write, key)?;
Ok(())
}
/// Writes the batch to db
/// Writes the batch to db.
pub fn commit(self) -> Result<(), Error> {
self.tx.commit()?;
self.write.commit()?;
Ok(())
}
/// Creates a child of this batch. It will be merged with its parent on
/// commit, abandoned otherwise.
pub fn child(&mut self) -> Result<Batch<'_>, Error> {
let write = self.store.env.nested_write_txn(&mut self.write)?;
Ok(Batch {
store: self.store,
tx: self.tx.child_tx()?,
write,
})
}
}
/// An iterator based on key prefix.
/// Caller is responsible for deserialization of the data.
pub struct PrefixIterator<F, T>
pub struct PrefixIterator<'a, F, T>
where
F: Fn(&[u8], &[u8]) -> Result<T, Error>,
{
tx: Arc<lmdb::ReadTransaction<'static>>,
cursor: Arc<lmdb::Cursor<'static, 'static>>,
seek: bool,
db: Arc<Option<Arc<Database<Bytes, Bytes>>>>,
read: Arc<RoTxn<'a, WithoutTls>>,
skip: usize,
prefix: Vec<u8>,
deserialize: F,
}
impl<F, T> Iterator for PrefixIterator<F, T>
impl<'a, F, T> Iterator for PrefixIterator<'a, F, T>
where
F: Fn(&[u8], &[u8]) -> Result<T, Error>,
{
type Item = T;
fn next(&mut self) -> Option<Self::Item> {
let access = self.tx.access();
let cursor = Arc::get_mut(&mut self.cursor).expect("failed to get cursor");
let kv: Result<(&[u8], &[u8]), _> = if self.seek {
cursor.next(&access)
} else {
self.seek = true;
cursor.seek_range_k(&access, &self.prefix[..])
};
kv.ok()
.filter(|(k, _)| k.starts_with(self.prefix.as_slice()))
.map(|(k, v)| match (self.deserialize)(k, v) {
Ok(v) => Some(v),
Err(_) => None,
})
.flatten()
let db = self.db.as_ref();
if let Ok(iter) = db.as_ref().unwrap().iter(&self.read) {
let kv = iter
.filter(|i| {
if let Ok(i) = i {
return i.0.starts_with(&self.prefix);
}
false
})
.skip(self.skip)
.next()
.transpose()
.unwrap_or(None);
self.skip += 1;
if let Some((k, v)) = kv {
return match (self.deserialize)(k, v) {
Ok(v) => Some(v),
Err(_) => None,
};
}
}
None
}
}
impl<F, T> PrefixIterator<F, T>
impl<'a, F, T> PrefixIterator<'a, F, T>
where
F: Fn(&[u8], &[u8]) -> Result<T, Error>,
{
/// Initialize a new prefix iterator.
pub fn new(
tx: Arc<lmdb::ReadTransaction<'static>>,
cursor: Arc<lmdb::Cursor<'static, 'static>>,
db: Arc<Option<Arc<Database<Bytes, Bytes>>>>,
read: RoTxn<'a, WithoutTls>,
prefix: &[u8],
deserialize: F,
) -> PrefixIterator<F, T> {
) -> PrefixIterator<'a, F, T> {
PrefixIterator {
tx,
cursor,
seek: false,
db,
read: Arc::new(read),
skip: 0,
prefix: prefix.to_vec(),
deserialize,
}
store/tests/lmdb.rs
+4 -4
View File
@@ -76,7 +76,7 @@ fn test_exists() -> Result<(), store::Error> {
let value = [1, 1, 1, 1];
// Start new batch and insert a new key/value entry.
let batch = store.batch()?;
let mut batch = store.batch()?;
batch.put(&key, &value)?;
// Check we can see the new entry in uncommitted batch.
@@ -105,7 +105,7 @@ fn test_iter() -> Result<(), store::Error> {
let value = [1, 1, 1, 1];
// Start new batch and insert a new key/value entry.
let batch = store.batch()?;
let mut batch = store.batch()?;
batch.put(&key, &value)?;
// TODO - This is not currently possible (and we need to be aware of this).
@@ -144,7 +144,7 @@ fn lmdb_allocate() -> Result<(), store::Error> {
println!("Allocating chunk: {}", i);
let chunk = PhatChunkStruct::new();
let key_val = format!("phat_chunk_set_1_{}", i);
let batch = store.batch()?;
let mut batch = store.batch()?;
let key = store::to_key(b'P', &key_val);
batch.put_ser(&key, &chunk)?;
batch.commit()?;
@@ -160,7 +160,7 @@ fn lmdb_allocate() -> Result<(), store::Error> {
println!("Allocating chunk: {}", i);
let chunk = PhatChunkStruct::new();
let key_val = format!("phat_chunk_set_2_{}", i);
let batch = store.batch()?;
let mut batch = store.batch()?;
let key = store::to_key(b'P', &key_val);
batch.put_ser(&key, &chunk)?;
batch.commit()?;