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

Changeset - fb814548c7d5

Parent rev.

Child rev.

[Not reviewed]

0 14 0

mh - 3 years ago 2022-04-12 09:30:48
contact@maxhenger.nl

Add tcp component to standard library

13 files changed with 61 insertions and 38 deletions:

src/protocol/ast.rs

src/protocol/eval/executor.rs

src/protocol/parser/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/component/component_pdl.rs

src/runtime2/component/component_random.rs

src/runtime2/poll/mod.rs

src/runtime2/runtime.rs

src/runtime2/stdlib/internet.rs

std/std.internet.pdl

0 comments (0 inline, 0 general)

src/protocol/ast.rs

➞

Show inline comments

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

src/protocol/eval/executor.rs

➞

Show inline comments

@@ @@ -755,7 +755,7 @@ impl Prompt { @@
                                         },
+                                    }
                                 },
                                 Method::ComponentRandomU32 => {
+                                Method::ComponentRandomU32 | Method::ComponentTcpClient => {
                                     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/parser/mod.rs

➞

Show inline comments

@@ @@ -298,8 +298,9 @@ impl Parser { @@
         use std::fs;
         // Pair is (name, add_to_global_namespace)
-        const FILES: [(&'static str, bool); 2] = [
+        const FILES: [(&'static str, bool); 3] = [
             ("std.global.pdl", true),
             ("std.internet.pdl", false),
             ("std.random.pdl", false),
         ];

src/protocol/parser/pass_definitions.rs

➞

Show inline comments

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

src/protocol/parser/pass_validation_linking.rs

➞

Show inline comments

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

src/runtime2/component/component.rs

➞

Show inline comments

@@ @@ -3,9 +3,10 @@ use crate::protocol::*; @@
 use crate::runtime2::*;
 use crate::runtime2::communication::*;
 use super::{CompCtx, CompPDL};
+use super::{CompCtx, CompPDL, CompId};
 use super::component_context::*;
 use super::component_random::*;
 use super::component_internet::*;
 use super::control_layer::*;
 use super::consensus::*;
@@ @@ -19,7 +20,7 @@ pub enum CompScheduling { @@
 /// Generic representation of a component (as viewed by a scheduler).
 pub(crate) trait Component {
     /// Called upon the creation of the component.
     fn on_creation(&mut self, sched_ctx: &SchedulerCtx);
+    fn on_creation(&mut self, comp_id: CompId, sched_ctx: &SchedulerCtx);
     /// Called if the component is created by another component and the messages
     /// are being transferred between the two.
@@ @@ -116,8 +117,9 @@ pub(crate) fn create_component( @@
     if definition.source.is_builtin() {
         // Builtin component
         let component = match definition.source {
+        let component: Box<dyn Component> = match definition.source {
             ProcedureSource::CompRandomU32 => Box::new(ComponentRandomU32::new(arguments)),
             ProcedureSource::CompTcpClient => Box::new(ComponentTcpClient::new(arguments)),
             _ => unreachable!(),
         };

src/runtime2/component/component_internet.rs

➞

Show inline comments

 use crate::protocol::eval::{ValueGroup, Value, EvalError};
 use crate::runtime2::*;
 use crate::runtime2::component::CompCtx;
+use crate::runtime2::component::{CompCtx, CompId};
 use crate::runtime2::stdlib::internet::*;
 use crate::runtime2::poll::*;
 use super::component::{self, *};
 use super::control_layer::*;
@@ @@ -38,6 +39,7 @@ pub struct ComponentTcpClient { @@
     // Properties for the tcp socket
     socket_state: SocketState,
     sync_state: SyncState,
     poll_ticket: Option<PollTicket>,
     inbox_main: Option<DataMessage>,
     inbox_backup: Vec<DataMessage>,
     pdl_input_port_id: PortId, // input from PDL, so transmitted over socket
@@ @@ -45,6 +47,7 @@ pub struct ComponentTcpClient { @@
     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,
@@ @@ -54,7 +57,8 @@ pub struct ComponentTcpClient { @@
+}
 impl Component for ComponentTcpClient {
     fn on_creation(&mut self, sched_ctx: &SchedulerCtx) {
     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");
@@ @@ -64,6 +68,17 @@ impl Component for ComponentTcpClient { @@
         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 SocketState::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");
             debug_assert!(self.poll_ticket.is_none());
             self.poll_ticket = Some(poll_ticket);
+        }
+    }
     fn adopt_message(&mut self, _comp_ctx: &mut CompCtx, message: DataMessage) {
@@ @@ -150,6 +165,9 @@ impl Component for ComponentTcpClient { @@
                                     // Component requires us to end the sync round
                                     let decision = self.consensus.notify_sync_end(sched_ctx, comp_ctx);
                                     component::default_handle_sync_decision(&mut self.exec_state, decision, &mut self.consensus);
                                 } else if tag_value == self.input_union_shutdown_tag_value {
                                     // Component wants to close the connection
                                     todo!("implement clean shutdown, don't forget to unregister to poll ticket");
+                                }
                             } else {
                                 todo!("handle sync failure due to message deadlock");
@@ @@ -268,11 +286,13 @@ impl ComponentTcpClient { @@
         return Self{
             socket_state: SocketState::Connected(socket.unwrap()),
             sync_state: SyncState::AwaitingCmd,
             poll_ticket: None,
             inbox_main: 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(),

src/runtime2/component/component_pdl.rs

➞

Show inline comments

@@ @@ -7,6 +7,7 @@ use crate::protocol::eval::{ @@
     EvalContinuation, EvalResult, EvalError
 };
 use crate::runtime2::runtime::CompId;
 use crate::runtime2::scheduler::SchedulerCtx;
 use crate::runtime2::communication::*;
@@ @@ -222,7 +223,7 @@ pub(crate) struct CompPDL { @@
+}
 impl Component for CompPDL {
     fn on_creation(&mut self, _sched_ctx: &SchedulerCtx) {
+    fn on_creation(&mut self, _id: CompId, _sched_ctx: &SchedulerCtx) {
         // Intentionally empty
+    }
@@ @@ -591,6 +592,12 @@ impl CompPDL { @@
     // Handling ports
     // -------------------------------------------------------------------------
     /// Creates a new component and transfers ports. Because of the stepwise
     /// process in which memory is allocated, ports are transferred, messages
     /// are exchanged, component lifecycle methods are called, etc. This
     /// function facilitates a lot of implicit assumptions (e.g. when the
     /// `Component::on_creation` method is called, the component is already
     /// registered at the runtime).
     fn create_component_and_transfer_ports(
         &mut self,
         sched_ctx: &SchedulerCtx, creator_ctx: &mut CompCtx,
@@ @@ -709,7 +716,7 @@ impl CompPDL { @@
         let (created_key, component) = sched_ctx.runtime.finish_create_pdl_component(
             reservation, component, created_ctx, false,
         );
         component.component.on_creation(sched_ctx);
+        component.component.on_creation(created_key.downgrade(), sched_ctx);
         // Now modify the creator's ports: remove every transferred port and
         // potentially remove the peer component.

src/runtime2/component/component_random.rs

➞

Show inline comments

@@ @@ -27,7 +27,7 @@ pub struct ComponentRandomU32 { @@
+}
 impl Component for ComponentRandomU32 {
     fn on_creation(&mut self, _sched_ctx: &SchedulerCtx) {
+    fn on_creation(&mut self, _id: CompId, _sched_ctx: &SchedulerCtx) {
+    }
     fn adopt_message(&mut self, _comp_ctx: &mut CompCtx, _message: DataMessage) {

src/runtime2/poll/mod.rs

➞

Show inline comments

@@ @@ -295,7 +295,7 @@ pub(crate) struct PollingClient { @@
+}
 impl PollingClient {
     fn register<F: AsFileDescriptor>(&self, entity: F, handle: CompHandle, read: bool, write: bool) -> Result<PollTicket, RtError> {
+    pub(crate) fn register<F: AsFileDescriptor>(&self, entity: &F, handle: CompHandle, read: bool, write: bool) -> Result<PollTicket, RtError> {
         let generation = self.generation_counter.fetch_add(1, Ordering::Relaxed);
         let user_data = user_data_for_component(handle.id().0, generation);
         self.queue.push(PollCmd::Register(handle, user_data));
@@ @@ -307,7 +307,7 @@ impl PollingClient { @@
         return Ok(PollTicket(file_descriptor, user_data.0));
+    }
     fn unregister(&self, ticket: PollTicket) -> Result<(), RtError> {
+    pub(crate) fn unregister(&self, ticket: PollTicket) -> Result<(), RtError> {
         let file_descriptor = ticket.0;
         let user_data = UserData(ticket.1);
         self.queue.push(PollCmd::Unregister(file_descriptor, user_data));

src/runtime2/runtime.rs

➞

Show inline comments

@@ @@ -306,12 +306,15 @@ impl RuntimeInner { @@
         return CompHandle::new(id, &component.public);
+    }
     /// Will remove a component and its memory from the runtime. May only be
     /// called if the necessary conditions for destruction have been met.
     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);
+    }

src/runtime2/stdlib/internet.rs

➞

Show inline comments

@@ @@ -9,6 +9,8 @@ use libc::{ @@
 };
 use mio::{event, Interest, Registry, Token};
 use crate::runtime2::poll::{AsFileDescriptor, FileDescriptor};
 #[derive(Debug)]
 pub enum SocketError {
     Opening,
@@ @@ -83,6 +85,12 @@ impl Drop for SocketTcpClient { @@
+    }
+}
 impl AsFileDescriptor for SocketTcpClient {
     fn as_file_descriptor(&self) -> FileDescriptor {
         return self.socket_handle;
+    }
+}
 /// Raw socket receiver. Essentially a listener that accepts a single connection
 struct SocketRawRx {
     listen_handle: c_int,
@@ @@ -310,27 +318,3 @@ fn socket_family_from_ip(ip: IpAddr) -> libc::c_int { @@
 fn htons(port: u16) -> u16 {
     return port.to_be();
+}
@@ \ No newline at end of file @@
 mod tests {
     use std::net::*;
     use super::*;
     // #[test] @nocommit Remove this
     // fn test_inet_thingo() {
     //     const SIZE: usize = 1024;
     //
     //     let s = SocketTcpClient::new(IpAddr::V4(Ipv4Addr::new(142, 250, 179, 163)), 80).expect("connect");
     //     s.send(b"GET / HTTP/1.1\r\n\r\n").expect("sending");
     //     let mut total = Vec::<u8>::new();
     //     let mut buffer = [0; SIZE];
     //     let mut received = SIZE;
     //
     //     while received > 0 {
     //         received = s.receive(&mut buffer).expect("receiving");
     //         println!("DEBUG: Received {} bytes", received);
     //         total.extend_from_slice(&buffer[..received]);
     //     }
     //     let as_str = String::from_utf8_lossy(total.as_slice());
     //     println!("Yay! Got {} bytes:\n{}", as_str.len(), as_str);
     // }
+}
@@ \ No newline at end of file @@

std/std.internet.pdl

➞

Show inline comments

@@ @@ -4,6 +4,7 @@ union Cmd { @@
     Send(u8[]),
     Receive,
     Finish,
     Shutdown,
+}
 primitive tcp_client(u8[] ip, u16 port, in<Cmd> cmds, out<u8[]> rx) {

0 comments (0 inline, 0 general)