class UnilateralSortMerger implements Sorter<E> {

	@Override
	public void close() {
		// check if the sorter has been closed before
		synchronized (this) {
			if (this.closed) {
				return;
			}
			
			// mark as closed
			this.closed = true;
		}
		
		// from here on, the code is in a try block, because even through errors might be thrown in this block,
		// we need to make sure that all the memory is released.
		try {
			// if the result iterator has not been obtained yet, set the exception
			synchronized (this.iteratorLock) {
				if (this.iteratorException == null) {
					this.iteratorException = new IOException("The sorter has been closed.");
					this.iteratorLock.notifyAll();
				}
			}
			
			// stop all the threads
			if (this.readThread != null) {
				try {
					this.readThread.shutdown();
				} catch (Throwable t) {
					LOG.error("Error shutting down reader thread: " + t.getMessage(), t);
				}
			}
			if (this.sortThread != null) {
				try {
					this.sortThread.shutdown();
				} catch (Throwable t) {
					LOG.error("Error shutting down sorter thread: " + t.getMessage(), t);
				}
			}
			if (this.spillThread != null) {
				try {
					this.spillThread.shutdown();
				} catch (Throwable t) {
					LOG.error("Error shutting down spilling thread: " + t.getMessage(), t);
				}
			}

			try {
				if (this.readThread != null) {
					this.readThread.join();
				}
				
				if (this.sortThread != null) {
					this.sortThread.join();
				}
				
				if (this.spillThread != null) {
					this.spillThread.join();
				}
			}
			catch (InterruptedException iex) {
				LOG.debug("Closing of sort/merger was interrupted. " +
						"The reading/sorting/spilling threads may still be working.", iex);
			}
		}
		finally {

			// Dispose all in memory sorter in order to clear memory references
			for (InMemorySorter<?> inMemorySorter : inMemorySorters) {
				inMemorySorter.dispose();
			}

			// RELEASE ALL MEMORY. If the threads and channels are still running, this should cause
			// exceptions, because their memory segments are freed
			try {
				if (!this.writeMemory.isEmpty()) {
					this.memoryManager.release(this.writeMemory);
				}
				this.writeMemory.clear();
			}
			catch (Throwable t) {}
			
			try {
				if (!this.sortReadMemory.isEmpty()) {
					this.memoryManager.release(this.sortReadMemory);
				}
				this.sortReadMemory.clear();
			}
			catch (Throwable t) {}
			
			// we have to loop this, because it may fail with a concurrent modification exception
			while (!this.openChannels.isEmpty()) {
				try {
					for (Iterator<FileIOChannel> channels = this.openChannels.iterator(); channels.hasNext(); ) {
						final FileIOChannel channel = channels.next();
						channels.remove();
						channel.closeAndDelete();
					}
				}
				catch (Throwable t) {}
			}
			
			// we have to loop this, because it may fail with a concurrent modification exception
			while (!this.channelsToDeleteAtShutdown.isEmpty()) {
				try {
					for (Iterator<FileIOChannel.ID> channels = this.channelsToDeleteAtShutdown.iterator(); channels.hasNext(); ) {
						final FileIOChannel.ID channel = channels.next();
						channels.remove();
						try {
							final File f = new File(channel.getPath());
							if (f.exists()) {
								f.delete();
							}
						} catch (Throwable t) {}
					}
				}
				catch (Throwable t) {}
			}
			
			try {
				if (this.largeRecordHandler != null) {
					this.largeRecordHandler.close();
				}
			} catch (Throwable t) {}
		}
	}

	public  UnilateralSortMerger(MemoryManager memoryManager, IOManager ioManager,
			MutableObjectIterator<E> input, AbstractInvokable parentTask,
			TypeSerializerFactory<E> serializerFactory, TypeComparator<E> comparator,
			double memoryFraction, int maxNumFileHandles, float startSpillingFraction,
			boolean handleLargeRecords, boolean objectReuseEnabled);
	public  UnilateralSortMerger(MemoryManager memoryManager, IOManager ioManager,
			MutableObjectIterator<E> input, AbstractInvokable parentTask,
			TypeSerializerFactory<E> serializerFactory, TypeComparator<E> comparator,
			double memoryFraction, int numSortBuffers, int maxNumFileHandles,
			float startSpillingFraction, boolean handleLargeRecords, boolean objectReuseEnabled);
	public  UnilateralSortMerger(MemoryManager memoryManager, List<MemorySegment> memory,
			IOManager ioManager,
			MutableObjectIterator<E> input, AbstractInvokable parentTask, 
			TypeSerializerFactory<E> serializerFactory, TypeComparator<E> comparator,
			int numSortBuffers, int maxNumFileHandles,
			float startSpillingFraction, boolean handleLargeRecords, boolean objectReuseEnabled);
	protected  UnilateralSortMerger(MemoryManager memoryManager,
			IOManager ioManager,
			MutableObjectIterator<E> input, AbstractInvokable parentTask, 
			TypeSerializerFactory<E> serializerFactory, TypeComparator<E> comparator,
			double memoryFraction, int numSortBuffers, int maxNumFileHandles,
			float startSpillingFraction, boolean noSpillingMemory, boolean handleLargeRecords,
			boolean objectReuseEnabled);
	protected  UnilateralSortMerger(MemoryManager memoryManager, List<MemorySegment> memory,
			IOManager ioManager,
			MutableObjectIterator<E> input, AbstractInvokable parentTask, 
			TypeSerializerFactory<E> serializerFactory, TypeComparator<E> comparator,
			int numSortBuffers, int maxNumFileHandles,
			float startSpillingFraction, boolean noSpillingMemory, boolean handleLargeRecords,
			boolean objectReuseEnabled);
	protected  UnilateralSortMerger(
			MemoryManager memoryManager,
			List<MemorySegment> memory,
			IOManager ioManager,
			MutableObjectIterator<E> input,
			AbstractInvokable parentTask,
			TypeSerializerFactory<E> serializerFactory,
			TypeComparator<E> comparator,
			int numSortBuffers,
			int maxNumFileHandles,
			float startSpillingFraction,
			boolean noSpillingMemory,
			boolean handleLargeRecords,
			boolean objectReuseEnabled,
			InMemorySorterFactory<E> inMemorySorterFactory);

}