@Test public void testInMemorySorterDisposal() throws Exception { final TestingInMemorySorterFactory<Tuple2<Integer, Integer>> inMemorySorterFactory = new TestingInMemorySorterFactory<>(); final int numPages = 32; final MemoryManager memoryManager = new MemoryManager(MemoryManager.DEFAULT_PAGE_SIZE * numPages, 1); final IOManagerAsync ioManager = new IOManagerAsync(); final DummyInvokable parentTask = new DummyInvokable(); try { final List<MemorySegment> memory = memoryManager.allocatePages(parentTask, numPages); final UnilateralSortMerger<Tuple2<Integer, Integer>> unilateralSortMerger = new UnilateralSortMerger<>( memoryManager, memory, ioManager, EmptyMutableObjectIterator.get(), parentTask, TestData.getIntIntTupleSerializerFactory(), TestData.getIntIntTupleComparator(), 10, 2, 1.0f, true, false, false, inMemorySorterFactory); final Collection<TestingInMemorySorter<?>> inMemorySorters = inMemorySorterFactory.getInMemorySorters(); assertThat(inMemorySorters, is(not(empty()))); unilateralSortMerger.close(); assertThat(unilateralSortMerger.closed, is(true)); for (TestingInMemorySorter<?> inMemorySorter : inMemorySorters) { assertThat(inMemorySorter.isDisposed(), is(true)); } } finally { ioManager.shutdown(); memoryManager.shutdown(); } }
@Override public void close() { synchronized (this) { if (this.closed) { return; } this.closed = true; } try { synchronized (this.iteratorLock) { if (this.iteratorException == null) { this.iteratorException = new IOException("The sorter has been closed."); this.iteratorLock.notifyAll(); } } 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 { for (InMemorySorter<?> inMemorySorter : inMemorySorters) { inMemorySorter.dispose(); } 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) {} 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) {} } 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) {} } }
UnilateralSortMerger implements Sorter<E> { @Override public void close() { synchronized (this) { if (this.closed) { return; } this.closed = true; } try { synchronized (this.iteratorLock) { if (this.iteratorException == null) { this.iteratorException = new IOException("The sorter has been closed."); this.iteratorLock.notifyAll(); } } 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 { for (InMemorySorter<?> inMemorySorter : inMemorySorters) { inMemorySorter.dispose(); } 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) {} 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) {} } 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) {} } } }
UnilateralSortMerger implements Sorter<E> { @Override public void close() { synchronized (this) { if (this.closed) { return; } this.closed = true; } try { synchronized (this.iteratorLock) { if (this.iteratorException == null) { this.iteratorException = new IOException("The sorter has been closed."); this.iteratorLock.notifyAll(); } } 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 { for (InMemorySorter<?> inMemorySorter : inMemorySorters) { inMemorySorter.dispose(); } 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) {} 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) {} } 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) {} } } 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);  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);  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); }
UnilateralSortMerger implements Sorter<E> { @Override public void close() { synchronized (this) { if (this.closed) { return; } this.closed = true; } try { synchronized (this.iteratorLock) { if (this.iteratorException == null) { this.iteratorException = new IOException("The sorter has been closed."); this.iteratorLock.notifyAll(); } } 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 { for (InMemorySorter<?> inMemorySorter : inMemorySorters) { inMemorySorter.dispose(); } 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) {} 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) {} } 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) {} } } 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);  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);  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); @Override void close(); @Override MutableObjectIterator<E> getIterator(); }
UnilateralSortMerger implements Sorter<E> { @Override public void close() { synchronized (this) { if (this.closed) { return; } this.closed = true; } try { synchronized (this.iteratorLock) { if (this.iteratorException == null) { this.iteratorException = new IOException("The sorter has been closed."); this.iteratorLock.notifyAll(); } } 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 { for (InMemorySorter<?> inMemorySorter : inMemorySorters) { inMemorySorter.dispose(); } 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) {} 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) {} } 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) {} } } 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);  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);  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); @Override void close(); @Override MutableObjectIterator<E> getIterator();  }