2. One `get` in the synchronous round.

As a preliminary remark: note that encountering an error is nothing special: the component can simply print an error to `stdout` and stop executing. The handling of the error by peers is of importance! If an interaction is made impossible because a peer has stopped executing, then the component that wishes to perform that interaction should error out itself!

### Handling Errors Outside of a Sync Block

If a component `E` encounters a critical error outside of a sync block. Then we can be sure that if it had a lat synchronous round, that it succeeded. However, there might be future synchronous rounds for component `E`, likewise a peer component `C` might have already put a message in `E`'s inbox.

The requirement for the outside-sync error of `E` is that any future sync interactions by `C` will fail (but, if `C` has no future interactions, it shouldn't fail either!). 

Note that `E` cannot perform `put`/`get` requests, because we're assuming `E` is outside of a sync block. Hence the only possible failing interaction is that `C` has performed a `put`, or is attempting a `get`. In the case the `C` `put`s to `E`, then `E` might not have figured out the identity of `C` yet (see earlier remarks on the eventual consistency of peer detection). Hence `C` is responsible for ensuring its own correct shutdown due to a failing `put`. Likewise for a `get`: `C` cannot receive from `E` if it is failing. So if `C` is waiting on a message to arrive, or if it will call `get` in the future, then `C` must fail as well.

In this case it is sufficient for `E` to send around a `ClosePort` message. As detailed in another chapter of this document. However, a particular race condition might occur. We have assumed that `E` is not in a sync block. But `C` is not aware of this fact. `C` might not be able to distinguish between the following three cases:

1. Regular shutdown: Components `C` and `E` are not in a sync round.

   - `E` broadcasts `ClosePort`.

   - `C` receives `ClosePort`.

2. Shutdown within a sync round, `ClosePort` leads `Solution`: A leader component `L`, peer component `C` and failing component `E`. Assume that all are/were busy in a synchronous round with one another.

   - `L` broadcasts `Solution` for the current sync round.

   - `E` receives `Solution`, finishes round. 

   - `E` encounters an error, so sends `ClosePort` to `C`.

   - `C` receives `ClosePort` from `E`.

   - `C` receives `Solution` from `L`.

3. Shutdown within a sync round, `Solution` leads `ClosePort`: Same components `L`, `C` and `E`.

   - `L` broadcasts `Solution` for the current sync round.

   - `E` receives `Solution` finishes round.

   - `E` encounters an error, so sends `ClosePort` to `C`.

   - `C` receives `Solution` from `L`.

   - `C` receives `ClosePort` from `E`.

In all described cases `E` encounters an error after finishing a sync round. But from the point of view of `C` it is unsure whether the `ClosePort` message pertains to the current synchronous round or not. In case 1 and 3 nothing is out of the ordinary. But in case 2 we have that `C` is at a particular point in time aware of the `ClosePort` from `E`, but not yet of the `Solution` from `L`. `C` should not fail the sync round, as it is completed, but it is unaware of this fact.

As a rather simple solution, since components that are participating with one another in a sync round move in lock-step at the end of the sync block, we send a boolean along with the `ClosePort`, e.g. `ClosePort(nonsync)`. This boolean indicates whether `E` was inside or outside of a sync block during it encountering an error. Now `C` can distinguish between the three cases: in all cases it agrees that `E` was not in a sync block (and hence: the sync round in cases 2 and 3 can be completed).

### Handling Errors Inside of a Sync Block

If `E` is inside of a sync block. Then it has interacted with other components. Our requirement now is that the sync round fails (and ofcourse, that all of the peers are notified that `E` will no longer be present in the runtime). There are two things that are complicating this type of failure:

1. Suppose that in the successful case of the synchronous interaction, there are a large number of components interacting with one another. Now it might be that `E` fails very early in its sync block, such that it cannot interact with several components. This lack of interaction might cause the single sync block to break up into several smaller sync blocks. Each of these separated regions is supposed to fail.

2. Within a particular synchronous interaction we might have that the leader `L` has a reference to the component `E` without it being a direct peer. There is a reference counting system in place that makes sure that `L` can always send messages to `E`. But we still need to make sure that those references stay alive for as long as needed. 

Suppose a synchronous region is (partially) established, and the component `E` encounters a critical error. The two points given above imply that two processes need to be initiated. For the first error-handling process, we simply use the same scheme as described in the case where `E` is not in a synchronous region. However now we broadcast `ClosePort(sync)` instead of `ClosePort(nonsync)` messages. Consider the following two cases: 

1. Component `C` is not part of the same synchronous region as `E`. And component `C` has tried `put`ting to `E`. If `C` receives a `ClosePort(sync)`, then it knows that its interaction should fail. Note: it might be that `E` wasn't planning on `get`ting from `C` in the sync round in which `E` failed, but much later. In that case it still makes sense for `C` to fail; it would have failed in the future. A small inconsistency here (within the current infinitely-deadlocking implementation) is that if `E` would *never* `get` from `C`, then `C` would deadlock instead of crash (one could argue that this implies that deadlocking should lead to crashing through a timeout mechanism).

2. Component `C` is not part of the same synchronous region as `E`. And if `E` wouldn't have crashed, then it would've `put` a message to `C`. In this case it is still proper for `C` to crash. The reasoning works the same as above.

So that is to say that this `ClosePort(sync)` causes instant failure of `C` if it has used the closed port in a round without consensus, or if it uses that port in the future. Note that this `ClosePort(sync)` system causes cascading failures throughout the disjoint synchronous regions. This is as intended: once one component's PDL program can no longer be executed, we cannot depend on the discovery of all the peers that constitute the intended synchronous region. So instead we rely on a peer-to-peer mechanism to make sure that every component is notified of failure.

        },

        ControlMessageContent::BlockPort(port_id) => {

            // One of our messages was accepted, but the port should be

            // blocked.

            let port_handle = comp_ctx.get_port_handle(port_id);

            let port_info = comp_ctx.get_port(port_handle);

            debug_assert_eq!(port_info.kind, PortKind::Putter);

            if port_info.state == PortState::Open {

                // only when open: we don't do this when closed, and we we don't do this if we're blocked due to peer changes

                comp_ctx.set_port_state(port_handle, PortState::BlockedDueToFullBuffers);

            }

        },

        ControlMessageContent::ClosePort(content) => {

            // Request to close the port. We immediately comply and remove

            // the component handle as well

            let port_handle = comp_ctx.get_port_handle(content.port_to_close);

            // We're closing the port, so we will always update the peer of the

            // port (in case of error messages)

            let port_info = comp_ctx.get_port_mut(port_handle);

            port_info.peer_comp_id = message.sender_comp_id;

            let port_info = comp_ctx.get_port(port_handle);

            let peer_comp_id = port_info.peer_comp_id;

            let peer_handle = comp_ctx.get_peer_handle(peer_comp_id);

            // One exception to sending an `Ack` is if we just closed the

            // port ourselves, meaning that the `ClosePort` messages got

            // sent to one another.

            if let Some(control_id) = control.has_close_port_entry(port_handle, comp_ctx) {

                // The two components (sender and this component) are closing

                // the channel at the same time.

                default_handle_ack(control, control_id, sched_ctx, comp_ctx);

            } else {

                // Respond to the message

                let last_instruction = port_info.last_instruction;

                let port_was_used = last_instruction != PortInstruction::None;

                default_send_ack(message.id, peer_handle, sched_ctx, comp_ctx);

                comp_ctx.remove_peer(sched_ctx, port_handle, peer_comp_id, false); // do not remove if closed

                comp_ctx.set_port_state(port_handle, PortState::Closed); // now set to closed

                // Make sure that we've not reached an error condition. Note

                // that if this condition is not met, then we don't error out

                // now, but we may error out in the next sync block when we

                // try to `put`/`get` on the port. This condition makes sure

                // that if we have a successful sync round, followed by the peer

                // closing the port, that we don't consider the sync round to

                // have failed by mistake.

                if content.closed_in_sync_round && exec_state.mode.is_in_sync_block() && port_was_used {

                    return Err((

                        last_instruction,

                        format!("Peer component (id:{}) shut down, so previous communication cannot have succeeded", peer_comp_id.0)

                    ));

                }

            }

        },

        ControlMessageContent::UnblockPort(port_id) => {

            // We were previously blocked (or already closed)

            let port_handle = comp_ctx.get_port_handle(port_id);

            let port_info = comp_ctx.get_port(port_handle);

            debug_assert_eq!(port_info.kind, PortKind::Putter);

            if port_info.state == PortState::BlockedDueToFullBuffers {

                default_handle_unblock_put(exec_state, consensus, port_handle, sched_ctx, comp_ctx);

            }

        },

        ControlMessageContent::PortPeerChangedBlock(port_id) => {

            // The peer of our port has just changed. So we are asked to

            // temporarily block the port (while our original recipient is

            // potentially rerouting some of the in-flight messages) and

            // Ack. Then we wait for the `unblock` call.

            debug_assert_eq!(message.target_port_id, Some(port_id));

            let port_handle = comp_ctx.get_port_handle(port_id);

            comp_ctx.set_port_state(port_handle, PortState::BlockedDueToPeerChange);

            let port_info = comp_ctx.get_port(port_handle);

            let peer_handle = comp_ctx.get_peer_handle(port_info.peer_comp_id);

            default_send_ack(message.id, peer_handle, sched_ctx, comp_ctx);

        },

        ControlMessageContent::PortPeerChangedUnblock(new_port_id, new_comp_id) => {

            let port_handle = comp_ctx.get_port_handle(message.target_port_id.unwrap());

            let port_info = comp_ctx.get_port(port_handle);

            debug_assert!(port_info.state == PortState::BlockedDueToPeerChange);

            let old_peer_id = port_info.peer_comp_id;

            comp_ctx.remove_peer(sched_ctx, port_handle, old_peer_id, false);

            let port_info = comp_ctx.get_port_mut(port_handle);

            port_info.peer_comp_id = new_comp_id;

            port_info.peer_port_id = new_port_id;

            comp_ctx.add_peer(port_handle, sched_ctx, new_comp_id, None);

            default_handle_unblock_put(exec_state, consensus, port_handle, sched_ctx, comp_ctx);

        }

    }

    return Ok(());

                                let message = self.inbox_main.take().unwrap();

                                let target_port_id = message.data_header.target_port;

                                let receive_result = component::default_handle_received_data_message(

                                    target_port_id, PortInstruction::NoSource,

                                    &mut self.inbox_main, &mut self.inbox_backup,

                                    comp_ctx, sched_ctx, &mut self.control

                                );

                                if let Err(location_and_message) = receive_result {

                                    component::default_handle_error_for_builtin(&mut self.exec_state, sched_ctx, location_and_message);

                                    return CompScheduling::Immediate;

                                } else {

                                    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 CompScheduling::Immediate;

                                    } else if tag_value == self.input_union_receive_tag_value {

                                        // Component requires a `recv`

                                        self.sync_state = SyncState::Getting;

                                        return CompScheduling::Immediate;

                                    } else if tag_value == self.input_union_finish_tag_value {

                                        // Component requires us to end the sync round

                                        self.sync_state = SyncState::FinishSync;

                                        return CompScheduling::Immediate;

                                    } else if tag_value == self.input_union_shutdown_tag_value {

                                        // Component wants to close the connection

                                        self.sync_state = SyncState::FinishSyncThenQuit;

                                        return CompScheduling::Immediate;

                                    } else {

                                        unreachable!("got tag_value {}", tag_value)

                                    }

                                }

                            } else {

                                todo!("handle sync failure due to message deadlock");

                                return CompScheduling::Sleep;

                            }

                        } else {

                            self.exec_state.set_as_blocked_get(self.pdl_input_port_id);

                            return 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 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_success(sched_ctx, comp_ctx);

                        self.exec_state.mode = CompMode::SyncEnd;

                        component::default_handle_sync_decision(sched_ctx, &mut self.exec_state, decision, &mut self.consensus);

                        return CompScheduling::Immediate;

                    },

                    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();

                        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, 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;

    }

    fn run(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx) -> CompScheduling {

        use EvalContinuation as EC;

        sched_ctx.log(&format!("Running component (mode: {:?})", self.exec_state.mode));

        // Depending on the mode don't do anything at all, take some special

        // actions, or fall through and run the PDL code.

        match self.exec_state.mode {

            CompMode::NonSync | CompMode::Sync => {

                // continue and run PDL code

            },

            CompMode::SyncEnd | CompMode::BlockedGet | CompMode::BlockedPut | CompMode::BlockedSelect => {

                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, &self.control, sched_ctx

            ),

            CompMode::Exit => return component::default_handle_exit(&self.exec_state),

        }

        let run_result = self.execute_prompt(&sched_ctx);

        if let Err(error) = run_result {

            self.handle_component_error(sched_ctx, CompError::Executor(error));

            return CompScheduling::Immediate;

        }

        let run_result = run_result.unwrap();

        match run_result {

            EC::Stepping => unreachable!(), // execute_prompt runs until this is no longer returned

            EC::BranchInconsistent | EC::NewFork | EC::BlockFires(_) => todo!("remove these"),

            // Results that can be returned in sync mode

            EC::SyncBlockEnd => {

                debug_assert_eq!(self.exec_state.mode, CompMode::Sync);

                self.handle_sync_end(sched_ctx, comp_ctx);

                return CompScheduling::Immediate;

            },

            EC::BlockGet(expr_id, port_id) => {

                debug_assert_eq!(self.exec_state.mode, CompMode::Sync);

                debug_assert!(self.exec_ctx.stmt.is_none());

                let port_id = port_id_from_eval(port_id);

                let port_handle = comp_ctx.get_port_handle(port_id);

                let port_index = comp_ctx.get_port_index(port_handle);

                if let Some(message) = &self.inbox_main[port_index] {

                    // Check if we can actually receive the message

                    if self.consensus.try_receive_data_message(sched_ctx, comp_ctx, message) {

                        // Message was received. Make sure any blocked peers and

                        // pending messages are handled.

                        let message = self.inbox_main[port_index].take().unwrap();

                        let receive_result = component::default_handle_received_data_message(

                            port_id, PortInstruction::SourceLocation(expr_id),

                            &mut self.inbox_main[port_index], &mut self.inbox_backup,

                            comp_ctx, sched_ctx, &mut self.control

                        );

                        if let Err(location_and_message) = receive_result {

                            self.handle_generic_component_error(sched_ctx, location_and_message);

                            return CompScheduling::Immediate

                        } else {

                            self.exec_ctx.stmt = ExecStmt::PerformedGet(message.content);

                            return CompScheduling::Immediate;

                        }

                    } else {

                        todo!("handle sync failure due to message deadlock");

                        return CompScheduling::Sleep;

                    }

                } else {

                    // We need to wait

                    self.exec_state.set_as_blocked_get(port_id);

                    return CompScheduling::Sleep;

                }

            },

            EC::Put(expr_id, port_id, value) => {

                debug_assert_eq!(self.exec_state.mode, CompMode::Sync);

                sched_ctx.log(&format!("Putting value {:?}", value));

                // Send the message

                let target_port_id = port_id_from_eval(port_id);

                let send_result = component::default_send_data_message(

                    &mut self.exec_state, target_port_id,

                    PortInstruction::SourceLocation(expr_id), value,

                    sched_ctx, &mut self.consensus, comp_ctx

                );

                if let Err(location_and_message) = send_result {

                    self.handle_generic_component_error(sched_ctx, location_and_message);

                    return CompScheduling::Immediate;

                } else {

                    // When `run` is called again (potentially after becoming

                    // unblocked) we need to instruct the executor that we performed

                    // the `put`

                    let scheduling = send_result.unwrap();

        // Whatever we do, glean information from headers in message

        if self.exec_state.mode.is_in_sync_block() {

            self.consensus.handle_incoming_data_message(comp_ctx, &message);

        }

        let port_handle = comp_ctx.get_port_handle(message.data_header.target_port);

        let port_index = comp_ctx.get_port_index(port_handle);

        match component::default_handle_incoming_data_message(

            &mut self.exec_state, &mut self.inbox_main[port_index], comp_ctx, message,

            sched_ctx, &mut self.control

        ) {

            IncomingData::PlacedInSlot => {

                if self.exec_state.mode == CompMode::BlockedSelect {

                    let select_decision = self.select_state.handle_updated_inbox(&self.inbox_main, comp_ctx);

                    if let SelectDecision::Case(case_index) = select_decision {

                        self.exec_ctx.stmt = ExecStmt::PerformedSelectWait(case_index);

                        self.exec_state.mode = CompMode::Sync;

                    }

                }

            },

            IncomingData::SlotFull(message) => {

                self.inbox_backup.push(message);

            }

        }

    }

    fn handle_incoming_sync_message(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, message: SyncMessage) {

        let decision = self.consensus.receive_sync_message(sched_ctx, comp_ctx, message);

        component::default_handle_sync_decision(sched_ctx, &mut self.exec_state, decision, &mut self.consensus);

    }

    /// Handles an error coming from the generic `component::handle_xxx`

    /// functions. Hence accepts argument as a tuple.

    fn handle_generic_component_error(&mut self, sched_ctx: &SchedulerCtx, location_and_message: (PortInstruction, String)) {

        // Retrieve location and message, display in terminal

        let (location, message) = location_and_message;

        let error = match location {

            PortInstruction::None => CompError::Component(message),

            PortInstruction::NoSource => unreachable!(), // for debugging: all in-sync errors are associated with a source location

            PortInstruction::SourceLocation(expression_id) => {

                let protocol = &sched_ctx.runtime.protocol;

                CompError::Executor(EvalError::new_error_at_expr(

                    &self.prompt, &protocol.modules, &protocol.heap,

                    expression_id, message

                ))

            }

        };

    }

    fn handle_component_error(&mut self, sched_ctx: &SchedulerCtx, error: CompError) {

        sched_ctx.error(&format!("{}", error));

        // Set state to handle subsequent error

        let exit_reason = if self.exec_state.mode.is_in_sync_block() {

            ExitReason::ErrorInSync

        } else {

            ExitReason::ErrorNonSync

        };

        self.exec_state.set_as_start_exit(exit_reason);

    }

    // -------------------------------------------------------------------------

    // 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,

        definition_id: ProcedureDefinitionId, type_id: TypeId, mut arguments: ValueGroup

    ) {

        struct PortPair{

            creator_handle: LocalPortHandle,

            creator_id: PortId,

            created_handle: LocalPortHandle,

            created_id: PortId,

        }

        let mut opened_port_id_pairs = Vec::new();

        let mut closed_port_id_pairs = Vec::new();

        let reservation = sched_ctx.runtime.start_create_pdl_component();

        let mut created_ctx = CompCtx::new(&reservation);

        let other_proc = &sched_ctx.runtime.protocol.heap[definition_id];

        let self_proc = &sched_ctx.runtime.protocol.heap[self.prompt.frames[0].definition];

        // dbg_code!({

        //     sched_ctx.log(&format!(

        //         "DEBUG: Comp '{}' (ID {:?}) is creating comp '{}' (ID {:?})",

use crate::protocol::*;

use crate::protocol::eval::*;

use crate::runtime2::runtime::*;

use crate::runtime2::component::{CompCtx, CompPDL};

    let prompt = rt.inner.protocol.new_component(

        module_name.as_bytes(), routine_name.as_bytes(), args

    ).expect("create prompt");

    let reserved = rt.inner.start_create_pdl_component();

    let ctx = CompCtx::new(&reserved);

    let component = Box::new(CompPDL::new(prompt, 0));

    let (key, _) = rt.inner.finish_create_pdl_component(reserved, component, ctx, false);

    rt.inner.enqueue_work(key);

}

#[test]

fn test_component_creation() {

    let pd = ProtocolDescription::parse(b"

    primitive nothing_at_all() {

        s32 a = 5;

        auto b = 5 + a;

    }

    ").expect("compilation");

    let rt = Runtime::new(1, true, pd).unwrap();

    for _i in 0..20 {

        create_component(&rt, "", "nothing_at_all", no_args());

    }

}

#[test]

fn test_component_communication() {

    let pd = ProtocolDescription::parse(b"

    primitive sender(out<u32> o, u32 outside_loops, u32 inside_loops) {

        u32 outside_index = 0;

        while (outside_index < outside_loops) {

            u32 inside_index = 0;

            sync while (inside_index < inside_loops) {

                put(o, inside_index);

                inside_index += 1;

            }

            outside_index += 1;

        }

    }

    primitive receiver(in<u32> i, u32 outside_loops, u32 inside_loops) {

        u32 outside_index = 0;

        while (outside_index < outside_loops) {

            u32 inside_index = 0;

            sync while (inside_index < inside_loops) {

                auto val = get(i);

                while (val != inside_index) {} // infinite loop if incorrect value is received

                inside_index += 1;

            }

            outside_index += 1;

        }

    }

    composite constructor() {

        channel o_orom -> i_orom;

        channel o_mrom -> i_mrom;

        channel o_ormm -> i_ormm;