HCOLL-352 Changed garbage-free Chronicle-Map iterations API from old to new Contexts; Add SegmentContexts API

This commit is contained in:
leventov
2015-07-24 20:11:58 +03:00
parent 912a3cc2fc
commit 9fe1dab477
32 changed files with 196 additions and 231 deletions
@@ -33,8 +33,8 @@ import java.util.function.Predicate;
* This interface defines common {@link ChronicleMap} and {@link ChronicleSet}, related to off-heap
* memory management and file-mapping. Not usable by itself.
*/
public interface ChronicleHash<K, C extends KeyContext<K>, EQC extends ExternalHashQueryContext<K>>
extends Closeable {
public interface ChronicleHash<K, C extends HashEntry<K>, SC extends HashSegmentContext<K, ?>,
EQC extends ExternalHashQueryContext<K>> extends Closeable {
/**
* Returns the file this hash container mapped to, i. e. when it is created by
* {@link ChronicleHashBuilder#create()} call, or {@code null} if it is purely in-memory,
@@ -81,6 +81,8 @@ public interface ChronicleHash<K, C extends KeyContext<K>, EQC extends ExternalH
* @return the context to perform operations with the key
*/
@NotNull EQC queryContext(Data<K> key);
SC segmentContext(int segmentIndex);
/**
* Checks the given predicate on each entry in this {@code ChronicleHash} until all entries
@@ -65,7 +65,7 @@ import java.util.concurrent.TimeUnit;
* @param <B> the concrete builder type, i. e. {@link ChronicleMapBuilder}
* or {@link ChronicleSetBuilder}
*/
public interface ChronicleHashBuilder<K, H extends ChronicleHash<K, ?, ?>,
public interface ChronicleHashBuilder<K, H extends ChronicleHash<K, ?, ?, ?>,
B extends ChronicleHashBuilder<K, H, B>> extends Cloneable {
/**
@@ -27,5 +27,5 @@ public interface HashContext<K> {
/**
* Returns the accessed {@code ChronicleHash}.
*/
ChronicleHash<K, ?, ?> hash();
ChronicleHash<K, ?, ?, ?> hash();
}
@@ -0,0 +1,33 @@
/*
* Copyright 2015 Higher Frequency Trading
*
* http://www.higherfrequencytrading.com
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package net.openhft.chronicle.hash;
import net.openhft.chronicle.core.io.Closeable;
import java.util.function.Consumer;
import java.util.function.Predicate;
public interface HashSegmentContext<K, E extends HashEntry<K>> extends HashContext<K>, Closeable {
void forEachSegmentEntry(Consumer<? super E> action);
boolean forEachSegmentEntryWhile(Predicate<? super E> predicate);
long size();
}
@@ -22,7 +22,7 @@ import net.openhft.chronicle.hash.ChronicleHashBuilder;
import net.openhft.chronicle.hash.serialization.internal.SerializationBuilder;
//TODO remove this temporary interface
public interface ChronicleHashBuilderImpl<K, H extends ChronicleHash<K, ?, ?>,
public interface ChronicleHashBuilderImpl<K, H extends ChronicleHash<K, ?, ?, ?>,
B extends ChronicleHashBuilder<K, H, B>> extends ChronicleHashBuilder<K, H, B> {
SerializationBuilder<K> keyBuilder();
@@ -16,9 +16,7 @@
package net.openhft.chronicle.hash.impl;
import net.openhft.chronicle.hash.ChronicleHash;
import net.openhft.chronicle.hash.ExternalHashQueryContext;
import net.openhft.chronicle.hash.KeyContext;
import net.openhft.chronicle.hash.*;
import net.openhft.chronicle.hash.impl.hashlookup.HashLookup;
import net.openhft.chronicle.hash.impl.util.BuildVersion;
import net.openhft.chronicle.hash.serialization.BytesReader;
@@ -44,8 +42,9 @@ import static java.lang.Math.max;
import static net.openhft.lang.MemoryUnit.*;
public abstract class VanillaChronicleHash<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
C extends KeyContext<K>, ECQ extends ExternalHashQueryContext<K>>
implements ChronicleHash<K, C, ECQ>, Serializable {
C extends HashEntry<K>, SC extends HashSegmentContext<K, ?>,
ECQ extends ExternalHashQueryContext<K>>
implements ChronicleHash<K, C, SC, ECQ>, Serializable {
private static final long serialVersionUID = 0L;
@@ -21,5 +21,5 @@ package net.openhft.chronicle.hash.impl;
import net.openhft.chronicle.hash.serialization.internal.MetaBytesInterop;
public interface VanillaChronicleHashHolder<K, KI, MKI extends MetaBytesInterop<K, ? super KI>> {
VanillaChronicleHash<K, KI, MKI, ?, ?> h();
VanillaChronicleHash<K, KI, MKI, ?, ?, ?> h();
}
@@ -360,7 +360,7 @@ public abstract class SegmentStages implements SegmentLock {
}
void initSegment() {
VanillaChronicleHash<?, ?, ?, ?, ?> h = hh.h();
VanillaChronicleHash<?, ?, ?, ?, ?, ?> h = hh.h();
long hashLookupOffset = h.segmentOffset(segmentIndex);
long freeListOffset = hashLookupOffset + h.segmentHashLookupOuterSize;
freeListBytes.storePositionAndSize(h.ms, freeListOffset, h.segmentFreeListInnerSize);
@@ -375,7 +375,7 @@ public abstract class SegmentStages implements SegmentLock {
//TODO refactor/optimize
public long alloc(int chunks) {
VanillaChronicleHash<?, ?, ?, ?, ?> h = hh.h();
VanillaChronicleHash<?, ?, ?, ?, ?, ?> h = hh.h();
if (chunks > h.maxChunksPerEntry)
throw new IllegalArgumentException("Entry is too large: requires " + chunks +
" entry size chucks, " + h.maxChunksPerEntry + " is maximum.");
@@ -27,6 +27,7 @@ import net.openhft.chronicle.hash.impl.stage.hash.CheckOnEachPublicOperation;
import net.openhft.sg.StageRef;
import net.openhft.sg.Staged;
import java.util.function.Consumer;
import java.util.function.Predicate;
@Staged
@@ -50,7 +51,7 @@ public abstract class HashSegmentIteration<K, E extends HashEntry<K>> implements
this.entryRemovedOnThisIteration = entryRemovedOnThisIteration;
}
public boolean forEachRemoving(Predicate<? super E> action) {
public boolean forEachSegmentEntryWhile(Predicate<? super E> action) {
s.innerUpdateLock.lock();
try {
long size = s.size();
@@ -86,6 +87,13 @@ public abstract class HashSegmentIteration<K, E extends HashEntry<K>> implements
initEntryRemovedOnThisIteration(false);
}
}
public void forEachSegmentEntry(Consumer<? super E> action) {
forEachSegmentEntryWhile(e -> {
action.accept(e);
return true;
});
}
public void checkEntryNotRemovedOnThisIteration() {
if (entryRemovedOnThisIterationInit())
@@ -80,7 +80,7 @@ interface AbstractChronicleMap<K, V> extends ChronicleMap<K, V>, Serializable {
@Override
default boolean containsValue(Object value) {
return !forEachEntryWhile(c -> !c.valueEqualTo((V) value));
return !forEachEntryWhile(c -> !c.value().equals(c.context().wrapValueAsData((V) value)));
}
@Override
@@ -90,7 +90,7 @@ interface AbstractChronicleMap<K, V> extends ChronicleMap<K, V>, Serializable {
@Override
default void forEach(BiConsumer<? super K, ? super V> action) {
forEachEntry(c -> action.accept(c.key(), c.get()));
forEachEntry(c -> action.accept(c.key().get(), c.value().get()));
}
@Override
@@ -147,7 +147,7 @@ interface AbstractChronicleMap<K, V> extends ChronicleMap<K, V>, Serializable {
@Override
public void forEach(java.util.function.Consumer<? super V> action) {
AbstractChronicleMap.this.forEachEntry(c -> action.accept(c.get()));
AbstractChronicleMap.this.forEachEntry(c -> action.accept(c.value().get()));
}
};
}
@@ -181,7 +181,7 @@ interface AbstractChronicleMap<K, V> extends ChronicleMap<K, V>, Serializable {
@Override
public void forEach(java.util.function.Consumer<? super K> action) {
AbstractChronicleMap.this.forEachEntry(c -> action.accept(c.key()));
AbstractChronicleMap.this.forEachEntry(c -> action.accept(c.key().get()));
}
};
}
@@ -198,9 +198,9 @@ interface AbstractChronicleMap<K, V> extends ChronicleMap<K, V>, Serializable {
try {
return forEachEntryWhile(c -> {
K k = c.key();
K k = c.key().get();
V v = (V) m.get(k instanceof CharSequence ? k.toString() : k);
return v != null && c.valueEqualTo(v);
return v != null && c.value().equals(c.context().wrapValueAsData(v));
});
} catch (ClassCastException unused) {
return false;
@@ -247,25 +247,21 @@ interface AbstractChronicleMap<K, V> extends ChronicleMap<K, V>, Serializable {
@Override
default void forEachEntry(final Consumer<? super MapKeyContext<K, V>> action) {
default void forEachEntry(final Consumer<? super MapEntry<K, V>> action) {
forEachEntryWhile(c -> {
action.accept(c);
return true;
});
}
IterationContextInterface<K, V> iterationContext();
@Override
default boolean forEachEntryWhile(final Predicate<? super MapKeyContext<K, V>> action) {
default boolean forEachEntryWhile(final Predicate<? super MapEntry<K, V>> action) {
boolean interrupt = false;
iteration:
try (IterationContextInterface<K, V> c = iterationContext()) {
for (int segmentIndex = actualSegments() - 1; segmentIndex >= 0; segmentIndex--) {
c.initTheSegmentIndex(segmentIndex);
if (!c.forEachRemoving(e -> action.test(c.deprecatedMapKeyContextOnIteration()))) {
for (int i = actualSegments() - 1; i >= 0; i--) {
try (MapSegmentContext<K, V, ?> c = segmentContext(i)) {
if (!c.forEachSegmentEntryWhile(action)) {
interrupt = true;
break iteration;
break;
}
}
}
@@ -30,7 +30,6 @@ import java.io.Closeable;
import java.io.File;
import java.io.IOException;
import java.io.Serializable;
import java.lang.Object;
import java.util.Map;
import java.util.concurrent.ConcurrentMap;
@@ -54,9 +53,9 @@ import java.util.concurrent.ConcurrentMap;
* @param <K> the map key type
* @param <V> the map value type
*/
public interface ChronicleMap<K, V>
extends ConcurrentMap<K, V>,
ChronicleHash<K, MapKeyContext<K, V>, ExternalMapQueryContext<K, V, ?>> {
public interface ChronicleMap<K, V> extends ConcurrentMap<K, V>,
ChronicleHash<K, MapEntry<K, V>, MapSegmentContext<K, V, ?>,
ExternalMapQueryContext<K, V, ?>> {
/**
* Returns the value to which the specified key is mapped, or {@code null} if this map contains
@@ -16,24 +16,22 @@
package net.openhft.chronicle.map;
import net.openhft.chronicle.map.impl.IterationContextInterface;
import java.util.ArrayDeque;
import java.util.Iterator;
import java.util.Map.Entry;
import java.util.NoSuchElementException;
import java.util.Queue;
import java.util.function.Consumer;
/**
* Very inefficient (esp. if segments are large), but CORRECT implementation
*/
abstract class ChronicleMapIterator<K, V, E> implements Iterator<E> {
abstract class ChronicleMapIterator<K, V, E> implements Iterator<E>, Consumer<MapEntry<K, V>> {
final AbstractChronicleMap<K, V> map;
private final Thread ownerThread = Thread.currentThread();
private final Queue<E> entryBuffer = new ArrayDeque<>();
private int segmentIndex;
IterationContextInterface<K, V> context;
E returned;
ChronicleMapIterator(AbstractChronicleMap<K, V> map) {
@@ -54,24 +52,23 @@ abstract class ChronicleMapIterator<K, V, E> implements Iterator<E> {
while (true) {
if (segmentIndex < 0)
return;
try (IterationContextInterface<K, V> c = map.iterationContext()) {
context = c;
c.initTheSegmentIndex(segmentIndex);
try (MapSegmentContext<K, V, ?> c = map.segmentContext(segmentIndex)) {
segmentIndex--;
if (c.size() == 0)
continue;
c.forEachRemoving(e -> {
entryBuffer.add(read());
return true;
});
c.forEachSegmentEntry(this);
return;
} finally {
context = null;
}
}
}
abstract E read();
@Override
public void accept(MapEntry<K, V> e) {
entryBuffer.add(read(e));
}
abstract E read(MapEntry<K, V> entry);
@Override
public boolean hasNext() {
@@ -108,9 +105,9 @@ abstract class ChronicleMapIterator<K, V, E> implements Iterator<E> {
}
@Override
Entry<K, V> read() {
K key = context.key().getUsing(null);
V value = context.value().getUsing(null);
Entry<K, V> read(MapEntry<K, V> entry) {
K key = entry.key().getUsing(null);
V value = entry.value().getUsing(null);
return new WriteThroughEntry<>(map, key, value);
}
@@ -127,8 +124,8 @@ abstract class ChronicleMapIterator<K, V, E> implements Iterator<E> {
}
@Override
K read() {
return context.key().getUsing(null);
K read(MapEntry<K, V> entry) {
return entry.key().getUsing(null);
}
@Override
@@ -0,0 +1,25 @@
/*
* Copyright 2015 Higher Frequency Trading
*
* http://www.higherfrequencytrading.com
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package net.openhft.chronicle.map;
import net.openhft.chronicle.hash.HashSegmentContext;
public interface MapSegmentContext<K, V, R>
extends HashSegmentContext<K, MapEntry<K, V>>, MapContext<K, V, R> {
}
@@ -18,7 +18,6 @@
package net.openhft.chronicle.map;
import net.openhft.chronicle.hash.KeyContext;
import net.openhft.chronicle.hash.replication.AbstractReplication;
import net.openhft.chronicle.hash.replication.ReplicableEntry;
import net.openhft.chronicle.hash.replication.TimeProvider;
@@ -218,7 +217,7 @@ public class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? sup
@Override
public void clear() {
forEachEntry(KeyContext::remove);
forEachEntry(c -> c.context().remove(c));
}
void addCloseable(Closeable closeable) {
@@ -220,13 +220,13 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
}
@Override
public boolean forEachEntryWhile(Predicate<? super MapKeyContext<K, V>> predicate) {
public boolean forEachEntryWhile(Predicate<? super MapEntry<K, V>> predicate) {
// TODO implement!
throw new UnsupportedOperationException();
}
@Override
public void forEachEntry(Consumer<? super MapKeyContext<K, V>> action) {
public void forEachEntry(Consumer<? super MapEntry<K, V>> action) {
// TODO implement!
throw new UnsupportedOperationException();
}
@@ -554,6 +554,11 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
throw new UnsupportedOperationException("Contexts are not supported by stateless clients");
}
@Override
public MapSegmentContext<K, V, ?> segmentContext(int segmentIndex) {
throw new UnsupportedOperationException();
}
public V get(Object key) {
return fetchObject(vClass, GET, (K) key);
}
@@ -49,7 +49,7 @@ import static net.openhft.chronicle.map.ChronicleMapBuilder.greatestCommonDiviso
public class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
V, VI, MVI extends MetaBytesInterop<V, ? super VI>, R>
extends VanillaChronicleHash<K, KI, MKI, MapKeyContext<K, V>,
extends VanillaChronicleHash<K, KI, MKI, MapEntry<K, V>, MapSegmentContext<K, V, ?>,
ExternalMapQueryContext<K, V, ?>>
implements AbstractChronicleMap<K, V> {
@@ -299,7 +299,6 @@ public class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super
throw new IllegalArgumentException("acquire*() MUST reuse the given " +
"value. Given value " + using + " cannot be reused to read " + acquired);
}
}
@NotNull
@@ -325,7 +324,7 @@ public class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super
@Override
public void clear() {
forEachEntry(KeyContext::remove);
forEachEntry(c -> c.context().remove(c));
}
public final long readValueSize(Bytes entry) {
@@ -348,8 +347,7 @@ public class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super
return (CompiledMapIterationContext<K, KI, MKI, V, VI, MVI, R>) iterCxt.get();
}
@Override
public IterationContextInterface<K, V> iterationContext() {
public IterationContextInterface<K, V, ?> iterationContext() {
CompiledMapIterationContext<K, KI, MKI, V, VI, MVI, R> c = i().getContext();
c.initUsed(true);
return c;
@@ -373,6 +371,13 @@ public class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super
return q;
}
@Override
public MapSegmentContext<K, V, ?> segmentContext(int segmentIndex) {
IterationContextInterface<K, V, ?> c = iterationContext();
c.initTheSegmentIndex(segmentIndex);
return c;
}
@Override
public V get(Object key) {
try (QueryContextInterface<K, V, R> q = queryContext(key)) {
@@ -1,21 +1,11 @@
package net.openhft.chronicle.map.impl;
import net.openhft.chronicle.map.MapEntry;
import net.openhft.chronicle.map.MapKeyContext;
import net.openhft.chronicle.map.MapSegmentContext;
import java.util.function.Predicate;
public interface IterationContextInterface<K, V> extends MapEntry<K, V>, AutoCloseable {
public interface IterationContextInterface<K, V, R> extends MapEntry<K, V>,
MapSegmentContext<K, V, R> {
long pos();
void initTheSegmentIndex(int segmentIndex);
MapKeyContext<K, V> deprecatedMapKeyContextOnIteration();
boolean forEachRemoving(Predicate<? super MapEntry<K, V>> action);
@Override
void close();
long size();
}
@@ -26,7 +26,6 @@ import net.openhft.chronicle.hash.impl.stage.hash.ThreadLocalCopiesHolder;
import net.openhft.chronicle.map.impl.stage.data.bytes.EntryValueBytesData;
import net.openhft.chronicle.map.impl.stage.data.instance.WrappedValueInstanceData;
import net.openhft.chronicle.map.impl.stage.entry.MapEntryStages;
import net.openhft.chronicle.map.impl.stage.iter.DeprecatedMapKeyContextOnIteration;
import net.openhft.chronicle.map.impl.stage.iter.IterationCheckOnEachPublicOperation;
import net.openhft.chronicle.map.impl.stage.iter.MapSegmentIteration;
import net.openhft.chronicle.map.impl.stage.map.MapEntryOperationsDelegation;
@@ -65,8 +64,6 @@ import net.openhft.sg.Staged;
EntryValueBytesData.class,
WrappedValueInstanceData.class,
DeprecatedMapKeyContextOnIteration.class,
})
public class MapIterationContext {
}
@@ -4,7 +4,8 @@ import net.openhft.chronicle.hash.replication.ReplicableEntry;
import java.util.function.Consumer;
public interface ReplicatedIterationContextInterface<K, V> extends IterationContextInterface<K, V> {
public interface ReplicatedIterationContextInterface<K, V, R>
extends IterationContextInterface<K, V, R> {
void forEachReplicableEntry(Consumer<? super ReplicableEntry> action);
@@ -27,7 +27,6 @@ import net.openhft.chronicle.map.impl.stage.data.DummyValueZeroData;
import net.openhft.chronicle.map.impl.stage.data.bytes.EntryValueBytesData;
import net.openhft.chronicle.map.impl.stage.data.instance.WrappedValueInstanceData;
import net.openhft.chronicle.map.impl.stage.entry.ReplicatedMapEntryStages;
import net.openhft.chronicle.map.impl.stage.iter.DeprecatedMapKeyContextOnIteration;
import net.openhft.chronicle.map.impl.stage.iter.IterationCheckOnEachPublicOperation;
import net.openhft.chronicle.map.impl.stage.iter.ReplicatedMapSegmentIteration;
import net.openhft.chronicle.map.impl.stage.map.MapEntryOperationsDelegation;
@@ -70,8 +69,6 @@ import net.openhft.sg.Staged;
WrappedValueInstanceData.class,
DeprecatedMapKeyContextOnIteration.class,
DummyValueZeroData.class,
})
public class ReplicatedMapIterationContext {
@@ -31,7 +31,7 @@ public interface VanillaChronicleMapHolder<K, KI, MKI extends MetaBytesInterop<K
VanillaChronicleMap<K, KI, MKI, V, VI, MVI, R> m();
@Override
default VanillaChronicleHash<K, KI, MKI, ?, ?> h() {
default VanillaChronicleHash<K, KI, MKI, ?, ?, ?> h() {
return m();
}
@@ -1,128 +0,0 @@
package net.openhft.chronicle.map.impl.stage.iter;
import net.openhft.chronicle.hash.Data;
import net.openhft.chronicle.hash.impl.stage.entry.SegmentStages;
import net.openhft.chronicle.hash.impl.stage.hash.CheckOnEachPublicOperation;
import net.openhft.chronicle.hash.locks.InterProcessLock;
import net.openhft.chronicle.map.MapKeyContext;
import net.openhft.chronicle.map.impl.stage.entry.MapEntryStages;
import net.openhft.chronicle.map.impl.stage.map.MapEntryOperationsDelegation;
import net.openhft.lang.io.Bytes;
import net.openhft.sg.StageRef;
import net.openhft.sg.Staged;
import org.jetbrains.annotations.NotNull;
@Staged
public class DeprecatedMapKeyContextOnIteration<K, V> implements MapKeyContext<K, V> {
@StageRef CheckOnEachPublicOperation checkOnEachPublicOperation;
@StageRef SegmentStages s;
@StageRef MapSegmentIteration<K, V> it;
@StageRef MapEntryOperationsDelegation<K, V, ?> ops;
@StageRef MapEntryStages<K, V> e;
@Override
public long valueOffset() {
checkOnEachPublicOperation.checkOnEachPublicOperation();
return e.valueOffset;
}
@Override
public long valueSize() {
checkOnEachPublicOperation.checkOnEachPublicOperation();
return e.valueSize;
}
@Override
public boolean valueEqualTo(V value) {
checkOnEachPublicOperation.checkOnEachPublicOperation();
return Data.bytesEquivalent(e.entryValue, it.context().wrapValueAsData(value));
}
@Override
public V get() {
checkOnEachPublicOperation.checkOnEachPublicOperation();
return it.value().get();
}
@Override
public V getUsing(V usingValue) {
checkOnEachPublicOperation.checkOnEachPublicOperation();
return it.value().getUsing(usingValue);
}
@Override
public boolean put(V newValue) {
checkOnEachPublicOperation.checkOnEachPublicOperation();
ops.replaceValue(it, it.context().wrapValueAsData(newValue));
return true;
}
@NotNull
@Override
public K key() {
checkOnEachPublicOperation.checkOnEachPublicOperation();
return it.key().get();
}
@NotNull
@Override
public Bytes entry() {
checkOnEachPublicOperation.checkOnEachPublicOperation();
return e.entryBytes;
}
@Override
public long keyOffset() {
checkOnEachPublicOperation.checkOnEachPublicOperation();
return e.keyOffset;
}
@Override
public long keySize() {
checkOnEachPublicOperation.checkOnEachPublicOperation();
return e.keySize;
}
@Override
public boolean containsKey() {
checkOnEachPublicOperation.checkOnEachPublicOperation();
return true;
}
@Override
public boolean remove() {
checkOnEachPublicOperation.checkOnEachPublicOperation();
ops.remove(it);
return true;
}
@Override
public void close() {
throw new UnsupportedOperationException("close() is not supported during iteration");
}
@NotNull
@Override
public InterProcessLock readLock() {
throw unsupportedLocks();
}
@NotNull
@Override
public InterProcessLock updateLock() {
throw unsupportedLocks();
}
@NotNull
@Override
public InterProcessLock writeLock() {
throw unsupportedLocks();
}
@NotNull
private UnsupportedOperationException unsupportedLocks() {
return new UnsupportedOperationException(
"Lock operations are not supported (and not needed!) during iteration");
}
}
@@ -25,7 +25,7 @@ import net.openhft.sg.Staged;
@Staged
public class IterationCheckOnEachPublicOperation extends CheckOnEachPublicOperation {
@StageRef MapSegmentIteration<?, ?> iteration;
@StageRef MapSegmentIteration<?, ?, ?> iteration;
@Override
public void checkOnEachPublicOperation() {
@@ -30,11 +30,10 @@ import net.openhft.sg.Staged;
import org.jetbrains.annotations.NotNull;
@Staged
public abstract class MapSegmentIteration<K, V> extends HashSegmentIteration<K, MapEntry<K, V>>
implements MapEntry<K, V>, IterationContextInterface<K, V> {
public abstract class MapSegmentIteration<K, V, R> extends HashSegmentIteration<K, MapEntry<K, V>>
implements MapEntry<K, V>, IterationContextInterface<K, V, R> {
@StageRef MapEntryStages<K, V> entry;
@StageRef public DeprecatedMapKeyContextOnIteration<K, V> deprecatedMapKeyContextOnIteration;
@StageRef WrappedValueInstanceValueHolder<K, V, ?> wrappedValueInstanceValueHolder;
@Override
@@ -12,8 +12,8 @@ import net.openhft.sg.Staged;
import java.util.function.Consumer;
@Staged
public abstract class ReplicatedMapSegmentIteration<K, V> extends MapSegmentIteration<K, V>
implements ReplicatedIterationContextInterface<K, V>, ReplicableEntry {
public abstract class ReplicatedMapSegmentIteration<K, V, R> extends MapSegmentIteration<K, V, R>
implements ReplicatedIterationContextInterface<K, V, R>, ReplicableEntry {
@StageRef ReplicatedMapEntryStages<K, V, ?> e;
@StageRef ReplicationUpdate<K> ru;
@@ -24,5 +24,6 @@ import net.openhft.chronicle.hash.KeyContext;
import java.util.Set;
public interface ChronicleSet<K>
extends Set<K>, ChronicleHash<K, KeyContext<K>, ExternalSetQueryContext<K, ?>> {
extends Set<K>, ChronicleHash<K, SetEntry<K>, SetSegmentContext<K, ?>,
ExternalSetQueryContext<K, ?>> {
}
@@ -143,13 +143,18 @@ class SetFromMap<E> extends AbstractSet<E>
}
@Override
public boolean forEachEntryWhile(Predicate<? super KeyContext<E>> predicate) {
return m.forEachEntryWhile(predicate);
public SetSegmentContext<E, ?> segmentContext(int segmentIndex) {
throw new UnsupportedOperationException();
}
@Override
public void forEachEntry(Consumer<? super KeyContext<E>> action) {
m.forEachEntry(action);
public boolean forEachEntryWhile(Predicate<? super SetEntry<E>> predicate) {
throw new UnsupportedOperationException();
}
@Override
public void forEachEntry(Consumer<? super SetEntry<E>> action) {
throw new UnsupportedOperationException();
}
@Override
@@ -0,0 +1,25 @@
/*
* Copyright 2015 Higher Frequency Trading
*
* http://www.higherfrequencytrading.com
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package net.openhft.chronicle.set;
import net.openhft.chronicle.hash.HashSegmentContext;
public interface SetSegmentContext<K, R>
extends HashSegmentContext<K, SetEntry<K>>, SetContext<K, R> {
}
@@ -36,15 +36,15 @@ public class VanillaChronicleMapConverter<K, V> extends AbstractChronicleMapConv
@Override
public void marshal(Object o, final HierarchicalStreamWriter writer, final MarshallingContext
marshallingContext) {
((ChronicleMap<K, V>) o).forEachEntry(c -> {
((ChronicleMap<K, V>) o).forEachEntry(e -> {
writer.startNode("entry");
{
final Object key = c.key();
final Object key = e.key().get();
writer.startNode(key.getClass().getName());
marshallingContext.convertAnother(key);
writer.endNode();
Object value = c.get();
Object value = e.value().get();
writer.startNode(value.getClass().getName());
marshallingContext.convertAnother(value);
writer.endNode();