-
Notifications
You must be signed in to change notification settings - Fork 1
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Generalized BlockSource to build upon any SeekableReadableChannelSour…
…ce (SRCS) rather than just java.nio.Path. Added initial read tracking for SRCSs.
- Loading branch information
Showing
19 changed files
with
578 additions
and
254 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
87 changes: 87 additions & 0 deletions
87
...arent/aksw-commons-io-buffers/src/main/java/org/aksw/commons/io/input/ChannelMonitor.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,87 @@ | ||
| package org.aksw.commons.io.input; | ||
|
|
||
| import com.google.common.collect.Range; | ||
| import com.google.common.collect.RangeMap; | ||
| import com.google.common.collect.RangeSet; | ||
| import com.google.common.collect.TreeRangeMap; | ||
| import com.google.common.collect.TreeRangeSet; | ||
|
|
||
| public class ChannelMonitor { | ||
| public class RangeTracker { | ||
| protected long totalDurationNanos; | ||
| // protected Set<Integer> readLengths; | ||
| protected int minReadLength; | ||
| protected int maxReadLength; | ||
| // protected long totalReadlength; | ||
| protected long readCount; | ||
|
|
||
| public RangeTracker(int readLength, long totalDurationNanos) { | ||
| this(readLength, readLength, totalDurationNanos, 1); | ||
| } | ||
|
|
||
| public RangeTracker(int minReadLength, int maxReadLength, long totalDurationNanos, long readCount) { | ||
| super(); | ||
| this.totalDurationNanos = totalDurationNanos; | ||
| this.minReadLength = minReadLength; | ||
| this.maxReadLength = maxReadLength; | ||
| this.readCount = readCount; | ||
| } | ||
|
|
||
| public long getTotalDurationNanos() { | ||
| return totalDurationNanos; | ||
| } | ||
|
|
||
| public int getMinReadLength() { | ||
| return minReadLength; | ||
| } | ||
|
|
||
| public int getMaxReadLength() { | ||
| return maxReadLength; | ||
| } | ||
|
|
||
| public long getReadCount() { | ||
| return readCount; | ||
| } | ||
|
|
||
| public void add(RangeTracker contrib) { | ||
| this.totalDurationNanos += contrib.totalDurationNanos; | ||
| this.maxReadLength = Math.max(this.maxReadLength, contrib.maxReadLength); | ||
| this.minReadLength = this.minReadLength == -1 | ||
| ? contrib.minReadLength | ||
| : Math.min(this.minReadLength, contrib.minReadLength); | ||
| ++this.readCount; | ||
| } | ||
|
|
||
| @Override | ||
| protected RangeTracker clone() { | ||
| return new RangeTracker(minReadLength, maxReadLength, totalDurationNanos, readCount); | ||
| } | ||
| } | ||
|
|
||
| protected RangeMap<Long, RangeTracker> trackedReads; | ||
|
|
||
| public ChannelMonitor() { | ||
| this.trackedReads = TreeRangeMap.create(); | ||
| } | ||
|
|
||
| public RangeMap<Long, RangeTracker> getTrackedReads() { | ||
| return trackedReads; | ||
| } | ||
|
|
||
| public synchronized void submitReadStats(long readStartPos, long readEndPos, int readLength, long durationNanos) { | ||
| if (readLength > 0) { // Skip lengths that are <= 0 | ||
| RangeTracker contribution = new RangeTracker(readLength, readLength, durationNanos, 1); | ||
|
|
||
| Range<Long> span = Range.openClosed(readStartPos, readEndPos); | ||
|
|
||
| RangeMap<Long, RangeTracker> subMap = trackedReads.subRangeMap(span); | ||
| subMap.asMapOfRanges().values().forEach(tracker -> tracker.add(contribution)); | ||
|
|
||
| RangeSet<Long> rangeSet = TreeRangeSet.create(subMap.asMapOfRanges().keySet()); | ||
| RangeSet<Long> gaps = rangeSet.complement().subRangeSet(span); | ||
| gaps.asRanges().forEach(range -> { | ||
| trackedReads.put(range, contribution.clone()); | ||
| }); | ||
| } | ||
| } | ||
| } |
88 changes: 88 additions & 0 deletions
88
...sw-commons-io-buffers/src/main/java/org/aksw/commons/io/input/InputStreamOverChannel.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,88 @@ | ||
| package org.aksw.commons.io.input; | ||
|
|
||
| import java.io.IOException; | ||
| import java.io.InputStream; | ||
| import java.nio.ByteBuffer; | ||
| import java.nio.channels.IllegalBlockingModeException; | ||
| import java.nio.channels.ReadableByteChannel; | ||
| import java.nio.channels.SelectableChannel; | ||
| import java.util.Objects; | ||
|
|
||
|
|
||
|
|
||
| /** | ||
| * This is {@link sun.nio.ch.ChannelInputStream} without relying on the channel's size() method. | ||
| */ | ||
| public class InputStreamOverChannel | ||
| extends InputStream | ||
| { | ||
| public static int read(ReadableByteChannel ch, ByteBuffer bb, | ||
| boolean block) | ||
| throws IOException | ||
| { | ||
| if (ch instanceof SelectableChannel) { | ||
| SelectableChannel sc = (SelectableChannel)ch; | ||
| synchronized (sc.blockingLock()) { | ||
| boolean bm = sc.isBlocking(); | ||
| if (!bm) | ||
| throw new IllegalBlockingModeException(); | ||
| if (bm != block) | ||
| sc.configureBlocking(block); | ||
| int n = ch.read(bb); | ||
| if (bm != block) | ||
| sc.configureBlocking(bm); | ||
| return n; | ||
| } | ||
| } else { | ||
| return ch.read(bb); | ||
| } | ||
| } | ||
|
|
||
| protected final ReadableByteChannel ch; | ||
| private ByteBuffer bb = null; | ||
| private byte[] bs = null; // Invoker's previous array | ||
| private byte[] b1 = null; | ||
|
|
||
| public InputStreamOverChannel(ReadableByteChannel ch) { | ||
| this.ch = ch; | ||
| } | ||
|
|
||
| @Override | ||
| public synchronized int read() throws IOException { | ||
| if (b1 == null) | ||
| b1 = new byte[1]; | ||
| int n = this.read(b1); | ||
| if (n == 1) | ||
| return b1[0] & 0xff; | ||
| return -1; | ||
| } | ||
|
|
||
| @Override | ||
| public synchronized int read(byte[] bs, int off, int len) | ||
| throws IOException | ||
| { | ||
| Objects.checkFromIndexSize(off, len, bs.length); | ||
| if (len == 0) | ||
| return 0; | ||
|
|
||
| ByteBuffer bb = ((this.bs == bs) | ||
| ? this.bb | ||
| : ByteBuffer.wrap(bs)); | ||
| bb.limit(Math.min(off + len, bb.capacity())); | ||
| bb.position(off); | ||
| this.bb = bb; | ||
| this.bs = bs; | ||
| return read(bb); | ||
| } | ||
|
|
||
| protected int read(ByteBuffer bb) | ||
| throws IOException | ||
| { | ||
| return read(ch, bb, true); | ||
| } | ||
|
|
||
| @Override | ||
| public void close() throws IOException { | ||
| ch.close(); | ||
| } | ||
| } |
8 changes: 8 additions & 0 deletions
8
...sw-commons-io-buffers/src/main/java/org/aksw/commons/io/input/ReadableChannelBuilder.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,8 @@ | ||
| package org.aksw.commons.io.input; | ||
|
|
||
| public interface ReadableChannelBuilder<A, X extends ReadableChannel<A>, B extends ReadableChannelBuilder<A, X, B>> { | ||
| B setStart(long start); | ||
| B setEnd(long end); | ||
| B setAdvertiseBlocks(boolean offOrOn); | ||
| X build(); | ||
| } |
34 changes: 34 additions & 0 deletions
34
...o-buffers/src/main/java/org/aksw/commons/io/input/SeekableReadableChannelWithMonitor.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,34 @@ | ||
| package org.aksw.commons.io.input; | ||
|
|
||
| import java.io.IOException; | ||
| import java.util.Objects; | ||
|
|
||
| public class SeekableReadableChannelWithMonitor<A, X extends SeekableReadableChannel<A>> | ||
| extends SeekableReadableChannelDecoratorBase<A, X> | ||
| { | ||
| protected ChannelMonitor monitor; | ||
|
|
||
| public SeekableReadableChannelWithMonitor(X delegate, ChannelMonitor monitor) { | ||
| super(delegate); | ||
| this.monitor = Objects.requireNonNull(monitor); | ||
| } | ||
|
|
||
| @Override | ||
| public int read(A array, int position, int length) throws IOException { | ||
| // Include positioning in the time so that long times may be discovered | ||
| long startTimestamp = System.nanoTime(); | ||
| long startPos = super.position(); | ||
| int result = super.read(array, position, length); | ||
| long endPos = super.position(); | ||
| long endTimestamp = System.nanoTime(); | ||
| long duration = endTimestamp - startTimestamp; | ||
| monitor.submitReadStats(startPos, endPos, result, duration); | ||
|
|
||
| return result; | ||
| } | ||
|
|
||
| @Override | ||
| public void position(long pos) throws IOException { | ||
| super.position(pos); | ||
| } | ||
| } |
57 changes: 57 additions & 0 deletions
57
...ers/src/main/java/org/aksw/commons/io/input/SeekableReadableSourceWrapperWithMonitor.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,57 @@ | ||
| package org.aksw.commons.io.input; | ||
|
|
||
| import java.io.IOException; | ||
| import java.util.Objects; | ||
|
|
||
| import org.aksw.commons.io.buffer.array.ArrayOps; | ||
|
|
||
| public class SeekableReadableSourceWrapperWithMonitor<A> | ||
| implements SeekableReadableChannelSource<A> | ||
| { | ||
| protected SeekableReadableChannelSource<A> delegate; | ||
| protected ChannelMonitor channelMonitor; | ||
|
|
||
| public SeekableReadableSourceWrapperWithMonitor(SeekableReadableChannelSource<A> delegate) { | ||
| this(delegate, new ChannelMonitor()); | ||
| } | ||
|
|
||
| public SeekableReadableSourceWrapperWithMonitor(SeekableReadableChannelSource<A> delegate, | ||
| ChannelMonitor channelMonitor) { | ||
| super(); | ||
| this.delegate = Objects.requireNonNull(delegate); | ||
| this.channelMonitor = Objects.requireNonNull(channelMonitor); | ||
| } | ||
|
|
||
| public SeekableReadableChannelSource<A> getDelegate() { | ||
| return delegate; | ||
| } | ||
|
|
||
| @Override | ||
| public long size() throws IOException { | ||
| return getDelegate().size(); | ||
| } | ||
|
|
||
| @Override | ||
| public ArrayOps<A> getArrayOps() { | ||
| return getDelegate().getArrayOps(); | ||
| } | ||
|
|
||
| @Override | ||
| public SeekableReadableChannel<A> newReadableChannel() throws IOException { | ||
| return wrap(getDelegate().newReadableChannel()); | ||
| } | ||
|
|
||
| @Override | ||
| public SeekableReadableChannel<A> newReadableChannel(long start) throws IOException { | ||
| return wrap(getDelegate().newReadableChannel(start)); | ||
| } | ||
|
|
||
| // @Override | ||
| // public SeekableReadableChannel<A> newReadableChannel(long start, long end) throws IOException { | ||
| // return wrap(getDelegate().newReadableChannel(start, end)); | ||
| // } | ||
|
|
||
| protected SeekableReadableChannel<A> wrap(SeekableReadableChannel<A> delegate) { | ||
| return new SeekableReadableChannelWithMonitor<>(delegate, channelMonitor); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.