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

Changeset - 9a4f36415198

Parent rev.

Child rev.

[Not reviewed]

0 3 0

MH - 3 years ago 2022-05-14 10:27:09
contact@maxhenger.nl

WIP on TCP listener component

3 files changed with 286 insertions and 24 deletions:

src/protocol/mod.rs

src/runtime2/component/component_internet.rs

274

std/std.internet.pdl

0 comments (0 inline, 0 general)

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_writer;
 mod token_writer;
 use std::sync::Mutex;
 use crate::collections::{StringPool, StringRef};
 pub use crate::protocol::ast::*;
 use crate::protocol::eval::*;
 use crate::protocol::input_source::*;
 use crate::protocol::parser::*;
 use crate::protocol::type_table::*;
 pub use parser::type_table::TypeId;
 /// 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 ProtocolDescription {
     pub fn parse(buffer: &[u8]) -> Result<Self, String> {
         let source = InputSource::new(String::new(), Vec::from(buffer));
         let mut parser = Parser::new(None)?;
         parser.feed(source).expect("failed to feed source");
         if let Err(err) = parser.parse() {
             println!("ERROR:\n{}", err);
             return Err(format!("{}", err))
+        }
         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_by_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_procedure() {
             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_procedure();
         if !ast_definition.poly_vars.is_empty() || ast_definition.kind == ProcedureKind::Function {
             return Err(ComponentCreationError::DefinitionNotComponent);
+        }
         // - check number of arguments by retrieving the one instantiated
         //   monomorph
         let concrete_type = ConcreteType{ parts: vec![ConcreteTypePart::Component(ast_definition.this, 0)] };
         let procedure_type_id = self.types.get_monomorph_type_id(&definition_id, &concrete_type.parts).unwrap();
         let procedure_monomorph_index = self.types.get_monomorph(procedure_type_id).variant.as_procedure().monomorph_index;
         let monomorph_info = &ast_definition.monomorphs[procedure_monomorph_index as usize];
         if monomorph_info.argument_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_id = monomorph_info.argument_types[arg_idx];
             let expected_type = &self.types.get_monomorph(expected_type_id).concrete_type;
             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, ast_definition.this, procedure_type_id, arguments));
+    }
     /// A somewhat temporary method. Can be used by components to lookup type
     /// definitions by their name (to have their implementation somewhat
     /// resistant to changes in the standard library)
     pub(crate) fn find_type<'a>(&'a self, module_name: &[u8], type_name: &[u8]) -> Option<TypeInspector<'a>> {
         // Lookup type definition in module
         let root_id = self.lookup_module_root(module_name)?;
         let module = &self.heap[root_id];
         let definition_id = module.get_definition_by_ident(&self.heap, type_name)?;
         let definition = &self.heap[definition_id];
         // Make sure type is not polymorphic and is not a procedure
         if !definition.poly_vars().is_empty() {
             return None;
+        }
         if definition.is_procedure() {
             return None;
+        }
         // Lookup type in type table
         let type_parts = [ConcreteTypePart::Instance(definition_id, 0)];
         let type_id = self.types.get_monomorph_type_id(&definition_id, &type_parts)
             .expect("type ID for non-polymorphic type");
         let type_monomorph = self.types.get_monomorph(type_id);
         return Some(TypeInspector{
             heap: definition,
             type_table: type_monomorph
         });
+    }
     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::Pointer | 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
     fn performed_select_wait(&mut self) -> Option<u32>; // None if not yet notified runtime of select blocker
+}
 pub struct ProtocolDescriptionBuilder {
     parser: Parser,
+}
 impl ProtocolDescriptionBuilder {
     pub fn new(std_lib_dir: Option<String>) -> Result<Self, String> {
         let mut parser = Parser::new(std_lib_dir)?;
         return Ok(Self{ parser })
+    }
     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),
         });
+    }
+}
 pub struct TypeInspector<'a> {
     heap: &'a Definition,
     type_table: &'a MonoType,
+}
 impl<'a> TypeInspector<'a> {
     pub fn as_union(&'a self) -> UnionTypeInspector<'a> {
         let heap = self.heap.as_union();
         let type_table = self.type_table.variant.as_union();
         return UnionTypeInspector{ heap, type_table };
+    }
     pub fn as_struct(&'a self) -> StructTypeInspector<'a> {
         let heap = self.heap.as_struct();
         let type_table = self.type_table.variant.as_struct();
         return StructTypeInspector{ heap, type_table };
+    }
+}
 pub struct UnionTypeInspector<'a> {
     heap: &'a UnionDefinition,
     type_table: &'a UnionMonomorph,
+}
 impl UnionTypeInspector<'_> {
     /// Retrieves union variant tag value.
     pub fn get_variant_tag_value(&self, variant_name: &[u8]) -> Option<i64> {
         let variant_index = self.heap.variants.iter()
             .position(|v| v.identifier.value.as_bytes() == variant_name)?;
         return Some(variant_index as i64);
+    }
+}
 pub struct StructTypeInspector<'a> {
     heap: &'a StructDefinition,
     type_table: &'a StructMonomorph,
+}
 impl StructTypeInspector<'_> {
     /// Retrieves number of struct fields
     pub fn get_num_struct_fields(&self) -> usize {
         return self.heap.fields.len();
+    }
     /// Retrieves struct field index
     pub fn get_struct_field_index(&self, field_name: &[u8]) -> Option<usize> {
         let field_index = self.heap.fields.iter()
             .position(|v| v.field.value.as_bytes() == field_name)?;
         return Some(field_index);
+    }
+}
@@ \ No newline at end of file @@

src/runtime2/component/component_internet.rs

➞

Show inline comments

 use crate::protocol::eval::{ValueGroup, Value};
 use crate::runtime2::*;
 use crate::runtime2::component::{CompCtx, CompId, PortInstruction};
 use crate::runtime2::stdlib::internet::*;
 use crate::runtime2::poll::*;
 use super::component::{self, *};
 use super::control_layer::*;
 use super::consensus::*;
 use std::io::ErrorKind as IoErrorKind;
 use std::net::{IpAddr, Ipv4Addr};
 // -----------------------------------------------------------------------------
 // ComponentTcpClient
 // -----------------------------------------------------------------------------
 enum ClientSocketState {
     Connected(SocketTcpClient),
     Error,
+}
 impl ClientSocketState {
     fn get_socket(&self) -> &SocketTcpClient {
         match self {
             ClientSocketState::Connected(v) => v,
             ClientSocketState::Error => unreachable!(),
+        }
+    }
+}
 /// States from the point of view of the component that is connecting to this
 /// TCP component (i.e. from the point of view of attempting to interface with
 /// a socket).
 #[derive(PartialEq, Debug)]
 enum ClientSyncState {
     AwaitingCmd,
     Getting,
     Putting,
     FinishSync,
     FinishSyncThenQuit,
+}
 pub struct ComponentTcpClient {
     // Properties for the tcp socket
     socket_state: ClientSocketState,
     sync_state: ClientSyncState,
     poll_ticket: Option<PollTicket>,
     inbox_main: InboxMain,
     inbox_backup: InboxBackup,
     pdl_input_port_id: PortId, // input from PDL, so transmitted over socket
     pdl_output_port_id: PortId, // output towards PDL, so received over socket
     // Information about union tags, extracted from PDL
     input_union_send_tag_value: i64,
     input_union_receive_tag_value: i64,
     input_union_finish_tag_value: i64,
     input_union_shutdown_tag_value: i64,
     // Generic component state
     exec_state: CompExecState,
     control: ControlLayer,
     consensus: Consensus,
     // Temporary variables
     byte_buffer: Vec<u8>,
+}
 impl Component for ComponentTcpClient {
     fn on_creation(&mut self, id: CompId, sched_ctx: &SchedulerCtx) {
         // Retrieve type information for messages we're going to receive
         let pd = &sched_ctx.runtime.protocol;
         let cmd_type = pd.find_type(b"std.internet", b"Cmd")
             .expect("'Cmd' type in the 'std.internet' module");
         let cmd_type = pd.find_type(b"std.internet", b"ClientCmd")
             .expect("'ClientCmd' type in the 'std.internet' module");
         let cmd_type = cmd_type
             .as_union();
         self.input_union_send_tag_value = cmd_type.get_variant_tag_value(b"Send").unwrap();
         self.input_union_receive_tag_value = cmd_type.get_variant_tag_value(b"Receive").unwrap();
         self.input_union_finish_tag_value = cmd_type.get_variant_tag_value(b"Finish").unwrap();
         self.input_union_shutdown_tag_value = cmd_type.get_variant_tag_value(b"Shutdown").unwrap();
         // Register socket for async events
         if let ClientSocketState::Connected(socket) = &self.socket_state {
             let self_handle = sched_ctx.runtime.get_component_public(id);
             let poll_ticket = sched_ctx.polling.register(socket, self_handle, true, true)
-                .expect("registering tcp component");
+                .expect("registering tcp client");
             debug_assert!(self.poll_ticket.is_none());
             self.poll_ticket = Some(poll_ticket);
+        }
+    }
     fn on_shutdown(&mut self, sched_ctx: &SchedulerCtx) {
         if let Some(poll_ticket) = self.poll_ticket.take() {
             sched_ctx.polling.unregister(poll_ticket)
                 .expect("unregistering tcp component");
+        }
+    }
     fn adopt_message(&mut self, _comp_ctx: &mut CompCtx, message: DataMessage) {
         let slot = &mut self.inbox_main[0];
         if slot.is_none() {
             *slot = Some(message);
         } else {
             self.inbox_backup.push(message);
+        }
+    }
     fn handle_message(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx, message: Message) {
         match message {
             Message::Data(message) => {
                 self.handle_incoming_data_message(sched_ctx, comp_ctx, message);
             },
             Message::Sync(message) => {
                 let decision = self.consensus.receive_sync_message(sched_ctx, comp_ctx, message);
                 component::default_handle_sync_decision(sched_ctx, &mut self.exec_state, comp_ctx, decision, &mut self.consensus);
             },
             Message::Control(message) => {
                 if let Err(location_and_message) = component::default_handle_control_message(
                     &mut self.exec_state, &mut self.control, &mut self.consensus,
                     message, sched_ctx, comp_ctx, &mut self.inbox_main, &mut self.inbox_backup
                 ) {
                     component::default_handle_error_for_builtin(&mut self.exec_state, sched_ctx, location_and_message);
+                }
             },
             Message::Poll => {
                 sched_ctx.info("Received polling event");
             },
+        }
+    }
     fn run(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx) -> CompScheduling {
         sched_ctx.info(&format!("Running component ComponentTcpClient (mode: {:?}, sync state: {:?})", self.exec_state.mode, self.sync_state));
         match self.exec_state.mode {
             CompMode::BlockedSelect |
             CompMode::PutPortsBlockedTransferredPorts |
             CompMode::PutPortsBlockedAwaitingAcks |
             CompMode::PutPortsBlockedSendingPort |
             CompMode::NewComponentBlocked => {
                 // Not possible: we never enter this state
                 unreachable!();
             },
             CompMode::NonSync => {
                 // When in non-sync mode
-                match &mut self.socket_state {
                 match &self.socket_state {
                     ClientSocketState::Connected(_socket) => {
                         if self.sync_state == ClientSyncState::FinishSyncThenQuit {
                             // Previous request was to let the component shut down
                             self.exec_state.set_as_start_exit(ExitReason::Termination);
                         } else {
                             // Reset for a new request
                             self.sync_state = ClientSyncState::AwaitingCmd;
                             component::default_handle_sync_start(
                                 &mut self.exec_state, &mut self.inbox_main, sched_ctx, comp_ctx, &mut self.consensus
                             );
+                        }
                         return CompScheduling::Immediate;
                     },
                     ClientSocketState::Error => {
                         // Could potentially send an error message to the
                         // connected component.
                         self.exec_state.set_as_start_exit(ExitReason::ErrorNonSync);
                         return CompScheduling::Immediate;
+                    }
+                }
             },
             CompMode::Sync => {
                 // When in sync mode: wait for a command to come in
                 match self.sync_state {
                     ClientSyncState::AwaitingCmd => {
                         match component::default_attempt_get(
                             &mut self.exec_state, self.pdl_input_port_id, PortInstruction::NoSource,
                             &mut self.inbox_main, &mut self.inbox_backup, sched_ctx, comp_ctx,
                             &mut self.control, &mut self.consensus
                         ) {
                             GetResult::Received(message) => {
                                 let (tag_value, embedded_heap_pos) = message.content.values[0].as_union();
                                 if tag_value == self.input_union_send_tag_value {
                                     // Retrieve bytes from the message
                                     self.byte_buffer.clear();
                                     let union_content = &message.content.regions[embedded_heap_pos as usize];
                                     debug_assert_eq!(union_content.len(), 1);
                                     let array_heap_pos = union_content[0].as_array();
                                     let array_values = &message.content.regions[array_heap_pos as usize];
                                     self.byte_buffer.reserve(array_values.len());
                                     for value in array_values {
                                         self.byte_buffer.push(value.as_uint8());
+                                    }
                                     self.sync_state = ClientSyncState::Putting;
                                 } else if tag_value == self.input_union_receive_tag_value {
                                     // Component requires a `recv`
                                     self.sync_state = ClientSyncState::Getting;
                                 } else if tag_value == self.input_union_finish_tag_value {
                                     // Component requires us to end the sync round
                                     self.sync_state = ClientSyncState::FinishSync;
                                 } else if tag_value == self.input_union_shutdown_tag_value {
                                     // Component wants to close the connection
                                     self.sync_state = ClientSyncState::FinishSyncThenQuit;
                                 } else {
                                     unreachable!("got tag_value {}", tag_value)
+                                }
                                 return CompScheduling::Immediate;
                             },
                             GetResult::NoMessage => {
                                 return CompScheduling::Sleep;
                             },
                             GetResult::Error(location_and_message) => {
                                 component::default_handle_error_for_builtin(&mut self.exec_state, sched_ctx, location_and_message);
                                 return CompScheduling::Immediate;
+                            }
+                        }
                     },
                     ClientSyncState::Putting => {
                         // We're supposed to send a user-supplied message fully
                         // over the socket. But we might end up blocking. In
                         // that case the component goes to sleep until it is
                         // polled.
                         let socket = self.socket_state.get_socket();
                         while !self.byte_buffer.is_empty() {
                             match socket.send(&self.byte_buffer) {
                                 Ok(bytes_sent) => {
                                     self.byte_buffer.drain(..bytes_sent);
                                 },
                                 Err(err) => {
                                     if err.kind() == IoErrorKind::WouldBlock {
                                         return CompScheduling::Sleep; // wait until notified
                                     } else {
                                         todo!("handle socket.send error {:?}", err)
+                                    }
+                                }
+                            }
+                        }
                         // If here then we're done putting the data, we can
                         // finish the sync round
                         component::default_handle_sync_end(&mut self.exec_state, sched_ctx, comp_ctx, &mut self.consensus);
                         return CompScheduling::Requeue;
                     },
                     ClientSyncState::Getting => {
                         // We're going to try and receive a single message. If
                         // this causes us to end up blocking the component
                         // goes to sleep until it is polled.
                         const BUFFER_SIZE: usize = 1024; // TODO: Move to config
                         let socket = self.socket_state.get_socket();
                         self.byte_buffer.resize(BUFFER_SIZE, 0);
                         match socket.receive(&mut self.byte_buffer) {
                             Ok(num_received) => {
                                 self.byte_buffer.resize(num_received, 0);
                                 let message_content = self.bytes_to_data_message_content(&self.byte_buffer);
                                 let send_result = component::default_send_data_message(
                                     &mut self.exec_state, self.pdl_output_port_id, PortInstruction::NoSource,
                                     message_content, sched_ctx, &mut self.consensus, &mut self.control, comp_ctx
                                 );
                                 if let Err(location_and_message) = send_result {
                                     component::default_handle_error_for_builtin(&mut self.exec_state, sched_ctx, location_and_message);
                                     return CompScheduling::Immediate;
                                 } else {
                                     let scheduling = send_result.unwrap();
                                     self.sync_state = ClientSyncState::AwaitingCmd;
                                     return scheduling;
+                                }
                             },
                             Err(err) => {
                                 if err.kind() == IoErrorKind::WouldBlock {
                                     return CompScheduling::Sleep; // wait until polled
                                 } else {
                                     todo!("handle socket.receive error {:?}", err)
+                                }
+                            }
+                        }
                     },
                     ClientSyncState::FinishSync | ClientSyncState::FinishSyncThenQuit => {
                         component::default_handle_sync_end(&mut self.exec_state, sched_ctx, comp_ctx, &mut self.consensus);
                         return CompScheduling::Requeue;
                     },
+                }
             },
             CompMode::BlockedGet => {
                 // Entered when awaiting a new command
                 debug_assert_eq!(self.sync_state, ClientSyncState::AwaitingCmd);
                 return CompScheduling::Sleep;
             },
             CompMode::SyncEnd | CompMode::BlockedPut =>
                 return CompScheduling::Sleep,
             CompMode::StartExit =>
                 return component::default_handle_start_exit(&mut self.exec_state, &mut self.control, sched_ctx, comp_ctx, &mut self.consensus),
             CompMode::BusyExit =>
                 return component::default_handle_busy_exit(&mut self.exec_state, &mut self.control, sched_ctx),
             CompMode::Exit =>
                 return component::default_handle_exit(&self.exec_state),
+        }
+    }
+}
 impl ComponentTcpClient {
     pub(crate) fn new(arguments: ValueGroup) -> Self {
         use std::net::{IpAddr, Ipv4Addr};
         debug_assert_eq!(arguments.values.len(), 4);
         // Parsing arguments
         let ip_heap_pos = arguments.values[0].as_array();
         let ip_elements = &arguments.regions[ip_heap_pos as usize];
         if ip_elements.len() != 4 {
             todo!("friendly error reporting: ip contains 4 octects");
+        }
         let ip_address = IpAddr::V4(Ipv4Addr::new(
             ip_elements[0].as_uint8(), ip_elements[1].as_uint8(),
             ip_elements[2].as_uint8(), ip_elements[3].as_uint8()
         ));
         let port = arguments.values[1].as_uint16();
         let (ip_address, port) = ip_addr_and_port_from_args(&arguments, 0, 1);
         let input_port = component::port_id_from_eval(arguments.values[2].as_input());
         let output_port = component::port_id_from_eval(arguments.values[3].as_output());
         let socket = SocketTcpClient::new(ip_address, port);
         if let Err(socket) = socket {
             todo!("friendly error reporting: failed to open socket (reason: {:?})", socket);
+        }
         return Self{
             socket_state: ClientSocketState::Connected(socket.unwrap()),
             sync_state: ClientSyncState::AwaitingCmd,
             poll_ticket: None,
             inbox_main: vec![None],
             inbox_main: vec![None, None],
             inbox_backup: Vec::new(),
             input_union_send_tag_value: -1,
             input_union_receive_tag_value: -1,
             input_union_finish_tag_value: -1,
             input_union_shutdown_tag_value: -1,
             pdl_input_port_id: input_port,
             pdl_output_port_id: output_port,
             exec_state: CompExecState::new(),
             control: ControlLayer::default(),
             consensus: Consensus::new(),
             byte_buffer: Vec::new(),
+        }
+    }
     pub(crate) fn new_with_existing_connection(socket: SocketTcpClient, input_port: PortId, output_port: PortId) -> Self {
         return Self{
             socket_state: ClientSocketState::Connected(socket);,
             sync_state: ClientSyncState::AwaitingCmd,
             poll_ticket: None,
             inbox_main: vec![None, None],
             inbox_backup: Vec::new(),
             input_union_send_tag_value: -1,
             input_union_receive_tag_value: -1,
             input_union_finish_tag_value: -1,
             input_union_shutdown_tag_value: -1,
             pdl_input_port_id: input_port,
             pdl_output_port_id: output_port,
             exec_state: CompExecState::new(),
             control: ControlLayer::default(),
             consensus: Consensus::new(),
             byte_buffer: Vec::new(),
+        }
+    }
     // Handles incoming data from the PDL side (hence, going into the socket)
     fn handle_incoming_data_message(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, message: DataMessage) {
         if self.exec_state.mode.is_in_sync_block() {
             self.consensus.handle_incoming_data_message(comp_ctx, &message);
+        }
         match component::default_handle_incoming_data_message(
             &mut self.exec_state, &mut self.inbox_main, comp_ctx, message, sched_ctx, &mut self.control
         ) {
             IncomingData::PlacedInSlot => {},
             IncomingData::SlotFull(message) => {
                 self.inbox_backup.push(message);
+            }
+        }
+    }
     fn data_message_to_bytes(&self, message: DataMessage, bytes: &mut Vec<u8>) {
         debug_assert_eq!(message.data_header.target_port, self.pdl_input_port_id);
         debug_assert_eq!(message.content.values.len(), 1);
         if let Value::Array(array_pos) = message.content.values[0] {
             let region = &message.content.regions[array_pos as usize];
             bytes.reserve(region.len());
             for value in region {
                 bytes.push(value.as_uint8());
+            }
         } else {
             unreachable!();
+        }
+    }
     fn bytes_to_data_message_content(&self, buffer: &[u8]) -> ValueGroup {
         // Turn bytes into silly executor-style array
         let mut values = Vec::with_capacity(buffer.len());
         for byte in buffer.iter().copied() {
             values.push(Value::UInt8(byte));
+        }
         // Put in a value group
         let mut value_group = ValueGroup::default();
         value_group.regions.push(values);
         value_group.values.push(Value::Array(0));
         return value_group;
+    }
+}
 // -----------------------------------------------------------------------------
 // ComponentTcpListener
 // -----------------------------------------------------------------------------
 enum ListenerSocketState {
     Connected(SocketTcpListener),
     Error,
+}
 impl ListenerSocketState {
     fn get_socket(&self) -> &SocketTcpListener {
         match self {
             ListenerSocketState::Connected(v) => return v,
             ListenerSocketState::Error => unreachable!(),
+        }
+    }
+}
 #[derive(PartialEq, Debug)]
 enum ListenerSyncState {
     AwaitingCmd,
     Accept,
     FinishSyncThenQuit,
+}
 pub struct ComponentTcpListener {
     // Properties for the tcp socket
     socket_state: ListenerSocketState,
     sync_state: ListenerSyncState,
     poll_ticket: Option<PollTicket>,
     inbox_main: InboxMain,
     inbox_backup: InboxBackup,
     pdl_output_port_id: PortId, // output port, sends
     pdl_input_port_id: PortId, // input port, receives commands
     pdl_output_port_id: PortId, // output port, sends connections
     // Information about union tags
     output_struct_rx_index: i64,
     output_struct_tx_index: i64,
     input_union_accept_tag: i64,
     input_union_shutdown_tag: i64,
     output_struct_rx_index: usize,
     output_struct_tx_index: usize,
     // Generic component state
+}
@@ \ No newline at end of file @@
     exec_state: CompExecState,
     control: ControlLayer,
     consensus: Consensus,
+}
 impl Component for ComponentTcpListener {
     fn on_creation(&mut self, id: CompId, sched_ctx: &SchedulerCtx) {
         // Retrieve type information for the message with ports we're going to send
         let pd = &sched_ctx.runtime.protocol;
         let cmd_type = pd.find_type(b"std.internet", b"ListenerCmd")
             .expect("'ListenerCmd' type in the 'std.internet' module");
         let cmd_type = cmd_type.as_union();
         self.input_union_accept_tag = cmd_type.get_variant_tag_value(b"Accept").unwrap();
         self.input_union_shutdown_tag = cmd_type.get_variant_tag_value(b"Shutdown").unwrap();
         let conn_type = pd.find_type(b"std.internet", b"TcpConnection")
             .expect("'TcpConnection' type in the 'std.internet' module");
         let conn_type = conn_type.as_struct();
         assert_eq!(conn_type.get_num_struct_fields(), 2);
         self.output_struct_rx_index = conn_type.get_struct_field_index(b"rx").unwrap();
         self.output_struct_tx_index = conn_type.get_struct_field_index(b"tx").unwrap();
         // Register socket for async events
         if let ListenerSocketState::Connected(socket) = &self.socket_state {
             let self_handle = sched_ctx.runtime.get_component_public(id);
             let poll_ticket = sched_ctx.polling.register(socket, self_handle, true, false)
                 .expect("registering tcp listener");
             debug_assert!(self.poll_ticket.is_none());
             self.poll_ticket = Some(poll_ticket);
+        }
+    }
     fn on_shutdown(&mut self, sched_ctx: &SchedulerCtx) {
         if let Some(poll_ticket) = self.poll_ticket.take() {
             sched_ctx.polling.unregister(poll_ticket);
+        }
+    }
     fn adopt_message(&mut self, _comp_ctx: &mut CompCtx, _message: DataMessage) {
         unreachable!();
+    }
     fn handle_message(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx, message: Message) {
         match message {
             Message::Data(_message) => unreachable!(),
             Message::Sync(message) => {
                 let decision = self.consensus.receive_sync_message(sched_ctx, comp_ctx, message);
                 component::default_handle_sync_decision(sched_ctx, &mut self.exec_state, comp_ctx, decision, &mut self.consensus);
             },
             Message::Control(message) => {
                 if let Err(location_and_message) = component::default_handle_control_message(
                     &mut self.exec_state, &mut self.control, &mut self.consensus,
                     message, sched_ctx, comp_ctx, &mut self.inbox_main, &mut self.inbox_backup
                 ) {
                     component::default_handle_error_for_builtin(&mut self.exec_state, sched_ctx, location_and_message);
+                }
             },
             Message::Poll => {
                 sched_ctx.info("Received polling event");
             },
+        }
+    }
     fn run(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx) -> CompScheduling {
         sched_ctx.info(&format!("Running comonent ComponentTcpListener (mode: {:?})", self.exec_state.mode));
         match self.exec_state.mode {
             CompMode::BlockedSelect
                 => unreachable!(),
             CompMode::PutPortsBlockedTransferredPorts |
             CompMode::PutPortsBlockedAwaitingAcks |
             CompMode::PutPortsBlockedSendingPort |
             CompMode::NewComponentBlocked
                 => return CompScheduling::Sleep,
             CompMode::NonSync => {
                 match &self.socket_state {
                     ListenerSocketState::Connected(_socket) => {
                         if self.sync_state == ListenerSyncState::FinishSyncThenQuit {
                             self.exec_state.set_as_start_exit(ExitReason::Termination);
                         } else {
                             self.sync_state = ListenerSyncState::AwaitingCmd;
                             component::default_handle_sync_start(
                                 &mut self.exec_state, &mut self.inbox_main, sched_ctx, comp_ctx, &mut self.consensus
                             );
+                        }
                         return CompScheduling::Immediate;
                     },
                     ListenerSocketState::Error => {
                         self.exec_state.set_as_start_exit(ExitReason::ErrorNonSync);
                         return CompScheduling::Immediate;
+                    }
+                }
             },
             CompMode::Sync => {
                 match self.sync_state {
                     ListenerSyncState::AwaitingCmd => {
                         match component::default_attempt_get(
                             &mut self.exec_state, self.pdl_input_port_id, PortInstruction::NoSource,
                             &mut self.inbox_main, &mut self.inbox_backup, sched_ctx, comp_ctx,
                             &mut self.control, &mut self.consensus
                         ) {
                             GetResult::Received(message) => {
                                 let (tag_value, _) = message.content.values[0].as_union();
                                 if tag_value == self.input_union_accept_tag {
                                     self.sync_state = ListenerSyncState::Accept;
                                 } else if tag_value == self.input_union_shutdown_tag {
                                     self.sync_state = ListenerSyncState::FinishSyncThenQuit;
                                 } else {
                                     unreachable!("got tag_value {}", tag_value);
+                                }
                                 return CompScheduling::Immediate;
                             },
                             GetResult::NoMessage => {
                                 return CompScheduling::Sleep;
                             },
                             GetResult::Error(location_and_message) => {
                                 component::default_handle_error_for_builtin(&mut self.exec_state, sched_ctx, location_and_message);
                                 return CompScheduling::Immediate;
+                            }
+                        }
                     },
                     ListenerSyncState::Accept => {
                         let socket = self.socket_state.get_socket();
                         match socket.accept() {
                             Ok(client) => {
                                 // TODO: Continue here, somehow precreate component, but do the transfer correctly
                                 let client = client.unwrap();
                                 component::default_start_create_component()
                                 todo!("actually create the component")
                             },
                             Err(err) => {
                                 if err.kind() == IoErrorKind::WouldBlock {
                                     return CompScheduling::Sleep;
                                 } else {
                                     todo!("handle listener.accept error {:?}", err);
+                                }
+                            }
+                        }
                     },
                     ListenerSyncState::FinishSyncThenQuit => {
                         component::default_handle_sync_end(&mut self.exec_state, sched_ctx, comp_ctx, &mut self.consensus);
                         return CompScheduling::Requeue;
+                    }
+                }
                 return CompScheduling::Sleep;
             },
             CompMode::BlockedGet => {
                 debug_assert_eq!(self.sync_state, ListenerSyncState::AwaitingCmd);
                 return CompScheduling::Sleep;
             },
             CompMode::SyncEnd | CompMode::BlockedPut
                 => return CompScheduling::Sleep,
             CompMode::StartExit =>
                 return component::default_handle_start_exit(&mut self.exec_state, &mut self.control, sched_ctx, comp_ctx, &mut self.consensus),
             CompMode::BusyExit =>
                 return component::default_handle_busy_exit(&mut self.exec_state, &mut self.control, sched_ctx),
             CompMode::Exit =>
                 return component::default_handle_exit(&self.exec_state),
+        }
+    }
+}
 impl ComponentTcpListener {
     pub(crate) fn new(arguments: ValueGroup) -> Self {
         debug_assert_eq!(arguments.values.len(), 4);
         // Parsing arguments
         let (ip_address, port) = ip_addr_and_port_from_args(&arguments, 0, 1);
         let input_port = component::port_id_from_eval(arguments.values[2].as_input());
         let output_port = component::port_id_from_eval(arguments.values[3].as_output());
         let socket = SocketTcpListener::new(ip_address, port);
         if let Err(socket) = socket {
             todo!("friendly error reporting: failed to open socket (reason: {:?})", socket);
+        }
         return Self {
             socket_state: ListenerSocketState::Connected(socket.unwrap()),
             sync_state: ListenerSyncState::AwaitingCmd,
             poll_ticket: None,
             inbox_main: vec![None, None],
             inbox_backup: InboxBackup::new(),
             pdl_input_port_id: input_port,
             pdl_output_port_id: output_port,
             input_union_accept_tag: -1,
             input_union_shutdown_tag: -1,
             output_struct_tx_index: 0,
             output_struct_rx_index: 0,
             exec_state: CompExecState::new(),
             control: ControlLayer::default(),
             consensus: Consensus::new(),
+        }
+    }
+}
 fn ip_addr_and_port_from_args(
     arguments: &ValueGroup, ip_index: usize, port_index: usize
 ) -> (IpAddr, u16) {
     debug_assert!(ip_index < arguments.values.len());
     debug_assert!(port_index < arguments.values.len());
     // Parsing IP address
     let ip_heap_pos = arguments.values[0].as_array();
     let ip_elements = &arguments.regions[ip_heap_pos as usize];
     let ip_address = match ip_elements.len() {
 => IpAddr::V4(Ipv4Addr::UNSPECIFIED),
 => IpAddr::V4(Ipv4Addr::new(
             ip_elements[0].as_uint8(), ip_elements[1].as_uint8(),
             ip_elements[2].as_uint8(), ip_elements[3].as_uint8()
         )),
         _ => todo!("friendly error reporting: ip should contain 4 octets (or 0 for unspecified)")
     };
     let port = arguments.values[port_index].as_uint16();
     return (ip_address, port);
+}

std/std.internet.pdl

➞

Show inline comments

 #module std.internet
-union Cmd {
+union ClientCmd {
     Send(u8[]),
     Receive,
     Finish,
     Shutdown,
+}
 comp tcp_client(u8[] ip, u16 port, in<Cmd> cmds, out<u8[]> rx) {
     #builtin
+}
 union ListenerCmd {
     Accept,
     Shutdown,
+}
 struct TcpConnection {
     out<Cmd> tx,
     in<u8[]> rx,
+}
 /* comp tcp_listener(u8[] ip, u16 port, out<TcpConnection> rx) {
+/* comp tcp_listener(u8[] ip, u16 port, in<ListenerCmd> cmds, out<TcpConnection> rx) {
     #builtin
 } */
@@ \ No newline at end of file @@

0 comments (0 inline, 0 general)