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

Changeset - 339b493050e2

Parent rev.

Child rev.

[Not reviewed]

0 9 0

MH - 3 years ago 2022-05-14 21:18:24
contact@maxhenger.nl

WIP on tcp listener/socket test

9 files changed with 202 insertions and 50 deletions:

src/protocol/ast.rs

src/protocol/eval/executor.rs

src/protocol/mod.rs

src/protocol/parser/pass_definitions.rs

src/protocol/parser/pass_validation_linking.rs

src/runtime2/component/component.rs

src/runtime2/component/component_internet.rs

src/runtime2/tests/internet.rs

104

std/std.internet.pdl

0 comments (0 inline, 0 general)

src/protocol/ast.rs

➞

Show inline comments

@@ @@ -1086,6 +1086,7 @@ pub enum ProcedureSource { @@
     // Builtin components, available to user
     CompRandomU32, // TODO: Remove, temporary thing
     CompTcpClient,
     CompTcpListener,
+}
 impl ProcedureSource {
@@ @@ -1852,6 +1853,7 @@ pub enum Method { @@
     // Builtin component,
     ComponentRandomU32,
     ComponentTcpClient,
     ComponentTcpListener,
     // User-defined
     UserFunction,
     UserComponent,
@@ @@ -1862,7 +1864,7 @@ impl Method { @@
         use Method::*;
         match self {
             Get | Put | Fires | Create | Length | Assert | Print => true,
             ComponentRandomU32 | ComponentTcpClient => true,
+            ComponentRandomU32 | ComponentTcpClient | ComponentTcpListener => true,
             _ => false,
+        }
+    }

src/protocol/eval/executor.rs

➞

Show inline comments

@@ @@ -770,7 +770,7 @@ impl Prompt { @@
                                         },
+                                    }
                                 },
                                 Method::ComponentRandomU32 | Method::ComponentTcpClient => {
+                                Method::ComponentRandomU32 | Method::ComponentTcpClient | Method::ComponentTcpListener => {
                                     debug_assert_eq!(heap[expr.procedure].parameters.len(), cur_frame.expr_values.len());
                                     debug_assert_eq!(heap[cur_frame.position].as_new().expression, expr.this);
                                 },

src/protocol/mod.rs

➞

Show inline comments

@@ @@ -160,6 +160,32 @@ impl ProtocolDescription { @@
         });
+    }
     /// Again a somewhat temporary method. Can be used by components to look up
     /// the definition of a particular procedure. Intended use is to find the
     /// DefinitionId/TypeId of builtin components.
     pub(crate) fn find_procedure(&self, module_name: &[u8], proc_name: &[u8]) -> Option<(ProcedureDefinitionId, TypeId)> {
         // 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, proc_name)?;
         let definition = &self.heap[definition_id];
         // Make sure the procedure is not polymorphic
         if !definition.poly_vars().is_empty() {
             return None;
+        }
         if !definition.is_procedure() {
             return None;
+        }
         // Lookup in type table
         let definition = definition.as_procedure();
         let type_parts = [ConcreteTypePart::Component(definition.this, 0)];
         let type_id = self.types.get_monomorph_type_id(&definition.this.upcast(), &type_parts)
             .expect("type ID for non-polymorphic procedure");
         return Some((definition.this, type_id));
+    }
     fn lookup_module_root(&self, module_name: &[u8]) -> Option<RootId> {
         for module in self.modules.iter() {
             match &module.name {

src/protocol/parser/pass_definitions.rs

➞

Show inline comments

@@ @@ -378,6 +378,7 @@ impl PassDefinitions { @@
                 ("std.global", "print") => ProcedureSource::FuncPrint,
                 ("std.random", "random_u32") => ProcedureSource::CompRandomU32,
                 ("std.internet", "tcp_client") => ProcedureSource::CompTcpClient,
                 ("std.internet", "tcp_listener") => ProcedureSource::CompTcpListener,
                 _ => panic!(
                     "compiler error: unknown builtin procedure '{}' in module '{}'",
                     procedure_name, module_name
@@ @@ -1680,6 +1681,7 @@ impl PassDefinitions { @@
                                     ProcedureSource::FuncPrint => Method::Print,
                                     ProcedureSource::CompRandomU32 => Method::ComponentRandomU32,
                                     ProcedureSource::CompTcpClient => Method::ComponentTcpClient,
                                     ProcedureSource::CompTcpListener => Method::ComponentTcpListener,
                                     _ => todo!("other procedure sources"),
                                 };

src/protocol/parser/pass_validation_linking.rs

➞

Show inline comments

@@ @@ -1158,7 +1158,8 @@ impl Visitor for PassValidationLinking { @@
             | Method::SelectRegisterCasePort
             | Method::SelectWait => unreachable!(), // not usable by programmer directly
             Method::ComponentRandomU32
             | Method::ComponentTcpClient => {
             | Method::ComponentTcpClient
             | Method::ComponentTcpListener => {
                 expecting_wrapping_new_stmt = true;
             },
             Method::UserFunction => {}

src/runtime2/component/component.rs

➞

Show inline comments

@@ @@ -241,6 +241,7 @@ pub(crate) fn create_component( @@
         let component: Box<dyn Component> = match definition.source {
             ProcedureSource::CompRandomU32 => Box::new(ComponentRandomU32::new(arguments)),
             ProcedureSource::CompTcpClient => Box::new(ComponentTcpClient::new(arguments)),
             ProcedureSource::CompTcpListener => Box::new(ComponentTcpListener::new(arguments)),
             _ => unreachable!(),
         };

src/runtime2/component/component_internet.rs

➞

Show inline comments

@@ @@ -8,8 +8,10 @@ use super::component::{self, *}; @@
 use super::control_layer::*;
 use super::consensus::*;
 use std::io::ErrorKind as IoErrorKind;
 use std::net::{IpAddr, Ipv4Addr};
 use crate::protocol::{ProcedureDefinitionId, TypeId};
 // -----------------------------------------------------------------------------
 // ComponentTcpClient
@@ @@ -413,43 +415,32 @@ impl ListenerSocketState { @@
+    }
+}
 struct PendingComponent {
     client: SocketTcpClient,
     cmd_rx: PortId,
     data_tx: PortId,
+}
 enum ListenerSyncState {
     AwaitingCmd,
     AcceptCommandReceived,
     AcceptChannelGenerated{ client: SocketTcpClient, cmd_rx: PortId, data_tx: PortId },
     AcceptCommandReceived, // just received `Accept` command
     AcceptChannelGenerated, // created channel, waiting to end the sync round
     AcceptGenerateComponent, // sync ended, back in non-sync, now generate component
     FinishSyncThenQuit,
+}
 impl ListenerSyncState {
     // Bit of a hacky solution: keeps the listener sync state intact, except
     // if it is `AcceptChannelGenerated`, that will be replaced with
     // `AwaitingCmd`. The reason is to move the `client` out.
     fn take(&mut self) -> ListenerSyncState {
         use ListenerSyncState::*;
         match self {
             AwaitingCmd => return AwaitingCmd,
             AcceptCommandReceived => return AcceptCommandReceived,
             FinishSyncThenQuit => return FinishSyncThenQuit,
             AcceptChannelGenerated{ .. } => {
                 let mut swapped = ListenerSyncState::AwaitingCmd;
                 std::mem::swap(self, &mut swapped);
                 return swapped;
+            }
+        }
+    }
+}
 pub struct ComponentTcpListener {
     // Properties for the tcp socket
     socket_state: ListenerSocketState,
     sync_state: ListenerSyncState,
     pending_component: Option<PendingComponent>,
     poll_ticket: Option<PollTicket>,
     inbox_main: InboxMain,
     inbox_backup: InboxBackup,
     pdl_input_port_id: PortId, // input port, receives commands
     pdl_output_port_id: PortId, // output port, sends connections
     // Information about union tags
     // Type information extracted from protocol
     tcp_client_definition: (ProcedureDefinitionId, TypeId),
     input_union_accept_tag: i64,
     input_union_shutdown_tag: i64,
     output_struct_rx_index: usize,
@@ @@ -503,7 +494,9 @@ impl Component for ComponentTcpListener { @@
     fn handle_message(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx, message: Message) {
         match message {
             Message::Data(_message) => unreachable!(),
             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);
@@ @@ -523,7 +516,7 @@ impl Component for ComponentTcpListener { @@
+    }
     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));
+        sched_ctx.info(&format!("Running component ComponentTcpListener (mode: {:?})", self.exec_state.mode));
         match self.exec_state.mode {
             CompMode::BlockedSelect
@@ @@ -536,21 +529,25 @@ impl Component for ComponentTcpListener { @@
             CompMode::NonSync => {
                 match &self.socket_state {
                     ListenerSocketState::Connected(_socket) => {
-                        match self.sync_state.take() {
                         match 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
                                 );
                             },
                             ListenerSyncState::AcceptCommandReceived => unreachable!(),
                             ListenerSyncState::AcceptChannelGenerated{ client, cmd_rx, data_tx } => {
                             ListenerSyncState::AcceptCommandReceived |
                             ListenerSyncState::AcceptChannelGenerated => unreachable!(),
                             ListenerSyncState::AcceptGenerateComponent => {
                                 // Now that we're outside the sync round, create the tcp client
                                 // component
                                 let socket_component: Box<dyn Component> = Box::new(ComponentTcpClient::new_with_existing_connection(client, cmd_rx, data_tx));
                                 let pending = self.pending_component.take().unwrap();
                                 let socket_component: Box<dyn Component> = Box::new(ComponentTcpClient::new_with_existing_connection(
                                     pending.client, pending.cmd_rx, pending.data_tx
                                 ));
                                 component::special_create_component(
                                     &mut self.exec_state, sched_ctx, comp_ctx, &mut self.control,
                                     &mut self.inbox_main, &mut self.inbox_backup, socket_component,
                                     vec![cmd_rx, data_tx]
+                                    vec![pending.cmd_rx, pending.data_tx]
                                 );
                                 self.sync_state = ListenerSyncState::AwaitingCmd; // superfluous, see ListenerSyncState.take()
                             },
@@ @@ -619,11 +616,11 @@ impl Component for ComponentTcpListener { @@
                                 // Construct the message containing the appropriate ports that will
                                 // be sent to the component commanding this listener.
                                 let mut values = ValueGroup::new_stack(vec![
                                     Value::Unassigned, Value::Unassigned
                                 ]);
                                 values.values[self.output_struct_tx_index] = Value::Output(port_id_to_eval(cmd_channel.putter_id));
                                 values.values[self.output_struct_rx_index] = Value::Input(port_id_to_eval(data_channel.getter_id));
                                 let mut values = ValueGroup::new_stack(Vec::with_capacity(1));
                                 values.values.push(Value::Struct(0));
                                 values.regions.push(vec![Value::Unassigned, Value::Unassigned]);
                                 values.regions[0][self.output_struct_tx_index] = Value::Output(port_id_to_eval(cmd_channel.putter_id));
                                 values.regions[0][self.output_struct_rx_index] = Value::Input(port_id_to_eval(data_channel.getter_id));
                                 if let Err(location_and_message) = component::default_send_data_message(
                                     &mut self.exec_state, self.pdl_output_port_id, PortInstruction::NoSource, values,
                                     sched_ctx, &mut self.consensus, &mut self.control, comp_ctx
@@ @@ -633,17 +630,15 @@ impl Component for ComponentTcpListener { @@
                                     );
+                                }
                                 // And finish the consensus round
                                 component::default_handle_sync_end(&mut self.exec_state, sched_ctx, comp_ctx, &mut self.consensus);
                                 // Enter a state such that when we leave the consensus round and
                                 // go back to the nonsync state, that we will actually create the
                                 // tcp client component.
                                 self.sync_state = ListenerSyncState::AcceptChannelGenerated {
                                 // Prepare for finishing the consensus round, and once finished,
                                 // create the tcp client component
                                 self.sync_state = ListenerSyncState::AcceptChannelGenerated;
                                 debug_assert!(self.pending_component.is_none());
                                 self.pending_component = Some(PendingComponent{
                                     client,
                                     cmd_rx: cmd_channel.getter_id,
                                     data_tx: data_channel.putter_id
                                 };
+                                });
                                 return CompScheduling::Requeue;
                             },
@@ @@ -656,11 +651,16 @@ impl Component for ComponentTcpListener { @@
+                            }
+                        }
                     },
                     ListenerSyncState::AcceptChannelGenerated{ .. } => unreachable!(),
                     ListenerSyncState::FinishSyncThenQuit => {
                     ListenerSyncState::AcceptChannelGenerated => {
                         component::default_handle_sync_end(&mut self.exec_state, sched_ctx, comp_ctx, &mut self.consensus);
                         self.sync_state = ListenerSyncState::AcceptGenerateComponent;
                         return CompScheduling::Requeue;
+                    }
                     ListenerSyncState::FinishSyncThenQuit => {
                         component::default_handle_sync_end(&mut self.exec_state, sched_ctx, comp_ctx, &mut self.consensus);
                         return CompScheduling::Requeue;
                     },
                     ListenerSyncState::AcceptGenerateComponent => unreachable!(),
+                }
             },
             CompMode::BlockedGet => {
@@ @@ -695,6 +695,7 @@ impl ComponentTcpListener { @@
         return Self {
             socket_state: ListenerSocketState::Connected(socket.unwrap()),
             sync_state: ListenerSyncState::AwaitingCmd,
             pending_component: None,
             poll_ticket: None,
             inbox_main: vec![None, None],
             inbox_backup: InboxBackup::new(),
@@ @@ -709,6 +710,21 @@ impl ComponentTcpListener { @@
             consensus: Consensus::new(),
+        }
+    }
     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 ip_addr_and_port_from_args(

src/runtime2/tests/internet.rs

➞

Show inline comments

@@ @@ -72,3 +72,107 @@ fn test_stdlib_file() { @@
+    }
     ", "constructor", no_args());
+}
 #[test]
 fn test_tcp_listener_and_client() {
     compile_and_create_component("
     import std.internet::*;
     func listen_port() -> u16 {
         return 2393;
+    }
     comp server(u32 num_connections) {
         // Start tcp listener
         channel listen_cmd_tx -> listen_cmd_rx;
         channel listen_conn_tx -> listen_conn_rx;
         new tcp_listener({}, listen_port(), listen_cmd_rx, listen_conn_tx);
         // Fake channels such that we can create a dummy connection variable
         channel client_cmd_tx -> unused_client_cmd_rx;
         channel unused_client_data_tx -> client_data_rx;
         auto new_connection = TcpConnection{
             tx: client_cmd_tx,
             rx: client_data_rx,
         };
         auto connection_counter = 0;
         while (connection_counter < num_connections) {
             // Wait until we get a connection
             print(\"server: waiting for an accepted connection\");
             sync {
                 put(listen_cmd_tx, ListenerCmd::Accept);
                 new_connection = get(listen_conn_rx);
+            }
             // We have a new connection, spawn an 'echoer' for it
             print(\"server: spawning an echo'ing component\");
             new echo_machine(new_connection);
             connection_counter += 1;
+        }
         // Shut down the listener
         print(\"server: shutting down listener\");
+    }
     // Waits for a single TCP byte (to simplify potentially having to
     // concatenate requests) and echos it
     comp echo_machine(TcpConnection conn) {
         auto data_to_echo = {};
         // Wait for a message
         sync {
             print(\"echo: receiving data\");
             put(conn.tx, ClientCmd::Receive);
             data_to_echo = get(conn.rx);
             put(conn.tx, ClientCmd::Finish);
+        }
         // Echo the message
         print(\"echo: sending back data\");
         sync put(conn.tx, ClientCmd::Send(data_to_echo));
         // Ask the tcp connection to shut down
         print(\"echo: shutting down\");
         sync put(conn.tx, ClientCmd::Shutdown);
+    }
     comp echo_requester(u8 byte_to_send) {
         channel cmd_tx -> cmd_rx;
         channel data_tx -> data_rx;
         new tcp_client({127, 0, 0, 1}, listen_port(), cmd_rx, data_tx);
         // Send the message
         print(\"requester: sending bytes\");
         sync put(cmd_tx, ClientCmd::Send({ byte_to_send }));
         // Receive the echo'd byte
         auto received_byte = byte_to_send + 1;
         sync {
             print(\"requester: receiving echo response\");
             put(cmd_tx, ClientCmd::Receive);
             received_byte = get(data_rx)[0];
             put(cmd_tx, ClientCmd::Finish);
+        }
         // Silly check, as always
         while (byte_to_send != received_byte) {
             print(\"requester: Oh no! The echo is an otherworldly distorter\");
+        }
         // Shut down the TCP connection
         print(\"requester: shutting down TCP component\");
         sync put(cmd_tx, ClientCmd::Shutdown);
+    }
     comp constructor() {
         auto num_connections = 1;
         new server(num_connections);
         auto connection_index = 0;
         while (connection_index < num_connections) {
             new echo_requester(cast(connection_index));
+        }
+    }
     ", "constructor", no_args());
+}
@@ \ No newline at end of file @@

std/std.internet.pdl

➞

Show inline comments

@@ @@ -21,6 +21,6 @@ struct TcpConnection { @@
     in<u8[]> rx,
+}
-/* comp tcp_listener(u8[] ip, u16 port, in<ListenerCmd> cmds, out<TcpConnection> rx) {
 comp tcp_listener(u8[] ip, u16 port, in<ListenerCmd> cmds, out<TcpConnection> rx) {
     #builtin
 } */
@@ \ No newline at end of file @@
+}
@@ \ No newline at end of file @@

0 comments (0 inline, 0 general)