Clean up Rpc exit signal
This commit is contained in:
		@@ -168,7 +168,7 @@ impl Fullnode {
 | 
			
		||||
            drone_addr,
 | 
			
		||||
            storage_state.clone(),
 | 
			
		||||
            config.rpc_config.clone(),
 | 
			
		||||
            exit.clone(),
 | 
			
		||||
            &exit,
 | 
			
		||||
        );
 | 
			
		||||
 | 
			
		||||
        let subscriptions = Arc::new(RpcSubscriptions::default());
 | 
			
		||||
@@ -293,9 +293,6 @@ impl Fullnode {
 | 
			
		||||
        // which is the sole initiator of rotations.
 | 
			
		||||
        self.poh_recorder.lock().unwrap().clear_bank();
 | 
			
		||||
        self.poh_service.exit();
 | 
			
		||||
        if let Some(ref rpc_service) = self.rpc_service {
 | 
			
		||||
            rpc_service.exit();
 | 
			
		||||
        }
 | 
			
		||||
        self.node_services.exit();
 | 
			
		||||
    }
 | 
			
		||||
 | 
			
		||||
 
 | 
			
		||||
@@ -57,13 +57,13 @@ impl JsonRpcRequestProcessor {
 | 
			
		||||
    pub fn new(
 | 
			
		||||
        storage_state: StorageState,
 | 
			
		||||
        config: JsonRpcConfig,
 | 
			
		||||
        fullnode_exit: Arc<AtomicBool>,
 | 
			
		||||
        fullnode_exit: &Arc<AtomicBool>,
 | 
			
		||||
    ) -> Self {
 | 
			
		||||
        JsonRpcRequestProcessor {
 | 
			
		||||
            bank: None,
 | 
			
		||||
            storage_state,
 | 
			
		||||
            config,
 | 
			
		||||
            fullnode_exit,
 | 
			
		||||
            fullnode_exit: fullnode_exit.clone(),
 | 
			
		||||
        }
 | 
			
		||||
    }
 | 
			
		||||
 | 
			
		||||
@@ -428,7 +428,7 @@ mod tests {
 | 
			
		||||
        let request_processor = Arc::new(RwLock::new(JsonRpcRequestProcessor::new(
 | 
			
		||||
            StorageState::default(),
 | 
			
		||||
            JsonRpcConfig::default(),
 | 
			
		||||
            exit,
 | 
			
		||||
            &exit,
 | 
			
		||||
        )));
 | 
			
		||||
        request_processor.write().unwrap().set_bank(&bank);
 | 
			
		||||
        let cluster_info = Arc::new(RwLock::new(ClusterInfo::new(NodeInfo::default())));
 | 
			
		||||
@@ -458,7 +458,7 @@ mod tests {
 | 
			
		||||
        let bank = Arc::new(Bank::new(&genesis_block));
 | 
			
		||||
        let exit = Arc::new(AtomicBool::new(false));
 | 
			
		||||
        let mut request_processor =
 | 
			
		||||
            JsonRpcRequestProcessor::new(StorageState::default(), JsonRpcConfig::default(), exit);
 | 
			
		||||
            JsonRpcRequestProcessor::new(StorageState::default(), JsonRpcConfig::default(), &exit);
 | 
			
		||||
        request_processor.set_bank(&bank);
 | 
			
		||||
        thread::spawn(move || {
 | 
			
		||||
            let blockhash = bank.last_blockhash();
 | 
			
		||||
@@ -631,7 +631,7 @@ mod tests {
 | 
			
		||||
                let mut request_processor = JsonRpcRequestProcessor::new(
 | 
			
		||||
                    StorageState::default(),
 | 
			
		||||
                    JsonRpcConfig::default(),
 | 
			
		||||
                    exit,
 | 
			
		||||
                    &exit,
 | 
			
		||||
                );
 | 
			
		||||
                request_processor.set_bank(&bank);
 | 
			
		||||
                Arc::new(RwLock::new(request_processor))
 | 
			
		||||
@@ -707,11 +707,8 @@ mod tests {
 | 
			
		||||
    #[test]
 | 
			
		||||
    fn test_rpc_request_processor_config_default_trait_fullnode_exit_fails() {
 | 
			
		||||
        let exit = Arc::new(AtomicBool::new(false));
 | 
			
		||||
        let request_processor = JsonRpcRequestProcessor::new(
 | 
			
		||||
            StorageState::default(),
 | 
			
		||||
            JsonRpcConfig::default(),
 | 
			
		||||
            exit.clone(),
 | 
			
		||||
        );
 | 
			
		||||
        let request_processor =
 | 
			
		||||
            JsonRpcRequestProcessor::new(StorageState::default(), JsonRpcConfig::default(), &exit);
 | 
			
		||||
        assert_eq!(request_processor.fullnode_exit(), Ok(false));
 | 
			
		||||
        assert_eq!(exit.load(Ordering::Relaxed), false);
 | 
			
		||||
    }
 | 
			
		||||
@@ -721,7 +718,7 @@ mod tests {
 | 
			
		||||
        let request_processor = JsonRpcRequestProcessor::new(
 | 
			
		||||
            StorageState::default(),
 | 
			
		||||
            JsonRpcConfig::DefaultConfig,
 | 
			
		||||
            exit.clone(),
 | 
			
		||||
            &exit,
 | 
			
		||||
        );
 | 
			
		||||
        assert_eq!(request_processor.fullnode_exit(), Ok(false));
 | 
			
		||||
        assert_eq!(exit.load(Ordering::Relaxed), false);
 | 
			
		||||
@@ -733,7 +730,7 @@ mod tests {
 | 
			
		||||
        let request_processor = JsonRpcRequestProcessor::new(
 | 
			
		||||
            StorageState::default(),
 | 
			
		||||
            JsonRpcConfig::TestOnlyAllowRpcFullnodeExit,
 | 
			
		||||
            exit.clone(),
 | 
			
		||||
            &exit,
 | 
			
		||||
        );
 | 
			
		||||
        assert_eq!(request_processor.fullnode_exit(), Ok(true));
 | 
			
		||||
        assert_eq!(exit.load(Ordering::Relaxed), true);
 | 
			
		||||
 
 | 
			
		||||
@@ -17,7 +17,6 @@ pub const RPC_PORT: u16 = 8899;
 | 
			
		||||
 | 
			
		||||
pub struct JsonRpcService {
 | 
			
		||||
    thread_hdl: JoinHandle<()>,
 | 
			
		||||
    exit: Arc<AtomicBool>,
 | 
			
		||||
    pub request_processor: Arc<RwLock<JsonRpcRequestProcessor>>, // Used only by tests...
 | 
			
		||||
}
 | 
			
		||||
 | 
			
		||||
@@ -28,13 +27,13 @@ impl JsonRpcService {
 | 
			
		||||
        drone_addr: SocketAddr,
 | 
			
		||||
        storage_state: StorageState,
 | 
			
		||||
        config: JsonRpcConfig,
 | 
			
		||||
        exit: Arc<AtomicBool>,
 | 
			
		||||
        exit: &Arc<AtomicBool>,
 | 
			
		||||
    ) -> Self {
 | 
			
		||||
        info!("rpc bound to {:?}", rpc_addr);
 | 
			
		||||
        let request_processor = Arc::new(RwLock::new(JsonRpcRequestProcessor::new(
 | 
			
		||||
            storage_state,
 | 
			
		||||
            config,
 | 
			
		||||
            exit.clone(),
 | 
			
		||||
            exit,
 | 
			
		||||
        )));
 | 
			
		||||
        let request_processor_ = request_processor.clone();
 | 
			
		||||
 | 
			
		||||
@@ -71,7 +70,6 @@ impl JsonRpcService {
 | 
			
		||||
            .unwrap();
 | 
			
		||||
        Self {
 | 
			
		||||
            thread_hdl,
 | 
			
		||||
            exit,
 | 
			
		||||
            request_processor,
 | 
			
		||||
        }
 | 
			
		||||
    }
 | 
			
		||||
@@ -79,15 +77,6 @@ impl JsonRpcService {
 | 
			
		||||
    pub fn set_bank(&mut self, bank: &Arc<Bank>) {
 | 
			
		||||
        self.request_processor.write().unwrap().set_bank(bank);
 | 
			
		||||
    }
 | 
			
		||||
 | 
			
		||||
    pub fn exit(&self) {
 | 
			
		||||
        self.exit.store(true, Ordering::Relaxed);
 | 
			
		||||
    }
 | 
			
		||||
 | 
			
		||||
    pub fn close(self) -> thread::Result<()> {
 | 
			
		||||
        self.exit();
 | 
			
		||||
        self.join()
 | 
			
		||||
    }
 | 
			
		||||
}
 | 
			
		||||
 | 
			
		||||
impl Service for JsonRpcService {
 | 
			
		||||
@@ -127,7 +116,7 @@ mod tests {
 | 
			
		||||
            drone_addr,
 | 
			
		||||
            StorageState::default(),
 | 
			
		||||
            JsonRpcConfig::default(),
 | 
			
		||||
            exit,
 | 
			
		||||
            &exit,
 | 
			
		||||
        );
 | 
			
		||||
        rpc_service.set_bank(&Arc::new(bank));
 | 
			
		||||
        let thread = rpc_service.thread_hdl.thread();
 | 
			
		||||
@@ -142,7 +131,7 @@ mod tests {
 | 
			
		||||
                .get_balance(alice.pubkey())
 | 
			
		||||
                .unwrap()
 | 
			
		||||
        );
 | 
			
		||||
 | 
			
		||||
        rpc_service.close().unwrap();
 | 
			
		||||
        exit.store(true, Ordering::Relaxed);
 | 
			
		||||
        rpc_service.join().unwrap();
 | 
			
		||||
    }
 | 
			
		||||
}
 | 
			
		||||
 
 | 
			
		||||
		Reference in New Issue
	
	Block a user