Cleanup generator retry logic from PipelineIteration

Move it to Generator itself.
Using super.run() allows us to do profiled runs for generator retries.