Merge branch 'master' into fix/net-slow-lock

Author: jcnelson

src/burnchains/burnchain.rs

@@ -1291,32 +1291,36 @@ impl Burnchain {
         // TODO: don't re-process blocks.  See if the block hash is already present in the burn db,
         // and if so, do nothing.
         let download_thread: thread::JoinHandle<Result<(), burnchain_error>> =
-            thread::spawn(move || {
-                while let Ok(Some(ipc_header)) = downloader_recv.recv() {
-                    debug!("Try recv next header");
-
-                    let download_start = get_epoch_time_ms();
-                    let ipc_block = downloader.download(&ipc_header)?;
-                    let download_end = get_epoch_time_ms();
-
-                    debug!(
-                        "Downloaded block {} in {}ms",
-                        ipc_block.height(),
-                        download_end.saturating_sub(download_start)
-                    );
+            thread::Builder::new()
+                .name("burnchain-downloader".to_string())
+                .spawn(move || {
+                    while let Ok(Some(ipc_header)) = downloader_recv.recv() {
+                        debug!("Try recv next header");
+
+                        let download_start = get_epoch_time_ms();
+                        let ipc_block = downloader.download(&ipc_header)?;
+                        let download_end = get_epoch_time_ms();
+
+                        debug!(
+                            "Downloaded block {} in {}ms",
+                            ipc_block.height(),
+                            download_end.saturating_sub(download_start)
+                        );
 
+                        parser_send
+                            .send(Some(ipc_block))
+                            .map_err(|_e| burnchain_error::ThreadChannelError)?;
+                    }
                     parser_send
-                        .send(Some(ipc_block))
+                        .send(None)
                         .map_err(|_e| burnchain_error::ThreadChannelError)?;
-                }
-                parser_send
-                    .send(None)
-                    .map_err(|_e| burnchain_error::ThreadChannelError)?;
-                Ok(())
-            });
+                    Ok(())
+                })
+                .unwrap();
 
-        let parse_thread: thread::JoinHandle<Result<(), burnchain_error>> =
-            thread::spawn(move || {
+        let parse_thread: thread::JoinHandle<Result<(), burnchain_error>> = thread::Builder::new()
+            .name("burnchain-parser".to_string())
+            .spawn(move || {
                 while let Ok(Some(ipc_block)) = parser_recv.recv() {
                     debug!("Try recv next block");
 
@@ -1338,34 +1342,38 @@ impl Burnchain {
                     .send(None)
                     .map_err(|_e| burnchain_error::ThreadChannelError)?;
                 Ok(())
-            });
+            })
+            .unwrap();
 
         let db_thread: thread::JoinHandle<Result<BurnchainBlockHeader, burnchain_error>> =
-            thread::spawn(move || {
-                let mut last_processed = burn_chain_tip;
-                while let Ok(Some(burnchain_block)) = db_recv.recv() {
-                    debug!("Try recv next parsed block");
+            thread::Builder::new()
+                .name("burnchain-db".to_string())
+                .spawn(move || {
+                    let mut last_processed = burn_chain_tip;
+                    while let Ok(Some(burnchain_block)) = db_recv.recv() {
+                        debug!("Try recv next parsed block");
+
+                        if burnchain_block.block_height() == 0 {
+                            continue;
+                        }
 
-                    if burnchain_block.block_height() == 0 {
-                        continue;
-                    }
+                        let insert_start = get_epoch_time_ms();
+                        last_processed =
+                            Burnchain::process_block(&myself, &mut burnchain_db, &burnchain_block)?;
+                        if !coord_comm.announce_new_burn_block() {
+                            return Err(burnchain_error::CoordinatorClosed);
+                        }
+                        let insert_end = get_epoch_time_ms();
 
-                    let insert_start = get_epoch_time_ms();
-                    last_processed =
-                        Burnchain::process_block(&myself, &mut burnchain_db, &burnchain_block)?;
-                    if !coord_comm.announce_new_burn_block() {
-                        return Err(burnchain_error::CoordinatorClosed);
+                        debug!(
+                            "Inserted block {} in {}ms",
+                            burnchain_block.block_height(),
+                            insert_end.saturating_sub(insert_start)
+                        );
                     }
-                    let insert_end = get_epoch_time_ms();
-
-                    debug!(
-                        "Inserted block {} in {}ms",
-                        burnchain_block.block_height(),
-                        insert_end.saturating_sub(insert_start)
-                    );
-                }
-                Ok(last_processed)
-            });
+                    Ok(last_processed)
+                })
+                .unwrap();
 
         // feed the pipeline!
         let input_headers = indexer.read_headers(start_block + 1, end_block + 1)?;

src/util/log.rs

@@ -50,7 +50,10 @@ fn print_msg_header(mut rd: &mut dyn RecordDecorator, record: &Record) -> io::Re
     write!(rd, " ")?;
     write!(rd, "[{}:{}]", record.file(), record.line())?;
     write!(rd, " ")?;
-    write!(rd, "[{:?}]", thread::current().id())?;
+    match thread::current().name() {
+        None => write!(rd, "[{:?}]", thread::current().id())?,
+        Some(name) => write!(rd, "[{}]", name)?,
+    }
 
     rd.start_whitespace()?;
     write!(rd, " ")?;

testnet/stacks-node/src/neon_node.rs

@@ -574,117 +574,123 @@ fn spawn_peer(
     // buffer up blocks to store without stalling the p2p thread
     let mut results_with_data = VecDeque::new();
 
-    let server_thread = thread::spawn(move || {
-        let handler_args = RPCHandlerArgs {
-            exit_at_block_height: exit_at_block_height.as_ref(),
-            genesis_chainstate_hash: Sha256Sum::from_hex(stx_genesis::GENESIS_CHAINSTATE_HASH)
-                .unwrap(),
-            ..RPCHandlerArgs::default()
-        };
+    let server_thread = thread::Builder::new()
+        .name("p2p".to_string())
+        .spawn(move || {
+            let handler_args = RPCHandlerArgs {
+                exit_at_block_height: exit_at_block_height.as_ref(),
+                genesis_chainstate_hash: Sha256Sum::from_hex(stx_genesis::GENESIS_CHAINSTATE_HASH)
+                    .unwrap(),
+                ..RPCHandlerArgs::default()
+            };
 
-        let mut disconnected = false;
-        let mut num_p2p_state_machine_passes = 0;
-        let mut num_inv_sync_passes = 0;
-        let mut mblock_deadline = 0;
+            let mut disconnected = false;
+            let mut num_p2p_state_machine_passes = 0;
+            let mut num_inv_sync_passes = 0;
+            let mut mblock_deadline = 0;
 
-        while !disconnected {
-            let download_backpressure = results_with_data.len() > 0;
-            let poll_ms = if !download_backpressure && this.has_more_downloads() {
-                // keep getting those blocks -- drive the downloader state-machine
-                debug!(
-                    "P2P: backpressure: {}, more downloads: {}",
+            while !disconnected {
+                let download_backpressure = results_with_data.len() > 0;
+                let poll_ms = if !download_backpressure && this.has_more_downloads() {
+                    // keep getting those blocks -- drive the downloader state-machine
+                    debug!(
+                        "P2P: backpressure: {}, more downloads: {}",
+                        download_backpressure,
+                        this.has_more_downloads()
+                    );
+                    100
+                } else {
+                    cmp::min(poll_timeout, config.node.microblock_frequency)
+                };
+
+                let mut expected_attachments = match attachments_rx.try_recv() {
+                    Ok(expected_attachments) => expected_attachments,
+                    _ => {
+                        debug!("Atlas: attachment channel is empty");
+                        HashSet::new()
+                    }
+                };
+
+                let _ = Relayer::setup_unconfirmed_state_readonly(&mut chainstate, &sortdb);
+                recv_unconfirmed_txs(&mut chainstate, unconfirmed_txs.clone());
+
+                let network_result = match this.run(
+                    &sortdb,
+                    &mut chainstate,
+                    &mut mem_pool,
+                    Some(&mut dns_client),
                     download_backpressure,
-                    this.has_more_downloads()
-                );
-                100
-            } else {
-                cmp::min(poll_timeout, config.node.microblock_frequency)
-            };
+                    poll_ms,
+                    &handler_args,
+                    &mut expected_attachments,
+                ) {
+                    Ok(res) => res,
+                    Err(e) => {
+                        error!("P2P: Failed to process network dispatch: {:?}", &e);
+                        panic!();
+                    }
+                };
 
-            let mut expected_attachments = match attachments_rx.try_recv() {
-                Ok(expected_attachments) => expected_attachments,
-                _ => {
-                    debug!("Atlas: attachment channel is empty");
-                    HashSet::new()
+                if num_p2p_state_machine_passes < network_result.num_state_machine_passes {
+                    // p2p state-machine did a full pass. Notify anyone listening.
+                    sync_comms.notify_p2p_state_pass();
+                    num_p2p_state_machine_passes = network_result.num_state_machine_passes;
                 }
-            };
 
-            let _ = Relayer::setup_unconfirmed_state_readonly(&mut chainstate, &sortdb);
-            recv_unconfirmed_txs(&mut chainstate, unconfirmed_txs.clone());
-
-            let network_result = match this.run(
-                &sortdb,
-                &mut chainstate,
-                &mut mem_pool,
-                Some(&mut dns_client),
-                download_backpressure,
-                poll_ms,
-                &handler_args,
-                &mut expected_attachments,
-            ) {
-                Ok(res) => res,
-                Err(e) => {
-                    error!("P2P: Failed to process network dispatch: {:?}", &e);
-                    panic!();
+                if num_inv_sync_passes < network_result.num_inv_sync_passes {
+                    // inv-sync state-machine did a full pass. Notify anyone listening.
+                    sync_comms.notify_inv_sync_pass();
+                    num_inv_sync_passes = network_result.num_inv_sync_passes;
                 }
-            };
 
-            if num_p2p_state_machine_passes < network_result.num_state_machine_passes {
-                // p2p state-machine did a full pass. Notify anyone listening.
-                sync_comms.notify_p2p_state_pass();
-                num_p2p_state_machine_passes = network_result.num_state_machine_passes;
-            }
-
-            if num_inv_sync_passes < network_result.num_inv_sync_passes {
-                // inv-sync state-machine did a full pass. Notify anyone listening.
-                sync_comms.notify_inv_sync_pass();
-                num_inv_sync_passes = network_result.num_inv_sync_passes;
-            }
-
-            if network_result.has_data_to_store() {
-                results_with_data.push_back(RelayerDirective::HandleNetResult(network_result));
-            }
+                if network_result.has_data_to_store() {
+                    results_with_data.push_back(RelayerDirective::HandleNetResult(network_result));
+                }
 
-            if mblock_deadline < get_epoch_time_ms() {
-                results_with_data.push_back(RelayerDirective::RunMicroblockTenure);
-                mblock_deadline = get_epoch_time_ms() + (config.node.microblock_frequency as u128);
-            }
+                if mblock_deadline < get_epoch_time_ms() {
+                    results_with_data.push_back(RelayerDirective::RunMicroblockTenure);
+                    mblock_deadline = get_epoch_time_ms() + (config.node.microblock_frequency as u128);
+                }
 
-            while let Some(next_result) = results_with_data.pop_front() {
-                // have blocks, microblocks, and/or transactions (don't care about anything else),
-                // or a directive to mine microblocks
-                if let Err(e) = relay_channel.try_send(next_result) {
-                    debug!(
-                        "P2P: {:?}: download backpressure detected",
-                        &this.local_peer
-                    );
-                    match e {
-                        TrySendError::Full(directive) => {
-                            if let RelayerDirective::RunMicroblockTenure = directive {
-                                // can drop this
-                            } else {
-                                // don't lose this data -- just try it again
-                                results_with_data.push_front(directive);
+                while let Some(next_result) = results_with_data.pop_front() {
+                    // have blocks, microblocks, and/or transactions (don't care about anything else),
+                    // or a directive to mine microblocks
+                    if let Err(e) = relay_channel.try_send(next_result) {
+                        debug!(
+                            "P2P: {:?}: download backpressure detected",
+                            &this.local_peer
+                        );
+                        match e {
+                            TrySendError::Full(directive) => {
+                                if let RelayerDirective::RunMicroblockTenure = directive {
+                                    // can drop this
+                                } else {
+                                    // don't lose this data -- just try it again
+                                    results_with_data.push_front(directive);
+                                }
+                                break;
+                            }
+                            TrySendError::Disconnected(_) => {
+                                info!("P2P: Relayer hang up with p2p channel");
+                                disconnected = true;
+                                break;
                             }
-                            break;
-                        }
-                        TrySendError::Disconnected(_) => {
-                            info!("P2P: Relayer hang up with p2p channel");
-                            disconnected = true;
-                            break;
                         }
+                    } else {
+                        debug!("P2P: Dispatched result to Relayer!");
                     }
-                } else {
-                    debug!("P2P: Dispatched result to Relayer!");
                 }
             }
-        }
-        debug!("P2P thread exit!");
-    });
+            debug!("P2P thread exit!");
+        })
+        .unwrap();
 
-    let _jh = thread::spawn(move || {
-        dns_resolver.thread_main();
-    });
+    let _jh = thread::Builder::new()
+        .name("dns-resolver".to_string())
+        .spawn(move || {
+            dns_resolver.thread_main();
+        })
+        .unwrap();
 
     Ok(server_thread)
 }
@@ -734,7 +740,7 @@ fn spawn_miner_relayer(
     let mut microblock_miner_state = None;
     let mut miner_tip = None;
 
-    let _relayer_handle = thread::spawn(move || {
+    let _relayer_handle = thread::Builder::new().name("relayer".to_string()).spawn(move || {
         let mut did_register_key = false;
         let mut key_registered_at_block = 0;
         while let Ok(mut directive) = relay_channel.recv() {
@@ -945,7 +951,7 @@ fn spawn_miner_relayer(
             }
         }
         debug!("Relayer exit!");
-    });
+    }).unwrap();
 
     Ok(())
 }