@Test @Ignore("For local testing only. Runs indefinitely") public void testPerformanceOfRandomAccessReads() throws Exception { final File dir = new File("target/storage/" + UUID.randomUUID().toString()); final File journalFile = new File(dir, "/4.prov.gz"); final File tocFile = TocUtil.getTocFile(journalFile); final int blockSize = 1024 * 32; try (final RecordWriter writer = createWriter(journalFile, new StandardTocWriter(tocFile, true, false), true, blockSize)) { writer.writeHeader(0L); for (int i = 0; i < 100_000; i++) { writer.writeRecord(TestUtil.createEvent()); } } final Long[] eventIds = new Long[] { 4L, 80L, 1024L, 1025L, 1026L, 1027L, 1028L, 1029L, 1030L, 40_000L, 80_000L, 99_000L }; final RecordReaderFactory readerFactory = (file, logs, maxChars) -> RecordReaders.newRecordReader(file, logs, maxChars); final List<File> files = new ArrayList<>(); files.add(new File(dir, "0.prov")); files.add(new File(dir, "0.prov")); files.add(new File(dir, "1.prov")); files.add(new File(dir, "2.prov")); files.add(new File(dir, "3.prov")); files.add(journalFile); files.add(new File(dir, "100000000.prov")); boolean loopForever = true; while (loopForever) { final long start = System.nanoTime(); for (int i = 0; i < 1000; i++) { final SelectiveRecordReaderEventIterator iterator = new SelectiveRecordReaderEventIterator( Collections.singletonList(journalFile), readerFactory, Arrays.asList(eventIds), 32 * 1024); for (final long id : eventIds) { time(() -> { return iterator.nextEvent().orElse(null); }, id); } } final long ms = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start); System.out.println(ms + " ms total"); } }
@Override public Optional<ProvenanceEventRecord> nextEvent() throws IOException { if (closed) { throw new IOException("EventIterator is already closed"); } final long start = System.nanoTime(); try { while (idIterator.hasNext()) { final long eventId = idIterator.next(); final File fileForEvent = getFileForEventId(eventId); if (fileForEvent == null) { continue; } if (!fileForEvent.equals(currentFile)) { if (reader != null) { try { reader.close(); } catch (final Exception e) { logger.warn("Failed to close {}; some resources may not be cleaned up appropriately", reader); } } reader = readerFactory.newRecordReader(fileForEvent, Collections.emptyList(), maxAttributeChars); this.currentFile = fileForEvent; } final Optional<ProvenanceEventRecord> eventOption = reader.skipToEvent(eventId); if (eventOption.isPresent() && eventOption.get().getEventId() == eventId) { reader.nextRecord(); return eventOption; } continue; } return Optional.empty(); } finally { final long ms = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start); logger.trace("Took {} ms to read next event", ms); } }
SelectiveRecordReaderEventIterator implements EventIterator { @Override public Optional<ProvenanceEventRecord> nextEvent() throws IOException { if (closed) { throw new IOException("EventIterator is already closed"); } final long start = System.nanoTime(); try { while (idIterator.hasNext()) { final long eventId = idIterator.next(); final File fileForEvent = getFileForEventId(eventId); if (fileForEvent == null) { continue; } if (!fileForEvent.equals(currentFile)) { if (reader != null) { try { reader.close(); } catch (final Exception e) { logger.warn("Failed to close {}; some resources may not be cleaned up appropriately", reader); } } reader = readerFactory.newRecordReader(fileForEvent, Collections.emptyList(), maxAttributeChars); this.currentFile = fileForEvent; } final Optional<ProvenanceEventRecord> eventOption = reader.skipToEvent(eventId); if (eventOption.isPresent() && eventOption.get().getEventId() == eventId) { reader.nextRecord(); return eventOption; } continue; } return Optional.empty(); } finally { final long ms = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start); logger.trace("Took {} ms to read next event", ms); } } }
SelectiveRecordReaderEventIterator implements EventIterator { @Override public Optional<ProvenanceEventRecord> nextEvent() throws IOException { if (closed) { throw new IOException("EventIterator is already closed"); } final long start = System.nanoTime(); try { while (idIterator.hasNext()) { final long eventId = idIterator.next(); final File fileForEvent = getFileForEventId(eventId); if (fileForEvent == null) { continue; } if (!fileForEvent.equals(currentFile)) { if (reader != null) { try { reader.close(); } catch (final Exception e) { logger.warn("Failed to close {}; some resources may not be cleaned up appropriately", reader); } } reader = readerFactory.newRecordReader(fileForEvent, Collections.emptyList(), maxAttributeChars); this.currentFile = fileForEvent; } final Optional<ProvenanceEventRecord> eventOption = reader.skipToEvent(eventId); if (eventOption.isPresent() && eventOption.get().getEventId() == eventId) { reader.nextRecord(); return eventOption; } continue; } return Optional.empty(); } finally { final long ms = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start); logger.trace("Took {} ms to read next event", ms); } } SelectiveRecordReaderEventIterator(final List<File> filesToRead, final RecordReaderFactory readerFactory, final List<Long> eventIds, final int maxAttributeChars); }
SelectiveRecordReaderEventIterator implements EventIterator { @Override public Optional<ProvenanceEventRecord> nextEvent() throws IOException { if (closed) { throw new IOException("EventIterator is already closed"); } final long start = System.nanoTime(); try { while (idIterator.hasNext()) { final long eventId = idIterator.next(); final File fileForEvent = getFileForEventId(eventId); if (fileForEvent == null) { continue; } if (!fileForEvent.equals(currentFile)) { if (reader != null) { try { reader.close(); } catch (final Exception e) { logger.warn("Failed to close {}; some resources may not be cleaned up appropriately", reader); } } reader = readerFactory.newRecordReader(fileForEvent, Collections.emptyList(), maxAttributeChars); this.currentFile = fileForEvent; } final Optional<ProvenanceEventRecord> eventOption = reader.skipToEvent(eventId); if (eventOption.isPresent() && eventOption.get().getEventId() == eventId) { reader.nextRecord(); return eventOption; } continue; } return Optional.empty(); } finally { final long ms = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start); logger.trace("Took {} ms to read next event", ms); } } SelectiveRecordReaderEventIterator(final List<File> filesToRead, final RecordReaderFactory readerFactory, final List<Long> eventIds, final int maxAttributeChars); @Override void close(); @Override Optional<ProvenanceEventRecord> nextEvent(); }
SelectiveRecordReaderEventIterator implements EventIterator { @Override public Optional<ProvenanceEventRecord> nextEvent() throws IOException { if (closed) { throw new IOException("EventIterator is already closed"); } final long start = System.nanoTime(); try { while (idIterator.hasNext()) { final long eventId = idIterator.next(); final File fileForEvent = getFileForEventId(eventId); if (fileForEvent == null) { continue; } if (!fileForEvent.equals(currentFile)) { if (reader != null) { try { reader.close(); } catch (final Exception e) { logger.warn("Failed to close {}; some resources may not be cleaned up appropriately", reader); } } reader = readerFactory.newRecordReader(fileForEvent, Collections.emptyList(), maxAttributeChars); this.currentFile = fileForEvent; } final Optional<ProvenanceEventRecord> eventOption = reader.skipToEvent(eventId); if (eventOption.isPresent() && eventOption.get().getEventId() == eventId) { reader.nextRecord(); return eventOption; } continue; } return Optional.empty(); } finally { final long ms = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start); logger.trace("Took {} ms to read next event", ms); } } SelectiveRecordReaderEventIterator(final List<File> filesToRead, final RecordReaderFactory readerFactory, final List<Long> eventIds, final int maxAttributeChars); @Override void close(); @Override Optional<ProvenanceEventRecord> nextEvent();  }