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

Changeset - 314ab231a7d1

Parent rev.

Child rev.

[Not reviewed]

0 2 0

MH - 3 years ago 2022-05-13 21:22:28
contact@maxhenger.nl

Implement TCP listener

2 files changed with 104 insertions and 19 deletions:

src/runtime2/component/component_internet.rs

src/runtime2/stdlib/internet.rs

0 comments (0 inline, 0 general)

src/runtime2/component/component_internet.rs

➞

Show inline comments

@@ @@ -15,40 +15,44 @@ enum SocketState { @@
     Error,
+}
 impl SocketState {
     fn get_socket(&self) -> &SocketTcpClient {
         match self {
             SocketState::Connected(v) => v,
             SocketState::Error => unreachable!(),
+        }
+    }
+}
 // -----------------------------------------------------------------------------
 // ComponentTcpClient
 // -----------------------------------------------------------------------------
 /// 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 SyncState {
+enum ClientSyncState {
     AwaitingCmd,
     Getting,
     Putting,
     FinishSync,
     FinishSyncThenQuit,
+}
 pub struct ComponentTcpClient {
     // Properties for the tcp socket
     socket_state: SocketState,
-    sync_state: SyncState,
+    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
     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,
@@ @@ -128,93 +132,93 @@ impl Component for ComponentTcpClient { @@
             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 {
                     SocketState::Connected(_socket) => {
-                        if self.sync_state == SyncState::FinishSyncThenQuit {
+                        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 = SyncState::AwaitingCmd;
+                            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;
                     },
                     SocketState::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 {
-                    SyncState::AwaitingCmd => {
+                    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 = SyncState::Putting;
+                                    self.sync_state = ClientSyncState::Putting;
                                 } else if tag_value == self.input_union_receive_tag_value {
                                     // Component requires a `recv`
-                                    self.sync_state = SyncState::Getting;
+                                    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 = SyncState::FinishSync;
+                                    self.sync_state = ClientSyncState::FinishSync;
                                 } else if tag_value == self.input_union_shutdown_tag_value {
                                     // Component wants to close the connection
-                                    self.sync_state = SyncState::FinishSyncThenQuit;
+                                    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;
+                            }
+                        }
                     },
-                    SyncState::Putting => {
+                    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 {
@@ @@ -222,68 +226,68 @@ impl Component for ComponentTcpClient { @@
                                     } 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;
                     },
-                    SyncState::Getting => {
+                    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 = SyncState::AwaitingCmd;
+                                    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)
+                                }
+                            }
+                        }
                     },
-                    SyncState::FinishSync | SyncState::FinishSyncThenQuit => {
+                    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, SyncState::AwaitingCmd);
+                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),
+        }
+    }
@@ @@ -308,25 +312,25 @@ impl ComponentTcpClient { @@
         let port = arguments.values[1].as_uint16();
         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: SocketState::Connected(socket.unwrap()),
-            sync_state: SyncState::AwaitingCmd,
+            sync_state: ClientSyncState::AwaitingCmd,
             poll_ticket: None,
             inbox_main: vec![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(),
@@ @@ -371,12 +375,17 @@ impl ComponentTcpClient { @@
         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
 // -----------------------------------------------------------------------------

src/runtime2/stdlib/internet.rs

➞

Show inline comments

@@ @@ -17,45 +17,58 @@ pub enum SocketError { @@
     Binding,
     Listening,
     Connecting,
     Accepted,
     Accepting,
+}
 enum SocketState {
     Opened,
     Listening,
+}
 /// TCP connection
 const SOCKET_BLOCKING: bool = false;
 /// TCP (client) connection
 pub struct SocketTcpClient {
     socket_handle: libc::c_int,
     is_blocking: bool,
+}
 impl SocketTcpClient {
     pub fn new(ip: IpAddr, port: u16) -> Result<Self, SocketError> {
         const BLOCKING: bool = false;
         let socket_handle = create_and_connect_socket(
             libc::SOCK_STREAM, libc::IPPROTO_TCP, ip, port
         )?;
-        if !set_socket_blocking(socket_handle, BLOCKING) {
+        if !set_socket_blocking(socket_handle, SOCKET_BLOCKING) {
             unsafe{ libc::close(socket_handle); }
             return Err(SocketError::Modifying);
+        }
         return Ok(SocketTcpClient{
             socket_handle,
             is_blocking: BLOCKING,
             is_blocking: SOCKET_BLOCKING,
         })
+    }
     fn new_from_handle(socket_handle: libc::c_int) -> Result<Self, SocketError> {
         if !set_socket_blocking(socket_handle, SOCKET_BLOCKING) {
             unsafe{ libc::close(socket_handle); }
             return Err(SocketError::Modifying);
+        }
         return Ok(SocketTcpClient{
             socket_handle,
             is_blocking: SOCKET_BLOCKING,
         })
+    }
     pub fn send(&self, message: &[u8]) -> Result<usize, IoError> {
         let result = unsafe{
             let message_pointer = message.as_ptr().cast();
             libc::send(self.socket_handle, message_pointer, message.len() as libc::size_t, 0)
         };
         if result < 0 {
             return Err(IoError::last_os_error());
+        }
@@ @@ -81,24 +94,75 @@ impl Drop for SocketTcpClient { @@
     fn drop(&mut self) {
         debug_assert!(self.socket_handle >= 0);
         unsafe{ close(self.socket_handle) };
+    }
+}
 impl AsFileDescriptor for SocketTcpClient {
     fn as_file_descriptor(&self) -> FileDescriptor {
         return self.socket_handle;
+    }
+}
 /// TCP listener. Yielding new connections
 pub struct SocketTcpListener {
     socket_handle: libc::c_int,
     is_blocking: bool,
+}
 impl SocketTcpListener {
     pub fn new(ip: IpAddr, port: u16) -> Result<Self, SocketError> {
         // Create and bind
         let socket_handle = create_and_bind_socket(
             libc::SOCK_STREAM, libc::IPPROTO_TCP, ip, port
         )?;
         if !set_socket_blocking(socket_handle, SOCKET_BLOCKING) {
             unsafe{ libc::close(socket_handle); }
             return Err(SocketError::Modifying);
+        }
         // Listen
         unsafe {
             let result = listen(socket_handle, libc::SOMAXCONN);
             if result < 0 {
                 unsafe{ libc::close(socket_handle); }
                 return Err(SocketError::Listening);
+            }
+        }
         return Ok(SocketTcpListener{
             socket_handle,
             is_blocking: SOCKET_BLOCKING,
         });
+    }
     pub fn accept(&self) -> Result<Result<SocketTcpClient, SocketError>, IoError> {
         let (mut address, mut address_size) = create_sockaddr_in_empty();
         let address_pointer = &mut address as *mut sockaddr_in;
         let socket_handle = unsafe { accept(self.socket_handle, address_pointer.cast(), &mut address_size) };
         if socket_handle < 0 {
             return Err(IoError::last_os_error());
+        }
         return Ok(SocketTcpClient::new_from_handle(socket_handle));
+    }
+}
 impl Drop for SocketTcpListener {
     fn drop(&mut self) {
         debug_assert!(self.socket_handle >= 0);
         unsafe{ close(self.socket_handle) };
+    }
+}
 /// Raw socket receiver. Essentially a listener that accepts a single connection
 struct SocketRawRx {
     listen_handle: c_int,
     accepted_handle: c_int,
+}
 impl SocketRawRx {
     pub fn new(ip: Option<Ipv4Addr>, port: u16) -> Result<Self, SocketError> {
         let ip = ip.unwrap_or(Ipv4Addr::UNSPECIFIED); // unspecified is the same as INADDR_ANY
         let address = unsafe{ in_addr{
             s_addr: std::mem::transmute(ip.octets()),
         }};
                 connect(socket_handle, socket_pointer.cast(), address_size)
+            }
         };
         if result < 0 {
             close(socket_handle);
             return Err(SocketError::Connecting);
+        }
         return Ok(socket_handle);
+    }
+}
 #[inline]
 fn create_sockaddr_in_empty() -> (sockaddr_in, libc::socklen_t) {
     let socket_address = sockaddr_in{
         sin_family: 0,
         sin_port: 0,
         sin_addr: in_addr { s_addr: 0 },
         sin_zero: [0; 8],
     };
     let address_size = size_of::<sockaddr_in>();
     return (socket_address, address_size as _);
+}
 #[inline]
 fn create_sockaddr_in_v4(ip: Ipv4Addr, port: u16) -> (sockaddr_in, libc::socklen_t) {
     let address = unsafe{
         in_addr{
             s_addr: std::mem::transmute(ip.octets())
+        }
     };
     let socket_address = sockaddr_in{
         sin_family: libc::AF_INET as libc::sa_family_t,
         sin_port: htons(port),
         sin_addr: address,

0 comments (0 inline, 0 general)