modified to handle input streams on each worker node