private void executeAllAddedBThreads ( Set < BThreadSyncSnapshot > nextRound , ExecutorService exSvc , BPEngineTask . Listener listener ) throws InterruptedException { // if any new bthreads are added, run and add their result Set < BThreadSyncSnapshot > addedBThreads = bprog . drainRecentlyRegisteredBthreads ( ) ; Set < ForkStatement > addedForks = bprog . drainRecentlyAddedForks ( ) ; while ( ( ( ! addedBThreads . isEmpty ( ) ) || ( ! addedForks . isEmpty ( ) ) ) && ! exSvc . isShutdown ( ) ) { Stream < BPEngineTask > threadStream = addedBThreads . stream ( ) . map ( bt -> new StartBThread ( bt , listener ) ) ; Stream < BPEngineTask > forkStream = addedForks . stream ( ) . flatMap ( f -> convertToTasks ( f , listener ) ) ; nextRound . addAll ( exSvc . invokeAll ( Stream . concat ( forkStream , threadStream ) . collect ( toList ( ) ) ) . stream ( ) . map ( f -> safeGet ( f ) ) . filter ( Objects :: nonNull ) . collect ( toList ( ) ) ) ; addedBThreads = bprog . drainRecentlyRegisteredBthreads ( ) ; addedForks = bprog . drainRecentlyAddedForks ( ) ; } }
Executes and adds all newly registered b - threads until no more new b - threads are registered .