docs: @Override
  	public void invoke() throws Exception {
  		this.headEventReader = new MutableRecordReader<>(
  				getEnvironment().getInputGate(0),
  				getEnvironment().getTaskManagerInfo().getTmpDirectories());
  
  		TaskConfig taskConfig = new TaskConfig(getTaskConfiguration());
  
  		// store all aggregators
  		this.aggregators = new HashMap<>();
  		for (AggregatorWithName<?> aggWithName : taskConfig.getIterationAggregators(getUserCodeClassLoader())) {
  			aggregators.put(aggWithName.getName(), aggWithName.getAggregator());
  		}
  
  		// store the aggregator convergence criterion
  		if (taskConfig.usesConvergenceCriterion()) {
  			convergenceCriterion = taskConfig.getConvergenceCriterion(getUserCodeClassLoader());
  			convergenceAggregatorName = taskConfig.getConvergenceCriterionAggregatorName();
  			Preconditions.checkNotNull(convergenceAggregatorName);
  		}
  
  		// store the default aggregator convergence criterion
  		if (taskConfig.usesImplicitConvergenceCriterion()) {
  			implicitConvergenceCriterion = taskConfig.getImplicitConvergenceCriterion(getUserCodeClassLoader());
  			implicitConvergenceAggregatorName = taskConfig.getImplicitConvergenceCriterionAggregatorName();
  			Preconditions.checkNotNull(implicitConvergenceAggregatorName);
  		}
  
  		maxNumberOfIterations = taskConfig.getNumberOfIterations();
  
  		// set up the event handler
  		int numEventsTillEndOfSuperstep = taskConfig.getNumberOfEventsUntilInterruptInIterativeGate(0);
  		eventHandler = new SyncEventHandler(numEventsTillEndOfSuperstep, aggregators,
  				getEnvironment().getUserClassLoader());
  		headEventReader.registerTaskEventListener(eventHandler, WorkerDoneEvent.class);
  
  		IntValue dummy = new IntValue();
  
  		while (!terminationRequested()) {
  
  			if (log.isInfoEnabled()) {
  				log.info(formatLogString("starting iteration [" + currentIteration + "]"));
  			}
  
  			// this call listens for events until the end-of-superstep is reached
  			readHeadEventChannel(dummy);
  
  			if (log.isInfoEnabled()) {
  				log.info(formatLogString("finishing iteration [" + currentIteration + "]"));
  			}
  
  			if (checkForConvergence()) {
  				if (log.isInfoEnabled()) {
  					log.info(formatLogString("signaling that all workers are to terminate in iteration ["
  						+ currentIteration + "]"));
  				}
  
  				requestTermination();
  				sendToAllWorkers(new TerminationEvent());
  			} else {
  				if (log.isInfoEnabled()) {
  					log.info(formatLogString("signaling that all workers are done in iteration [" + currentIteration
  						+ "]"));
  				}
  
  				AllWorkersDoneEvent allWorkersDoneEvent = new AllWorkersDoneEvent(aggregators);
  				sendToAllWorkers(allWorkersDoneEvent);
  
  				// reset all aggregators
  				for (Aggregator<?> agg : aggregators.values()) {
  					agg.reset();
  				}
  				currentIteration++;
  			}
  		}
  	}