gaspersic/solana
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Compiles/fmt and add assert for forward progress

Browse Source
This commit is contained in:
Stephen Akridge
2018-05-03 14:47:05 -07:00
parent 0aad71d46e
commit 2cdd515b12
1 changed files with 8 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
src/accountant_skel.rs
View File

@ -18,7 +18,6 @@ use result::Result;
use serde_json; use serde_json;
use signature::PublicKey; use signature::PublicKey;
use std::cmp::max; use std::cmp::max;
use std::collections::LinkedList;
use std::collections::VecDeque; use std::collections::VecDeque;
use std::io::{Cursor, Write}; use std::io::{Cursor, Write};
use std::mem::size_of; use std::mem::size_of;
@ -113,15 +112,10 @@ impl AccountantSkel {
"{}", "{}",
serde_json::to_string(&entry).unwrap() serde_json::to_string(&entry).unwrap()
).unwrap(); ).unwrap();
trace!("notify_entry_info entry");
Self::notify_entry_info_subscribers(obj, &entry); Self::notify_entry_info_subscribers(obj, &entry);
trace!("notify_entry_info done");
} }
fn receive_all<W: Write>( fn receive_all<W: Write>(obj: &SharedSkel, writer: &Arc<Mutex<W>>) -> Result<Vec<Entry>> {
obj: &SharedSkel,
writer: &Arc<Mutex<W>>,
) -> Result<Vec<Entry>> {
//TODO implement a serialize for channel that does this without allocations //TODO implement a serialize for channel that does this without allocations
let mut l = vec![]; let mut l = vec![];
let entry = obj.historian let entry = obj.historian
@ -129,16 +123,11 @@ impl AccountantSkel {
.lock() .lock()
.unwrap() .unwrap()
.recv_timeout(Duration::new(1, 0))?; .recv_timeout(Duration::new(1, 0))?;
trace!("obj.write 1 {:?}", entry);
Self::update_entry(obj, writer, &entry); Self::update_entry(obj, writer, &entry);
trace!("obj.write 1.end");
l.push(entry); l.push(entry);
while let Ok(entry) = obj.historian.receive() { while let Ok(entry) = obj.historian.receive() {
trace!("obj.write 2");
Self::update_entry(obj, writer, &entry); Self::update_entry(obj, writer, &entry);
trace!("obj.write 2.end");
l.push(entry); l.push(entry);
trace!("num: {}", num);
} }
Ok(l) Ok(l)
} }
@ -153,14 +142,13 @@ impl AccountantSkel {
exit: Arc<AtomicBool>, exit: Arc<AtomicBool>,
) -> Result<()> { ) -> Result<()> {
let mut q = VecDeque::new(); let mut q = VecDeque::new();
trace!("max: {}", max);
while let Ok(list) = Self::receive_all(&obj, writer) { while let Ok(list) = Self::receive_all(&obj, writer) {
trace!("New blobs? {}", list.len()); trace!("New blobs? {}", list.len());
let mut start = 0; let mut start = 0;
let mut end = 0; let mut end = 0;
while start < list.len() { while start < list.len() {
let total = 0; let mut total = 0;
for i in list[start..] { for i in &list[start..] {
total += size_of::<Event>() * i.events.len(); total += size_of::<Event>() * i.events.len();
total += size_of::<Entry>(); total += size_of::<Entry>();
if total >= BLOB_SIZE { if total >= BLOB_SIZE {
@ -168,11 +156,15 @@ impl AccountantSkel {
} }
end += 1; end += 1;
} }
// See that we made progress and a single
// vec of Events wasn't too big for a single packet
assert!(end > start);
let b = blob_recycler.allocate(); let b = blob_recycler.allocate();
let pos = { let pos = {
let mut bd = b.write().unwrap(); let mut bd = b.write().unwrap();
let mut out = Cursor::new(bd.data_mut()); let mut out = Cursor::new(bd.data_mut());
serialize_into(&mut out, &list[start .. end]).expect("failed to serialize output"); serialize_into(&mut out, &list[start..end])
.expect("failed to serialize output");
out.position() as usize out.position() as usize
}; };
assert!(pos < BLOB_SIZE); assert!(pos < BLOB_SIZE);