Boot EntryWriter's Mutex

Finally!
This commit is contained in:
Greg Fitzgerald 2018-07-01 14:31:13 -06:00 committed by Greg Fitzgerald
parent 4f3a291391
commit 669164bada
3 changed files with 16 additions and 18 deletions

View File

@ -10,7 +10,6 @@ use solana::mint::Mint;
use std::error; use std::error;
use std::io::{stdin, stdout, Read}; use std::io::{stdin, stdout, Read};
use std::process::exit; use std::process::exit;
use std::sync::Mutex;
fn main() -> Result<(), Box<error::Error>> { fn main() -> Result<(), Box<error::Error>> {
if is(Stream::Stdin) { if is(Stream::Stdin) {
@ -26,7 +25,7 @@ fn main() -> Result<(), Box<error::Error>> {
} }
let mint: Mint = serde_json::from_str(&buffer)?; let mint: Mint = serde_json::from_str(&buffer)?;
let writer = Mutex::new(stdout()); let mut writer = stdout();
EntryWriter::write_entries(&writer, &mint.create_entries())?; EntryWriter::write_entries(&mut writer, &mint.create_entries())?;
Ok(()) Ok(())
} }

View File

@ -6,40 +6,39 @@ use bank::Bank;
use entry::Entry; use entry::Entry;
use serde_json; use serde_json;
use std::io::{self, Write}; use std::io::{self, Write};
use std::sync::Mutex;
pub struct EntryWriter<'a, W> { pub struct EntryWriter<'a, W> {
bank: &'a Bank, bank: &'a Bank,
writer: Mutex<W>, writer: W,
} }
impl<'a, W: Write> EntryWriter<'a, W> { impl<'a, W: Write> EntryWriter<'a, W> {
/// Create a new Tpu that wraps the given Bank. /// Create a new Tpu that wraps the given Bank.
pub fn new(bank: &'a Bank, writer: Mutex<W>) -> Self { pub fn new(bank: &'a Bank, writer: W) -> Self {
EntryWriter { bank, writer } EntryWriter { bank, writer }
} }
fn write_entry(writer: &Mutex<W>, entry: &Entry) -> io::Result<()> { fn write_entry(writer: &mut W, entry: &Entry) -> io::Result<()> {
let serialized = serde_json::to_string(&entry).unwrap(); let serialized = serde_json::to_string(&entry).unwrap();
writeln!(writer.lock().unwrap(), "{}", serialized) writeln!(writer, "{}", serialized)
} }
pub fn write_entries(writer: &Mutex<W>, entries: &[Entry]) -> io::Result<()> { pub fn write_entries(writer: &mut W, entries: &[Entry]) -> io::Result<()> {
for entry in entries { for entry in entries {
Self::write_entry(writer, entry)?; Self::write_entry(writer, entry)?;
} }
Ok(()) Ok(())
} }
fn write_and_register_entry(&self, entry: &Entry) -> io::Result<()> { fn write_and_register_entry(&mut self, entry: &Entry) -> io::Result<()> {
trace!("write_and_register_entry entry"); trace!("write_and_register_entry entry");
if !entry.has_more { if !entry.has_more {
self.bank.register_entry_id(&entry.id); self.bank.register_entry_id(&entry.id);
} }
Self::write_entry(&self.writer, entry) Self::write_entry(&mut self.writer, entry)
} }
pub fn write_and_register_entries(&self, entries: &[Entry]) -> io::Result<()> { pub fn write_and_register_entries(&mut self, entries: &[Entry]) -> io::Result<()> {
for entry in entries { for entry in entries {
self.write_and_register_entry(&entry)?; self.write_and_register_entry(&entry)?;
} }
@ -61,8 +60,8 @@ mod tests {
let mint = Mint::new(1); let mint = Mint::new(1);
let bank = Bank::new(&mint); let bank = Bank::new(&mint);
let writer = Mutex::new(io::sink()); let writer = io::sink();
let entry_writer = EntryWriter::new(&bank, writer); let mut entry_writer = EntryWriter::new(&bank, writer);
let keypair = KeyPair::new(); let keypair = KeyPair::new();
let tx = Transaction::new(&mint.keypair(), keypair.pubkey(), 1, mint.last_id()); let tx = Transaction::new(&mint.keypair(), keypair.pubkey(), 1, mint.last_id());

View File

@ -12,7 +12,7 @@ use std::collections::VecDeque;
use std::io::Write; use std::io::Write;
use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::mpsc::{channel, Receiver}; use std::sync::mpsc::{channel, Receiver};
use std::sync::{Arc, Mutex}; use std::sync::Arc;
use std::thread::{Builder, JoinHandle}; use std::thread::{Builder, JoinHandle};
use std::time::Duration; use std::time::Duration;
use streamer::{BlobReceiver, BlobSender}; use streamer::{BlobReceiver, BlobSender};
@ -26,7 +26,7 @@ impl WriteStage {
/// Process any Entry items that have been published by the Historian. /// Process any Entry items that have been published by the Historian.
/// continuosly broadcast blobs of entries out /// continuosly broadcast blobs of entries out
pub fn write_and_send_entries<W: Write>( pub fn write_and_send_entries<W: Write>(
entry_writer: &EntryWriter<W>, entry_writer: &mut EntryWriter<W>,
blob_sender: &BlobSender, blob_sender: &BlobSender,
blob_recycler: &BlobRecycler, blob_recycler: &BlobRecycler,
entry_receiver: &Receiver<Vec<Entry>>, entry_receiver: &Receiver<Vec<Entry>>,
@ -55,10 +55,10 @@ impl WriteStage {
let thread_hdl = Builder::new() let thread_hdl = Builder::new()
.name("solana-writer".to_string()) .name("solana-writer".to_string())
.spawn(move || { .spawn(move || {
let entry_writer = EntryWriter::new(&bank, Mutex::new(writer)); let mut entry_writer = EntryWriter::new(&bank, writer);
loop { loop {
let _ = Self::write_and_send_entries( let _ = Self::write_and_send_entries(
&entry_writer, &mut entry_writer,
&blob_sender, &blob_sender,
&blob_recycler, &blob_recycler,
&entry_receiver, &entry_receiver,