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 106 insertions and 21 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

@@ @@ -21,28 +21,32 @@ impl SocketState { @@
             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,
@@ @@ -134,18 +138,18 @@ impl Component for ComponentTcpClient { @@
                 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;
                     },
@@ @@ -157,13 +161,13 @@ impl Component for ComponentTcpClient { @@
+                    }
+                }
             },
             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) => {
@@ @@ -177,22 +181,22 @@ impl Component for ComponentTcpClient { @@
                                     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;
                             },
@@ @@ -202,13 +206,13 @@ impl Component for ComponentTcpClient { @@
                             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() {
@@ @@ -228,13 +232,13 @@ impl Component for ComponentTcpClient { @@
                         // 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();
@@ @@ -250,34 +254,34 @@ impl Component for ComponentTcpClient { @@
                                 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),
@@ @@ -314,13 +318,13 @@ impl ComponentTcpClient { @@
         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,
@@ @@ -376,7 +380,12 @@ impl ComponentTcpClient { @@
         let mut value_group = ValueGroup::default();
         value_group.regions.push(values);
         value_group.values.push(Value::Array(0));
         return value_group;
+    }
+}
@@ \ No newline at end of file @@
+}
 // -----------------------------------------------------------------------------
 // ComponentTcpListener
 // -----------------------------------------------------------------------------

src/runtime2/stdlib/internet.rs

➞

Show inline comments

@@ @@ -23,33 +23,46 @@ pub enum SocketError { @@
 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();
@@ @@ -87,12 +100,63 @@ impl Drop for SocketTcpClient { @@
 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,
+}
+        }
         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())
+        }
@@ @@ -313,7 +389,7 @@ fn socket_family_from_ip(ip: IpAddr) -> libc::c_int { @@
     };
+}
 #[inline]
 fn htons(port: u16) -> u16 {
     return port.to_be();
+}
@@ \ No newline at end of file @@
+}

0 comments (0 inline, 0 general)