CSY/reowolf Changeset - f89fd9ceae05 · Centrum Wiskunde & Informatica (CWI)

Changeset - f89fd9ceae05

Parent rev.

Child rev.

[Not reviewed]

0 5 0

mh - 3 years ago 2022-02-08 14:38:21
contact@maxhenger.nl

Add basic binary for compilation

5 files changed with 140 insertions and 10 deletions:

bin-compiler/src/main.rs

src/lib.rs

src/protocol/mod.rs

src/runtime2/mod.rs

src/runtime2/runtime.rs

0 comments (0 inline, 0 general)

bin-compiler/src/main.rs

➞

Show inline comments

 use std::fs::File;
 use std::io::Read;
 use clap::{App, Arg};
 use reowolf_rs as rw;
 fn main() {
     let app = App::new("rwc")
         .author("Henger, M.")
         .version(env!("CARGO_PKG_VERSION"))
         .about("Reowolf compiler")
         .arg(
             Arg::new("input")
                 .long("input")
                 .short('i')
                 .help("input files")
                 .required(true)
                 .takes_value(true)
                 .multiple_occurrences(true)
+        )
         .arg(
             Arg::new("threads")
                 .long("threads")
                 .short('t')
                 .help("number of runtime threads")
                 .default_value("1")
                 .takes_value(true)
         );
     // Retrieve arguments and convert
     let app = app.get_matches();
     let input_files = app.values_of("input");
     if input_files.is_none() {
         println!("Expected at least one input file");
+        println!("ERROR: Expected at least one input file");
         return;
+    }
     let num_threads = app.value_of("threads").unwrap();
     let num_threads = match num_threads.parse::<i32>() {
         Ok(num_threads) => {
             if num_threads < 0 || num_threads > 255 {
                 println!("ERROR: Number of threads must be a number between 0 and 256");
                 return;
+            }
             num_threads as u32
         },
         Err(err) => {
             println!("ERROR: Failed to parse number of threads\nbecause: {}", err);
             return;
+        }
     };
     // Add input files to file buffer
     let input_files = input_files.unwrap();
     assert!(input_files.len() > 0); // because arg is required
     let mut builder = rw::ProtocolDescriptionBuilder::new();
     let mut file_buffer = Vec::with_capacity(4096);
     for input_file in input_files {
         println!("Got input: {}", input_file);
         print!("Adding file: {} ... ", input_file);
         let mut file = match File::open(input_file) {
             Ok(file) => file,
             Err(err) => {
                 println!("FAILED (to open file)\nbecause:\n{}", err);
                 return;
+            }
         };
         file_buffer.clear();
         if let Err(err) = file.read_to_end(&mut file_buffer) {
             println!("FAILED (to read file)\nbecause:\n{}", err);
             return;
+        }
         if let Err(err) = builder.add(input_file.to_string(), file_buffer.clone()) {
             println!("FAILED (to tokenize file)\nbecause:\n{}", err);
+        }
         println!("Success");
+    }
     // Compile the program
     print!("Compiling program ... ");
     let protocol_description = match builder.compile() {
         Ok(pd) => pd,
         Err(err) => {
             println!("FAILED\nbecause:\n{}", err);
             return;
+        }
     };
     println!("Success");
     // Make sure there is a nameless module with a main component
     print!("Creating main component ... ");
     let runtime = rw::runtime2::Runtime::new(num_threads, protocol_description);
     if let Err(err) = runtime.create_component(b"", b"main") {
         use rw::ComponentCreationError as CCE;
         let reason = match err {
             CCE::ModuleDoesntExist => "Input files did not contain a nameless module (that should contain the 'main' component)",
             CCE::DefinitionDoesntExist => "Input files did not contain a component called 'main'",
             CCE::DefinitionNotComponent => "Input file contained a 'main' function, but not a 'main' component",
             _ => "Unexpected error"
         };
         println!("FAILED\nbecause:\n{} (raw error: {:?})", reason, err);
         return;
+    }
     println!("Done!");
     println!("Success");
     println!("Now running until all components have exited");
+}
@@ \ No newline at end of file @@

src/lib.rs

➞

Show inline comments

 #[macro_use]
 mod macros;
 // mod common;
 mod protocol;
 pub mod runtime;
 pub mod runtime2;
 mod collections;
 pub use protocol::ProtocolDescription;
@@ \ No newline at end of file @@
 pub use protocol::{ProtocolDescription, ProtocolDescriptionBuilder, ComponentCreationError};
@@ \ No newline at end of file @@

src/protocol/mod.rs

➞

Show inline comments

 mod arena;
 pub(crate) mod eval;
 pub(crate) mod input_source;
 mod parser;
 #[cfg(test)] mod tests;
 pub(crate) mod ast;
 pub(crate) mod ast_printer;
 use std::sync::Mutex;
 use crate::collections::{StringPool, StringRef};
 use crate::protocol::ast::*;
 use crate::protocol::eval::*;
 use crate::protocol::input_source::*;
 use crate::protocol::parser::*;
 use crate::protocol::type_table::*;
 /// A protocol description module
 pub struct Module {
     pub(crate) source: InputSource,
     pub(crate) root_id: RootId,
     pub(crate) name: Option<StringRef<'static>>,
+}
 /// Description of a protocol object, used to configure new connectors.
 #[repr(C)]
 pub struct ProtocolDescription {
     pub(crate) modules: Vec<Module>,
     pub(crate) heap: Heap,
     pub(crate) types: TypeTable,
     pub(crate) pool: Mutex<StringPool>,
+}
 #[derive(Debug, Clone)]
 pub(crate) struct ComponentState {
     pub(crate) prompt: Prompt,
+}
 #[derive(Debug)]
 pub enum ComponentCreationError {
     ModuleDoesntExist,
     DefinitionDoesntExist,
     DefinitionNotComponent,
     InvalidNumArguments,
     InvalidArgumentType(usize),
     UnownedPort,
     InSync,
+}
 impl std::fmt::Debug for ProtocolDescription {
     fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
         write!(f, "(An opaque protocol description)")
+    }
+}
 impl ProtocolDescription {
     pub fn parse(buffer: &[u8]) -> Result<Self, String> {
         let source = InputSource::new(String::new(), Vec::from(buffer));
         let mut parser = Parser::new();
         parser.feed(source).expect("failed to feed source");
         if let Err(err) = parser.parse() {
             println!("ERROR:\n{}", err);
             return Err(format!("{}", err))
+        }
         debug_assert_eq!(parser.modules.len(), 1, "only supporting one module here for now");
         let modules: Vec<Module> = parser.modules.into_iter()
             .map(|module| Module{
                 source: module.source,
                 root_id: module.root_id,
                 name: module.name.map(|(_, name)| name)
             })
             .collect();
         return Ok(ProtocolDescription {
             modules,
             heap: parser.heap,
             types: parser.type_table,
             pool: Mutex::new(parser.string_pool),
         });
+    }
     pub(crate) fn new_component(
         &self, module_name: &[u8], identifier: &[u8], arguments: ValueGroup
     ) -> Result<Prompt, ComponentCreationError> {
         // Find the module in which the definition can be found
         let module_root = self.lookup_module_root(module_name);
         if module_root.is_none() {
             return Err(ComponentCreationError::ModuleDoesntExist);
+        }
         let module_root = module_root.unwrap();
         let root = &self.heap[module_root];
         let definition_id = root.get_definition_ident(&self.heap, identifier);
         if definition_id.is_none() {
             return Err(ComponentCreationError::DefinitionDoesntExist);
+        }
         let definition_id = definition_id.unwrap();
         let ast_definition = &self.heap[definition_id];
         if !ast_definition.is_component() {
             return Err(ComponentCreationError::DefinitionNotComponent);
+        }
         // Make sure that the types of the provided value group matches that of
         // the expected types.
         let ast_definition = ast_definition.as_component();
         if !ast_definition.poly_vars.is_empty() {
             return Err(ComponentCreationError::DefinitionNotComponent);
+        }
         // - check number of arguments by retrieving the one instantiated
         //   monomorph
         let concrete_type = ConcreteType{ parts: vec![ConcreteTypePart::Component(definition_id, 0)] };
         let mono_index = self.types.get_procedure_monomorph_index(&definition_id, &concrete_type.parts).unwrap();
         let mono_type = self.types.get_procedure_monomorph(mono_index);
         if mono_type.arg_types.len() != arguments.values.len() {
             return Err(ComponentCreationError::InvalidNumArguments);
+        }
         // - for each argument try to make sure the types match
         for arg_idx in 0..arguments.values.len() {
             let expected_type = &mono_type.arg_types[arg_idx];
             let provided_value = &arguments.values[arg_idx];
             if !self.verify_same_type(expected_type, 0, &arguments, provided_value) {
                 return Err(ComponentCreationError::InvalidArgumentType(arg_idx));
+            }
+        }
         // By now we're sure that all of the arguments are correct. So create
         // the connector.
         return Ok(Prompt::new(&self.types, &self.heap, definition_id, mono_index, arguments));
+    }
     fn lookup_module_root(&self, module_name: &[u8]) -> Option<RootId> {
         for module in self.modules.iter() {
             match &module.name {
                 Some(name) => if name.as_bytes() == module_name {
                     return Some(module.root_id);
                 },
                 None => if module_name.is_empty() {
                     return Some(module.root_id);
+                }
+            }
+        }
         return None;
+    }
     fn verify_same_type(&self, expected: &ConcreteType, expected_idx: usize, arguments: &ValueGroup, argument: &Value) -> bool {
         use ConcreteTypePart as CTP;
         match &expected.parts[expected_idx] {
             CTP::Void | CTP::Message | CTP::Slice | CTP::Function(_, _) | CTP::Component(_, _) => unreachable!(),
             CTP::Bool => if let Value::Bool(_) = argument { true } else { false },
             CTP::UInt8 => if let Value::UInt8(_) = argument { true } else { false },
             CTP::UInt16 => if let Value::UInt16(_) = argument { true } else { false },
             CTP::UInt32 => if let Value::UInt32(_) = argument { true } else { false },
             CTP::UInt64 => if let Value::UInt64(_) = argument { true } else { false },
             CTP::SInt8 => if let Value::SInt8(_) = argument { true } else { false },
             CTP::SInt16 => if let Value::SInt16(_) = argument { true } else { false },
             CTP::SInt32 => if let Value::SInt32(_) = argument { true } else { false },
             CTP::SInt64 => if let Value::SInt64(_) = argument { true } else { false },
             CTP::Character => if let Value::Char(_) = argument { true } else { false },
             CTP::String => {
                 // Match outer string type and embedded character types
                 if let Value::String(heap_pos) = argument {
                     for element in &arguments.regions[*heap_pos as usize] {
                         if let Value::Char(_) = element {} else {
                             return false;
+                        }
+                    }
                 } else {
                     return false;
+                }
                 return true;
             },
             CTP::Array => {
                 if let Value::Array(heap_pos) = argument {
                     let heap_pos = *heap_pos;
                     for element in &arguments.regions[heap_pos as usize] {
                         if !self.verify_same_type(expected, expected_idx + 1, arguments, element) {
                             return false;
+                        }
+                    }
                     return true;
                 } else {
                     return false;
+                }
             },
             CTP::Input => if let Value::Input(_) = argument { true } else { false },
             CTP::Output => if let Value::Output(_) = argument { true } else { false },
             CTP::Tuple(_) => todo!("implement full type checking on user-supplied arguments"),
             CTP::Instance(definition_id, _num_embedded) => {
                 let definition = self.types.get_base_definition(definition_id).unwrap();
                 match &definition.definition {
                     DefinedTypeVariant::Enum(definition) => {
                         if let Value::Enum(variant_value) = argument {
                             let is_valid = definition.variants.iter()
                                 .any(|v| v.value == *variant_value);
                             return is_valid;
+                        }
                     },
                     _ => todo!("implement full type checking on user-supplied arguments"),
+                }
                 return false;
             },
+        }
+    }
+}
 pub trait RunContext {
     fn performed_put(&mut self, port: PortId) -> bool;
     fn performed_get(&mut self, port: PortId) -> Option<ValueGroup>; // None if still waiting on message
     fn fires(&mut self, port: PortId) -> Option<Value>; // None if not yet branched
     fn performed_fork(&mut self) -> Option<bool>; // None if not yet forked
     fn created_channel(&mut self) -> Option<(Value, Value)>; // None if not yet prepared
+}
 pub struct ProtocolDescriptionBuilder {
     parser: Parser,
+}
 impl ProtocolDescriptionBuilder {
     pub fn new() -> Self {
         return Self{
             parser: Parser::new(),
+        }
+    }
     pub fn add(&mut self, filename: String, buffer: Vec<u8>) -> Result<(), ParseError> {
         let input = InputSource::new(filename, buffer);
         self.parser.feed(input)?;
         return Ok(())
+    }
     pub fn compile(mut self) -> Result<ProtocolDescription, ParseError> {
         self.parser.parse()?;
         let modules: Vec<Module> = self.parser.modules.into_iter()
             .map(|module| Module{
                 source: module.source,
                 root_id: module.root_id,
                 name: module.name.map(|(_, name)| name)
             })
             .collect();
         return Ok(ProtocolDescription {
             modules,
             heap: self.parser.heap,
             types: self.parser.type_table,
             pool: Mutex::new(self.parser.string_pool),
         });
+    }
+}

src/runtime2/mod.rs

➞

Show inline comments

 mod store;
 mod runtime;
 mod component;
 mod communication;
 mod scheduler;
 #[cfg(test)] mod tests;
@@ \ No newline at end of file @@
 #[cfg(test)] mod tests;
 pub use runtime::Runtime;
@@ \ No newline at end of file @@

src/runtime2/runtime.rs

➞

Show inline comments

 use std::sync::{Arc, Mutex, Condvar};
 use std::sync::atomic::{AtomicU32, AtomicBool, Ordering};
 use std::collections::VecDeque;
 use crate::protocol::*;
 use super::communication::Message;
 use super::component::{wake_up_if_sleeping, CompPDL, CompCtx};
 use super::store::{ComponentStore, ComponentReservation, QueueDynMpsc, QueueDynProducer};
 use super::scheduler::*;
 // -----------------------------------------------------------------------------
 // Component
 // -----------------------------------------------------------------------------
 /// Key to a component. Type system somewhat ensures that there can only be one
 /// of these. Only with a key one may retrieve privately-accessible memory for
 /// a component. Practically just a generational index, like `CompId` is.
 #[derive(Debug, Copy, Clone, PartialEq, Eq)]
 pub(crate) struct CompKey(pub u32);
 impl CompKey {
     pub(crate) fn downgrade(&self) -> CompId {
         return CompId(self.0);
+    }
+}
 /// Generational ID of a component
 #[derive(Debug, Copy, Clone, PartialEq, Eq)]
 pub struct CompId(pub u32);
 impl CompId {
     pub(crate) fn new_invalid() -> CompId {
         return CompId(u32::MAX);
+    }
     /// Upgrade component ID to component key. Unsafe because the caller needs
     /// to make sure that only one component key can exist at a time (to ensure
     /// a component can only be scheduled/executed by one thread).
     pub(crate) unsafe fn upgrade(&self) -> CompKey {
         return CompKey(self.0);
+    }
+}
 /// Handle to a component that is being created.
 pub(crate) struct CompReserved {
     reservation: ComponentReservation,
+}
 impl CompReserved {
     pub(crate) fn id(&self) -> CompId {
         return CompId(self.reservation.index)
+    }
+}
 /// Private fields of a component, may only be modified by a single thread at
 /// a time.
 pub(crate) struct RuntimeComp {
     pub public: CompPublic,
     pub code: CompPDL,
     pub ctx: CompCtx,
     pub inbox: QueueDynMpsc<Message>,
     pub exiting: bool,
+}
 /// Should contain everything that is accessible in a thread-safe manner
 // TODO: Do something about the `num_handles` thing. This needs to be a bit more
 //  "foolproof" to lighten the mental burden of using the `num_handles`
 //  variable.
 pub(crate) struct CompPublic {
     pub sleeping: AtomicBool,
     pub num_handles: AtomicU32, // manually modified (!)
     inbox: QueueDynProducer<Message>,
+}
 /// Handle to public part of a component. Would be nice if we could
 /// automagically manage the `num_handles` counter. But when it reaches zero we
 /// need to manually remove the handle from the runtime. So we just have debug
 /// code to make sure this actually happens.
 pub(crate) struct CompHandle {
     target: *const CompPublic,
     id: CompId, // TODO: @Remove after debugging
     #[cfg(debug_assertions)] decremented: bool,
+}
 impl CompHandle {
     fn new(id: CompId, public: &CompPublic) -> CompHandle {
         let handle = CompHandle{
             target: public,
             id,
             #[cfg(debug_assertions)] decremented: false,
         };
         handle.increment_users();
         return handle;
+    }
     pub(crate) fn send_message(&self, sched_ctx: &SchedulerCtx, message: Message, try_wake_up: bool) {
         sched_ctx.log(&format!("Sending message to [c:{:03}, wakeup:{}]: {:?}", self.id.0, try_wake_up, message));
         self.inbox.push(message);
         if try_wake_up {
             wake_up_if_sleeping(sched_ctx, self.id, self);
+        }
+    }
     fn increment_users(&self) {
         let old_count = self.num_handles.fetch_add(1, Ordering::AcqRel);
         debug_assert!(old_count > 0); // because we should never be able to retrieve a handle when the component is (being) destroyed
+    }
     /// Returns the `CompKey` to the component if it should be destroyed
     pub(crate) fn decrement_users(&mut self) -> Option<CompKey> {
         debug_assert!(!self.decremented, "illegal to 'decrement_users' twice");
         let old_count = self.num_handles.fetch_sub(1, Ordering::AcqRel);
         let new_count = old_count - 1;
         dbg_code!(self.decremented = true);
         if new_count == 0 {
             return Some(unsafe{ self.id.upgrade() });
+        }
         return None;
+    }
+}
 impl Clone for CompHandle {
     fn clone(&self) -> Self {
         debug_assert!(!self.decremented, "illegal to clone after 'decrement_users'");
         self.increment_users();
         return CompHandle{
             target: self.target,
             id: self.id,
             #[cfg(debug_assertions)] decremented: false,
         };
+    }
+}
 impl std::ops::Deref for CompHandle {
     type Target = CompPublic;
     fn deref(&self) -> &Self::Target {
         debug_assert!(!self.decremented); // cannot access if control is relinquished
         return unsafe{ &*self.target };
+    }
+}
 impl Drop for CompHandle {
     fn drop(&mut self) {
         debug_assert!(self.decremented, "need call to 'decrement_users' before dropping");
+    }
+}
 // -----------------------------------------------------------------------------
 // Runtime
 // -----------------------------------------------------------------------------
 pub struct Runtime {
     pub(crate) inner: Arc<RuntimeInner>,
     threads: Vec<std::thread::JoinHandle<()>>,
+}
 impl Runtime {
     pub fn new(num_threads: u32, protocol_description: ProtocolDescription) -> Runtime {
         assert!(num_threads > 0, "need a thread to perform work");
         let runtime_inner = Arc::new(RuntimeInner {
             protocol: protocol_description,
             components: ComponentStore::new(128),
             work_queue: Mutex::new(VecDeque::with_capacity(128)),
             work_condvar: Condvar::new(),
             active_elements: AtomicU32::new(1),
         });
         let mut runtime = Runtime {
             inner: runtime_inner,
             threads: Vec::with_capacity(num_threads as usize),
         };
         for thread_index in 0..num_threads {
             let mut scheduler = Scheduler::new(runtime.inner.clone(), thread_index);
             let thread_handle = std::thread::spawn(move || {
                 scheduler.run();
             });
             runtime.threads.push(thread_handle);
+        }
         return runtime;
+    }
     pub fn create_component(&self, module_name: &[u8], routine_name: &[u8]) -> Result<(), ComponentCreationError> {
         use crate::protocol::eval::ValueGroup;
         let prompt = self.inner.protocol.new_component(
             module_name, routine_name,
             ValueGroup::new_stack(Vec::new())
         )?;
         let reserved = self.inner.start_create_pdl_component();
         let ctx = CompCtx::new(&reserved);
         let (key, _) = self.inner.finish_create_pdl_component(reserved, CompPDL::new(prompt, 0), ctx, false);
         self.inner.enqueue_work(key);
         return Ok(())
+    }
+}
 impl Drop for Runtime {
     fn drop(&mut self) {
         self.inner.decrement_active_components();
         for handle in self.threads.drain(..) {
             handle.join().expect("join scheduler thread");
+        }
+    }
+}
 /// Memory that is maintained by "the runtime". In practice it is maintained by
 /// multiple schedulers, and this serves as the common interface to that memory.
 pub(crate) struct RuntimeInner {
     pub protocol: ProtocolDescription,
     components: ComponentStore<RuntimeComp>,
     work_queue: Mutex<VecDeque<CompKey>>,
     work_condvar: Condvar,
     active_elements: AtomicU32, // active components and APIs (i.e. component creators)
+}
 impl RuntimeInner {
     // Scheduling and retrieving work
     pub(crate) fn take_work(&self) -> Option<CompKey> {
         let mut lock = self.work_queue.lock().unwrap();
         while lock.is_empty() && self.active_elements.load(Ordering::Acquire) != 0 {
             lock = self.work_condvar.wait(lock).unwrap();
+        }
         // We have work, or the schedulers should exit.
         return lock.pop_front();
+    }
     pub(crate) fn enqueue_work(&self, key: CompKey) {
         let mut lock = self.work_queue.lock().unwrap();
         lock.push_back(key);
         self.work_condvar.notify_one();
+    }
     // Creating/destroying components
     pub(crate) fn start_create_pdl_component(&self) -> CompReserved {
         self.increment_active_components();
         let reservation = self.components.reserve();
         return CompReserved{ reservation };
+    }
     pub(crate) fn finish_create_pdl_component(
         &self, reserved: CompReserved,
         component: CompPDL, mut context: CompCtx, initially_sleeping: bool,
     ) -> (CompKey, &mut RuntimeComp) {
         let inbox_queue = QueueDynMpsc::new(16);
         let inbox_producer = inbox_queue.producer();
         let _id = reserved.id();
         context.id = reserved.id();
         let component = RuntimeComp {
             public: CompPublic{
                 sleeping: AtomicBool::new(initially_sleeping),
                 num_handles: AtomicU32::new(1), // the component itself acts like a handle
                 inbox: inbox_producer,
             },
             code: component,
             ctx: context,
             inbox: inbox_queue,
             exiting: false,
         };
         let index = self.components.submit(reserved.reservation, component);
         debug_assert_eq!(index, _id.0);
         let component = self.components.get_mut(index);
         return (CompKey(index), component);
+    }
     /// Creates a new component. Note that the public part will be properly
     /// initialized, but not all private fields are.
     pub(crate) fn create_pdl_component(&self, comp: CompPDL, ctx: CompCtx, initially_sleeping: bool) -> (CompKey, &mut RuntimeComp) {
         let inbox_queue = QueueDynMpsc::new(16);
         let inbox_producer = inbox_queue.producer();
         let comp = RuntimeComp{
             public: CompPublic{
                 sleeping: AtomicBool::new(initially_sleeping),
                 num_handles: AtomicU32::new(1), // the component itself acts like a handle
                 inbox: inbox_producer,
             },
             code: comp,
             ctx,
             inbox: inbox_queue,
             exiting: false,
         };
         let index = self.components.create(comp);
         // TODO: just do a reserve_index followed by a consume_index or something
         self.increment_active_components();
         let component = self.components.get_mut(index);
         component.ctx.id = CompId(index);
         return (CompKey(index), component);
+    }
     pub(crate) fn get_component(&self, key: CompKey) -> &mut RuntimeComp {
         let component = self.components.get_mut(key.0);
         return component;
+    }
     pub(crate) fn get_component_public(&self, id: CompId) -> CompHandle {
         let component = self.components.get(id.0);
         return CompHandle::new(id, &component.public);
+    }
     pub(crate) fn destroy_component(&self, key: CompKey) {
         dbg_code!({
             let component = self.get_component(key);
             debug_assert!(component.exiting);
             debug_assert_eq!(component.public.num_handles.load(Ordering::Acquire), 0);
         });
         self.decrement_active_components();
         self.components.destroy(key.0);
+    }
     // Tracking number of active interfaces and the active components
     #[inline]
     fn increment_active_components(&self) {
         let _old_val = self.active_elements.fetch_add(1, Ordering::AcqRel);
         debug_assert!(_old_val > 0); // can only create a component from a API/component, so can never be 0.
+    }
     fn decrement_active_components(&self) {
         let old_val = self.active_elements.fetch_sub(1, Ordering::AcqRel);
         debug_assert!(old_val > 0); // something wrong with incr/decr logic
         let new_val = old_val - 1;
         if new_val == 0 {
             // Just to be sure, in case the last thing that gets destroyed is an
             // API instead of a thread.
             let _lock = self.work_queue.lock();
             self.work_condvar.notify_all();
+        }
+    }
+}

0 comments (0 inline, 0 general)