Merge pull request #516 from ethcore/sync

Better memory management

ethcore/src/block_queue.rs

@@ -28,6 +28,31 @@ use service::*;
 use client::BlockStatus;
 use util::panics::*;
 
+known_heap_size!(0, UnVerifiedBlock, VerifyingBlock, PreVerifiedBlock);
+
+const MIN_MEM_LIMIT: usize = 16384;
+const MIN_QUEUE_LIMIT: usize = 512;
+
+/// Block queue configuration
+#[derive(Debug)]
+pub struct BlockQueueConfig {
+	/// Maximum number of blocks to keep in unverified queue.
+	/// When the limit is reached, is_full returns true.
+	pub max_queue_size: usize,
+	/// Maximum heap memory to use.
+	/// When the limit is reached, is_full returns true.
+	pub max_mem_use: usize,
+}
+
+impl Default for BlockQueueConfig {
+	fn default() -> Self {
+		BlockQueueConfig {
+			max_queue_size: 30000,
+			max_mem_use: 50 * 1024 * 1024,
+		}
+	}
+}
+
 /// Block queue status
 #[derive(Debug)]
 pub struct BlockQueueInfo {
@@ -37,6 +62,12 @@ pub struct BlockQueueInfo {
 	pub verified_queue_size: usize,
 	/// Number of blocks being verified
 	pub verifying_queue_size: usize,
+	/// Configured maximum number of blocks in the queue
+	pub max_queue_size: usize,
+	/// Configured maximum number of bytes to use
+	pub max_mem_use: usize,
+	/// Heap memory used in bytes
+	pub mem_used: usize,
 }
 
 impl BlockQueueInfo {
@@ -48,7 +79,8 @@ impl BlockQueueInfo {
 
 	/// Indicates that queue is full
 	pub fn is_full(&self) -> bool {
-		self.unverified_queue_size + self.verified_queue_size + self.verifying_queue_size > MAX_UNVERIFIED_QUEUE_SIZE
+		self.unverified_queue_size + self.verified_queue_size + self.verifying_queue_size > self.max_queue_size ||
+			self.mem_used > self.max_mem_use
 	}
 
 	/// Indicates that queue is empty
@@ -68,7 +100,9 @@ pub struct BlockQueue {
 	deleting: Arc<AtomicBool>,
 	ready_signal: Arc<QueueSignal>,
 	empty: Arc<Condvar>,
-	processing: RwLock<HashSet<H256>>
+	processing: RwLock<HashSet<H256>>,
+	max_queue_size: usize,
+	max_mem_use: usize,
 }
 
 struct UnVerifiedBlock {
@@ -106,11 +140,9 @@ struct Verification {
 	bad: HashSet<H256>,
 }
 
-const MAX_UNVERIFIED_QUEUE_SIZE: usize = 50000;
-
 impl BlockQueue {
 	/// Creates a new queue instance.
-	pub fn new(engine: Arc<Box<Engine>>, message_channel: IoChannel<NetSyncMessage>) -> BlockQueue {
+	pub fn new(config: BlockQueueConfig, engine: Arc<Box<Engine>>, message_channel: IoChannel<NetSyncMessage>) -> BlockQueue {
 		let verification = Arc::new(Mutex::new(Verification::default()));
 		let more_to_verify = Arc::new(Condvar::new());
 		let ready_signal = Arc::new(QueueSignal { signalled: AtomicBool::new(false), message_channel: message_channel });
@@ -133,7 +165,7 @@ impl BlockQueue {
 				.name(format!("Verifier #{}", i))
 				.spawn(move || {
 					panic_handler.catch_panic(move || {
-					  BlockQueue::verify(verification, engine, more_to_verify, ready_signal, deleting, empty)
+						BlockQueue::verify(verification, engine, more_to_verify, ready_signal, deleting, empty)
 					}).unwrap()
 				})
 				.expect("Error starting block verification thread")
@@ -149,6 +181,8 @@ impl BlockQueue {
 			deleting: deleting.clone(),
 			processing: RwLock::new(HashSet::new()),
 			empty: empty.clone(),
+			max_queue_size: max(config.max_queue_size, MIN_QUEUE_LIMIT),
+			max_mem_use: max(config.max_mem_use, MIN_MEM_LIMIT),
 		}
 	}
 
@@ -340,8 +374,26 @@ impl BlockQueue {
 			verified_queue_size: verification.verified.len(),
 			unverified_queue_size: verification.unverified.len(),
 			verifying_queue_size: verification.verifying.len(),
+			max_queue_size: self.max_queue_size,
+			max_mem_use: self.max_mem_use,
+			mem_used:
+				verification.unverified.heap_size_of_children()
+				+ verification.verifying.heap_size_of_children()
+				+ verification.verified.heap_size_of_children(),
+				// TODO: https://github.com/servo/heapsize/pull/50
+				//+ self.processing.read().unwrap().heap_size_of_children(),
 		}
 	}
+
+	pub fn collect_garbage(&self) { 
+		{
+			let mut verification = self.verification.lock().unwrap();
+			verification.unverified.shrink_to_fit();
+			verification.verifying.shrink_to_fit();
+			verification.verified.shrink_to_fit();
+		}
+		self.processing.write().unwrap().shrink_to_fit();
+	}
 }
 
 impl MayPanic for BlockQueue {
@@ -373,7 +425,7 @@ mod tests {
 	fn get_test_queue() -> BlockQueue {
 		let spec = get_test_spec();
 		let engine = spec.to_engine().unwrap();
-		BlockQueue::new(Arc::new(engine), IoChannel::disconnected())
+		BlockQueue::new(BlockQueueConfig::default(), Arc::new(engine), IoChannel::disconnected())
 	}
 
 	#[test]
@@ -381,7 +433,7 @@ mod tests {
 		// TODO better test
 		let spec = Spec::new_test();
 		let engine = spec.to_engine().unwrap();
-		let _ = BlockQueue::new(Arc::new(engine), IoChannel::disconnected());
+		let _ = BlockQueue::new(BlockQueueConfig::default(), Arc::new(engine), IoChannel::disconnected());
 	}
 
 	#[test]
@@ -437,4 +489,19 @@ mod tests {
 
 		assert!(queue.queue_info().is_empty());
 	}
+
+	#[test]
+	fn test_mem_limit() {
+		let spec = get_test_spec();
+		let engine = spec.to_engine().unwrap();
+		let mut config = BlockQueueConfig::default();
+		config.max_mem_use = super::MIN_MEM_LIMIT;  // empty queue uses about 15000
+		let mut queue = BlockQueue::new(config, Arc::new(engine), IoChannel::disconnected());
+		assert!(!queue.queue_info().is_full());
+		let mut blocks = get_good_dummy_block_seq(50);
+		for b in blocks.drain(..) {
+			queue.import_block(b).unwrap();
+		}
+		assert!(queue.queue_info().is_full());
+	}
 }