Files
@ 644bbf1ed134
Branch filter:
Location: CSY/reowolf/src/runtime2/component/component_internet.rs
644bbf1ed134
13.4 KiB
application/rls-services+xml
WIP on TCP component implementation, changes to interface
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 | use crate::protocol::eval::{ValueGroup, Value, EvalError};
use crate::runtime2::*;
use crate::runtime2::component::CompCtx;
use crate::runtime2::stdlib::internet::*;
use super::component::{self, *};
use super::control_layer::*;
use super::consensus::*;
use std::io::ErrorKind as IoErrorKind;
enum SocketState {
Connected(SocketTcpClient),
Error,
}
impl SocketState {
fn get_socket(&self) -> &SocketTcpClient {
match self {
SocketState::Connected(v) => v,
SocketState::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).
enum SyncState {
AwaitingCmd,
Getting,
Putting
}
pub struct ComponentTcpClient {
// Properties for the tcp socket
socket_state: SocketState,
sync_state: SyncState,
pending_recv: Vec<DataMessage>, // on the input port
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,
// 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, sched_ctx: &SchedulerCtx) {
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")
.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();
}
fn adopt_message(&mut self, _comp_ctx: &mut CompCtx, message: DataMessage) {
self.handle_incoming_data_message(message);
}
fn handle_message(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx, message: Message) {
match mesage {
Message::Data(message) => {
self.handle_incoming_data_message(message);
},
Message::Sync(message) => {
let decision = self.consensus.receive_sync_message(sched_ctx, comp_ctx, message);
component::default_handle_sync_decision(&mut self.exec_state, decision, &mut self.consensus);
},
Message::Control(message) => {
component::default_handle_control_message(
&mut self.exec_state, &mut self.control, &mut self.consensus,
message, sched_ctx, comp_ctx
);
}
}
}
fn run(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx) -> Result<CompScheduling, EvalError> {
sched_ctx.log(&format!("Running component ComponentTcpClient (mode: {:?}", self.exec_state.mode));
match self.exec_state.mode {
CompMode::BlockedSelect => {
// Not possible: we never enter this state
unreachable!();
},
CompMode::NonSync => {
// When in non-sync mode
match &mut self.socket_state {
SocketState::Connected(_socket) => {
// Always move into the sync-state
self.sync_state = SyncState::AwaitingCmd;
self.consensus.notify_sync_start(comp_ctx);
self.exec_state.mode = CompMode::Sync;
},
SocketState::Error => {
// Could potentially send an error message to the
// connected component.
self.exec_state.mode = CompMode::StartExit;
return Ok(CompScheduling::Immediate);
}
}
},
CompMode::Sync => {
// When in sync mode: wait for a command to come in
match self.sync_state {
SyncState::AwaitingCmd => {
if let Some(message) = self.pending_recv.pop() {
if self.consensus.try_receive_data_message(sched_ctx, comp_ctx, &message) {
// Check which command we're supposed to execute.
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;
return Ok(CompScheduling::Immediate);
} else if tag_value == self.input_union_receive_tag_value {
// Component requires a `recv`
self.sync_state = SyncState::Getting;
return Ok(CompScheduling::Immediate);
} else if tag_value == self.input_union_finish_tag_value {
// 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 {
todo!("handle sync failure due to message deadlock");
return Ok(CompScheduling::Sleep);
}
} else {
self.exec_state.set_as_blocked_get(self.pdl_input_port_id);
return Ok(CompScheduling::Sleep);
}
},
SyncState::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 Ok(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
let decision = self.consensus.notify_sync_end(sched_ctx, comp_ctx);
component::default_handle_sync_decision(&mut self.exec_state, decision, &mut self.consensus);
},
SyncState::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();
debug_assert!(self.byte_buffer.is_empty());
self.byte_buffer.resize(BUFFER_SIZE, 0);
match socket.receive(&mut self.byte_buffer) {
Ok(num_received) => {
self.byte_buffer.resize(num_received);
let message_content = self.bytes_to_data_message_content(&self.byte_buffer);
let port_handle = comp_ctx.get_port_handle(self.pdl_output_port_id);
let port_info = comp_ctx.get_port(port_handle);
let message = self.consensus.annotate_data_message(comp_ctx, port_info, message_content);
},
Err(err) => {
if err.kind() == IoErrorKind::WouldBlock {
return Ok(CompScheduling::Sleep); // wait until polled
} else {
todo!("handle socket.receive error {:?}", err);
}
}
}
},
}
},
CompMode::BlockedGet => {
// Entered when awaiting a new command
debug_assert_eq!(self.sync_state, SyncState::AwaitingCmd);
if self.
},
CompMode::SyncEnd | CompMode::BlockedPut =>
return Ok(CompScheduling::Sleep),
CompMode::StartExit =>
return Ok(component::default_handle_start_exit(&mut self.exec_state, &mut self.control, sched_ctx, comp_ctx)),
CompMode::BusyExit =>
return Ok(component::default_handle_busy_exit(&mut self.exec_state, &mut self.control, sched_ctx)),
CompMode::Exit =>
return Ok(component::default_handle_exit(&self.exec_state)),
}
return Ok(CompScheduling::Immediate);
}
}
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 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 {:?}", socket);
}
return Self{
socket_state: SocketState::Connected(socket.unwrap()),
input_union_send_tag_value: -1,
input_union_receive_tag_value: -1,
input_union_finish_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(),
}
}
// Handles incoming data from the PDL side (hence, going into the socket)
fn handle_incoming_data_message(&mut self, message: DataMessage) {
// Input message is an array of bytes (u8)
self.pending_recv.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;
}
}
|