use(channels). pipeline_stage(i:var(chan(void)), o:var(chan(void)), del:val(int), name:val(string)) : expr(void) = ( newstringvalparam(name); forever( i?null; bview_set_colour(name,"red"); wait_for(del); o!null; bview_set_colour(name,"white") ); endstringvalparam(name) ). pipeline_fork(i:var(chan(void)), o1:var(chan(void)), o2:var(chan(void)), del:val(int), name:val(string)) : expr(void) = ( newstringvalparam(name); forever( i?null; bview_set_colour(name,"red"); wait_for(del); ( o1!null | o2!null ); bview_set_colour(name,"white") ); endstringvalparam(name) ). pipeline_join(i1:var(chan(void)), i2:var(chan(void)), o:var(chan(void)), del:val(int), name:val(string)) : expr(void) = ( newstringvalparam(name); forever( ( i1?null | i2?null ); bview_set_colour(name,"red"); wait_for(del); o!null; bview_set_colour(name,"white") ); endstringvalparam(name) ). C1:var(chan(void)). C2:var(chan(void)). C3:var(chan(void)). C4:var(chan(void)). C5:var(chan(void)). C6:var(chan(void)). C7:var(chan(void)). C8:var(chan(void)). init_chan(C1,"C1"); init_chan(C2,"C2"); init_chan(C3,"C3"); init_chan(C4,"C4"); init_chan(C5,"C5"); init_chan(C6,"C6"); init_chan(C7,"C7"); init_chan(C8,"C8"); ( forever(C1!null) | pipeline_fork(C1,C2,C5,10,"fork") | pipeline_stage(C2,C3,10,"A1") | pipeline_stage(C3,C4,10,"A2") | pipeline_stage(C5,C6,10,"B1") | pipeline_join(C4,C6,C7,10,"join") | pipeline_stage(C7,C8,10,"C1") | forever(C8?null) )