code: synchronized LocatableInputSplit getNextInputSplit(final AbstractInstance instance) {
  
  		final Queue<QueueElem> instanceSplitList = getInstanceSplitList(instance);
  
  		while (true) {
  
  			final QueueElem candidate = instanceSplitList.poll();
  			if (candidate == null) {
  				return null;
  			}
  
  			if (this.masterSet.remove(candidate.getInputSplit())) {
  				if (LOG.isInfoEnabled()) {
  					if (candidate.distance == 0) {
  						LOG.info(instance + " receives local file input split");
  					} else {
  						LOG.info(instance + " receives remote file input split (distance " + candidate.distance + ")");
  					}
  				}
  				return candidate.getInputSplit();
  			}
  
  			if (this.masterSet.isEmpty()) {
  				return null;
  			}
  		}
  	}
docs: Returns the next locatable input split to be consumed by the given instance. The returned input split is selected
  in a
  way that the distance between the split's storage location and the requesting {@link AbstractInstance} is as
  short as possible.
  
  @param instance
  the instance requesting the next file input split
  @return the next input split to be consumed by the given instance or <code>null</code> if all input splits have
  already been consumed.