Enable actions to be scheduled asyncronously (#280)

* Enable actions to be scheduled asyncronously

* Format

* Remove unnecessary info printing

* Format

Author: erlingrj

include/reactor-uc/macros_api.h

@@ -58,10 +58,11 @@
 #define lf_schedule_with_val(action, offset, val)                                                                      \
   do {                                                                                                                 \
     __typeof__(val) __val = (val);                                                                                     \
-    lf_ret_t ret = (action)->super.schedule(&(action)->super, (offset), (const void *)&__val);                         \
+    Action *__a = (Action *)(action);                                                                                  \
+    lf_ret_t ret = __a->schedule(__a, (offset), (const void *)&__val);                                                 \
     if (ret == LF_FATAL) {                                                                                             \
       LF_ERR(TRIG, "Scheduling an value, that doesn't have value!");                                                   \
-      Scheduler *sched = (action)->super.super.parent->env->scheduler;                                                 \
+      Scheduler *sched = __a->super.parent->env->scheduler;                                                            \
       sched->do_shutdown(sched, sched->current_tag(sched));                                                            \
       throw("Tried to schedule a value onto an action without a type!");                                               \
     }                                                                                                                  \
@@ -70,7 +71,8 @@
 /// @private
 #define lf_schedule_without_val(action, offset)                                                                        \
   do {                                                                                                                 \
-    (action)->super.schedule(&(action)->super, (offset), NULL);                                                        \
+    Action *__a = (Action *)(action);                                                                                  \
+    __a->schedule(__a, (offset), NULL);                                                                                \
   } while (0)
 
 /// @private
@@ -104,11 +106,12 @@
  */
 #define lf_schedule_array(action, offset, array)                                                                       \
   do {                                                                                                                 \
-    lf_ret_t ret = (action)->super.schedule(&(action)->super, (offset), (const void *)array);                          \
-    if (ret == LF_FATAL) {                                                                                             \
+    Action *__a = (Action *)(action);                                                                                  \
+    lf_ret_t __ret = __a->schedule(__a, (offset), (const void *)array);                                                \
+    if (__ret == LF_FATAL) {                                                                                           \
       LF_ERR(TRIG, "Scheduling an value, that doesn't have value!");                                                   \
-      Scheduler *sched = (action)->super.super.parent->env->scheduler;                                                 \
-      sched->do_shutdown(sched, sched->current_tag(sched));                                                            \
+      Scheduler *__sched = __a->super.parent->env->scheduler;                                                          \
+      __sched->do_shutdown(__sched, __sched->current_tag(__sched));                                                    \
     }                                                                                                                  \
   } while (0)
 

src/environments/unfederated_environment.c

@@ -110,5 +110,5 @@ void Environment_ctor(Environment *self, Reactor *main, Scheduler *scheduler, bo
 
 void Environment_free(Environment *self) {
   (void)self;
-  LF_INFO(ENV, "Freeing top-level environment.");
+  LF_DEBUG(ENV, "Freeing top-level environment.");
 }

src/federated.c

@@ -138,8 +138,8 @@ void FederatedInputConnection_ctor(FederatedInputConnection *self, Reactor *pare
 // a TaggedMessage available.
 void FederatedConnectionBundle_handle_tagged_msg(FederatedConnectionBundle *self, const FederateMessage *_msg) {
   const TaggedMessage *msg = &_msg->message.tagged_message;
-  LF_INFO(FED, "Callback on FedConnBundle %p for message of size=%u with tag:" PRINTF_TAG, self, msg->payload.size,
-          msg->tag);
+  LF_DEBUG(FED, "Callback on FedConnBundle %p for message of size=%u with tag:" PRINTF_TAG, self, msg->payload.size,
+           msg->tag);
   assert(((size_t)msg->conn_id) < self->inputs_size);
   lf_ret_t ret;
   FederatedInputConnection *input = self->inputs[msg->conn_id];

src/platform/posix/tcp_ip_channel.c

@@ -121,7 +121,7 @@ static lf_ret_t _TcpIpChannel_reset_socket(TcpIpChannel *self) {
 
 static void _TcpIpChannel_spawn_worker_thread(TcpIpChannel *self) {
   int res;
-  TCP_IP_CHANNEL_INFO("Spawning worker thread");
+  TCP_IP_CHANNEL_DEBUG("Spawning worker thread");
 
   memset(&self->worker_thread_stack, 0, TCP_IP_CHANNEL_RECV_THREAD_STACK_SIZE);
   if (pthread_attr_init(&self->worker_thread_attr) != 0) {
@@ -414,7 +414,7 @@ static void *_TcpIpChannel_worker_thread(void *untyped_self) {
   lf_ret_t ret;
   int res;
 
-  TCP_IP_CHANNEL_INFO("Starting worker thread");
+  TCP_IP_CHANNEL_DEBUG("Starting worker thread");
 
   while (true) {
     // Check if we have any pending cancel requests from the runtime.

src/startup_coordinator.c

@@ -14,7 +14,7 @@ static lf_ret_t StartupCoordinator_connect_to_neighbors_blocking(StartupCoordina
   FederatedEnvironment *env_fed = (FederatedEnvironment *)self->env;
   validate(self->state == StartupCoordinationState_UNINITIALIZED);
   self->state = StartupCoordinationState_CONNECTING;
-  LF_INFO(FED, "%s connecting to %zu federated peers", self->env->main->name, env_fed->net_bundles_size);
+  LF_DEBUG(FED, "%s connecting to %zu federated peers", self->env->main->name, env_fed->net_bundles_size);
   lf_ret_t ret;
 
   // Open all connections.
@@ -50,8 +50,8 @@ static lf_ret_t StartupCoordinator_connect_to_neighbors_blocking(StartupCoordina
     }
   }
 
-  LF_INFO(FED, "%s Established connection to all %zu federated peers", self->env->main->name,
-          env_fed->net_bundles_size);
+  LF_DEBUG(FED, "%s Established connection to all %zu federated peers", self->env->main->name,
+           env_fed->net_bundles_size);
   self->state = StartupCoordinationState_HANDSHAKING;
   return LF_OK;
 }
@@ -184,7 +184,7 @@ static void StartupCoordinator_handle_startup_handshake_response(StartupCoordina
     }
 
     if (all_received) {
-      LF_INFO(FED, "Handshake completed with %zu federated peers", self->num_neighbours);
+      LF_DEBUG(FED, "Handshake completed with %zu federated peers", self->num_neighbours);
       self->state = StartupCoordinationState_NEGOTIATING;
       // Schedule the start time negotiation to occur immediately.
       StartupCoordinator_schedule_system_self_event(self, self->env->get_physical_time(self->env) + MSEC(50),
@@ -289,8 +289,8 @@ static void StartupCoordinator_handle_start_time_proposal(StartupCoordinator *se
   }
 
   if (iteration_completed) {
-    LF_INFO(FED, "Start time negotiation round %d completed. Current start time: " PRINTF_TIME,
-            self->start_time_proposal_step, self->start_time_proposal);
+    LF_DEBUG(FED, "Start time negotiation round %d completed. Current start time: " PRINTF_TIME,
+             self->start_time_proposal_step, self->start_time_proposal);
     if (self->start_time_proposal_step == self->longest_path) {
       LF_INFO(FED, "Start time negotiation completed Starting at " PRINTF_TIME, self->start_time_proposal);
       self->state = StartupCoordinationState_RUNNING;