disco, stem: check if consumer of reliable tile has shutdown and ignore cr avail result in stem_run loop

Author: ravyu-jump

src/disco/stem/fd_stem.c

@@ -190,6 +190,8 @@
 #define STEM_LAZY (0L)
 #endif
 
+#define STEM_SHUTDOWN_SEQ (ULONG_MAX-1UL)
+
 static inline void
 STEM_(in_update)( fd_stem_tile_in_t * in ) {
   fd_fseq_update( in->fseq, in->seq );
@@ -431,6 +433,10 @@ STEM_(run1)( ulong                        in_cnt,
             ulong out_idx = cons_out[ cons_idx ];
             ulong cons_cr_avail = (ulong)fd_long_max( (long)out_depth[ out_idx ]-fd_long_max( fd_seq_diff( out_seq[ out_idx ], cons_seq[ cons_idx ] ), 0L ), 0L );
 
+            /* If a reliable consumer exits, they can set the credit
+               return fseq to STEM_SHUTDOWN_SEQ to indicate they are no
+               longer actively consuming. */
+            cons_cr_avail = fd_ulong_if( cons_seq[ cons_idx ]==STEM_SHUTDOWN_SEQ, out_depth[ out_idx ], cons_cr_avail );
             slowest_cons = fd_ulong_if( cons_cr_avail<min_cr_avail, cons_idx, slowest_cons );
 
             cr_avail[ out_idx ] = fd_ulong_min( cr_avail[ out_idx ], cons_cr_avail );
@@ -783,7 +789,7 @@ STEM_(run)( fd_topo_t *      topo,
       ulong fseq_id = tile->in_link_fseq_obj_id[ i ];
       ulong * fseq = fd_fseq_join( fd_topo_obj_laddr( topo, fseq_id ) );
       FD_TEST( fseq );
-      fd_fseq_update( fseq, ULONG_MAX );
+      fd_fseq_update( fseq, STEM_SHUTDOWN_SEQ );
     }
   }
 }