Implement KVStore for TestStore, drop process_events_async_with_kv_store_sync
#4069
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -57,7 +57,7 @@ use crate::util::dyn_signer::{ | |
| use crate::util::logger::{Logger, Record}; | ||
| #[cfg(feature = "std")] | ||
| use crate::util::mut_global::MutGlobal; | ||
| use crate::util::persist::{KVStore, KVStoreSync, MonitorName}; | ||
| use crate::util::ser::{Readable, ReadableArgs, Writeable, Writer}; | ||
| use crate::util::test_channel_signer::{EnforcementState, TestChannelSigner}; | ||
|
|
||
|
|
@@ -84,7 +84,10 @@ use crate::io; | |
| use crate::prelude::*; | ||
| use crate::sign::{EntropySource, NodeSigner, RandomBytes, Recipient, SignerProvider}; | ||
| use crate::sync::{Arc, Mutex}; | ||
| use alloc::boxed::Box; | ||
| use core::future::Future; | ||
| use core::mem; | ||
| use core::pin::Pin; | ||
| use core::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; | ||
| use core::time::Duration; | ||
|
|
||
|
|
@@ -863,10 +866,8 @@ impl TestStore { | |
| let persisted_bytes = Mutex::new(new_hash_map()); | ||
| Self { persisted_bytes, read_only } | ||
| } | ||
|
|
||
| fn read_internal( | ||
| &self, primary_namespace: &str, secondary_namespace: &str, key: &str, | ||
| ) -> io::Result<Vec<u8>> { | ||
| let persisted_lock = self.persisted_bytes.lock().unwrap(); | ||
|
|
@@ -888,7 +889,7 @@ impl KVStoreSync for TestStore { | |
| } | ||
| } | ||
|
|
||
| fn write_internal( | ||
| &self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: Vec<u8>, | ||
| ) -> io::Result<()> { | ||
| if self.read_only { | ||
|
|
@@ -911,7 +912,7 @@ impl KVStoreSync for TestStore { | |
| Ok(()) | ||
| } | ||
|
|
||
| fn remove_internal( | ||
| &self, primary_namespace: &str, secondary_namespace: &str, key: &str, _lazy: bool, | ||
| ) -> io::Result<()> { | ||
| if self.read_only { | ||
|
|
@@ -935,7 +936,9 @@ impl KVStoreSync for TestStore { | |
| Ok(()) | ||
| } | ||
|
|
||
| fn list_internal( | ||
| &self, primary_namespace: &str, secondary_namespace: &str, | ||
| ) -> io::Result<Vec<String>> { | ||
| let mut persisted_lock = self.persisted_bytes.lock().unwrap(); | ||
|
|
||
| let prefixed = if secondary_namespace.is_empty() { | ||
|
|
@@ -950,6 +953,89 @@ impl KVStoreSync for TestStore { | |
| } | ||
| } | ||
|
|
||
| impl KVStore for TestStore { | ||
| fn read( | ||
| &self, primary_namespace: &str, secondary_namespace: &str, key: &str, | ||
| ) -> Pin<Box<dyn Future<Output = Result<Vec<u8>, io::Error>> + 'static + Send>> { | ||
| let res = self.read_internal(&primary_namespace, &secondary_namespace, &key); | ||
|
There was a problem hiding this comment. Hmmmmm, would be nice to be able to race reads and writes rather than always immediately completing..... There was a problem hiding this comment. Right, I agree it would be nice to eventually extend this for improved test coverage. But see above: given it's not actually used anywhere it's a bit out-of-scope here. |
||
| Box::pin(async move { TestStoreFuture::new(res).await }) | ||
| } | ||
| fn write( | ||
| &self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: Vec<u8>, | ||
| ) -> Pin<Box<dyn Future<Output = Result<(), io::Error>> + 'static + Send>> { | ||
| let res = self.write_internal(&primary_namespace, &secondary_namespace, &key, buf); | ||
| Box::pin(async move { TestStoreFuture::new(res).await }) | ||
| } | ||
| fn remove( | ||
| &self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool, | ||
| ) -> Pin<Box<dyn Future<Output = Result<(), io::Error>> + 'static + Send>> { | ||
| let res = self.remove_internal(&primary_namespace, &secondary_namespace, &key, lazy); | ||
| Box::pin(async move { TestStoreFuture::new(res).await }) | ||
| } | ||
| fn list( | ||
| &self, primary_namespace: &str, secondary_namespace: &str, | ||
| ) -> Pin<Box<dyn Future<Output = Result<Vec<String>, io::Error>> + 'static + Send>> { | ||
| let res = self.list_internal(primary_namespace, secondary_namespace); | ||
| Box::pin(async move { TestStoreFuture::new(res).await }) | ||
| } | ||
| } | ||
|
|
||
| impl KVStoreSync for TestStore { | ||
| fn read( | ||
| &self, primary_namespace: &str, secondary_namespace: &str, key: &str, | ||
| ) -> io::Result<Vec<u8>> { | ||
| self.read_internal(primary_namespace, secondary_namespace, key) | ||
| } | ||
|
|
||
| fn write( | ||
| &self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: Vec<u8>, | ||
| ) -> io::Result<()> { | ||
| self.write_internal(primary_namespace, secondary_namespace, key, buf) | ||
| } | ||
|
|
||
| fn remove( | ||
| &self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool, | ||
| ) -> io::Result<()> { | ||
| self.remove_internal(primary_namespace, secondary_namespace, key, lazy) | ||
| } | ||
|
|
||
| fn list(&self, primary_namespace: &str, secondary_namespace: &str) -> io::Result<Vec<String>> { | ||
| self.list_internal(primary_namespace, secondary_namespace) | ||
| } | ||
| } | ||
|
|
||
| // A `Future` that returns the result only on the second poll. | ||
| pub(crate) struct TestStoreFuture<R> { | ||
| inner: Mutex<(Option<core::task::Waker>, Option<io::Result<R>>)>, | ||
| } | ||
|
|
||
| impl<R> TestStoreFuture<R> { | ||
| fn new(res: io::Result<R>) -> Self { | ||
| let inner = Mutex::new((None, Some(res))); | ||
| Self { inner } | ||
| } | ||
| } | ||
|
|
||
| impl<R> Future for TestStoreFuture<R> { | ||
| type Output = Result<R, io::Error>; | ||
| fn poll( | ||
| self: Pin<&mut Self>, cx: &mut core::task::Context<'_>, | ||
| ) -> core::task::Poll<Self::Output> { | ||
| let mut inner_lock = self.inner.lock().unwrap(); | ||
| let first_poll = inner_lock.0.is_none(); | ||
| if first_poll { | ||
| (*inner_lock).0 = Some(cx.waker().clone()); | ||
| core::task::Poll::Pending | ||
|
There was a problem hiding this comment. OH hmm this is more complicated than I thought. Technically this is a malformed future - after we switch to There was a problem hiding this comment.
Ah, good point. Now added a fixup that wakes after dropping the lock.
Yeah, tbh. this PR is mostly to make lightningdevkit/ldk-node#633 compile, and not even there we'd actually currently use the async |
||
| } else { | ||
| let waker = inner_lock.0.take().expect("We should never poll more than twice"); | ||
| let res = inner_lock.1.take().expect("We should never poll more than twice"); | ||
| drop(inner_lock); | ||
| waker.wake(); | ||
| core::task::Poll::Ready(res) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| unsafe impl Sync for TestStore {} | ||
| unsafe impl Send for TestStore {} | ||
|
|
||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Hmm, not quite sure I understand why we want to drop this? ldk-node might not use it, but I imagine some others might? Its the equivalent of our previous async BP loop and keeping it makes upgrades easier for those who might not want to switch to partial-async-kvstore immediately?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
IMO it's a very odd middleground API that was introduced when the
KVStoreSyncWrapperwasn't public. I fear users might find it confusing, and just usingKVStoreSyncWrapperwhen needed seems way more consistent (as they'd already need to do that for some of the other types they'd hand into that method anyways).There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
No comment. It's so little code that I think it's fine either way.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
IMO we should prefer users not use
KVStoreSyncWrapperdirectly. The docs even explicitly say "It is not necessary to use this type directly." (and I feel like we should#[doc(hidden)]it?).There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Hmm, but at least in LDK Node using
KVStoreSyncWrapperwas unavoidable. I can drop the drop commit if you insist, but IMO it's a pretty awkward confusing API.