exec_state: &mut CompExecState, inbox_main: &mut InboxMain,

    comp_ctx: &mut CompCtx, incoming_message: DataMessage,

    sched_ctx: &SchedulerCtx, control: &mut ControlLayer

) -> IncomingData {

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

    let port_index = comp_ctx.get_port_index(port_handle);

    let port_value_slot = &mut inbox_main[port_index];

    let target_port_id = incoming_message.data_header.target_port;

    if port_value_slot.is_none() {

        // We can put the value in the slot

        *port_value_slot = Some(incoming_message);

                default_handle_ack(exec_state, control, control_id, sched_ctx, comp_ctx, consensus, inbox_main, inbox_backup)?;

            } else {

                // Respond to the message

                let port_info = comp_ctx.get_port(port_handle);

                let last_registered_round = port_info.last_registered_round;

                let last_instruction = port_info.last_instruction;

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

                comp_ctx.change_port_peer(sched_ctx, port_handle, None);

                // Handle any possible error conditions (which boil down to: the

                // port has been used, but the peer has died). If not in sync

                // mode then we close the port immediately.

    // and notify the consensus handler of the present messages

    consensus.notify_sync_start(comp_ctx);

    for (port_index, message) in inbox_main.iter().enumerate() {

        if let Some(message) = message {

            consensus.handle_incoming_data_message(comp_ctx, message);

            let port_info = comp_ctx.get_port_by_index_mut(port_index);

        }

    }

    // Modify execution state

    debug_assert_eq!(exec_state.mode, CompMode::NonSync);

    exec_state.mode = CompMode::Sync;

    // Generic operating state

    pub kind: PortKind,

    pub state: PortState,

    // State tracking for error detection and error handling

    pub last_registered_round: Option<u32>,

    pub last_instruction: PortInstruction, // used during sync round to detect port-closed-during-sync errors

    pub close_at_sync_end: bool, // set during sync round when receiving a port-closed-after-sync message

    pub(crate) associated_with_peer: bool,

}

pub struct Peer {

    pub id: CompId,

            kind: PortKind::Putter,

            state: PortState::new(),

            peer_comp_id: self.id,

            last_registered_round: None,

            last_instruction: PortInstruction::None,

            close_at_sync_end: false,

            associated_with_peer: false,

        });

        self.ports.push(Port{

            self_id: getter_id,

            peer_port_id: putter_id,

            kind: PortKind::Getter,

            state: PortState::new(),

            peer_comp_id: self.id,

            last_registered_round: None,

            last_instruction: PortInstruction::None,

            close_at_sync_end: false,

            associated_with_peer: false,

        });

        return Channel{ putter_id, getter_id };

    }

        let self_id = PortId(self.take_port_id());

        self.ports.push(Port{

            self_id, peer_comp_id, peer_port_id, kind, state,

            last_registered_round: None,

            last_instruction: PortInstruction::None,

            close_at_sync_end: false,

            associated_with_peer: false,

        });

        return LocalPortHandle(self_id);

    }

    /// Adds a self-reference. Called by the runtime/scheduler