CSY/reowolf Changeset - aebcee16d6bc · Centrum Wiskunde & Informatica (CWI)

Changeset - aebcee16d6bc

Parent rev.

Child rev.

[Not reviewed]

1 7 6

Christopher Esterhuyse - 5 years ago 2020-06-26 11:16:39
christopher.esterhuyse@gmail.com

fixed cyclic routing bug. Was actually a one-liner

14 files changed with 151 insertions and 92 deletions:

examples/6_amy_log.txt

examples/6_atomic/amy.c

examples/7_recovery/amy.c

examples/7_recovery/make.sh

examples/8_net_ports/amy.c

examples/8_net_ports/make.sh

examples/9_net_self_putget/amy.c

examples/9_net_self_putget/make.sh

examples/reowolf_rs.dll

bin+mod

reowolf.h

src/runtime/communication.rs

src/runtime/ffi.rs

src/runtime/setup.rs

src/runtime/tests.rs

0 comments (0 inline, 0 general)

examples/6_amy_log.txt

➞

Show inline comments

deleted file

examples/6_atomic/amy.c

➞

Show inline comments

@@ @@ -16,24 +16,14 @@ int main(int argc, char** argv) { @@
 	connector_connect(c, -1);
 	connector_print_debug(c);
 	printf("Let's try to put without get\n");
 	connector_put_bytes(c, putter, "hello", 5);
 	// connector_get(c, getter);
 	int err = connector_sync(c, 5000);
 	printf("Error code %d with string `%s`\n", err, reowolf_error_peek(NULL));
 	printf("Let's try again, doing both\n");
 	connector_put_bytes(c, putter, "hello", 5);
 	connector_get(c, getter);
 	err = connector_sync(c, 5000);
 	printf("Error code %d with string `%s`\n", err, reowolf_error_peek(NULL));
 	size_t msg_len;
 	const char * msg_ptr = connector_gotten_bytes(c, getter, &msg_len);
 	printf("Got msg `%.*s`\n", msg_len, msg_ptr);
 	protocol_description_destroy(pd);
 	connector_destroy(c);
 	free(pdl_ptr);
 	return 0;
+}
@@ \ No newline at end of file @@

examples/7_recovery/amy.c

➞

Show inline comments

@@ new file 100644 @@
 #include <stdio.h>
 #include <string.h>
 #include "../../reowolf.h"
 #include "../utility.c"
 int main(int argc, char** argv) {
 	char * pdl_ptr = buffer_pdl("eg_protocols.pdl");
 	size_t pdl_len = strlen(pdl_ptr);
 	Arc_ProtocolDescription * pd = protocol_description_parse(pdl_ptr, pdl_len);
 	char logpath[] = "./7_amy_log.txt";
 	Connector * c = connector_new_logging(pd, logpath, sizeof(logpath)-1);
 	printf("Err %s\n", reowolf_error_peek(NULL));
 	PortId putter, getter;
 	connector_add_port_pair(c, &putter, &getter);
 	connector_connect(c, -1);
 	connector_print_debug(c);
 	printf("Let's try to put without get\n");
 	connector_put_bytes(c, putter, "hello", 5);
 	// connector_get(c, getter);
 	int err = connector_sync(c, 5000);
 	printf("Error code %d with string `%s`\n", err, reowolf_error_peek(NULL));
 	printf("\nLet's try again, doing both\n");
 	connector_put_bytes(c, putter, "hello", 5);
 	connector_get(c, getter);
 	err = connector_sync(c, 5000);
 	printf("Error code %d with string `%s`\n", err, reowolf_error_peek(NULL));
 	size_t msg_len;
 	const char * msg_ptr = connector_gotten_bytes(c, getter, &msg_len);
 	printf("Got msg `%.*s`\n", msg_len, msg_ptr);
 	protocol_description_destroy(pd);
 	connector_destroy(c);
 	free(pdl_ptr);
 	return 0;
+}
@@ \ No newline at end of file @@

examples/7_recovery/make.sh

➞

Show inline comments

@@ new file 100644 @@
 #!/bin/sh
 LIB_PATH="../../target/release"
 gcc -L $LIB_PATH -lreowolf_rs -Wl,-R$LIB_PATH amy.c -o ./amy

examples/8_net_ports/amy.c

➞

Show inline comments

@@ new file 100644 @@
 #include <stdio.h>
 #include <string.h>
 #include "../../reowolf.h"
 #include "../utility.c"
 int main(int argc, char** argv) {
 	Arc_ProtocolDescription * pd = protocol_description_parse("", 0);
 	char logpath[] = "./8_amy_log.txt";
 	Connector * c = connector_new_logging(pd, logpath, sizeof(logpath)-1);
 	printf("Error str `%s`\n", reowolf_error_peek(NULL));
 	PortId putter, getter;
 	char addr_str[] = "127.0.0.1:8000";
 	connector_add_net_port(
 		c, &putter, addr_str, sizeof(addr_str)-1, Putter, Active);
 	printf("Error str `%s`\n", reowolf_error_peek(NULL));
 	connector_add_net_port(
 		c, &getter, addr_str, sizeof(addr_str)-1, Getter, Passive);
 	printf("Error str `%s`\n", reowolf_error_peek(NULL));
 	connector_connect(c, 4000);
 	printf("Error str `%s`\n", reowolf_error_peek(NULL));
 	protocol_description_destroy(pd);
 	connector_destroy(c);
 	return 0;
+}
@@ \ No newline at end of file @@

examples/8_net_ports/make.sh

➞

Show inline comments

@@ new file 100644 @@
 #!/bin/sh
 LIB_PATH="../../target/release"
 gcc -L $LIB_PATH -lreowolf_rs -Wl,-R$LIB_PATH amy.c -o ./amy

examples/9_net_self_putget/amy.c

➞

Show inline comments

@@ new file 100644 @@
 #include <stdio.h>
 #include <string.h>
 #include "../../reowolf.h"
 #include "../utility.c"
 int main(int argc, char** argv) {
 	Arc_ProtocolDescription * pd = protocol_description_parse("", 0);
 	char logpath[] = "./9_amy_log.txt";
 	Connector * c = connector_new_logging(pd, logpath, sizeof(logpath)-1);
 	printf("Error str `%s`\n", reowolf_error_peek(NULL));
 	PortId putter, getter;
 	char addr_str[] = "127.0.0.1:8000";
 	connector_add_net_port(
 		c, &putter, addr_str, sizeof(addr_str)-1, Putter, Active);
 	printf("Error str `%s`\n", reowolf_error_peek(NULL));
 	connector_add_net_port(
 		c, &getter, addr_str, sizeof(addr_str)-1, Getter, Passive);
 	printf("Error str `%s`\n", reowolf_error_peek(NULL));
 	connector_connect(c, 4000);
 	printf("Error str `%s`\n", reowolf_error_peek(NULL));
 	connector_put_bytes(c, putter, "hi", 2);
 	printf("Error str `%s`\n", reowolf_error_peek(NULL));
 	connector_get(c, getter);
 	printf("Error str `%s`\n", reowolf_error_peek(NULL));
 	connector_sync(c, 4000);
 	printf("Error str `%s`\n", reowolf_error_peek(NULL));
 	protocol_description_destroy(pd);
 	connector_destroy(c);
 	return 0;
+}
@@ \ No newline at end of file @@

examples/9_net_self_putget/make.sh

➞

Show inline comments

@@ new file 100644 @@
 #!/bin/sh
 LIB_PATH="../../target/release"
 gcc -L $LIB_PATH -lreowolf_rs -Wl,-R$LIB_PATH amy.c -o ./amy

examples/reowolf_rs.dll

➞

Show inline comments

binary diff not shown

reowolf.h

➞

Show inline comments

@@ @@ -55,17 +55,17 @@ ErrorCode connector_add_component(Connector *connector, @@
  * - a utf-8 encoded socket address,
  * - the logical polarity of P,
  * - the "physical" polarity in {Active, Passive} of the endpoint through which P's peer will be discovered,
  * returns P, a port newly added to the native interface.
  */
 ErrorCode connector_add_net_port(Connector *connector,
                                  PortId *port,
                                  const uint8_t *addr_str_ptr,
                                  uintptr_t addr_str_len,
                                  Polarity port_polarity,
                                  EndpointPolarity endpoint_polarity,
                                  PortId *port);
                                  EndpointPolarity endpoint_polarity);
 /**
  * Given an initialized connector in setup or connecting state,
  * - Creates a new directed port pair with logical channel putter->getter,
  * - adds the ports to the native component's interface,
  * - and returns them using the given out pointers.

src/runtime/communication.rs

➞

Show inline comments

@@ @@ -394,15 +394,18 @@ impl Connector { @@
         log!(cu.logger, "All proto components are blocked");
         log!(cu.logger, "Entering decision loop...");
         comm.endpoint_manager.undelay_all();
         'undecided: loop {
             // drain payloads_to_get, sending them through endpoints / feeding them to components
             log!(cu.logger, "Decision loop! have {} messages to recv", payloads_to_get.len());
             while let Some((getter, send_payload_msg)) = payloads_to_get.pop() {
                 assert!(cu.port_info.polarities.get(&getter).copied() == Some(Getter));
                 match cu.port_info.routes.get(&getter) {
                 let route = cu.port_info.routes.get(&getter);
                 log!(cu.logger, "Routing msg {:?} to {:?}", &send_payload_msg, &route);
                 match route {
                     None => {
                         log!(
                             cu.logger,
                             "Delivery to getter {:?} msg {:?} failed. Physical route unmapped!",
                             getter,
                             &send_payload_msg

src/runtime/ffi.rs

➞

Show inline comments

@@ @@ -214,17 +214,17 @@ pub unsafe extern "C" fn connector_add_component( @@
 /// - the logical polarity of P,
 /// - the "physical" polarity in {Active, Passive} of the endpoint through which P's peer will be discovered,
 /// returns P, a port newly added to the native interface.
 #[no_mangle]
 pub unsafe extern "C" fn connector_add_net_port(
     connector: &mut Connector,
     port: *mut PortId,
     addr_str_ptr: *const u8,
     addr_str_len: usize,
     port_polarity: Polarity,
     endpoint_polarity: EndpointPolarity,
     port: *mut PortId,
 ) -> ErrorCode {
     StoredError::tl_clear();
     let addr_bytes = &*slice_from_parts(addr_str_ptr, addr_str_len);
     let addr_str = match std::str::from_utf8(addr_bytes) {
         Ok(addr_str) => addr_str,
         Err(err) => {

src/runtime/setup.rs

➞

Show inline comments

@@ @@ -144,13 +144,15 @@ fn new_endpoint_manager( @@
+    }
     ////////////////////////////////////////////
     // 1. Start to construct EndpointManager
     const WAKER_TOKEN: Token = Token(usize::MAX);
     const WAKER_PERIOD: Duration = Duration::from_millis(90);
     assert!(endpoint_setups.len() < WAKER_TOKEN.0); // using MAX usize as waker token
     let mut waker_continue_signal: Option<Arc<AtomicBool>> = None;
     let mut poll = Poll::new().map_err(|_| PollInitFailed)?;
     let mut events = Events::with_capacity(endpoint_setups.len() * 2 + 4);
     let mut polled_undrained = IndexSet::default();
     let mut delayed_messages = vec![];
@@ @@ -291,13 +293,25 @@ fn new_endpoint_manager( @@
                                 todo.recv_peer_port = Some(peer_info.port);
                                 // 1. finally learned the peer of this port!
                                 port_info.peers.insert(todo.local_port, peer_info.port);
                                 // 2. learned the info of this peer port
                                 port_info.polarities.insert(peer_info.port, peer_info.polarity);
                                 port_info.peers.insert(peer_info.port, todo.local_port);
                                 port_info.routes.insert(peer_info.port, Route::Endpoint { index });
                                 if let Some(route) = port_info.routes.get(&peer_info.port) {
                                     // check just for logging purposes
                                     log!(
                                         logger,
                                         "Special case! Route to peer {:?} already known to be {:?}. Leave untouched",
                                         peer_info.port,
                                         route
                                     );
+                                }
                                 port_info
                                     .routes
                                     .entry(peer_info.port)
                                     .or_insert(Route::Endpoint { index });
+                            }
                             Some(inappropriate_msg) => {
                                 log!(
                                     logger,
                                     "delaying msg {:?} during channel setup phase",
                                     inappropriate_msg

src/runtime/tests.rs

➞

Show inline comments

@@ @@ -501,6 +501,19 @@ fn chain_connect() { @@
             c.new_net_port(Getter, sock_addrs[3], Active).unwrap();
             c.connect(Some(Duration::from_secs(1))).unwrap();
         });
     })
     .unwrap();
+}
 #[test]
 fn net_self_loop() {
     let test_log_path = Path::new("./logs/net_self_loop");
     let sock_addr = next_test_addr();
     let mut c = file_logged_connector(0, test_log_path);
     let p = c.new_net_port(Putter, sock_addr, Active).unwrap();
     let g = c.new_net_port(Getter, sock_addr, Passive).unwrap();
     c.connect(Some(Duration::from_secs(1))).unwrap();
     c.put(p, TEST_MSG.clone()).unwrap();
     c.get(g).unwrap();
     c.sync(Some(Duration::from_millis(500))).unwrap();
+}

0 comments (0 inline, 0 general)