Changeset - 3d271f7cfd7e
Parent rev.
Child rev.
[Not reviewed]
0 4 0
MH - 3 years ago 2022-02-02 16:07:36
contact@maxhenger.nl
WIP: Fix bug due to conflicting closed-port states
4 files changed with 69 insertions and 52 deletions:
src/runtime2/communication.rs
11
1
src/runtime2/component/component_pdl.rs
20
15
src/runtime2/component/control_layer.rs
8
30
src/runtime2/tests/mod.rs
30
6
0 comments (0 inline, 0 general)
src/runtime2/communication.rs
➞
Show inline comments
 
@@ -23,16 +23,26 @@ pub enum PortKind {
 
    Getter,
 
}
 

			




	
	
	
		
 
#[derive(Debug, Clone, Copy, PartialEq, Eq)]

			




	
	
	
		
 
pub enum PortState {

			




	
	
	
		
 
    Open,

			




	
	
	
		
 
    Blocked,

			




	
	
	
		
 
    BlockedDueToPeerChange,

			




	
	
	
		
 
    BlockedDueToFullBuffers,

			




	
	
	
		
 
    Closed,

			




	
	
	
		
 
}

			




	
	
	
		
 

			




	
	
	
		
 
impl PortState {

			




	
	
	
		
 
    pub fn is_blocked(&self) -> bool {

			




	
	
	
		
 
        match self {

			




	
	
	
		
 
            PortState::BlockedDueToPeerChange | PortState::BlockedDueToFullBuffers => true,

			




	
	
	
		
 
            PortState::Open | PortState::Closed => false,

			




	
	
	
		
 
        }

			




	
	
	
		
 
    }

			




	
	
	
		
 
}

			




	
	
	
		
 

			




	
	
	
		
 
pub struct Channel {

			




	
	
	
		
 
    pub putter_id: PortId,

			




	
	
	
		
 
    pub getter_id: PortId,

			




	
	
	
		
 
}

			




	
	
	
		
 

			




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

			




        

    


    
    

    

        

                

                    src/runtime2/component/component_pdl.rs
                

                

                  
                      
                        

                      ➞

                      
                        
                  

                  
                      
                  
                      
                  
                      
                  
                      
                  
                  
                

                

                    Show inline comments
                    
                

        

        

            



	
	
		
 
@@ -230,18 +230,23 @@ impl CompPDL {

			




	
	
	
		
 
            },

			




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

			




	
	
	
		
 
                debug_assert_eq!(self.mode, Mode::Sync);

			




	
	
	
		
 
                let port_id = port_id_from_eval(port_id);

			




	
	
	
		
 
                let port_handle = comp_ctx.get_port_handle(port_id);

			




	
	
	
		
 
                let port_info = comp_ctx.get_port(port_handle);

			




	
	
	
		
 
                if port_info.state == PortState::Blocked {

			




	
	
	
		
 
                    todo!("handle blocked port");

			




	
	
	
		
 
                }

			




	
	
	
		
 
                if port_info.state.is_blocked() {

			




	
	
	
		
 
                    self.mode = Mode::BlockedPut;

			




	
	
	
		
 
                    self.mode_port = port_id;

			




	
	
	
		
 
                    self.mode_value = value;

			




	
	
	
		
 
                    self.exec_ctx.stmt = ExecStmt::PerformedPut; // prepare for when we become unblocked

			




	
	
	
		
 
                    return Ok(CompScheduling::Sleep);

			




	
	
	
		
 
                } else {

			




	
	
	
		
 
                    self.send_data_message_and_wake_up(sched_ctx, comp_ctx, port_handle, value);

			




	
	
	
		
 
                    self.exec_ctx.stmt = ExecStmt::PerformedPut;

			




	
	
	
		
 
                    return Ok(CompScheduling::Immediate);

			




	
	
	
		
 
                }

			




	
	
	
		
 
            },

			




	
	
	
		
 
            // Results that can be returned outside of sync mode

			




	
	
	
		
 
            EC::ComponentTerminated => {

			




	
	
	
		
 
                self.mode = Mode::StartExit; // next call we'll take care of the exit

			




	
	
	
		
 
                return Ok(CompScheduling::Immediate);

			




	
	
	
		
 
            },

			




	
	
		
 
@@ -376,28 +381,28 @@ impl CompPDL {

			




	
	
	
		
 
        let port_index = comp_ctx.get_port_index(port_handle);

			




	
	
	
		
 
        if self.inbox_main[port_index].is_none() {

			




	
	
	
		
 
            self.inbox_main[port_index] = Some(message);

			




	
	
	
		
 

			




	
	
	
		
 
            // After direct insertion, check if this component's execution is 

			




	
	
	
		
 
            // blocked on receiving a message on that port

			




	
	
	
		
 
            debug_assert_ne!(comp_ctx.get_port(port_handle).state, PortState::Blocked); // because we could insert directly

			




	
	
	
		
 
            debug_assert!(!comp_ctx.get_port(port_handle).state.is_blocked()); // because we could insert directly

			




	
	
	
		
 
            if self.mode == Mode::BlockedGet && self.mode_port == target_port_id {

			




	
	
	
		
 
                // We were indeed blocked

			




	
	
	
		
 
                self.mode = Mode::Sync;

			




	
	
	
		
 
                self.mode_port = PortId::new_invalid();

			




	
	
	
		
 
            }

			




	
	
	
		
 
            

			




	
	
	
		
 
            return;

			




	
	
	
		
 
        }

			




	
	
	
		
 

			




	
	
	
		
 
        // The direct inbox is full, so the port will become (or was already) blocked

			




	
	
	
		
 
        let port_info = comp_ctx.get_port_mut(port_handle);

			




	
	
	
		
 
        debug_assert!(port_info.state == PortState::Open || port_info.state == PortState::Blocked);

			




	
	
	
		
 
        debug_assert!(port_info.state == PortState::Open || port_info.state.is_blocked());

			




	
	
	
		
 

			




	
	
	
		
 
        if port_info.state == PortState::Open {

			




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

			




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

			




	
	
	
		
 
            let (peer_handle, message) =

			




	
	
	
		
 
                self.control.initiate_port_blocking(comp_ctx, port_handle);

			




	
	
	
		
 

			




	
	
	
		
 
            let peer = comp_ctx.get_peer(peer_handle);

			




	
	
	
		
 
            peer.handle.send_message(sched_ctx, Message::Control(message), true);

			




	
	
	
		
 
        }

			




	
	
		
 
@@ -418,22 +423,22 @@ impl CompPDL {

			




	
	
	
		
 
        let port_info = comp_ctx.get_port(port_handle);

			




	
	
	
		
 
        for message_index in 0..self.inbox_backup.len() {

			




	
	
	
		
 
            let message = &self.inbox_backup[message_index];

			




	
	
	
		
 
            if message.data_header.target_port == port_info.self_id {

			




	
	
	
		
 
                // One more message for this port

			




	
	
	
		
 
                let message = self.inbox_backup.remove(message_index);

			




	
	
	
		
 
                debug_assert_eq!(comp_ctx.get_port(port_handle).state, PortState::Blocked); // since we had >1 message on the port

			




	
	
	
		
 
                debug_assert!(comp_ctx.get_port(port_handle).state.is_blocked()); // since we had >1 message on the port

			




	
	
	
		
 
                self.inbox_main[port_index] = Some(message);

			




	
	
	
		
 

			




	
	
	
		
 
                return;

			




	
	
	
		
 
            }

			




	
	
	
		
 
        }

			




	
	
	
		
 

			




	
	
	
		
 
        // Did not have any more messages. So if we were blocked, then we need

			




	
	
	
		
 
        // to send the "unblock" message.

			




	
	
	
		
 
        if port_info.state == PortState::Blocked {

			




	
	
	
		
 
        if port_info.state == PortState::BlockedDueToFullBuffers {

			




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

			




	
	
	
		
 
            let (peer_handle, message) = self.control.cancel_port_blocking(comp_ctx, port_handle);

			




	
	
	
		
 
            let peer_info = comp_ctx.get_peer(peer_handle);

			




	
	
	
		
 
            peer_info.handle.send_message(sched_ctx, Message::Control(message), true);

			




	
	
	
		
 
        }

			




	
	
	
		
 
    }

			




	
	
		
 
@@ -458,14 +463,15 @@ impl CompPDL {

			




	
	
	
		
 
            ControlMessageContent::BlockPort(port_id) => {

			




	
	
	
		
 
                // On 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::Closed {

			




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

			




	
	
	
		
 
                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(port_id) => {

			




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

			




	
	
	
		
 
                // the component handle as well

			




	
	
	
		
 
                let port_handle = comp_ctx.get_port_handle(port_id);

			




	
	
		
 
@@ -485,35 +491,34 @@ impl CompPDL {

			




	
	
	
		
 
            },

			




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

			




	
	
	
		
 
                debug_assert!(port_info.state == PortState::Blocked || port_info.state == PortState::Closed);

			




	
	
	
		
 
                if port_info.state == PortState::Blocked {

			




	
	
	
		
 
                if port_info.state == PortState::BlockedDueToFullBuffers {

			




	
	
	
		
 
                    self.handle_unblock_port_instruction(sched_ctx, comp_ctx, port_handle);

			




	
	
	
		
 
                }

			




	
	
	
		
 
            },

			




	
	
	
		
 
            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::Blocked);

			




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

			




	
	
	
		
 

			




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

			




	
	
	
		
 
            },

			




	
	
	
		
 
            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::Blocked);

			




	
	
	
		
 
                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;

			




	
	
		
 
@@ -572,13 +577,13 @@ impl CompPDL {

			




	
	
	
		
 

			




	
	
	
		
 
    /// Unblocks a port, potentially continuing execution of the component, in

			




	
	
	
		
 
    /// response to a message that told us to unblock a previously blocked

			




	
	
	
		
 
    fn handle_unblock_port_instruction(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, port_handle: LocalPortHandle) {

			




	
	
	
		
 
        let port_info = comp_ctx.get_port_mut(port_handle);

			




	
	
	
		
 
        let port_id = port_info.self_id;

			




	
	
	
		
 
        debug_assert_eq!(port_info.state, PortState::Blocked);

			




	
	
	
		
 
        debug_assert!(port_info.state.is_blocked());

			




	
	
	
		
 
        port_info.state = PortState::Open;

			




	
	
	
		
 

			




	
	
	
		
 
        if self.mode == Mode::BlockedPut && port_id == self.mode_port {

			




	
	
	
		
 
            // We were blocked on the port that just became unblocked, so

			




	
	
	
		
 
            // send the message.

			




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

			




        

    


    
    

    

        

                

                    src/runtime2/component/control_layer.rs
                

                

                  
                      
                        

                      ➞

                      
                        
                  

                  
                      
                  
                      
                  
                      
                  
                      
                  
                  
                

                

                    Show inline comments
                    
                

        

        

            



	
	
		
 
@@ -21,13 +21,12 @@ struct ControlEntry {

			




	
	
	
		
 
    content: ControlContent,

			




	
	
	
		
 
}

			




	
	
	
		
 

			




	
	
	
		
 
enum ControlContent {

			




	
	
	
		
 
    PeerChange(ContentPeerChange),

			




	
	
	
		
 
    ScheduleComponent(CompId),

			




	
	
	
		
 
    BlockedPort(PortId),

			




	
	
	
		
 
    ClosedPort(PortId),

			




	
	
	
		
 
}

			




	
	
	
		
 

			




	
	
	
		
 
struct ContentPeerChange {

			




	
	
	
		
 
    source_port: PortId,

			




	
	
	
		
 
    source_comp: CompId,

			




	
	
		
 
@@ -118,13 +117,12 @@ impl ControlLayer {

			




	
	
	
		
 
            },

			




	
	
	
		
 
            ControlContent::ScheduleComponent(to_schedule) => {

			




	
	
	
		
 
                // If all change-of-peers are `Ack`d, then we're ready to

			




	
	
	
		
 
                // schedule the component!

			




	
	
	
		
 
                return (AckAction::ScheduleComponent(to_schedule), None);

			




	
	
	
		
 
            },

			




	
	
	
		
 
            ControlContent::BlockedPort(_) => unreachable!(),

			




	
	
	
		
 
            ControlContent::ClosedPort(closed_port) => {

			




	
	
	
		
 
                // If a closed port is Ack'd, then we remove the reference to

			




	
	
	
		
 
                // that component.

			




	
	
	
		
 
                let port_handle = comp_ctx.get_port_handle(closed_port);

			




	
	
	
		
 
                let port_info = comp_ctx.get_port(port_handle);

			




	
	
	
		
 
                let port_peer_comp_id = port_info.peer_comp_id;

			




	
	
		
 
@@ -240,68 +238,48 @@ impl ControlLayer {

			




	
	
	
		
 
                target_port_id: Some(peer_port_id),

			




	
	
	
		
 
                content: ControlMessageContent::ClosePort(peer_port_id),

			




	
	
	
		
 
            }

			




	
	
	
		
 
        );

			




	
	
	
		
 
    }

			




	
	
	
		
 

			




	
	
	
		
 
    /// Adds a control entry to track that a port is blocked. Expects the caller

			




	
	
	
		
 
    /// to have set the port's state to blocking already. The returned tuple

			




	
	
	
		
 
    /// contains a message and the peer to send it to.

			




	
	
	
		
 
    /// Generates the control message used to indicate to a peer that a port

			




	
	
	
		
 
    /// should be blocked (expects the caller to have set the port's state to

			




	
	
	
		
 
    /// blocked).

			




	
	
	
		
 
    pub(crate) fn initiate_port_blocking(&mut self, comp_ctx: &CompCtx, port_handle: LocalPortHandle) -> (LocalPeerHandle, ControlMessage) {

			




	
	
	
		
 
        let port_info = comp_ctx.get_port(port_handle);

			




	
	
	
		
 
        debug_assert_eq!(port_info.kind, PortKind::Getter); // because we're telling the putter to block

			




	
	
	
		
 
        debug_assert_eq!(port_info.state, PortState::Blocked); // contract with caller

			




	
	
	
		
 
        debug_assert_eq!(port_info.state, PortState::BlockedDueToFullBuffers); // contract with caller

			




	
	
	
		
 

			




	
	
	
		
 
        let peer_port_id = port_info.peer_port_id;

			




	
	
	
		
 
        let peer_comp_id = port_info.peer_comp_id;

			




	
	
	
		
 
        let peer_handle = comp_ctx.get_peer_handle(peer_comp_id);

			




	
	
	
		
 

			




	
	
	
		
 
        let entry_id = self.take_id();

			




	
	
	
		
 
        self.entries.push(ControlEntry{

			




	
	
	
		
 
            id: entry_id,

			




	
	
	
		
 
            ack_countdown: 0,

			




	
	
	
		
 
            content: ControlContent::BlockedPort(port_info.self_id),

			




	
	
	
		
 
        });

			




	
	
	
		
 

			




	
	
	
		
 
        return (

			




	
	
	
		
 
            peer_handle,

			




	
	
	
		
 
            ControlMessage{

			




	
	
	
		
 
                id: entry_id,

			




	
	
	
		
 
                id: ControlId::new_invalid(),

			




	
	
	
		
 
                sender_comp_id: comp_ctx.id,

			




	
	
	
		
 
                target_port_id: Some(port_info.peer_port_id),

			




	
	
	
		
 
                content: ControlMessageContent::BlockPort(peer_port_id),

			




	
	
	
		
 
            }

			




	
	
	
		
 
        );

			




	
	
	
		
 
    }

			




	
	
	
		
 

			




	
	
	
		
 
    /// Removes the control entry that tracks that a port is blocked. Expects

			




	
	
	
		
 
    /// the caller to have already marked the port as unblocked. Again the

			




	
	
	
		
 
    /// returned tuple contains a message and the target it is intended for

			




	
	
	
		
 
    /// Generates a messages used to indicate to a peer that a port should be

			




	
	
	
		
 
    /// unblocked again.

			




	
	
	
		
 
    pub(crate) fn cancel_port_blocking(&mut self, comp_ctx: &CompCtx, port_handle: LocalPortHandle) -> (LocalPeerHandle, ControlMessage) {

			




	
	
	
		
 
        let port_info = comp_ctx.get_port(port_handle);

			




	
	
	
		
 
        debug_assert_eq!(port_info.kind, PortKind::Getter); // because we're initiating the unblocking

			




	
	
	
		
 
        debug_assert_eq!(port_info.state, PortState::Open); // contract with caller, the locally stored entry ensures we were blocked before

			




	
	
	
		
 

			




	
	
	
		
 
        let position = self.entries.iter()

			




	
	
	
		
 
            .position(|v| {

			




	
	
	
		
 
                if let ControlContent::BlockedPort(blocked_port_id) = &v.content {

			




	
	
	
		
 
                    if *blocked_port_id == port_info.self_id {

			




	
	
	
		
 
                        return true;

			




	
	
	
		
 
                    }

			




	
	
	
		
 
                }

			




	
	
	
		
 
                return false;

			




	
	
	
		
 
            })

			




	
	
	
		
 
            .unwrap();

			




	
	
	
		
 

			




	
	
	
		
 
        let entry = self.entries.remove(position);

			




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

			




	
	
	
		
 

			




	
	
	
		
 
        return (

			




	
	
	
		
 
            peer_handle,

			




	
	
	
		
 
            ControlMessage{

			




	
	
	
		
 
                id: entry.id,

			




	
	
	
		
 
                id: ControlId::new_invalid(),

			




	
	
	
		
 
                sender_comp_id: comp_ctx.id,

			




	
	
	
		
 
                target_port_id: Some(port_info.peer_port_id),

			




	
	
	
		
 
                content: ControlMessageContent::UnblockPort(port_info.peer_port_id)

			




	
	
	
		
 
            }

			




	
	
	
		
 
        );

			




	
	
	
		
 
    }

			




        

    


    
    

    

        

                

                    src/runtime2/tests/mod.rs
                

                

                  
                      
                        

                      ➞

                      
                        
                  

                  
                      
                  
                      
                  
                      
                  
                      
                  
                  
                

                

                    Show inline comments
                    
                

        

        

            



	
	
		
 
@@ -28,29 +28,53 @@ fn test_component_creation() {

			




	
	
	
		
 
    for i in 0..20 {

			




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

			




	
	
	
		
 
    }

			




	
	
	
		
 
}

			




	
	
	
		
 

			




	
	
	
		
 
#[test]

			




	
	
	
		
 
fn test_component_communication() {

			




	
	
	
		
 
fn test_component_communication_a() {

			




	
	
	
		
 
    let pd = ProtocolDescription::parse(b"

			




	
	
	
		
 
    primitive sender(out<u32> o) {

			




	
	
	
		
 
        print(\"sender\");

			




	
	
	
		
 
        sync put(o, 1);

			




	
	
	
		
 
    }

			




	
	
	
		
 
    primitive receiver(in<u32> i) {

			




	
	
	
		
 
        print(\"receiver\");

			




	
	
	
		
 
        sync auto a = get(i);

			




	
	
	
		
 
    }

			




	
	
	
		
 
    composite constructor() {

			




	
	
	
		
 
        channel o -> i;

			




	
	
	
		
 
        print(\"creating sender\");

			




	
	
	
		
 
        new sender(o);

			




	
	
	
		
 
        print(\"creating receiver\");

			




	
	
	
		
 
        new receiver(i);

			




	
	
	
		
 
        print(\"done\");

			




	
	
	
		
 
    }

			




	
	
	
		
 
    ").expect("compilation");

			




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

			




	
	
	
		
 

			




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

			




	
	
	
		
 
}

			




	
	
	
		
 

			




	
	
	
		
 
#[test]

			




	
	
	
		
 
fn test_component_communication_b() {

			




	
	
	
		
 
    let pd = ProtocolDescription::parse(b"

			




	
	
	
		
 
    primitive sender(out<u32> o, u32 rounds) {

			




	
	
	
		
 
        u32 index = 0;

			




	
	
	
		
 
        sync while (index < rounds) {

			




	
	
	
		
 
            put(o, index);

			




	
	
	
		
 
            index += 1;

			




	
	
	
		
 
        }

			




	
	
	
		
 
    }

			




	
	
	
		
 

			




	
	
	
		
 
    primitive receiver(in<u32> i, u32 rounds) {

			




	
	
	
		
 
        u32 index = 0;

			




	
	
	
		
 
        sync while (index < rounds) {

			




	
	
	
		
 
            auto val = get(i);

			




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

			




	
	
	
		
 
            index += 1;

			




	
	
	
		
 
        }

			




	
	
	
		
 
    }

			




	
	
	
		
 

			




	
	
	
		
 
    composite constructor() {

			




	
	
	
		
 
        channel o -> i;

			




	
	
	
		
 
        new sender(o, 5);

			




	
	
	
		
 
        new receiver(i, 5);

			




	
	
	
		
 
    }").expect("compilation");

			




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

			




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

			




	
	
	
		
 
}

			




	
	
		
 
\ No newline at end of file

			




        

    



    


    



    



      

      





    
    0 comments (0 inline, 0 general)
    





    


  

  







  


    




    








    
        Server instance: kalli-1039567
    
    
        This site is powered by
            Kallithea,
        which is
        © 2010–2022 by various authors & licensed under GPLv3.
            – Support