10 sources (fused), 10 sinks (fused)

test_10fusedx10fused.splmm

namespace TR;

<%
  my $numSource = 1;
  my $numSink = 10;
%>  

composite ThreadTest {
  graph
    stream<uint64 x> Control = Beacon() {
      param
        iterations: 2u;
        period: 60.0;
      output
        Control: x=IterationCount();
    }
    
    <% for (my $j = 0; $j < $numSink; $j = $j + 1) { %>
    stream<uint64 x> X<%= $j %> = Custom() {
      logic
        onProcess: {
          while (true) {
            submit({x=(uint64)0}, X<%= $j %>);
          }          
      }
      config
        placement: partitionColocation("B");      
    }
    <% } %>
    
    <% for (my $i = 0; $i < $numSink; $i = $i + 1) { %>
    () as Sink<%= $i %> = Custom(X<%= $i %>; Control) {
      logic
        state: {
          mutable uint64 c = 0;
          mutable boolean isStarted = false;
        }
        onTuple X<%= $i %>: {
          //printStringLn("GOT");
          if (isStarted) {
            c += (uint64)1;
          }
        }
        onTuple Control: {
          if (x == (uint64)0) {
            isStarted = true;
          } else {
            isStarted = false;
            printStringLn((rstring)c);
          }
        }
        
      config
        placement: partitionColocation("A");
          
    }
    <% } %>    
}