Centralized coordination fails with usage of broadcast connections

[erlingrj]

I am getting STP violations on centrally coordinated programs that are using broadcast connections. This might be solved by #2459

To reproduce run this program:

target C;

reactor Wrapper {
  output [2] out: int
  input [2] in: int

  r = new R()

  // This connection statement causes STP violations
  (r.out)+ -> out

  reaction(in) -> r.in {=
    lf_set(r.in, 0);
  =}

}

reactor R {
  output out: int
  input in: int
  preamble {=
    // Schedule an event roughly every 200 msec.
    void* external(void* a) {
      int i = 0;
      while (true) {
        lf_sleep(MSEC(500));
        lf_schedule_int(a, 0, i++);
      }
    }
  =}

  state thread_id: lf_thread_t = 0
  physical action a: int
  timer t (0, 10 msec)

  reaction(startup) -> a {=
    // Start a thread to schedule physical actions.
    lf_thread_create(&self->thread_id, &external, a);
  =}

  reaction(a) -> out {=
    interval_t elapsed_time = lf_time_logical_elapsed();
    printf("Action triggered with %d at logical time %lld nsec after start.\n", a->value, elapsed_time);
    lf_set(out, a->value);
  =}

  reaction(t) {= =}


  reaction(in) {=
    printf("Received input at logical time %lld nsec after start.\n", lf_time_logical_elapsed());
  =}
}


federated reactor {
  r = new [2] Wrapper()

  r.out -> interleaved(r.in)
}

The problem goes away by replacing the broadcast connection with a reaction:

target C;

reactor Wrapper {
  output [2] out: int
  input [2] in: int

  r = new R()

// Replacing broadcast connection with a reaction and it works
  reaction(r.out) -> out {=
    lf_set(out[0], r.out->value);
    lf_set(out[1], r.out->value);
  =}

  reaction(in) -> r.in {=
    lf_set(r.in, 0);
  =}

}

reactor R {
  output out: int
  input in: int
  preamble {=
    // Schedule an event roughly every 200 msec.
    void* external(void* a) {
      int i = 0;
      while (true) {
        lf_sleep(MSEC(500));
        lf_schedule_int(a, 0, i++);
      }
    }
  =}

  state thread_id: lf_thread_t = 0
  physical action a: int
  timer t (0, 10 msec)

  reaction(startup) -> a {=
    // Start a thread to schedule physical actions.
    lf_thread_create(&self->thread_id, &external, a);
  =}

  reaction(a) -> out {=
    interval_t elapsed_time = lf_time_logical_elapsed();
    printf("Action triggered with %d at logical time %lld nsec after start.\n", a->value, elapsed_time);
    lf_set(out, a->value);
  =}

  reaction(t) {= =}


  reaction(in) {=
    printf("Received input at logical time %lld nsec after start.\n", lf_time_logical_elapsed());
  =}
}


federated reactor {
  r = new [2] Wrapper()

  r.out -> interleaved(r.in)
}