Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
ISPN-5219 Expose Distributed Iterators over HotRod
- Loading branch information
Showing
87 changed files
with
2,500 additions
and
43 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
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
16 changes: 16 additions & 0 deletions
16
...t/hotrod-client/src/main/java/org/infinispan/client/hotrod/impl/iteration/KeyTracker.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,16 @@ | ||
| package org.infinispan.client.hotrod.impl.iteration; | ||
|
|
||
| import java.util.Set; | ||
|
|
||
| /** | ||
| * @author gustavonalle | ||
| * @since 8.0 | ||
| */ | ||
| public interface KeyTracker { | ||
|
|
||
| boolean track(byte[] key); | ||
|
|
||
| void segmentsFinished(byte[] finishedSegments); | ||
|
|
||
| Set<Integer> missedSegments(); | ||
| } |
21 changes: 21 additions & 0 deletions
21
...d-client/src/main/java/org/infinispan/client/hotrod/impl/iteration/KeyTrackerFactory.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,21 @@ | ||
| package org.infinispan.client.hotrod.impl.iteration; | ||
|
|
||
| import org.infinispan.client.hotrod.impl.consistenthash.ConsistentHash; | ||
| import org.infinispan.client.hotrod.impl.consistenthash.SegmentConsistentHash; | ||
|
|
||
| /** | ||
| * @author gustavonalle | ||
| * @since 8.0 | ||
| */ | ||
| public final class KeyTrackerFactory { | ||
|
|
||
| private KeyTrackerFactory() { | ||
| } | ||
|
|
||
| public static KeyTracker create(ConsistentHash hash, int topologyId) { | ||
| if (topologyId == -1) return new NoOpSegmentKeyTracker(); | ||
| if (hash == null) return new ReplKeyTracker(); | ||
| return new SegmentKeyTracker((SegmentConsistentHash) hash); | ||
| } | ||
|
|
||
| } |
26 changes: 26 additions & 0 deletions
26
...ient/src/main/java/org/infinispan/client/hotrod/impl/iteration/NoOpSegmentKeyTracker.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,26 @@ | ||
| package org.infinispan.client.hotrod.impl.iteration; | ||
|
|
||
| import java.util.Collections; | ||
| import java.util.Set; | ||
|
|
||
| /** | ||
| * @author gustavonalle | ||
| * @since 8.0 | ||
| */ | ||
| public class NoOpSegmentKeyTracker implements KeyTracker { | ||
|
|
||
| @Override | ||
| public boolean track(byte[] key) { | ||
| return true; | ||
| } | ||
|
|
||
| @Override | ||
| public void segmentsFinished(byte[] finishedSegments) { | ||
| } | ||
|
|
||
| @Override | ||
| public Set<Integer> missedSegments() { | ||
| return Collections.emptySet(); | ||
| } | ||
|
|
||
| } |
150 changes: 150 additions & 0 deletions
150
...nt/src/main/java/org/infinispan/client/hotrod/impl/iteration/RemoteCloseableIterator.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,150 @@ | ||
| package org.infinispan.client.hotrod.impl.iteration; | ||
|
|
||
| import net.jcip.annotations.NotThreadSafe; | ||
| import org.infinispan.client.hotrod.exceptions.TransportException; | ||
| import org.infinispan.client.hotrod.impl.operations.IterationEndResponse; | ||
| import org.infinispan.client.hotrod.impl.operations.IterationNextOperation; | ||
| import org.infinispan.client.hotrod.impl.operations.IterationNextResponse; | ||
| import org.infinispan.client.hotrod.impl.operations.IterationStartOperation; | ||
| import org.infinispan.client.hotrod.impl.operations.IterationStartResponse; | ||
| import org.infinispan.client.hotrod.impl.operations.OperationsFactory; | ||
| import org.infinispan.client.hotrod.impl.transport.Transport; | ||
| import org.infinispan.client.hotrod.logging.Log; | ||
| import org.infinispan.client.hotrod.logging.LogFactory; | ||
| import org.infinispan.commons.marshall.Marshaller; | ||
| import org.infinispan.commons.util.CloseableIterator; | ||
|
|
||
| import java.util.AbstractMap.SimpleEntry; | ||
| import java.util.LinkedList; | ||
| import java.util.Map.Entry; | ||
| import java.util.NoSuchElementException; | ||
| import java.util.Queue; | ||
| import java.util.Set; | ||
|
|
||
| import static org.infinispan.client.hotrod.impl.protocol.HotRodConstants.INVALID_ITERATION; | ||
| import static org.infinispan.client.hotrod.impl.protocol.HotRodConstants.NO_ERROR_STATUS; | ||
| import static org.infinispan.client.hotrod.marshall.MarshallerUtil.bytes2obj; | ||
|
|
||
| /** | ||
| * @author gustavonalle | ||
| * @since 8.0 | ||
| */ | ||
| @NotThreadSafe | ||
| public class RemoteCloseableIterator implements CloseableIterator<Entry<Object, Object>> { | ||
|
|
||
| private static final Log log = LogFactory.getLog(RemoteCloseableIterator.class); | ||
|
|
||
| private final OperationsFactory operationsFactory; | ||
| private final Marshaller marshaller; | ||
| private final String filterConverterFactory; | ||
| private final Set<Integer> segments; | ||
| private final int batchSize; | ||
|
|
||
| private KeyTracker segmentKeyTracker; | ||
| private Transport transport; | ||
| private String iterationId; | ||
| boolean endOfIteration = false; | ||
| private Queue<SimpleEntry<Object, Object>> nextElements = new LinkedList<>(); | ||
|
|
||
| public RemoteCloseableIterator(OperationsFactory operationsFactory, String filterConverterFactory, Set<Integer> segments, int batchSize, Marshaller marshaller) { | ||
| this.filterConverterFactory = filterConverterFactory; | ||
| this.segments = segments; | ||
| this.batchSize = batchSize; | ||
| this.operationsFactory = operationsFactory; | ||
| this.marshaller = marshaller; | ||
| } | ||
|
|
||
| @Override | ||
| public void close() { | ||
| IterationEndResponse endResponse = operationsFactory.newIterationEndOperation(iterationId, transport).execute(); | ||
| short status = endResponse.getStatus(); | ||
|
|
||
| if (status == NO_ERROR_STATUS) { | ||
| log.iterationClosed(iterationId); | ||
| } | ||
| if (endResponse.getStatus() == INVALID_ITERATION) { | ||
| throw log.errorClosingIteration(iterationId); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public boolean hasNext() { | ||
| if (nextElements.isEmpty()) { | ||
| fetch(); | ||
| } | ||
| return !endOfIteration; | ||
| } | ||
|
|
||
| @Override | ||
| public Entry<Object, Object> next() { | ||
| if (!hasNext()) throw new NoSuchElementException(); | ||
| return nextElements.remove(); | ||
| } | ||
|
|
||
| private void fetch() { | ||
| try { | ||
| IterationNextOperation iterationNextOperation = operationsFactory.newIterationNextOperation(iterationId, transport); | ||
|
|
||
| while (nextElements.isEmpty() && !endOfIteration) { | ||
| IterationNextResponse iterationNextResponse = iterationNextOperation.execute(); | ||
| short status = iterationNextResponse.getStatus(); | ||
| if (status == INVALID_ITERATION) { | ||
| throw log.errorRetrievingNext(iterationId); | ||
| } | ||
| Entry<byte[], byte[]>[] entries = iterationNextResponse.getEntries(); | ||
|
|
||
| if (entries.length == 0) { | ||
| endOfIteration = true; | ||
| break; | ||
| } | ||
| for (Entry<byte[], byte[]> entry : entries) { | ||
| if (segmentKeyTracker.track(entry.getKey())) { | ||
| nextElements.add(new SimpleEntry<>(unmarshall(entry.getKey()), unmarshall(entry.getValue()))); | ||
| } | ||
| } | ||
| segmentKeyTracker.segmentsFinished(iterationNextResponse.getFinishedSegments()); | ||
| } | ||
|
|
||
| } catch (TransportException e) { | ||
| log.warnf(e, "Error reaching the server during iteration"); | ||
| restartIteration(segmentKeyTracker.missedSegments()); | ||
| fetch(); | ||
| } | ||
| } | ||
|
|
||
|
|
||
| private Object unmarshall(byte[] bytes) { | ||
| return bytes2obj(marshaller, bytes); | ||
| } | ||
|
|
||
| private void restartIteration(Set<Integer> missedSegments) { | ||
| startInternal(missedSegments); | ||
| } | ||
|
|
||
| private void start(Set<Integer> fromSegments) { | ||
| IterationStartResponse startResponse = startInternal(fromSegments); | ||
|
|
||
| this.segmentKeyTracker = KeyTrackerFactory.create(startResponse.getSegmentConsistentHash(), startResponse.getTopologyId()); | ||
| } | ||
|
|
||
| private IterationStartResponse startInternal(Set<Integer> fromSegments) { | ||
| if (log.isDebugEnabled()) { | ||
| log.debugf("Staring iteration with segments %s", fromSegments); | ||
| } | ||
| IterationStartOperation iterationStartOperation = operationsFactory.newIterationStartOperation(filterConverterFactory, fromSegments, batchSize); | ||
| IterationStartResponse startResponse = iterationStartOperation.execute(); | ||
| this.transport = startResponse.getTransport(); | ||
| if (log.isDebugEnabled()) { | ||
| log.debugf("Obtained transport", this.transport); | ||
| } | ||
| this.iterationId = startResponse.getIterationId(); | ||
| if (log.isDebugEnabled()) { | ||
| log.debugf("IterationId:", this.iterationId); | ||
| } | ||
| return startResponse; | ||
| } | ||
|
|
||
| public void start() { | ||
| start(segments); | ||
| } | ||
| } |
37 changes: 37 additions & 0 deletions
37
...trod-client/src/main/java/org/infinispan/client/hotrod/impl/iteration/ReplKeyTracker.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,37 @@ | ||
| package org.infinispan.client.hotrod.impl.iteration; | ||
|
|
||
|
|
||
| import java.nio.ByteBuffer; | ||
| import java.util.Collections; | ||
| import java.util.HashSet; | ||
| import java.util.Set; | ||
|
|
||
| import static java.nio.ByteBuffer.wrap; | ||
| import org.infinispan.commons.equivalence.ByteArrayEquivalence; | ||
| import org.infinispan.commons.util.CollectionFactory; | ||
|
|
||
| /** | ||
| * Tracks all keys seen during iteration. Depends on ISPN-5451 to be done more efficiently, by | ||
| * discarding segments as soon as they are completed iterating. | ||
| * | ||
| * @author gustavonalle | ||
| * @since 8.0 | ||
| */ | ||
| public class ReplKeyTracker implements KeyTracker { | ||
|
|
||
| private Set<byte[]> keys = CollectionFactory.makeSet(ByteArrayEquivalence.INSTANCE); | ||
|
|
||
| @Override | ||
| public boolean track(byte[] key) { | ||
| return keys.add(key); | ||
| } | ||
|
|
||
| @Override | ||
| public void segmentsFinished(byte[] finishedSegments) { | ||
| } | ||
|
|
||
| @Override | ||
| public Set<Integer> missedSegments() { | ||
| return Collections.emptySet(); | ||
| } | ||
| } |
Oops, something went wrong.