10 sources (fused, 1 sink

test_10fusedx1.splmm

namespace TR;

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

composite ThreadTest {
  graph
    stream<uint64 x> Control = Beacon() {
      param
        iterations: 2u;
        period: 60.0;
      output
        Control: x=IterationCount();
    }
    
    <% for (my $j = 0; $j < $numSource; $j = $j + 1) { %>
    stream<uint64 x> X<%= $j %> = Custom() {
      logic
        onProcess: {
          while (true) {
            submit({x=(uint64)0}, X<%= $j %>);
          }          
      }
      config
        placement: host(NodePool);      
    }
    <% } %>
    
    <% for (my $i = 0; $i < $numSink; $i = $i + 1) { %>
    () as Sink<%= $i %> = Custom(
      <% for (my $j = 0; $j < $numSource; $j = $j + 1) { %>
        X<%= $j %>;
      <% } %>
    Control) {
      logic
        state: {
          mutable uint64 c = 0;
          mutable boolean isStarted = false;
        }
        onTuple X0: {
          //printStringLn("GOT");
          if (isStarted) {
            c += (uint64)1;
          }
        }
        onTuple Control: {
          if (x == (uint64)0) {
            isStarted = true;
          } else {
            isStarted = false;
            printStringLn((rstring)c);
          }
        }
        
      config
        placement: host(RfaPool);
          
    }
    <% } %>  
    
  config
    hostPool:
      NodePool=createPool({tags=["nodejs"]}, Sys.Shared),
      RfaPool=createPool({tags=["rfa"]}, Sys.Shared);
      
}