update actix to version 0.13

update tokio to version 1
rewrite stest

account/provider/src/provider/mod.rs

@@ -21,7 +21,7 @@ impl ProviderFactory {
                 config
                     .account_dir
                     .as_ref()
-                    .ok_or(anyhow!("expect dir for local account"))?,
+                    .ok_or_else(|| anyhow!("expect dir for local account"))?,
                 chain_id,
             ) {
                 Ok(p) => Ok(Box::new(p)),

account/service/Cargo.toml

@@ -19,7 +19,7 @@ starcoin-account-api = { path = "../api", features = ["mock"] }
 starcoin-account = { path = "../" }
 starcoin-crypto = { git = "https://github.com/starcoinorg/starcoin-crypto", rev = "d871dfb4216f034ee334a575926c101574d9d6dc"}
 starcoin-chain-notify = { path = "../../chain/chain-notify" }
-
+tokio = "1"
 
 [dev-dependencies]
 stest = { path = "../../commons/stest" }

account/service/src/service.rs

@@ -199,7 +199,6 @@ mod tests {
     use starcoin_account_api::AccountAsyncService;
     use starcoin_config::NodeConfig;
     use starcoin_service_registry::{RegistryAsyncService, RegistryService};
-
     #[stest::test]
     async fn test_actor_launch() -> Result<()> {
         let config = Arc::new(NodeConfig::random_for_test());

block-relayer/Cargo.toml

@@ -28,6 +28,6 @@ starcoin-network-rpc-api = { path = "../network-rpc/api" }
 starcoin-service-registry = { path = "../commons/service-registry" }
 
 [dev-dependencies]
-tokio = { version = "0.2", features = ["full"] }
+tokio = {version = "^1", features = ["full"] }
 hex = "0.4.3"
 stest = { path = "../commons/stest"}

chain/Cargo.toml

@@ -36,7 +36,7 @@ rand_core = { version = "0.6.3", default-features = false }
 stdlib = { path = "../vm/stdlib" }
 stest = { path = "../commons/stest" }
 starcoin-account-api = { path = "../account/api" }
-tokio = { version = "0.2", features = ["full"] }
+tokio = {version = "^1", features = ["full"] }
 starcoin-genesis = { path = "../genesis" }
 proptest = "1.0.0"
 proptest-derive = { version = "0.3.0" }

chain/service/Cargo.toml

@@ -25,6 +25,7 @@ starcoin-chain-api = { path = "../api" }
 starcoin-logger = { path = "../../commons/logger" }
 starcoin-state-api = { path = "../../state/api" }
 starcoin-chain = { path = "../" }
+tokio = "1"
 
 [dev-dependencies]
 stest = { path = "../../commons/stest" }

cmd/airdrop/Cargo.toml

@@ -21,9 +21,9 @@ serde = "1.0.130"
 serde_json = "~1"
 csv = "~1"
 hex = "0.4"
-jsonrpc-core-client="~17"
+jsonrpc-core-client="18"
 rpassword = "~5"
-tokio={version="0.2", features=["full"]}
+tokio = {version = "^1", features=["full"]}
 futures-util = "~0.3"
-futures-retry="~0.5"
+futures-retry="0.6"
 

cmd/airdrop/src/main.rs

@@ -235,7 +235,7 @@ async fn main() -> Result<()> {
                 None => {
                     println!("wait txn to be mined, {}", txn_hash);
                     // sleep 10s.
-                    tokio::time::delay_for(Duration::from_secs(5)).await;
+                    tokio::time::sleep(Duration::from_secs(5)).await;
                     continue;
                 }
                 Some(txn_info) => {

cmd/faucet/Cargo.toml

@@ -11,7 +11,7 @@ anyhow = "1.0.41"
 tiny_http = "0.6"
 futures = "0.3.12"
 hex = { version = "0.4.3", default-features = false }
-tokio = { version = "0.2", features = ["full"] }
+tokio = {version = "^1", features = ["full"] }
 tokio-executor = { version = "0.2.0-alpha.6", features = ["blocking"] }
 futures-timer = "3.0"
 ascii = "0.8"

cmd/genesis-nft-miner/Cargo.toml

@@ -17,6 +17,6 @@ clap = { version = "3.0.0-rc.7", features=["derive"] }
 serde = "1.0.130"
 serde_json = "~1"
 hex = "0.4"
-jsonrpc-core-client="~17"
+jsonrpc-core-client="18"
 rpassword = "~5"
-tokio={version="0.2", features=["full"]}
+tokio = {version = "^1", features=["full"]}

cmd/genesis-nft-miner/src/main.rs

@@ -135,7 +135,7 @@ async fn main() -> Result<()> {
         match txn_info {
             None => {
                 println!("wait txn to be mined, {}", txn_hash);
-                tokio::time::delay_for(Duration::from_secs(5)).await;
+                tokio::time::sleep(Duration::from_secs(5)).await;
                 continue;
             }
             Some(txn_info) => {

cmd/indexer/Cargo.toml

@@ -19,10 +19,10 @@ starcoin-crypto = { git = "https://github.com/starcoinorg/starcoin-crypto", rev
 starcoin-types = {path= "../../types" }
 starcoin-rpc-api = {path = "../../rpc/api" }
 starcoin-logger = {path = "../../commons/logger"}
-jsonrpc-core-client="~17"
-tokio={version="0.2", features=["full"]}
+jsonrpc-core-client="18"
+tokio = {version = "^1", features=["full"]}
 futures-util = "~0.3"
-futures-retry="~0.5"
+futures-retry="0.6"
 
 [[bin]]
 name="starcoin_indexer"

cmd/indexer/src/es_sinker.rs

@@ -396,7 +396,7 @@ mod tests {
     use starcoin_crypto::HashValue;
     use std::str::FromStr;
 
-    #[tokio::test(threaded_scheduler)]
+    #[tokio::test(flavor = "multi_thread")]
     #[ignore]
     async fn test_update_tip_header() {
         let es_url = "http://localhost:9200";

cmd/indexer/src/main.rs

@@ -239,9 +239,8 @@ fn main() -> anyhow::Result<()> {
     let _log_handle = starcoin_logger::init();
     let opts: Options = Options::parse();
     info!("opts: {:?}", &opts);
-    let mut rt = runtime::Builder::new()
+    let rt = runtime::Builder::new_multi_thread()
         .thread_name("starcoin-indexer")
-        .threaded_scheduler()
         .enable_all()
         .build()?;
     let channel: ChainClient = rt

cmd/miner_client/Cargo.toml

@@ -6,20 +6,20 @@ edition = "2021"
 
 [dependencies]
 anyhow = "1.0.41"
-actix = "0.10.0"
-actix-rt = "1.1"
-async-std = "1.11"
+actix = "0.13"
+actix-rt = "2.6"
+async-std = "1.10"
 futures = "0.3.12"
 starcoin-config = { path = "../../config" }
 starcoin-types = { path = "../../types" }
 consensus = { path = "../../consensus", package = "starcoin-consensus" }
-tokio = { version = "0.3", features = ["full"] }
+tokio = {version = "^1", features = ["full"] }
 logger = { path = "../../commons/logger", package = "starcoin-logger" }
 futures-channel = "0.3"
 crypto = { package = "starcoin-crypto", git = "https://github.com/starcoinorg/starcoin-crypto", rev = "d871dfb4216f034ee334a575926c101574d9d6dc"}
-jsonrpc-core = { version = "17.0.0", features = ["arbitrary_precision"] }
-jsonrpc-server-utils = { version = "17.0.0" }
-jsonrpc-core-client = { version = "17.0.0" }
+jsonrpc-core = { version = "18", features = ["arbitrary_precision"] }
+jsonrpc-server-utils = { version = "18" }
+jsonrpc-core-client = { version = "18" }
 
 serde_json = { version = "1.0", features = ["arbitrary_precision"] }
 serde = { version = "1.0.130", features = ["derive"] }

cmd/miner_client/src/main.rs

@@ -39,10 +39,14 @@ fn main() {
         }
     };
     let user = opts.user;
-    let mut system = System::builder()
-        .stop_on_panic(true)
-        .name("starcoin-miner")
-        .build();
+    let system = System::with_tokio_rt(|| {
+        tokio::runtime::Builder::new_multi_thread()
+            .enable_all()
+            .on_thread_stop(|| println!("starcoin-miner thread stopped"))
+            .thread_name("starcoin-miner")
+            .build()
+            .expect("failed to create tokio runtime for starcoin-miner")
+    });
     if let Err(err) = system.block_on(async move {
         let registry = RegistryService::launch();
         let _ = registry.put_shared(config).await?;

cmd/starcoin/Cargo.toml

@@ -12,7 +12,7 @@ rand = "0.8.5"
 serde = { version = "1.0.130" }
 serde_json = { version="1.0", features = ["arbitrary_precision"]}
 crossbeam-channel = "0.5.4"
-tokio = { version = "0.2", features = ["full"] }
+tokio = {version = "^1", features = ["full"] }
 futures = "0.3.12"
 bcs-ext = { package="bcs-ext", path = "../../commons/bcs_ext" }
 structopt = "0.3.26"

cmd/starcoin/src/dev/subscribe_cmd.rs

@@ -1,7 +1,7 @@
 use crate::cli_state::CliState;
 use crate::StarcoinOpt;
 use anyhow::Result;
-use futures::{StreamExt, TryStream, TryStreamExt};
+use futures::{TryStream, TryStreamExt};
 use scmd::{CommandAction, ExecContext};
 use starcoin_rpc_api::types::pubsub::EventFilter;
 use starcoin_rpc_api::types::TypeTagView;
@@ -126,17 +126,16 @@ fn blocking_display_notification<T, F>(
 ) where
     F: Fn(&T) -> String,
 {
-    let mut rt = tokio::runtime::Builder::new()
-        .basic_scheduler()
+    let rt = tokio::runtime::Builder::new_multi_thread()
         .build()
         .expect("should able to create tokio runtime");
     let stdin = tokio::io::stdin();
     let mut lines = tokio::io::BufReader::new(stdin).lines();
     rt.block_on(async move {
         loop {
             tokio::select! {
-               maybe_quit = lines.next()  => {
-                   if let Some(Ok(q)) = maybe_quit {
+               maybe_quit = lines.next_line()  => {
+                   if let Ok(Some(q)) = maybe_quit {
                        if q.as_str() == "q" {
                            break;
                        }

cmd/tx-factory/Cargo.toml

@@ -15,7 +15,7 @@ anyhow = "1.0.41"
 structopt = "0.3.26"
 ctrlc = { version = "3.2.1", features = ["termination"] }
 futures = "0.3.12"
-tokio = { version = "0.2", features = ["full"] }
+tokio = {version = "^1", features = ["full"] }
 starcoin-logger = { path = "../../commons/logger" }
 starcoin-config = { path = "../../config"}
 starcoin-crypto = { git = "https://github.com/starcoinorg/starcoin-crypto", rev = "d871dfb4216f034ee334a575926c101574d9d6dc"}

commons/metrics/Cargo.toml

@@ -10,7 +10,7 @@ description = "Starcoin metrics"
 [dependencies]
 anyhow = "1.0.41"
 futures = "0.3.12"
-hyper = "0.13.9"
+hyper = { version = "0.14", features = ["full"]}
 starcoin-logger = { path = "../logger" }
 prometheus = { version = "0.13.0", default-features = false, features =["push"] }
 serde_json = { version="1.0", features = ["arbitrary_precision"]}

commons/metrics/src/metric_server.rs

@@ -4,8 +4,9 @@
 use crate::json_encoder::JsonEncoder;
 use futures::future;
 use hyper::{
+    server::Server,
     service::{make_service_fn, service_fn},
-    Body, Method, Request, Response, Server, StatusCode,
+    Body, Method, Request, Response, StatusCode,
 };
 use prometheus::{hostname_grouping_key, BasicAuthentication, Registry};
 use prometheus::{Encoder, TextEncoder};

commons/service-registry/Cargo.toml

@@ -11,8 +11,9 @@ edition = "2021"
 anyhow = "1.0"
 thiserror = "1.0"
 async-trait = "0.1"
-actix = "0.10.0"
-actix-rt = "1.1"
+tokio = "^1"
+actix = "0.13"
+actix-rt = "2.6"
 futures = "0.3.12"
 futures-timer = "3.0"
 serde = { version = "1.0.130", features = ["derive"] }

commons/service-registry/src/bus/tests.rs

@@ -15,7 +15,7 @@ async fn test_onshot() {
     let bus = registry.service_ref::<BusService>().await.unwrap();
     let bus2 = bus.clone();
     let arbiter = Arbiter::new();
-    arbiter.exec_fn(move || loop {
+    arbiter.spawn_fn(move || loop {
         let result = bus2.broadcast(MyMessage {}).is_ok();
         debug!("broadcast result: {}", result);
         sleep(Duration::from_millis(50));
@@ -32,7 +32,7 @@ async fn test_channel() {
     let bus = registry.service_ref::<BusService>().await.unwrap();
     let bus2 = bus.clone();
     let arbiter = Arbiter::new();
-    arbiter.exec_fn(move || loop {
+    arbiter.spawn_fn(move || loop {
         let result = bus2.broadcast(MyMessage {}).is_ok();
         debug!("broadcast result: {}", result);
         sleep(Duration::from_millis(50));

commons/service-registry/src/service.rs

@@ -6,8 +6,8 @@ use crate::service_actor::{EventMessage, ServiceActor};
 use crate::service_cache::ServiceCache;
 use crate::{RegistryAsyncService, RegistryService};
 use crate::{ServiceRef, ServiceRequest};
-use actix::fut::{wrap_future, IntoActorFuture};
-use actix::{ActorContext, ActorFuture, AsyncContext, Context};
+use actix::fut::wrap_future;
+use actix::{ActorContext, ActorFutureExt, AsyncContext, Context};
 use anyhow::{format_err, Result};
 use futures::channel::oneshot::{channel, Receiver};
 use futures::executor::block_on;
@@ -142,7 +142,7 @@ where
                     );
                 }
             });
-        self.ctx.wait(fut.into_future());
+        self.ctx.wait(fut);
     }
 
     pub fn add_stream<M, MS>(&mut self, stream: MS)
@@ -171,7 +171,7 @@ where
                     );
                 }
             });
-        self.ctx.wait(fut.into_future());
+        self.ctx.wait(fut);
     }
 
     pub fn broadcast<M>(&mut self, msg: M)

commons/service-registry/src/service_registry.rs

@@ -197,7 +197,7 @@ impl Registry {
 
         let arbiter = Arbiter::new();
         let registry_ref = self.service_ref.clone();
-        let addr = ServiceActor::start_in_arbiter(&arbiter, move |_ctx| f(registry_ref));
+        let addr = ServiceActor::start_in_arbiter(&arbiter.handle(), move |_ctx| f(registry_ref));
         let service_ref: ServiceRef<S> = addr.into();
         let holder = ServiceHolder::new(arbiter, service_ref.clone());
         self.services.push(Box::new(holder));
@@ -313,7 +313,7 @@ impl ServiceFactory<RegistryService> for RegistryService {
 impl RegistryService {
     pub fn launch() -> ServiceRef<Self> {
         let arbiter = Arbiter::new();
-        let addr = ServiceActor::start_in_arbiter(&arbiter, |ctx| {
+        let addr = ServiceActor::start_in_arbiter(&arbiter.handle(), |ctx| {
             let service_ref: ServiceRef<RegistryService> = ctx.address().into();
             ServiceActor::new::<RegistryService>(service_ref)
         });

commons/service-registry/tests/broadcast.rs

@@ -1,10 +1,11 @@
 // Copyright (c) The Starcoin Core Contributors
 // SPDX-License-Identifier: Apache-2.0
 
+use std::time::Duration;
+
 use crate::common::broadcast_process_service::{
     BMessage1, BMessage2, BroadcastProcessAsyncService, BroadcastProcessService,
 };
-use actix::clock::Duration;
 use futures_timer::Delay;
 use starcoin_service_registry::bus::{Bus, BusService};
 use starcoin_service_registry::{RegistryAsyncService, RegistryService};

commons/service-registry/tests/common/loop_task_service.rs

@@ -1,8 +1,9 @@
 // Copyright (c) The Starcoin Core Contributors
 // SPDX-License-Identifier: Apache-2.0
 
+use std::time::Duration;
+
 use crate::common::cal_service::{CalAsyncService, CalService};
-use actix::clock::Duration;
 use anyhow::Result;
 use futures_timer::Delay;
 use log::{error, info};

commons/service-registry/tests/loop_task.rs

@@ -1,7 +1,8 @@
 // Copyright (c) The Starcoin Core Contributors
 // SPDX-License-Identifier: Apache-2.0
 
-use actix::clock::Duration;
+use std::time::Duration;
+
 use common::cal_service::CalService;
 use common::loop_task_service::{GetTaskStatus, LoopTaskService, StartTaskEvent};
 use futures_timer::Delay;

commons/service-registry/tests/panic.rs

@@ -15,10 +15,14 @@ pub mod common;
 #[stest::test]
 fn test_service_panic() {
     //Fixme Expect the system stop on panic,but this feature not work as expect.
-    let mut sys = System::builder()
-        .stop_on_panic(true)
-        .name("panic_test")
-        .build();
+    let sys = System::with_tokio_rt(|| {
+        tokio::runtime::Builder::new_multi_thread()
+            .enable_all()
+            .on_thread_stop(|| println!("panic_test thread stopped"))
+            .thread_name("panic_test")
+            .build()
+            .expect("failed to create tokio runtime for panic_test")
+    });
     sys.block_on(async {
         let registry = RegistryService::launch();
         let service_ref = registry.register::<PanicService>().await.unwrap();

commons/stest/Cargo.toml

@@ -11,9 +11,9 @@ stest-macro = {path = "./stest-macro"}
 starcoin-logger = {path = "../logger"}
 timeout-join-handler = {path = "../timeout-join-handler"}
 log = "0.4.16"
-actix = "0.10.0"
-actix-rt = "1.1"
+actix = "0.13"
+actix-rt = "2.6"
 futures = "0.3.12"
-tokio = { version = "0.2", features = ["full"] }
+tokio = {version = "^1", features = ["full"] }
 anyhow = "1.0.41"