From e6babb07812b9d42598653c9607c712b5b625379 Mon Sep 17 00:00:00 2001 From: SeongChan Lee Date: Tue, 8 Jan 2019 13:46:57 +0900 Subject: [PATCH 1/6] Remove whitespaces in timer thread name --- util/timer/src/timer.rs | 7 ++----- 1 file changed, 2 insertions(+), 5 deletions(-) diff --git a/util/timer/src/timer.rs b/util/timer/src/timer.rs index 9f6ce02f91..fc7528505b 100644 --- a/util/timer/src/timer.rs +++ b/util/timer/src/timer.rs @@ -51,7 +51,7 @@ impl TimerLoop { { let scheduler = Arc::clone(&scheduler); thread::Builder::new() - .name("Timer scheduler".to_string()) + .name("timer.scheduler".to_string()) .spawn(move || scheduler.run(&worker_queue)) .unwrap(); } @@ -420,10 +420,7 @@ fn spawn_workers(size: usize, timers: &Arc>, queue: &A for i in 0..size { let queue = Arc::clone(queue); let timers = Arc::clone(timers); - thread::Builder::new() - .name(format!("Timer worker #{}", i)) - .spawn(move || worker_loop(&timers, &queue)) - .unwrap(); + thread::Builder::new().name(format!("timer.worker.{}", i)).spawn(move || worker_loop(&timers, &queue)).unwrap(); } } From 8cdee0ce8b45a7766e8007729faed179eb2d87fa Mon Sep 17 00:00:00 2001 From: SeongChan Lee Date: Wed, 9 Jan 2019 03:27:13 +0900 Subject: [PATCH 2/6] Use autoincremented unique TimerId instead of TimerName --- util/timer/src/timer.rs | 67 ++++++++++++++++++++++++++--------------- 1 file changed, 42 insertions(+), 25 deletions(-) diff --git a/util/timer/src/timer.rs b/util/timer/src/timer.rs index fc7528505b..c502c40daf 100644 --- a/util/timer/src/timer.rs +++ b/util/timer/src/timer.rs @@ -19,7 +19,7 @@ use std::collections::binary_heap::BinaryHeap; use std::collections::hash_map::{Entry, HashMap}; use std::collections::VecDeque; use std::string::ToString; -use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::sync::{Arc, Weak}; use std::thread; use std::time::{Duration, Instant}; @@ -33,11 +33,12 @@ pub trait TimeoutHandler: Send + Sync { fn on_timeout(&self, _token: TimerToken) {} } -type TimeoutHandlerMap = HashMap>; +type TimeoutHandlerMap = HashMap>; #[derive(Clone)] pub struct TimerLoop { timers: Arc>, + timer_id_nonce: Arc, scheduler: Arc, } @@ -58,6 +59,7 @@ impl TimerLoop { TimerLoop { timers, + timer_id_nonce: Arc::new(AtomicUsize::new(0)), scheduler, } } @@ -66,11 +68,13 @@ impl TimerLoop { where T: 'static + TimeoutHandler, { let mut timers = self.timers.write(); - match timers.entry(name) { + let timer_id = self.timer_id_nonce.fetch_add(1, Ordering::SeqCst); + match timers.entry(timer_id) { Entry::Occupied(_) => unreachable!("Timer name was already occupied: {}", name), Entry::Vacant(entry) => { entry.insert(handler); TimerApi { + timer_id, timer_name: name, scheduler: Arc::downgrade(&self.scheduler), } @@ -79,7 +83,10 @@ impl TimerLoop { } } +type TimerId = usize; + pub struct TimerApi { + timer_id: TimerId, timer_name: TimerName, scheduler: Weak, } @@ -93,17 +100,17 @@ pub enum ScheduleError { impl TimerApi { pub fn schedule_once(&self, after: Duration, timer_token: TimerToken) -> Result<(), ScheduleError> { let scheduler = self.scheduler.upgrade().ok_or(ScheduleError::TimerLoopDropped)?; - scheduler.schedule(TimerId(self.timer_name, timer_token), after, None) + scheduler.schedule(ScheduleId(self.timer_id, timer_token), after, None) } pub fn schedule_repeat(&self, after: Duration, timer_token: TimerToken) -> Result<(), ScheduleError> { let scheduler = self.scheduler.upgrade().ok_or(ScheduleError::TimerLoopDropped)?; - scheduler.schedule(TimerId(self.timer_name, timer_token), after, Some(after)) + scheduler.schedule(ScheduleId(self.timer_id, timer_token), after, Some(after)) } pub fn cancel(&self, timer_token: TimerToken) -> Result { let scheduler = self.scheduler.upgrade().ok_or(ScheduleError::TimerLoopDropped)?; - let result = scheduler.cancel(TimerId(self.timer_name, timer_token)); + let result = scheduler.cancel(ScheduleId(self.timer_id, timer_token)); Ok(result) } } @@ -121,16 +128,21 @@ impl Scheduler { } } - fn schedule(&self, timer_id: TimerId, after: Duration, repeat: Option) -> Result<(), ScheduleError> { + fn schedule( + &self, + schedule_id: ScheduleId, + after: Duration, + repeat: Option, + ) -> Result<(), ScheduleError> { let mut scheduler = self.inner.lock(); - scheduler.schedule(timer_id, after, repeat)?; + scheduler.schedule(schedule_id, after, repeat)?; self.condvar.notify_one(); Ok(()) } - fn cancel(&self, timer_id: TimerId) -> bool { + fn cancel(&self, schedule_id: ScheduleId) -> bool { let mut scheduler = self.inner.lock(); - let result = scheduler.cancel(timer_id); + let result = scheduler.cancel(schedule_id); self.condvar.notify_all(); result } @@ -157,7 +169,7 @@ impl Scheduler { /// Rule 1. All detached 'state_control' is in 'Cancelled' state. /// A detached one has no way to 'cancel' it. so it should be in 'Cancelled' state. /// Rule 2. All 'state_control' in 'self.states' are unique. -/// Otherwise, it leads to two different TimerId shares same 'state_control'. +/// Otherwise, it leads to two different ScheduleId shares same 'state_control'. /// Rule 3. All 'state_control' in 'Wait | WaitRepeating' state that contained in 'self.heap' are unique. /// Otherwise, cancelling a 'state_control' makes two different schedules being cancelled. /// @@ -165,8 +177,8 @@ impl Scheduler { /// Attached', 'Detached', 'Garbage' /// Lemma 2. All 'state_control' in 'Wait | WaitRepeating' state in 'ScheduleInner' is either an attached one or a garbage. /// A detached ones are all in 'Cancelled' state (Rule 1) so it is in either 'Attached' or 'Garbage' state. -/// Lemma 3. We can find all 'state_control' for a 'schedule' that are in 'Wait | WaitRepeating' using a TimerId. -/// We can find a 'state_control' for a TimerId (Rule 2), and all 'state_control' that +/// Lemma 3. We can find all 'state_control' for a 'schedule' that are in 'Wait | WaitRepeating' using a ScheduleId. +/// We can find a 'state_control' for a ScheduleId (Rule 2), and all 'state_control' that /// 'Wait | WaitRepeating' is either an attached one or a garbage (Lemma 2), /// but a garbage is not in 'self.heap' (Def 2). /// @@ -180,7 +192,7 @@ impl Scheduler { /// Note 2. Timeout, Cancelled states never revive. (to ease the complexity) struct SchedulerInner { - states: HashMap>, + states: HashMap>, heap: BinaryHeap>>, stop: bool, } @@ -193,8 +205,13 @@ impl SchedulerInner { } } - fn schedule(&mut self, timer_id: TimerId, after: Duration, repeat: Option) -> Result<(), ScheduleError> { - let state_control = match self.states.entry(timer_id) { + fn schedule( + &mut self, + schedule_id: ScheduleId, + after: Duration, + repeat: Option, + ) -> Result<(), ScheduleError> { + let state_control = match self.states.entry(schedule_id) { Entry::Vacant(entry) => { // unique one(Rule 2). it is going to be attached. let state_control = Arc::new(ScheduleStateControl::new_auto(repeat)); @@ -216,7 +233,7 @@ impl SchedulerInner { let schedule = Schedule { at: Instant::now() + after, - timer_id, + schedule_id, repeat, state_control, }; @@ -227,7 +244,7 @@ impl SchedulerInner { fn try_reschedule(&mut self, mut schedule: Schedule) { schedule.at = Instant::now() + schedule.repeat.expect("Schedule should have repeat interval"); - match self.states.entry(schedule.timer_id) { + match self.states.entry(schedule.schedule_id) { Entry::Vacant(_) => { // 'schedule.state_control' was detached one (Def 1). // Don't reschedule since it is Cancelled (Rule 1) @@ -251,9 +268,9 @@ impl SchedulerInner { }; } - fn cancel(&mut self, timer_id: TimerId) -> bool { + fn cancel(&mut self, schedule_id: ScheduleId) -> bool { // See Corollary 2. - match self.states.entry(timer_id) { + match self.states.entry(schedule_id) { Entry::Vacant(_) => false, Entry::Occupied(entry) => { // Detach and cancel it (Rule 1) @@ -278,7 +295,7 @@ impl SchedulerInner { self.try_reschedule(timed_out.clone()); worker_queue.enqueue(timed_out); } else { - let enqueue = match self.states.entry(timed_out.timer_id) { + let enqueue = match self.states.entry(timed_out.schedule_id) { Entry::Occupied(entry) => { if Arc::ptr_eq(entry.get(), &timed_out.state_control) { // 'timed_out.state_control' was attached one. (Def 1) @@ -304,7 +321,7 @@ impl SchedulerInner { } #[derive(Copy, Clone, Eq, PartialEq, Hash)] -struct TimerId(TimerName, TimerToken); +struct ScheduleId(TimerId, TimerToken); /* valid state transition: @@ -383,7 +400,7 @@ impl ScheduleStateControl { #[derive(Clone)] struct Schedule { at: Instant, - timer_id: TimerId, + schedule_id: ScheduleId, repeat: Option, state_control: Arc, } @@ -427,8 +444,8 @@ fn spawn_workers(size: usize, timers: &Arc>, queue: &A fn worker_loop(timers: &Arc>, queue: &Arc) { while let Some(schedule) = queue.dequeue() { let timers = timers.read(); - let TimerId(timer_name, timer_token) = schedule.timer_id; - if let Some(timer) = timers.get(timer_name) { + let ScheduleId(timer_id, timer_token) = schedule.schedule_id; + if let Some(timer) = timers.get(&timer_id) { schedule.state_control.within_lock(|state| { debug_assert_ne!(*state, ScheduleState::Timeout); if *state != ScheduleState::Cancelled { From dbeba7c79b9cbd0eb512a45ba22671d3bc2b48b6 Mon Sep 17 00:00:00 2001 From: SeongChan Lee Date: Wed, 9 Jan 2019 03:49:37 +0900 Subject: [PATCH 3/6] Remove TimerHandlerMap --- util/timer/src/timer.rs | 82 +++++++++++++++++++---------------------- 1 file changed, 38 insertions(+), 44 deletions(-) diff --git a/util/timer/src/timer.rs b/util/timer/src/timer.rs index c502c40daf..cc4f18848f 100644 --- a/util/timer/src/timer.rs +++ b/util/timer/src/timer.rs @@ -33,23 +33,21 @@ pub trait TimeoutHandler: Send + Sync { fn on_timeout(&self, _token: TimerToken) {} } -type TimeoutHandlerMap = HashMap>; - #[derive(Clone)] pub struct TimerLoop { - timers: Arc>, timer_id_nonce: Arc, scheduler: Arc, } impl TimerLoop { pub fn new(worker_size: usize) -> TimerLoop { - let timers = Arc::new(RwLock::new(HashMap::new())); let scheduler = Arc::new(Scheduler::new()); let worker_queue = Arc::new(WorkerQueue::new()); - spawn_workers(worker_size, &timers, &worker_queue); + + spawn_workers(worker_size, &worker_queue); { + let worker_queue = Arc::clone(&worker_queue); let scheduler = Arc::clone(&scheduler); thread::Builder::new() .name("timer.scheduler".to_string()) @@ -58,7 +56,6 @@ impl TimerLoop { } TimerLoop { - timers, timer_id_nonce: Arc::new(AtomicUsize::new(0)), scheduler, } @@ -67,18 +64,12 @@ impl TimerLoop { pub fn new_timer(&self, name: TimerName, handler: Arc) -> TimerApi where T: 'static + TimeoutHandler, { - let mut timers = self.timers.write(); let timer_id = self.timer_id_nonce.fetch_add(1, Ordering::SeqCst); - match timers.entry(timer_id) { - Entry::Occupied(_) => unreachable!("Timer name was already occupied: {}", name), - Entry::Vacant(entry) => { - entry.insert(handler); - TimerApi { - timer_id, - timer_name: name, - scheduler: Arc::downgrade(&self.scheduler), - } - } + TimerApi { + timer_id, + handler, + timer_name: name, + scheduler: Arc::downgrade(&self.scheduler), } } } @@ -88,6 +79,7 @@ type TimerId = usize; pub struct TimerApi { timer_id: TimerId, timer_name: TimerName, + handler: Arc, scheduler: Weak, } @@ -100,17 +92,17 @@ pub enum ScheduleError { impl TimerApi { pub fn schedule_once(&self, after: Duration, timer_token: TimerToken) -> Result<(), ScheduleError> { let scheduler = self.scheduler.upgrade().ok_or(ScheduleError::TimerLoopDropped)?; - scheduler.schedule(ScheduleId(self.timer_id, timer_token), after, None) + scheduler.schedule(self, timer_token, after, None) } pub fn schedule_repeat(&self, after: Duration, timer_token: TimerToken) -> Result<(), ScheduleError> { let scheduler = self.scheduler.upgrade().ok_or(ScheduleError::TimerLoopDropped)?; - scheduler.schedule(ScheduleId(self.timer_id, timer_token), after, Some(after)) + scheduler.schedule(self, timer_token, after, Some(after)) } pub fn cancel(&self, timer_token: TimerToken) -> Result { let scheduler = self.scheduler.upgrade().ok_or(ScheduleError::TimerLoopDropped)?; - let result = scheduler.cancel(ScheduleId(self.timer_id, timer_token)); + let result = scheduler.cancel(self, timer_token); Ok(result) } } @@ -130,19 +122,20 @@ impl Scheduler { fn schedule( &self, - schedule_id: ScheduleId, + requested_timer: &TimerApi, + timer_token: TimerToken, after: Duration, repeat: Option, ) -> Result<(), ScheduleError> { let mut scheduler = self.inner.lock(); - scheduler.schedule(schedule_id, after, repeat)?; + scheduler.schedule(requested_timer, timer_token, after, repeat)?; self.condvar.notify_one(); Ok(()) } - fn cancel(&self, schedule_id: ScheduleId) -> bool { + fn cancel(&self, requested_timer: &TimerApi, timer_token: TimerToken) -> bool { let mut scheduler = self.inner.lock(); - let result = scheduler.cancel(schedule_id); + let result = scheduler.cancel(requested_timer, timer_token); self.condvar.notify_all(); result } @@ -207,10 +200,12 @@ impl SchedulerInner { fn schedule( &mut self, - schedule_id: ScheduleId, + requested_timer: &TimerApi, + timer_token: TimerToken, after: Duration, repeat: Option, ) -> Result<(), ScheduleError> { + let schedule_id = ScheduleId(requested_timer.timer_id, timer_token); let state_control = match self.states.entry(schedule_id) { Entry::Vacant(entry) => { // unique one(Rule 2). it is going to be attached. @@ -236,6 +231,7 @@ impl SchedulerInner { schedule_id, repeat, state_control, + handler: Arc::clone(&requested_timer.handler), }; // state_control become an attached one (Def 1) self.heap.push(Reverse(TimeOrdered(schedule))); @@ -268,7 +264,8 @@ impl SchedulerInner { }; } - fn cancel(&mut self, schedule_id: ScheduleId) -> bool { + fn cancel(&mut self, requested_timer: &TimerApi, timer_token: TimerToken) -> bool { + let schedule_id = ScheduleId(requested_timer.timer_id, timer_token); // See Corollary 2. match self.states.entry(schedule_id) { Entry::Vacant(_) => false, @@ -403,6 +400,7 @@ struct Schedule { schedule_id: ScheduleId, repeat: Option, state_control: Arc, + handler: Arc, } struct TimeOrdered(T); @@ -433,31 +431,27 @@ impl std::cmp::PartialOrd for TimeOrdered { } } -fn spawn_workers(size: usize, timers: &Arc>, queue: &Arc) { +fn spawn_workers(size: usize, queue: &Arc) { for i in 0..size { let queue = Arc::clone(queue); - let timers = Arc::clone(timers); - thread::Builder::new().name(format!("timer.worker.{}", i)).spawn(move || worker_loop(&timers, &queue)).unwrap(); + thread::Builder::new().name(format!("timer.worker.{}", i)).spawn(move || worker_loop(&queue)).unwrap(); } } -fn worker_loop(timers: &Arc>, queue: &Arc) { +fn worker_loop(queue: &Arc) { while let Some(schedule) = queue.dequeue() { - let timers = timers.read(); - let ScheduleId(timer_id, timer_token) = schedule.schedule_id; - if let Some(timer) = timers.get(&timer_id) { - schedule.state_control.within_lock(|state| { - debug_assert_ne!(*state, ScheduleState::Timeout); - if *state != ScheduleState::Cancelled { - timer.on_timeout(timer_token); - let is_oneshot = schedule.repeat.is_none(); - debug_assert_eq!(is_oneshot, *state == ScheduleState::Wait); - if is_oneshot { - ScheduleStateControl::set_timeout(state); - } + let ScheduleId(_, timer_token) = schedule.schedule_id; + schedule.state_control.within_lock(|state| { + debug_assert_ne!(*state, ScheduleState::Timeout); + if *state != ScheduleState::Cancelled { + schedule.handler.on_timeout(timer_token); + let is_oneshot = schedule.repeat.is_none(); + debug_assert_eq!(is_oneshot, *state == ScheduleState::Wait); + if is_oneshot { + ScheduleStateControl::set_timeout(state); } - }); - } + } + }); } } From 112b9a7899e381a5cfa3d89296b6d8db07aaa371 Mon Sep 17 00:00:00 2001 From: SeongChan Lee Date: Wed, 9 Jan 2019 16:45:28 +0900 Subject: [PATCH 4/6] Change TimerApi to be created without TimeoutHandler --- codechain/run_node.rs | 7 +- core/src/client/client.rs | 55 ++++++------- core/src/client/mod.rs | 3 - core/src/client/test_client.rs | 4 +- core/src/service.rs | 4 +- network/src/client.rs | 5 +- util/timer/src/timer.rs | 141 ++++++++++++++++++++++++++------- 7 files changed, 145 insertions(+), 74 deletions(-) diff --git a/codechain/run_node.rs b/codechain/run_node.rs index af4ecf938b..cf867c5c28 100644 --- a/codechain/run_node.rs +++ b/codechain/run_node.rs @@ -96,11 +96,10 @@ fn client_start( let db_path = cfg.db_path.as_ref().map(|s| s.as_str()).unwrap(); let client_path = Path::new(db_path); let client_config = Default::default(); - let service = ClientService::start(&client_config, &scheme, &client_path, miner) + let reseal_timer = timer_loop.new_timer("Client reseal timer"); + let service = ClientService::start(&client_config, &scheme, &client_path, miner, reseal_timer.clone()) .map_err(|e| format!("Client service error: {}", e))?; - - let reseal_timer = timer_loop.new_timer("Client reseal timer", service.client()); - service.client().register_reseal_timer(reseal_timer); + reseal_timer.set_handler(&service.client()); Ok(service) } diff --git a/core/src/client/client.rs b/core/src/client/client.rs index dc57b10c45..fc9111b72e 100644 --- a/core/src/client/client.rs +++ b/core/src/client/client.rs @@ -84,7 +84,7 @@ pub struct Client { importer: Importer, /// Timer for reseal_min_period/reseal_max_period on miner client - reseal_timer: RwLock>, + reseal_timer: TimerApi, } impl Client { @@ -94,6 +94,7 @@ impl Client { db: Arc, miner: Arc, message_channel: IoChannel, + reseal_timer: TimerApi, ) -> Result, Error> { let journal_db = journaldb::new(Arc::clone(&db), journaldb::Algorithm::Archive, ::db::COL_STATE); let mut state_db = StateDB::new(journal_db); @@ -127,7 +128,7 @@ impl Client { queue_transactions: AtomicUsize::new(0), genesis_accounts, importer, - reseal_timer: RwLock::new(None), + reseal_timer, }); // ensure buffered changes are flushed. @@ -135,10 +136,6 @@ impl Client { Ok(client) } - pub fn register_reseal_timer(&self, timer: TimerApi) { - self.register_timer(timer); - } - /// Returns engine reference. pub fn engine(&self) -> &CodeChainEngine { &*self.engine @@ -279,38 +276,32 @@ impl TimeoutHandler for Client { } impl ResealTimer for Client { - fn register_timer(&self, timer: TimerApi) { - *self.reseal_timer.write() = Some(timer); - } - fn set_max_timer(&self) { - if let Some(reseal_timer) = self.reseal_timer.read().as_ref() { - reseal_timer.cancel(RESEAL_MAX_TIMER_TOKEN).expect("Reseal max timer clear succeeds"); - match reseal_timer - .schedule_once(self.importer.miner.get_options().reseal_max_period, RESEAL_MAX_TIMER_TOKEN) - { - Ok(_) => {} - Err(TimerScheduleError::TokenAlreadyScheduled) => { - // Since set_max_timer could be called in multi thread, ignore the TokenAlreadyScheduled error - } - Err(err) => unreachable!("Reseal max timer should not fail but failed with {:?}", err), + self.reseal_timer.cancel(RESEAL_MAX_TIMER_TOKEN).expect("Reseal max timer clear succeeds"); + match self + .reseal_timer + .schedule_once(self.importer.miner.get_options().reseal_max_period, RESEAL_MAX_TIMER_TOKEN) + { + Ok(_) => {} + Err(TimerScheduleError::TokenAlreadyScheduled) => { + // Since set_max_timer could be called in multi thread, ignore the TokenAlreadyScheduled error } - }; + Err(err) => unreachable!("Reseal max timer should not fail but failed with {:?}", err), + } } fn set_min_timer(&self) { - if let Some(reseal_timer) = self.reseal_timer.read().as_ref() { - reseal_timer.cancel(RESEAL_MIN_TIMER_TOKEN).expect("Reseal min timer clear succeeds"); - match reseal_timer - .schedule_once(self.importer.miner.get_options().reseal_min_period, RESEAL_MIN_TIMER_TOKEN) - { - Ok(_) => {} - Err(TimerScheduleError::TokenAlreadyScheduled) => { - // Since set_min_timer could be called in multi thread, ignore the TokenAlreadyScheduled error - } - Err(err) => unreachable!("Reseal min timer should not fail but failed with {:?}", err), + self.reseal_timer.cancel(RESEAL_MIN_TIMER_TOKEN).expect("Reseal min timer clear succeeds"); + match self + .reseal_timer + .schedule_once(self.importer.miner.get_options().reseal_min_period, RESEAL_MIN_TIMER_TOKEN) + { + Ok(_) => {} + Err(TimerScheduleError::TokenAlreadyScheduled) => { + // Since set_min_timer could be called in multi thread, ignore the TokenAlreadyScheduled error } - }; + Err(err) => unreachable!("Reseal min timer should not fail but failed with {:?}", err), + } } } diff --git a/core/src/client/mod.rs b/core/src/client/mod.rs index 2232718a7b..9519f7fe71 100644 --- a/core/src/client/mod.rs +++ b/core/src/client/mod.rs @@ -35,7 +35,6 @@ use ckey::{Address, PlatformAddress, Public}; use cmerkle::Result as TrieResult; use cnetwork::NodeId; use cstate::{AssetScheme, AssetSchemeAddress, FindActionHandler, OwnedAsset, Text, TopLevelState, TopStateView}; -use ctimer::TimerApi; use ctypes::invoice::Invoice; use ctypes::transaction::{AssetTransferInput, PartialHashing, ShardTransaction}; use ctypes::{BlockNumber, ShardId}; @@ -188,8 +187,6 @@ pub trait Shard { /// Provides a timer API for reseal_min_period/reseal_max_period on miner client pub trait ResealTimer { - /// Register timer API - fn register_timer(&self, timer: TimerApi); /// Set reseal min timer as reseal_min_period, for creating blocks with transactions which are pending because of reseal_min_period fn set_min_timer(&self); /// Set reseal max timer as reseal_max_period, for creating empty blocks every reseal_max_period diff --git a/core/src/client/test_client.rs b/core/src/client/test_client.rs index ef408b8075..1beec547f8 100644 --- a/core/src/client/test_client.rs +++ b/core/src/client/test_client.rs @@ -39,7 +39,7 @@ use ckey::{public_to_address, Address, Generator, NetworkId, PlatformAddress, Ra use cmerkle::skewed_merkle_root; use cnetwork::NodeId; use cstate::{FindActionHandler, StateDB}; -use ctimer::{TimeoutHandler, TimerApi, TimerToken}; +use ctimer::{TimeoutHandler, TimerToken}; use ctypes::invoice::Invoice; use ctypes::transaction::{Action, Transaction}; use ctypes::BlockNumber; @@ -532,8 +532,6 @@ impl TimeoutHandler for TestBlockChainClient { } impl ResealTimer for TestBlockChainClient { - fn register_timer(&self, _timer: TimerApi) {} - fn set_max_timer(&self) {} fn set_min_timer(&self) {} diff --git a/core/src/service.rs b/core/src/service.rs index f955749258..f98bdd1af3 100644 --- a/core/src/service.rs +++ b/core/src/service.rs @@ -19,6 +19,7 @@ use std::sync::Arc; use cio::{IoContext, IoHandler, IoHandlerResult, IoService}; use cnetwork::NodeId; +use ctimer::TimerApi; use kvdb_rocksdb::{Database, DatabaseConfig}; use primitives::Bytes; @@ -40,6 +41,7 @@ impl ClientService { scheme: &Scheme, client_path: &Path, miner: Arc, + reseal_timer: TimerApi, ) -> Result { let io_service = IoService::::start("Client")?; @@ -54,7 +56,7 @@ impl ClientService { .map_err(::client::Error::Database)?, ); - let client = Client::try_new(config, &scheme, db, miner, io_service.channel())?; + let client = Client::try_new(config, &scheme, db, miner, io_service.channel(), reseal_timer)?; let client_io = Arc::new(ClientIoHandler { client: client.clone(), diff --git a/network/src/client.rs b/network/src/client.rs index a063044a9d..12b422368d 100644 --- a/network/src/client.rs +++ b/network/src/client.rs @@ -124,12 +124,13 @@ impl Client { let name = extension.name(); let mut extensions = self.extensions.write(); let p2p_channel = self.p2p_channel.clone(); - let timer = self.timer_loop.new_timer(name, Arc::clone(&extension)); + let timer = self.timer_loop.new_timer(name); let api: Arc = Arc::new(ClientApi { extension: Arc::downgrade(&extension) as Weak, p2p_channel, - timer, + timer: timer.clone(), }); + timer.set_handler(&extension); extension.on_initialize(api); if extensions.insert(name, extension).is_some() { unreachable!("Duplicated extension name : {}", name) diff --git a/util/timer/src/timer.rs b/util/timer/src/timer.rs index cc4f18848f..e79de269c4 100644 --- a/util/timer/src/timer.rs +++ b/util/timer/src/timer.rs @@ -61,13 +61,11 @@ impl TimerLoop { } } - pub fn new_timer(&self, name: TimerName, handler: Arc) -> TimerApi - where - T: 'static + TimeoutHandler, { + pub fn new_timer(&self, name: TimerName) -> TimerApi { let timer_id = self.timer_id_nonce.fetch_add(1, Ordering::SeqCst); TimerApi { timer_id, - handler, + handler: Arc::new(RwLock::new(None)), timer_name: name, scheduler: Arc::downgrade(&self.scheduler), } @@ -76,10 +74,11 @@ impl TimerLoop { type TimerId = usize; +#[derive(Clone)] pub struct TimerApi { timer_id: TimerId, timer_name: TimerName, - handler: Arc, + handler: Arc>>>, scheduler: Weak, } @@ -90,6 +89,14 @@ pub enum ScheduleError { } impl TimerApi { + pub fn set_handler(&self, handler: &Arc) + where + T: 'static + TimeoutHandler, { + let mut handler_guard = self.handler.write(); + assert!(handler_guard.is_none(), "Timer handler cannot be changed once it is set"); + *handler_guard = Some(Arc::downgrade(handler) as Weak); + } + pub fn schedule_once(&self, after: Duration, timer_token: TimerToken) -> Result<(), ScheduleError> { let scheduler = self.scheduler.upgrade().ok_or(ScheduleError::TimerLoopDropped)?; scheduler.schedule(self, timer_token, after, None) @@ -206,6 +213,12 @@ impl SchedulerInner { repeat: Option, ) -> Result<(), ScheduleError> { let schedule_id = ScheduleId(requested_timer.timer_id, timer_token); + let handler = { + let guard = requested_timer.handler.read(); + let handler = guard.as_ref().expect("TimeoutHandler must be set"); + Weak::clone(handler) + }; + let state_control = match self.states.entry(schedule_id) { Entry::Vacant(entry) => { // unique one(Rule 2). it is going to be attached. @@ -231,7 +244,7 @@ impl SchedulerInner { schedule_id, repeat, state_control, - handler: Arc::clone(&requested_timer.handler), + handler, }; // state_control become an attached one (Def 1) self.heap.push(Reverse(TimeOrdered(schedule))); @@ -288,9 +301,11 @@ impl SchedulerInner { let Reverse(TimeOrdered(timed_out)) = self.heap.pop().expect("It always have an item"); if timed_out.repeat.is_some() { - // timed_out.state_control is re-pushed only after it is popped (Rule 3) - self.try_reschedule(timed_out.clone()); - worker_queue.enqueue(timed_out); + if let Some(callback) = Callback::from_schedule(&timed_out) { + // timed_out.state_control is re-pushed only after it is popped (Rule 3) + self.try_reschedule(timed_out.clone()); + worker_queue.enqueue(callback); + } } else { let enqueue = match self.states.entry(timed_out.schedule_id) { Entry::Occupied(entry) => { @@ -306,7 +321,9 @@ impl SchedulerInner { }; // timed_out is anyway removed. if enqueue { - worker_queue.enqueue(timed_out); + if let Some(callback) = Callback::from_schedule(&timed_out) { + worker_queue.enqueue(callback); + } } else { // 'timed_out.state_control' was detached one. (Def 1) // It already be cancelled (Rule 1) @@ -400,7 +417,7 @@ struct Schedule { schedule_id: ScheduleId, repeat: Option, state_control: Arc, - handler: Arc, + handler: Weak, } struct TimeOrdered(T); @@ -431,6 +448,28 @@ impl std::cmp::PartialOrd for TimeOrdered { } } +struct Callback { + schedule_id: ScheduleId, + from_oneshot_schedule: bool, + state_control: Arc, + handler: Arc, +} + +impl Callback { + fn from_schedule(schedule: &Schedule) -> Option { + if let Some(handler) = schedule.handler.upgrade() { + Some(Callback { + schedule_id: schedule.schedule_id, + from_oneshot_schedule: schedule.repeat.is_none(), + state_control: Arc::clone(&schedule.state_control), + handler, + }) + } else { + None + } + } +} + fn spawn_workers(size: usize, queue: &Arc) { for i in 0..size { let queue = Arc::clone(queue); @@ -439,15 +478,14 @@ fn spawn_workers(size: usize, queue: &Arc) { } fn worker_loop(queue: &Arc) { - while let Some(schedule) = queue.dequeue() { - let ScheduleId(_, timer_token) = schedule.schedule_id; - schedule.state_control.within_lock(|state| { + while let Some(callback) = queue.dequeue() { + let ScheduleId(_, timer_token) = callback.schedule_id; + callback.state_control.within_lock(|state| { debug_assert_ne!(*state, ScheduleState::Timeout); if *state != ScheduleState::Cancelled { - schedule.handler.on_timeout(timer_token); - let is_oneshot = schedule.repeat.is_none(); - debug_assert_eq!(is_oneshot, *state == ScheduleState::Wait); - if is_oneshot { + callback.handler.on_timeout(timer_token); + debug_assert_eq!(callback.from_oneshot_schedule, *state == ScheduleState::Wait); + if callback.from_oneshot_schedule { ScheduleStateControl::set_timeout(state); } } @@ -456,7 +494,7 @@ fn worker_loop(queue: &Arc) { } struct WorkerQueue { - queue: Mutex>, + queue: Mutex>, condvar: Condvar, stop: AtomicBool, } @@ -470,16 +508,16 @@ impl WorkerQueue { } } - fn enqueue(&self, schedule: Schedule) { + fn enqueue(&self, callback: Callback) { let mut queue = self.queue.lock(); if self.stop.load(Ordering::SeqCst) { return } - queue.push_back(schedule); + queue.push_back(callback); self.condvar.notify_one(); } - fn dequeue(&self) -> Option { + fn dequeue(&self) -> Option { let mut queue = self.queue.lock(); while queue.is_empty() { if self.stop.load(Ordering::SeqCst) { @@ -493,6 +531,8 @@ impl WorkerQueue { #[cfg(test)] mod tests { + use std::panic; + use super::*; use parking_lot::Mutex; @@ -528,6 +568,14 @@ mod tests { diff < tick_epsilon() } + fn new_timer(timer_loop: &TimerLoop, name: TimerName, handler: &Arc) -> TimerApi + where + T: 'static + Sized + TimeoutHandler, { + let timer = timer_loop.new_timer(name); + timer.set_handler(handler); + timer + } + #[test] fn test_timeout() { let timer_token = 100; @@ -542,7 +590,7 @@ mod tests { condvar.notify_all(); })) }; - let timer = timer_loop.new_timer("test", Arc::clone(&handler)); + let timer = new_timer(&timer_loop, "test", &handler); let begin = Instant::now(); timer.schedule_once(tick(), timer_token).unwrap(); @@ -570,7 +618,7 @@ mod tests { condvar.notify_all(); })) }; - let timer = timer_loop.new_timer("test", Arc::clone(&handler)); + let timer = new_timer(&timer_loop, "test", &handler); assert_eq!(timer.schedule_once(tick(), timer_token), Ok(())); assert_eq!(timer.cancel(timer_token), Ok(true)); @@ -581,12 +629,47 @@ mod tests { assert!(value.is_none()); } + #[test] + #[should_panic] + fn test_timer_set_hander_twice() { + let timer_loop = TimerLoop::new(1); + let handler = Arc::new(CallbackHandler(|_| {})); + let timer = timer_loop.new_timer("test"); + timer.set_handler(&handler); + timer.set_handler(&handler); + } + + #[test] + fn test_handler_drop() { + let timer_token = 100; + let timer_loop = TimerLoop::new(1); + let pair = Arc::new((Condvar::new(), Mutex::new(None))); + let handler = { + let pair = Arc::clone(&pair); + Arc::new(CallbackHandler(move |_| { + let (ref condvar, ref mutex) = *pair; + let mut value = mutex.lock(); + *value = Some(()); + condvar.notify_all(); + })) + }; + let timer = new_timer(&timer_loop, "test", &handler); + + assert_eq!(timer.schedule_once(tick(), timer_token), Ok(())); + drop(handler); + + let (ref condvar, ref mutex) = *pair; + let mut value = mutex.lock(); + condvar.wait_for(&mut value, long_tick()); + assert!(value.is_none()); + } + #[test] fn test_schedule_twice() { let timer_token = 100; let timer_loop = TimerLoop::new(1); let handler = Arc::new(CallbackHandler(|_| {})); - let timer = timer_loop.new_timer("test", Arc::clone(&handler)); + let timer = new_timer(&timer_loop, "test", &handler); assert_eq!(timer.schedule_once(tick(), timer_token), Ok(())); assert_eq!(timer.schedule_once(tick(), timer_token), Err(ScheduleError::TokenAlreadyScheduled)); @@ -598,7 +681,7 @@ mod tests { let timer_token_2 = 200; let timer_loop = TimerLoop::new(1); let handler = Arc::new(CallbackHandler(|_| {})); - let timer = timer_loop.new_timer("test", Arc::clone(&handler)); + let timer = new_timer(&timer_loop, "test", &handler); assert_eq!(timer.schedule_once(tick(), timer_token_1), Ok(())); assert_eq!(timer.schedule_once(tick(), timer_token_2), Ok(())); @@ -618,7 +701,7 @@ mod tests { condvar.notify_all(); })) }; - let timer = timer_loop.new_timer("test", Arc::clone(&handler)); + let timer = new_timer(&timer_loop, "test", &handler); assert_eq!(timer.schedule_once(tick(), timer_token), Ok(())); thread::sleep(long_tick()); @@ -649,7 +732,7 @@ mod tests { condvar.notify_all(); })) }; - let timer = timer_loop.new_timer("test", Arc::clone(&handler)); + let timer = new_timer(&timer_loop, "test", &handler); let begin = Instant::now(); assert_eq!(timer.schedule_once(tick(), timer_token), Ok(())); @@ -681,7 +764,7 @@ mod tests { condvar.notify_all(); })) }; - let timer = timer_loop.new_timer("test", Arc::clone(&handler)); + let timer = new_timer(&timer_loop, "test", &handler); let begin = Instant::now(); timer.schedule_repeat(tick(), timer_token).unwrap(); From 1f8c2235c7e962edb856695ab2f3d553d9145f6b Mon Sep 17 00:00:00 2001 From: SeongChan Lee Date: Sun, 13 Jan 2019 02:10:31 +0900 Subject: [PATCH 5/6] Fix not to reschedule detached Schedule Detached schedule is cancelled schedule. --- util/timer/src/timer.rs | 18 +++++++++++------- 1 file changed, 11 insertions(+), 7 deletions(-) diff --git a/util/timer/src/timer.rs b/util/timer/src/timer.rs index e79de269c4..9caa3ea996 100644 --- a/util/timer/src/timer.rs +++ b/util/timer/src/timer.rs @@ -251,30 +251,33 @@ impl SchedulerInner { Ok(()) } - fn try_reschedule(&mut self, mut schedule: Schedule) { + fn try_reschedule(&mut self, mut schedule: Schedule) -> bool { schedule.at = Instant::now() + schedule.repeat.expect("Schedule should have repeat interval"); match self.states.entry(schedule.schedule_id) { Entry::Vacant(_) => { // 'schedule.state_control' was detached one (Def 1). // Don't reschedule since it is Cancelled (Rule 1) // schedule is going to be removed + false } Entry::Occupied(entry) => { if Arc::ptr_eq(entry.get(), &schedule.state_control) { // schedule.state_control was attached one, (Def 1) // just re-push to heap. self.heap.push(Reverse(TimeOrdered(schedule))); + true } else if entry.get().is_cancelled() { // Detach the entry (Corollary 1) before it become garbage (Note 1), entry.remove(); - // 'schedule.state_control' was detached one (Def 1). - // Don't reschedule since it is Cancelled (Rule 1) - // schedule is going to be removed + // 'schedule.state_control' was detached one (Def 1). + // Don't reschedule since it is Cancelled (Rule 1) + // schedule is going to be removed + false } else { unreachable!("Rule 1 was violated"); } } - }; + } } fn cancel(&mut self, requested_timer: &TimerApi, timer_token: TimerToken) -> bool { @@ -303,8 +306,9 @@ impl SchedulerInner { if timed_out.repeat.is_some() { if let Some(callback) = Callback::from_schedule(&timed_out) { // timed_out.state_control is re-pushed only after it is popped (Rule 3) - self.try_reschedule(timed_out.clone()); - worker_queue.enqueue(callback); + if self.try_reschedule(timed_out.clone()) { + worker_queue.enqueue(callback); + } } } else { let enqueue = match self.states.entry(timed_out.schedule_id) { From 2102673a676ea51f6b21f7d281ddf30143b447ce Mon Sep 17 00:00:00 2001 From: SeongChan Lee Date: Sun, 13 Jan 2019 02:11:58 +0900 Subject: [PATCH 6/6] Add timer trace logs --- Cargo.lock | 2 + util/logger/src/macros.rs | 3 ++ util/timer/Cargo.toml | 2 + util/timer/src/lib.rs | 4 ++ util/timer/src/timer.rs | 82 ++++++++++++++++++++++++++++++++++++--- 5 files changed, 87 insertions(+), 6 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 91c19628da..8e13e93a10 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -547,6 +547,8 @@ dependencies = [ name = "codechain-timer" version = "0.1.0" dependencies = [ + "codechain-logger 0.1.0", + "log 0.4.6 (registry+https://github.com/rust-lang/crates.io-index)", "parking_lot 0.6.4 (registry+https://github.com/rust-lang/crates.io-index)", ] diff --git a/util/logger/src/macros.rs b/util/logger/src/macros.rs index a37fe22455..7bac0ba73d 100644 --- a/util/logger/src/macros.rs +++ b/util/logger/src/macros.rs @@ -85,6 +85,9 @@ macro_rules! log_target { (TEST_SCRIPT) => { "test_script" }; + (TIMER) => { + "timer" + }; (TRIE) => { "trie" }; diff --git a/util/timer/Cargo.toml b/util/timer/Cargo.toml index bbe5ca23e6..5b71f792e3 100644 --- a/util/timer/Cargo.toml +++ b/util/timer/Cargo.toml @@ -7,3 +7,5 @@ authors = ["CodeChain Team "] [dependencies] parking_lot = "0.6.0" +log = "0.4.6" +codechain-logger = { path = "../logger" } diff --git a/util/timer/src/lib.rs b/util/timer/src/lib.rs index 52ae64e9c0..e9cc7f74dc 100644 --- a/util/timer/src/lib.rs +++ b/util/timer/src/lib.rs @@ -15,6 +15,10 @@ // along with this program. If not, see . extern crate parking_lot; +#[macro_use] +extern crate log; +#[macro_use] +extern crate codechain_logger as clogger; mod timer; diff --git a/util/timer/src/timer.rs b/util/timer/src/timer.rs index 9caa3ea996..1de35c1b3c 100644 --- a/util/timer/src/timer.rs +++ b/util/timer/src/timer.rs @@ -239,12 +239,22 @@ impl SchedulerInner { } }; + ctrace!( + TIMER, + "schedule(TimerName({}), {:?}, after: {:?}, repeat: {:?})", + requested_timer.timer_name, + schedule_id, + after, + repeat, + ); + let schedule = Schedule { at: Instant::now() + after, schedule_id, repeat, state_control, handler, + timer_name: requested_timer.timer_name, }; // state_control become an attached one (Def 1) self.heap.push(Reverse(TimeOrdered(schedule))); @@ -287,6 +297,7 @@ impl SchedulerInner { Entry::Vacant(_) => false, Entry::Occupied(entry) => { // Detach and cancel it (Rule 1) + ctrace!(TIMER, "cancel(TimerName({}), {:?})", requested_timer.timer_name, schedule_id); let state_control = entry.remove(); state_control.cancel() } @@ -303,12 +314,35 @@ impl SchedulerInner { } let Reverse(TimeOrdered(timed_out)) = self.heap.pop().expect("It always have an item"); + let timer_name = timed_out.timer_name; + let schedule_id = timed_out.schedule_id; + if timed_out.repeat.is_some() { if let Some(callback) = Callback::from_schedule(&timed_out) { // timed_out.state_control is re-pushed only after it is popped (Rule 3) if self.try_reschedule(timed_out.clone()) { + ctrace!( + TIMER, + "handle_timeout(TimerName({}), {:?}, repeat): Enqueue to worker queue", + timer_name, + schedule_id + ); worker_queue.enqueue(callback); + } else { + ctrace!( + TIMER, + "handle_timeout(TimerName({}), {:?}, repeat): Cancelled schedule", + timer_name, + schedule_id + ); } + } else { + ctrace!( + TIMER, + "handle_timeout(TimerName({}), {:?}, repeat): Dropped TimeoutHandler", + timer_name, + schedule_id + ); } } else { let enqueue = match self.states.entry(timed_out.schedule_id) { @@ -324,21 +358,41 @@ impl SchedulerInner { _ => false, // also detached. }; // timed_out is anyway removed. - if enqueue { - if let Some(callback) = Callback::from_schedule(&timed_out) { + + if let Some(callback) = Callback::from_schedule(&timed_out) { + if enqueue { + ctrace!( + TIMER, + "handle_timeout(TimerName({}), {:?}, once): Enqueue to worker queue", + timer_name, + schedule_id + ); worker_queue.enqueue(callback); + } else { + // 'timed_out.state_control' was detached one. (Def 1) + // It already be cancelled (Rule 1) + debug_assert!(timed_out.state_control.is_cancelled()); + ctrace!( + TIMER, + "handle_timeout(TimerName({}), {:?}, once): Cancelled schedule", + timer_name, + schedule_id + ); } } else { - // 'timed_out.state_control' was detached one. (Def 1) - // It already be cancelled (Rule 1) - debug_assert!(timed_out.state_control.is_cancelled()); + ctrace!( + TIMER, + "handle_timeout(TimerName({}), {:?}, once): Dropped TimeoutHandler", + timer_name, + schedule_id + ); } } } } } -#[derive(Copy, Clone, Eq, PartialEq, Hash)] +#[derive(Copy, Clone, Eq, PartialEq, Hash, Debug)] struct ScheduleId(TimerId, TimerToken); /* @@ -422,6 +476,7 @@ struct Schedule { repeat: Option, state_control: Arc, handler: Weak, + timer_name: TimerName, } struct TimeOrdered(T); @@ -457,6 +512,7 @@ struct Callback { from_oneshot_schedule: bool, state_control: Arc, handler: Arc, + timer_name: TimerName, } impl Callback { @@ -467,6 +523,7 @@ impl Callback { from_oneshot_schedule: schedule.repeat.is_none(), state_control: Arc::clone(&schedule.state_control), handler, + timer_name: schedule.timer_name, }) } else { None @@ -487,11 +544,24 @@ fn worker_loop(queue: &Arc) { callback.state_control.within_lock(|state| { debug_assert_ne!(*state, ScheduleState::Timeout); if *state != ScheduleState::Cancelled { + ctrace!( + TIMER, + "worker_loop(TimerName({}), {:?}): Call on_timeout", + callback.timer_name, + callback.schedule_id + ); callback.handler.on_timeout(timer_token); debug_assert_eq!(callback.from_oneshot_schedule, *state == ScheduleState::Wait); if callback.from_oneshot_schedule { ScheduleStateControl::set_timeout(state); } + } else { + ctrace!( + TIMER, + "worker_loop(TimerName({}), {:?}): Cancelled callback", + callback.timer_name, + callback.schedule_id + ); } }); }