@@ -36,7 +36,6 @@ use vss_client::util::retry::{
3636use vss_client:: util:: storable_builder:: { EntropySource , StorableBuilder } ;
3737
3838use crate :: io:: utils:: check_namespace_key_validity;
39- use crate :: runtime:: Runtime ;
4039
4140type CustomRetryPolicy = FilteredRetryPolicy <
4241 JitteredRetryPolicy <
@@ -67,7 +66,6 @@ pub struct VssStore {
6766 // Version counter to ensure that writes are applied in the correct order. It is assumed that read and list
6867 // operations aren't sensitive to the order of execution.
6968 next_version : AtomicU64 ,
70- runtime : Arc < Runtime > ,
7169 // A VSS-internal runtime we use to avoid any deadlocks we could hit when waiting on a spawned
7270 // blocking task to finish while the blocked thread had acquired the reactor. In particular,
7371 // this works around a previously-hit case where a concurrent call to
@@ -80,7 +78,7 @@ pub struct VssStore {
8078impl VssStore {
8179 pub ( crate ) fn new (
8280 base_url : String , store_id : String , vss_seed : [ u8 ; 32 ] ,
83- header_provider : Arc < dyn VssHeaderProvider > , runtime : Arc < Runtime > ,
81+ header_provider : Arc < dyn VssHeaderProvider > ,
8482 ) -> Self {
8583 let next_version = AtomicU64 :: new ( 1 ) ;
8684 let internal_runtime = Some (
@@ -124,7 +122,7 @@ impl VssStore {
124122 key_obfuscator,
125123 ) ) ;
126124
127- Self { inner, next_version, runtime , internal_runtime }
125+ Self { inner, next_version, internal_runtime }
128126 }
129127
130128 // Same logic as for the obfuscated keys below, but just for locking, using the plaintext keys
@@ -171,13 +169,14 @@ impl KVStoreSync for VssStore {
171169 async move { inner. read_internal ( primary_namespace, secondary_namespace, key) . await } ;
172170 // TODO: We could drop the timeout here once we ensured vss-client's Retry logic always
173171 // times out.
174- let spawned_fut = internal_runtime. spawn ( async move {
175- tokio:: time:: timeout ( VSS_IO_TIMEOUT , fut) . await . map_err ( |_| {
176- let msg = "VssStore::read timed out" ;
177- Error :: new ( ErrorKind :: Other , msg)
178- } )
179- } ) ;
180- self . runtime . block_on ( spawned_fut) . expect ( "We should always finish" ) ?
172+ tokio:: task:: block_in_place ( move || {
173+ internal_runtime. block_on ( async move {
174+ tokio:: time:: timeout ( VSS_IO_TIMEOUT , fut) . await . map_err ( |_| {
175+ let msg = "VssStore::read timed out" ;
176+ Error :: new ( ErrorKind :: Other , msg)
177+ } )
178+ } ) ?
179+ } )
181180 }
182181
183182 fn write (
@@ -209,13 +208,14 @@ impl KVStoreSync for VssStore {
209208 } ;
210209 // TODO: We could drop the timeout here once we ensured vss-client's Retry logic always
211210 // times out.
212- let spawned_fut = internal_runtime. spawn ( async move {
213- tokio:: time:: timeout ( VSS_IO_TIMEOUT , fut) . await . map_err ( |_| {
214- let msg = "VssStore::write timed out" ;
215- Error :: new ( ErrorKind :: Other , msg)
216- } )
217- } ) ;
218- self . runtime . block_on ( spawned_fut) . expect ( "We should always finish" ) ?
211+ tokio:: task:: block_in_place ( move || {
212+ internal_runtime. block_on ( async move {
213+ tokio:: time:: timeout ( VSS_IO_TIMEOUT , fut) . await . map_err ( |_| {
214+ let msg = "VssStore::write timed out" ;
215+ Error :: new ( ErrorKind :: Other , msg)
216+ } )
217+ } ) ?
218+ } )
219219 }
220220
221221 fn remove (
@@ -247,13 +247,14 @@ impl KVStoreSync for VssStore {
247247 } ;
248248 // TODO: We could drop the timeout here once we ensured vss-client's Retry logic always
249249 // times out.
250- let spawned_fut = internal_runtime. spawn ( async move {
251- tokio:: time:: timeout ( VSS_IO_TIMEOUT , fut) . await . map_err ( |_| {
252- let msg = "VssStore::remove timed out" ;
253- Error :: new ( ErrorKind :: Other , msg)
254- } )
255- } ) ;
256- self . runtime . block_on ( spawned_fut) . expect ( "We should always finish" ) ?
250+ tokio:: task:: block_in_place ( move || {
251+ internal_runtime. block_on ( async move {
252+ tokio:: time:: timeout ( VSS_IO_TIMEOUT , fut) . await . map_err ( |_| {
253+ let msg = "VssStore::remove timed out" ;
254+ Error :: new ( ErrorKind :: Other , msg)
255+ } )
256+ } ) ?
257+ } )
257258 }
258259
259260 fn list ( & self , primary_namespace : & str , secondary_namespace : & str ) -> io:: Result < Vec < String > > {
@@ -268,13 +269,14 @@ impl KVStoreSync for VssStore {
268269 let fut = async move { inner. list_internal ( primary_namespace, secondary_namespace) . await } ;
269270 // TODO: We could drop the timeout here once we ensured vss-client's Retry logic always
270271 // times out.
271- let spawned_fut = internal_runtime. spawn ( async move {
272- tokio:: time:: timeout ( VSS_IO_TIMEOUT , fut) . await . map_err ( |_| {
273- let msg = "VssStore::list timed out" ;
274- Error :: new ( ErrorKind :: Other , msg)
275- } )
276- } ) ;
277- self . runtime . block_on ( spawned_fut) . expect ( "We should always finish" ) ?
272+ tokio:: task:: block_in_place ( move || {
273+ internal_runtime. block_on ( async move {
274+ tokio:: time:: timeout ( VSS_IO_TIMEOUT , fut) . await . map_err ( |_| {
275+ let msg = "VssStore::list timed out" ;
276+ Error :: new ( ErrorKind :: Other , msg)
277+ } )
278+ } ) ?
279+ } )
278280 }
279281}
280282
@@ -694,7 +696,6 @@ mod tests {
694696
695697 use super :: * ;
696698 use crate :: io:: test_utils:: do_read_write_remove_list_persist;
697- use crate :: logger:: Logger ;
698699
699700 #[ test]
700701 fn vss_read_write_remove_list_persist ( ) {
@@ -704,11 +705,7 @@ mod tests {
704705 let mut vss_seed = [ 0u8 ; 32 ] ;
705706 rng. fill_bytes ( & mut vss_seed) ;
706707 let header_provider = Arc :: new ( FixedHeaders :: new ( HashMap :: new ( ) ) ) ;
707- let logger = Arc :: new ( Logger :: new_log_facade ( ) ) ;
708- let runtime = Arc :: new ( Runtime :: new ( logger) . unwrap ( ) ) ;
709- let vss_store =
710- VssStore :: new ( vss_base_url, rand_store_id, vss_seed, header_provider, runtime) ;
711-
708+ let vss_store = VssStore :: new ( vss_base_url, rand_store_id, vss_seed, header_provider) ;
712709 do_read_write_remove_list_persist ( & vss_store) ;
713710 }
714711
@@ -720,10 +717,7 @@ mod tests {
720717 let mut vss_seed = [ 0u8 ; 32 ] ;
721718 rng. fill_bytes ( & mut vss_seed) ;
722719 let header_provider = Arc :: new ( FixedHeaders :: new ( HashMap :: new ( ) ) ) ;
723- let logger = Arc :: new ( Logger :: new_log_facade ( ) ) ;
724- let runtime = Arc :: new ( Runtime :: new ( logger) . unwrap ( ) ) ;
725- let vss_store =
726- VssStore :: new ( vss_base_url, rand_store_id, vss_seed, header_provider, runtime) ;
720+ let vss_store = VssStore :: new ( vss_base_url, rand_store_id, vss_seed, header_provider) ;
727721
728722 do_read_write_remove_list_persist ( & vss_store) ;
729723 drop ( vss_store)
@@ -737,10 +731,7 @@ mod tests {
737731 let mut vss_seed = [ 0u8 ; 32 ] ;
738732 rng. fill_bytes ( & mut vss_seed) ;
739733 let header_provider = Arc :: new ( FixedHeaders :: new ( HashMap :: new ( ) ) ) ;
740- let logger = Arc :: new ( Logger :: new_log_facade ( ) ) ;
741- let runtime = Arc :: new ( Runtime :: new ( logger) . unwrap ( ) ) ;
742- let vss_store =
743- VssStore :: new ( vss_base_url, rand_store_id, vss_seed, header_provider, runtime) ;
734+ let vss_store = VssStore :: new ( vss_base_url, rand_store_id, vss_seed, header_provider) ;
744735
745736 let primary_namespace = "test_namespace" ;
746737 let secondary_namespace = "" ;
0 commit comments