diff --git a/Cargo.lock b/Cargo.lock index 429a99f9..74db18b7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -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" diff --git a/chain/src/chain.rs b/chain/src/chain.rs index 273a4415..aabadc32 100644 --- a/chain/src/chain.rs +++ b/chain/src/chain.rs @@ -150,8 +150,8 @@ pub struct Chain { store: Arc, adapter: Arc, orphans: Arc, - txhashset: Arc>, - header_pmmr: Arc>>, + txhashset: Arc>, + header_pmmr: Arc>>, pibd_segmenter: Arc>>, pibd_desegmenter: Arc>>, // 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, - txhashset: &'a mut txhashset::TxHashSet, + batch: Batch<'a>, + header_pmmr: &'a mut PMMRHandle, + txhashset: &'a mut TxHashSet, ) -> Result, 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 { 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, + header_pmmr: &PMMRHandle, 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)) diff --git a/chain/src/linked_list.rs b/chain/src/linked_list.rs index 14039b66..79d22564 100644 --- a/chain/src/linked_list.rs +++ b/chain/src/linked_list.rs @@ -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: ::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::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::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, Error> { + fn pop_pos(&self, batch: &mut Batch<'_>, commit: Commitment) -> Result, 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 RewindableListIndex for MultiIndex { - 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 RewindableListIndex for MultiIndex { } impl PruneableListIndex for MultiIndex { - 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 PruneableListIndex for MultiIndex { /// 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 PruneableListIndex for MultiIndex { /// Pop off the back/tail of the linked list. /// Used when pruning old data. - fn pop_pos_back(&self, batch: &Batch<'_>, commit: Commitment) -> Result, Error> { + fn pop_pos_back(&self, batch: &mut Batch<'_>, commit: Commitment) -> Result, Error> { match self.get_list(batch, commit)? { None => Ok(None), Some(ListWrapper::Single { pos }) => { diff --git a/chain/src/pipe.rs b/chain/src/pipe.rs index e49ea579..820b3a07 100644 --- a/chain/src/pipe.rs +++ b/chain/src/pipe.rs @@ -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 { let extension = &mut ext.extension; diff --git a/chain/src/store.rs b/chain/src/store.rs index e11d1f48..56fb7060 100644 --- a/chain/src/store.rs +++ b/chain/src/store.rs @@ -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, CommitPos)>, Error> { + pub fn output_pos_iter( + &self, + ) -> Result, 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, Error> { + pub fn blocks_iter(&self) -> Result + '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, Vec)>, Error> { + pub fn blocks_raw_iter(&self) -> Result, Vec)> + 'a, Error> { let key = to_key(BLOCK_PREFIX, ""); self.db.iter(&key, |k, v| Ok((k.to_vec(), v.to_vec()))) } diff --git a/chain/src/txhashset/desegmenter.rs b/chain/src/txhashset/desegmenter.rs index ca391397..2efbfee3 100644 --- a/chain/src/txhashset/desegmenter.rs +++ b/chain/src/txhashset/desegmenter.rs @@ -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()?; diff --git a/chain/src/txhashset/txhashset.rs b/chain/src/txhashset/txhashset.rs index a0d6bc64..cbdb899f 100644 --- a/chain/src/txhashset/txhashset.rs +++ b/chain/src/txhashset/txhashset.rs @@ -538,7 +538,7 @@ impl TxHashSet { pub fn init_recent_kernel_pos_index( &self, header_pmmr: &PMMRHandle, - 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, - batch: &Batch<'_>, + batch: &mut Batch<'_>, status: Option>, stop_state: Option>, ) -> Result<(), Error> { @@ -635,7 +635,7 @@ impl TxHashSet { pub fn init_output_pos_index( &self, header_pmmr: &PMMRHandle, - batch: &Batch<'_>, + batch: &mut Batch<'_>, ) -> Result<(), Error> { let now = Instant::now(); @@ -733,10 +733,10 @@ pub fn extending_readonly( inner: F, ) -> Result where - F: FnOnce(&mut ExtensionPair<'_>, &Batch<'_>) -> Result, + F: FnOnce(&mut ExtensionPair<'_>, &mut Batch<'_>) -> Result, { 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 where - F: FnOnce(&mut ExtensionPair<'_>, &Batch<'_>) -> Result, + F: FnOnce(&mut ExtensionPair<'_>, &mut Batch<'_>) -> Result, { let sizes: (u64, u64, u64); let res: Result; @@ -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, +pub fn header_extending_readonly( + handle: &mut PMMRHandle, store: &ChainStore, inner: F, ) -> Result where - F: FnOnce(&mut HeaderExtension<'_>, &Batch<'_>) -> Result, + F: FnOnce(&mut HeaderExtension<'_>, &mut Batch<'_>) -> Result, { - 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 where - F: FnOnce(&mut HeaderExtension<'_>, &Batch<'_>) -> Result, + F: FnOnce(&mut HeaderExtension<'_>, &mut Batch<'_>) -> Result, { let size: u64; let res: Result; @@ -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, Error> { + fn rewind_single_block( + &mut self, + block: &Block, + batch: &mut Batch<'_>, + ) -> Result, 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(()); } diff --git a/chain/tests/store_indices.rs b/chain/tests/store_indices.rs index 5cc19356..c011ed6c 100644 --- a/chain/tests/store_indices.rs +++ b/chain/tests/store_indices.rs @@ -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. diff --git a/chain/tests/store_kernel_pos_index.rs b/chain/tests/store_kernel_pos_index.rs index 664e2818..d7cf6d30 100644 --- a/chain/tests/store_kernel_pos_index.rs +++ b/chain/tests/store_kernel_pos_index.rs @@ -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()?; diff --git a/chain/tests/test_block_known.rs b/chain/tests/test_block_known.rs index bdab3a81..d5c0694a 100644 --- a/chain/tests/test_block_known.rs +++ b/chain/tests/test_block_known.rs @@ -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(); diff --git a/p2p/src/peers.rs b/p2p/src/peers.rs index 0f10a84e..21340b10 100644 --- a/p2p/src/peers.rs +++ b/p2p/src/peers.rs @@ -275,22 +275,30 @@ impl Peers { } } - /// Iterator over all peers we know about (stored in our db). - pub fn peer_data_iter(&self) -> Result, 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 { - 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 { - 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![] @@ -445,24 +453,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 + }); + } } } diff --git a/p2p/src/store.rs b/p2p/src/store.rs index 4b4dde9f..85b30437 100644 --- a/p2p/src/store.rs +++ b/p2p/src/store.rs @@ -119,19 +119,19 @@ impl PeerStore { /// Instantiates a new peer store under the provided root path. pub fn new(db_root: &str) -> Result { 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) -> 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, 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, 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, Error> { - let peers: Vec = 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::(&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, 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 + '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, 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, Error> { + let peers: Vec = self.peers_iter()?.collect(); + Ok(peers) + } + /// Deletes peers from the storage that satisfy some condition `predicate` - pub fn delete_peers(&self, predicate: F) -> Result<(), Error> + pub fn delete_peers(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(()) diff --git a/servers/Cargo.toml b/servers/Cargo.toml index cb162771..a9a6978a 100644 --- a/servers/Cargo.toml +++ b/servers/Cargo.toml @@ -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" diff --git a/store/Cargo.toml b/store/Cargo.toml index 72f30ca8..9654cc1f 100644 --- a/store/Cargo.toml +++ b/store/Cargo.toml @@ -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" diff --git a/store/src/lmdb.rs b/store/src/lmdb.rs index b1173fbc..f2e0faac 100644 --- a/store/src/lmdb.rs +++ b/store/src/lmdb.rs @@ -17,9 +17,8 @@ use std::fs; use std::sync::Arc; -use lmdb_zero as lmdb; -use lmdb_zero::traits::CreateCursor; -use lmdb_zero::LmdbResultExt; +use heed::types::Bytes; +use heed::{Database, Env, EnvOpenOptions, RoTxn, RwTxn, WithoutTls}; use crate::grin_core::global; use crate::grin_core::ser::{self, DeserializationMode, ProtocolVersion}; @@ -42,7 +41,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 +53,9 @@ pub enum Error { OtherErr(String), } -impl From for Error { - fn from(e: lmdb::error::Error) -> Error { - Error::LmdbErr(e) +impl From for Error { + fn from(e: heed::Error) -> Error { + Error::LmdbErr(e.to_string()) } } @@ -83,8 +82,8 @@ 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, - db: Arc>>>>, + env: Arc>, + db: Arc>>>>, name: String, version: ProtocolVersion, alloc_chunk_size: usize, @@ -117,53 +116,53 @@ 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, }; - { - 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, } } @@ -172,97 +171,50 @@ 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 { - 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>, 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) - } - } + 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; + let new_size = { + let mut tot = env_info.map_size; while size_used as f32 / tot as f32 > RESIZE_MIN_TARGET_PERCENT { - tot += self.alloc_chunk_size; + tot += alloc_chunk_size; } tot }; - - // close - let mut w = self.db.write(); - *w = None; - - 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( - &self, - key: &[u8], - access: &lmdb::ConstAccessor<'_>, - db: &lmdb::Database<'_>, - deserialize: F, - ) -> Result, Error> + fn get_with(&self, key: &[u8], read: &RoTxn, deserialize: F) -> Result, Error> where F: Fn(&[u8], &[u8]) -> Result, { - 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 +228,73 @@ impl Store { key: &[u8], deser_mode: Option, ) -> Result, 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 { - 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(&self, prefix: &[u8], deserialize: F) -> Result, Error> + pub fn iter( + &'_ self, + prefix: &[u8], + deserialize: F, + ) -> Result, Error> where F: Fn(&[u8], &[u8]) -> Result, { - 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, Error> { - // check if the db needs resizing before returning the batch - if self.needs_resize()? { - self.do_resize()?; + let (resize, new_size) = Self::needs_resize(self.env.clone(), self.alloc_chunk_size); + if resize { + unsafe { + self.env.resize(new_size)?; + } } - let tx = lmdb::WriteTransaction::new(self.env.clone())?; - Ok(Batch { store: self, tx }) + 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, 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(&self, key: &[u8], value: &W) -> Result<(), Error> { + pub fn put_ser(&mut self, key: &[u8], value: &W) -> Result<(), Error> { self.put_ser_with_version(key, value, self.store.protocol_version()) } @@ -362,7 +306,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( - &self, + &mut self, key: &[u8], value: &W, version: ProtocolVersion, @@ -376,33 +320,26 @@ impl<'a> Batch<'a> { /// Low-level access for retrieving data by key. /// Takes a function for flexible deserialization. - pub fn get_with(&self, key: &[u8], deserialize: F) -> Result, Error> + fn get_with(&self, key: &[u8], deserialize: F) -> Result, Error> where F: Fn(&[u8], &[u8]) -> Result, { - 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 { - 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()?; - Ok(res.is_some()) + Ok(self.store.exists(key)?) } /// Produces an iterator from the provided key prefix. - pub fn iter(&self, prefix: &[u8], deserialize: F) -> Result, Error> + pub fn iter( + &self, + prefix: &[u8], + deserialize: F, + ) -> Result, Error> where F: Fn(&[u8], &[u8]) -> Result, { @@ -427,85 +364,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, 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 +pub struct PrefixIterator<'a, F, T> where F: Fn(&[u8], &[u8]) -> Result, { - tx: Arc>, - cursor: Arc>, - seek: bool, + db: Arc>>>, + read: Arc>, + skip: usize, prefix: Vec, deserialize: F, } -impl Iterator for PrefixIterator +impl<'a, F, T> Iterator for PrefixIterator<'a, F, T> where F: Fn(&[u8], &[u8]) -> Result, { type Item = T; fn next(&mut self) -> Option { - 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 PrefixIterator +impl<'a, F, T> PrefixIterator<'a, F, T> where F: Fn(&[u8], &[u8]) -> Result, { /// Initialize a new prefix iterator. pub fn new( - tx: Arc>, - cursor: Arc>, + db: Arc>>>, + read: RoTxn<'a, WithoutTls>, prefix: &[u8], deserialize: F, - ) -> PrefixIterator { + ) -> PrefixIterator<'a, F, T> { PrefixIterator { - tx, - cursor, - seek: false, + db, + read: Arc::new(read), + skip: 0, prefix: prefix.to_vec(), deserialize, } diff --git a/store/tests/lmdb.rs b/store/tests/lmdb.rs index dc3d2329..c9fe6ceb 100644 --- a/store/tests/lmdb.rs +++ b/store/tests/lmdb.rs @@ -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()?;