Alawn/Chronicle-Map
1
0
Fork 0
mirror of https://github.com/OpenHFT/Chronicle-Map.git synced 2021-09-13 01:26:12 +00:00
Code Releases Activity

Remove blank lines at the start of methods and loops.

Browse Source
This commit is contained in:
Peter Lawrey
2014-11-30 22:48:57 +02:00
parent adae564595
commit 1e40c765f8
36 changed files with 0 additions and 228 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/hash/replication/ReplicationHub.java
-1
View File
@@ -72,7 +72,6 @@ public final class ReplicationHub extends AbstractReplication {
@Override
public String toString() {
return "ReplicationHub{" + super.toString() +
"maxEntrySize=" + maxEntrySize +
", channels=" + Arrays.toString(channels) + '}';
src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java
-7
View File
@@ -149,7 +149,6 @@ abstract class AbstractChannelReplicator implements Closeable {
*/
private Selector openSelector(@NotNull final Selector selector,
@NotNull final SelectedSelectionKeySet selectedKeySet) {
try {
Class<?> selectorImplClass =
@@ -319,7 +318,6 @@ abstract class AbstractChannelReplicator implements Closeable {
long throttleIntervalInMillis,
long serializedEntrySize,
long bitsPerDay) {
this.selector = selector;
this.throttleInterval = throttleIntervalInMillis;
double bytesPerMs = ((double) bitsPerDay) / DAYS.toMillis(1) / BITS_IN_A_BYTE;
@@ -352,7 +350,6 @@ abstract class AbstractChannelReplicator implements Closeable {
LOG.debug("Restoring OP_WRITE on all channels");
for (SelectableChannel selectableChannel : channels) {
final SelectionKey selectionKey = selectableChannel.keyFor(selector);
if (selectionKey != null)
selectionKey.interestOps(selectionKey.interestOps() | OP_WRITE);
@@ -425,7 +422,6 @@ abstract class AbstractChannelReplicator implements Closeable {
@Override
public boolean onEntry(final Bytes entry, final int chronicleId) {
long pos0 = in.position();
// used to denote that this is not a stateless map event
@@ -500,9 +496,7 @@ abstract class AbstractChannelReplicator implements Closeable {
* @param reconnectionInterval the period to wait before connecting
*/
private void doConnect(final long reconnectionInterval) {
final Thread thread = new Thread(new Runnable() {
public void run() {
SelectableChannel socketChannel = null;
try {
@@ -514,7 +508,6 @@ abstract class AbstractChannelReplicator implements Closeable {
try {
closeables.add(socketChannel);
} catch (IllegalStateException e) {
// close could have be called from another thread, while we were in Thread.sleep()
// which would cause a IllegalStateException
closeQuietly(socketChannel);
src/main/java/net/openhft/chronicle/map/ChannelProvider.java
-9
View File
@@ -115,7 +115,6 @@ final class ChannelProvider implements Closeable {
try {
final int chronicleId = (int) source.readStopBit();
if (chronicleId < chronicleChannels.length) {
// this channel is has not currently been created so it updates will be ignored
if (channelEntryExternalizables[chronicleId] != null)
channelEntryExternalizables[chronicleId]
@@ -129,7 +128,6 @@ final class ChannelProvider implements Closeable {
};
private final byte localIdentifier;
final Replica asReplica = new Replica() {
@Override
public byte identifier() {
return localIdentifier;
@@ -138,7 +136,6 @@ final class ChannelProvider implements Closeable {
@Override
public ModificationIterator acquireModificationIterator(
final byte remoteIdentifier, final ModificationNotifier notifier) {
channelDataLock.writeLock().lock();
try {
final ModificationIterator result = modificationIterator.get(remoteIdentifier);
@@ -146,7 +143,6 @@ final class ChannelProvider implements Closeable {
return result;
final ModificationIterator result0 = new ModificationIterator() {
@Override
public boolean hasNext() {
channelDataLock.readLock().lock();
@@ -321,7 +317,6 @@ final class ChannelProvider implements Closeable {
}
private static ByteBufferBytes toBootstrapMessage(int chronicleChannel, final long lastModificationTime, final byte localIdentifier) {
final ByteBufferBytes writeBuffer = new ByteBufferBytes(ByteBuffer.allocate(1 + 1 + 2 + 8));
writeBuffer.writeByte(BOOTSTRAP_MESSAGE);
writeBuffer.writeByte(localIdentifier);
@@ -409,7 +404,6 @@ final class ChannelProvider implements Closeable {
class SystemQueue {
final Replica asReplica = new Replica() {
@Override
public byte identifier() {
throw new UnsupportedOperationException();
@@ -418,14 +412,12 @@ final class ChannelProvider implements Closeable {
@Override
public ModificationIterator acquireModificationIterator(
final byte remoteIdentifier, final ModificationNotifier modificationNotifier) {
final ModificationIterator result = systemModificationIterator.get(remoteIdentifier);
if (result != null)
return result;
final PayloadProvider iterator = new PayloadProvider() {
final Queue<Bytes> payloads = new LinkedTransferQueue<Bytes>();
@Override
@@ -495,7 +487,6 @@ final class ChannelProvider implements Closeable {
SystemQueue(DirectBitSet systemModificationIteratorBitSet,
AtomicReferenceArray<PayloadProvider> systemModificationIterator,
MessageHandler messageHandler) {
this.systemModificationIteratorBitSet = systemModificationIteratorBitSet;
this.systemModificationIterator = systemModificationIterator;
this.messageHandler = messageHandler;
src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java
-1
View File
@@ -51,7 +51,6 @@ public final class ChronicleMapBuilder<K, V> implements ChronicleMapBuilderI<K,
*/
public static <K, V> ChronicleMapBuilder<K, V> of(
@NotNull Class<K> keyClass, @NotNull Class<V> valueClass) {
if (valueClass.isEnum())
return new ChronicleMapBuilder<K, V>(OnHeapUpdatableChronicleMapBuilder.of(keyClass, valueClass));
src/main/java/net/openhft/chronicle/map/CloseablesManager.java
-1
View File
@@ -32,7 +32,6 @@ final class CloseablesManager implements Closeable {
private final List<Closeable> closeables = new ArrayList<>();
public synchronized void add(Closeable closeable) {
if (closeable == null)
throw new NullPointerException();
if (!closeables.contains(closeable))
src/main/java/net/openhft/chronicle/map/NodeDiscovery.java
-18
View File
@@ -116,12 +116,10 @@ public class NodeDiscovery {
= new AtomicReference<CountDownLatch>(new CountDownLatch(1));
final NodeDiscoveryEventListener nodeDiscoveryEventListener = new NodeDiscoveryEventListener() {
@Override
public void onRemoteNodeEvent(@NotNull final KnownNodes remoteNodes,
@NotNull final ConcurrentExpiryMap<AddressAndPort,
ProposedNodes> proposedIdentifiersWithHost) {
LOG.info("onRemoteNodeEvent " + remoteNodes + ", proposedIdentifiersWithHost=" + proposedIdentifiersWithHost);
knownHostPorts.addAll(remoteNodes.addressAndPorts());
@@ -131,7 +129,6 @@ public class NodeDiscovery {
for (ProposedNodes proposedIdentifierWithHost :
proposedIdentifiersWithHost.values()) {
if (!proposedIdentifierWithHost.addressAndPort().equals(ourAddressAndPort)) {
int proposedIdentifier = proposedIdentifierWithHost.identifier();
if (proposedIdentifier != -1) {
knownAndProposedIdentifiers.set(proposedIdentifier, true);
@@ -216,7 +213,6 @@ public class NodeDiscovery {
// we should make a local copy as this may change
final RemoteNodeValidator remoteNodeValidator = new RemoteNodeValidator() {
private final ConcurrentMap<Byte, SocketAddress> identifiers = new ConcurrentHashMap<Byte,
SocketAddress>();
@@ -286,7 +282,6 @@ public class NodeDiscovery {
*/
private DirectBitSet orBitSets(@NotNull final DirectBitSet source,
@NotNull final DirectBitSet destination) {
// merges the two bit-sets together
for (int i = (int) source.nextSetBit(0); i > 0;
i = (int) source.nextSetBit(i + 1)) {
@@ -320,7 +315,6 @@ public class NodeDiscovery {
int count = 0;
for (; ; ) {
if (knownIdentifiers.setIfClear(possible)) {
return possible;
}
@@ -404,7 +398,6 @@ class NodeDiscoveryBroadcaster extends UdpChannelReplicator {
*/
UdpSocketChannelEntryReader(final int serializedEntrySize,
@NotNull final BytesMarshallable externalizable) {
// we make the buffer twice as large just to give ourselves headroom
in = allocateDirect(serializedEntrySize * 2);
@@ -463,7 +456,6 @@ class NodeDiscoveryBroadcaster extends UdpChannelReplicator {
UdpSocketChannelEntryWriter(final int serializedEntrySize,
@NotNull final BytesMarshallable externalizable,
@NotNull final UdpChannelReplicator udpReplicator) {
this.externalizable = externalizable;
this.udpReplicator = udpReplicator;
@@ -517,7 +509,6 @@ class KnownNodes implements BytesMarshallable {
private ATSDirectBitSet atsDirectBitSet;
KnownNodes() {
this.activeIdentifiersBitSetBytes = new ByteBufferBytes(ByteBuffer.allocate(128 / 8));
this.addressAndPorts = new ConcurrentSkipListSet<AddressAndPort>();
this.atsDirectBitSet = new ATSDirectBitSet(this.activeIdentifiersBitSetBytes);
@@ -562,7 +553,6 @@ class KnownNodes implements BytesMarshallable {
@Override
public void writeMarshallable(@net.openhft.lang.model.constraints.NotNull Bytes out) {
// make a safe copy
final Set<AddressAndPort> addressAndPorts = new HashSet<AddressAndPort>(this.addressAndPorts);
@@ -743,7 +733,6 @@ class DiscoveryNodeBytesMarshallable implements BytesMarshallable {
*/
@Override
public void writeMarshallable(@net.openhft.lang.model.constraints.NotNull Bytes out) {
if (bootstrapRequired.getAndSet(false)) {
writeBootstrap(out);
return;
@@ -777,7 +766,6 @@ class DiscoveryNodeBytesMarshallable implements BytesMarshallable {
* @return returns true if the UDP message contains the text 'BOOTSTRAP'
*/
private ProposedNodes readBootstrapMessage(Bytes in) {
final long start = in.position();
try {
@@ -789,7 +777,6 @@ class DiscoveryNodeBytesMarshallable implements BytesMarshallable {
// reads the text bootstrap
for (int i = 0; i < size; i++) {
final byte byteRead = in.readByte();
final byte expectedByte = BOOTSTRAP_BYTES.readByte(i);
@@ -811,7 +798,6 @@ class DiscoveryNodeBytesMarshallable implements BytesMarshallable {
final ProposedNodes bootstrap = readBootstrapMessage(in);
if (bootstrap != null) {
if (bootstrap.addressAndPort().equals(this.ourAddressAndPort))
return;
@@ -948,7 +934,6 @@ class ConcurrentExpiryMap<K extends BytesMarshallable, V extends BytesMarshallab
private final Class<V> vClass;
ConcurrentExpiryMap(Class<K> kClass, Class<V> vClass) {
this.kClass = kClass;
this.vClass = vClass;
}
@@ -964,7 +949,6 @@ class ConcurrentExpiryMap<K extends BytesMarshallable, V extends BytesMarshallab
short size = in.readShort();
try {
for (int i = 0; i < size; i++) {
final K k = kClass.newInstance();
k.readMarshallable(in);
@@ -1002,7 +986,6 @@ class ConcurrentExpiryMap<K extends BytesMarshallable, V extends BytesMarshallab
map.put(k, v);
final W w = new W(v);
queue.add(new Map.Entry<K, W<V>>() {
@Override
public K getKey() {
return k;
@@ -1029,7 +1012,6 @@ class ConcurrentExpiryMap<K extends BytesMarshallable, V extends BytesMarshallab
void expireEntries(long timeOlderThan) {
for (; ; ) {
final Map.Entry<K, W<V>> e = this.queue.peek();
if (e == null)
src/main/java/net/openhft/chronicle/map/OffHeapUpdatableChronicleMapBuilder.java
-1
View File
@@ -29,7 +29,6 @@ final class OffHeapUpdatableChronicleMapBuilder<K, V>
public static <K, V> ChronicleMapBuilderI<K, V> of(
@NotNull Class<K> keyClass, @NotNull Class<V> valueClass) {
if (valueClass.isEnum())
return OnHeapUpdatableChronicleMapBuilder.of(keyClass, valueClass);
src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java
-7
View File
@@ -386,7 +386,6 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
@Override
public void writeExternalEntry(@NotNull Bytes entry, @NotNull Bytes destination,
int chronicleId) {
final long initialLimit = entry.limit();
final long keySize = keySizeMarshaller.readSize(entry);
@@ -462,7 +461,6 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
public void readExternalEntry(
@NotNull ThreadLocalCopies copies, @NotNull SegmentState segmentState,
@NotNull Bytes source) {
final long keySize = keySizeMarshaller.readSize(source);
final long valueSize = valueSizeMarshaller.readSize(source);
final long timeStamp = source.readStopBit();
@@ -496,7 +494,6 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
boolean debugEnabled = LOG.isDebugEnabled();
if (isDeleted) {
if (debugEnabled) {
LOG.debug("READING FROM SOURCE - into local-id={}, remote={}, remove(key={})",
localIdentifier, remoteIdentifier, source.toString().trim()
@@ -932,7 +929,6 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
*/
private boolean shouldIgnore(@NotNull final NativeBytes entry, final long timestamp,
final byte identifier) {
final long lastModifiedTimeStamp = entry.readLong();
// if the readTimeStamp is newer then we'll reject this put()
@@ -1173,14 +1169,12 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
public void dirtyEntries(final long timeStamp,
final ModificationIterator.EntryModifiableCallback callback,
final boolean bootstrapOnlyLocalEntries) {
readLock(null);
ThreadLocalCopies copies = SegmentState.getCopies(null);
try (SegmentState segmentState = SegmentState.get(copies)) {
final int index = Segment.this.getIndex();
final MultiStoreBytes tmpBytes = segmentState.tmpBytes;
hashLookup().forEach(new MultiMap.EntryConsumer() {
@Override
public void accept(long hash, long pos) {
final NativeBytes entry = reuse(tmpBytes, offsetFromPos(pos));
@@ -1411,7 +1405,6 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
segment.readLock(null);
try {
if (changes.clearIfSet(position)) {
entryCallback.onBeforeEntry();
final long segmentPos = position & posMask;
src/main/java/net/openhft/chronicle/map/ReplicationHubFindByName.java
-2
View File
@@ -63,7 +63,6 @@ class ReplicationHubFindByName<K> implements FindByName {
final MapEventListener<CharSequence, ChronicleMapBuilderWithChannelId> listener = new
MapEventListener<CharSequence, ChronicleMapBuilderWithChannelId>() {
// creates a map based on the details that are sent to the map of builders
@Override
public void onPut(CharSequence key, ChronicleMapBuilderWithChannelId value,
@@ -179,7 +178,6 @@ class ReplicationHubFindByName<K> implements FindByName {
private ChronicleHashInstanceConfig toReplicatedViaChannel(ChronicleHashBuilder<K, ?, ?> builder, int
withChannelId) {
return builder.instance().replicatedViaChannel(replicationHub.createChannel((short) withChannelId));
}
src/main/java/net/openhft/chronicle/map/Replicators.java
-2
View File
@@ -39,7 +39,6 @@ final class Replicators {
static Replicator tcp(final AbstractReplication replication) {
return new Replicator() {
@Override
protected Closeable applyTo(@NotNull final AbstractChronicleMapBuilder builder,
@NotNull final Replica replica,
@@ -64,7 +63,6 @@ final class Replicators {
static Replicator udp(
final UdpTransportConfig replicationConfig) {
return new Replicator() {
@Override
protected Closeable applyTo(@NotNull final AbstractChronicleMapBuilder builder,
@NotNull final Replica map,
src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java
-22
View File
@@ -179,7 +179,6 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
}
private ExecutorService lazyExecutorService() {
if (executorService == null) {
synchronized (this) {
if (executorService == null)
@@ -203,7 +202,6 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
long timeoutAt = System.currentTimeMillis() + timeoutMs;
for (; ; ) {
checkTimeout(timeoutAt);
// ensures that the excising connection are closed
@@ -259,7 +257,6 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
* closes the existing connections and establishes a new closeables
*/
private void closeExisting() {
// ensure that any excising connection are first closed
if (closeables != null)
closeables.closeQuietly();
@@ -397,7 +394,6 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
*/
@Override
public synchronized boolean equals(Object object) {
if (this == object) return true;
if (object == null || object.getClass().isAssignableFrom(Map.class))
return false;
@@ -488,7 +484,6 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
}
public synchronized V remove(Object key) {
if (key == null)
throw keyNotNullNPE();
@@ -547,7 +542,6 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
bytes.writeObject(function);
return;
} catch (IllegalStateException e) {
Throwable cause = e.getCause();
if (cause instanceof IOException && cause.getMessage().contains("Not enough available space")) {
@@ -560,7 +554,6 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
}
public synchronized void putAll(Map<? extends K, ? extends V> map) {
final long sizeLocation = putReturnsNull ? writeEvent(PUT_ALL_WITHOUT_ACC) :
writeEventAnSkip(PUT_ALL);
@@ -577,7 +570,6 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
long timeoutTime = config.timeoutMs() + System.currentTimeMillis();
try {
for (; ; ) {
if (clientChannel == null)
clientChannel = lazyConnect(config.timeoutMs(), config.remoteAddress());
@@ -597,7 +589,6 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
}
private synchronized void writeEntriesForPutAll(Map<? extends K, ? extends V> map) {
final int numberOfEntries = map.size();
int numberOfEntriesReadSoFar = 0;
@@ -648,7 +639,6 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
}
private void resizeIfRequired(int numberOfEntries, int numberOfEntriesReadSoFar, final long start) {
final long remaining = bytes.remaining();
if (remaining < maxEntrySize) {
@@ -670,7 +660,6 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
}
private void resizeBuffer(int size, long start) {
if (LOG.isDebugEnabled())
LOG.debug("resizing buffer to size=" + size);
@@ -710,7 +699,6 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
}
public synchronized void clear() {
// get the data back from the server
blockingFetch(writeEventAnSkip(CLEAR));
}
@@ -729,7 +717,6 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
BytesReader<V> valueReader = valueReaderWithSize.readerForLoop(null);
for (; ; ) {
final boolean hasMoreEntries = in.readBoolean();
// number of entries in this chunk
@@ -766,7 +753,6 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
final Map<K, V> result = new HashMap<K, V>();
for (; ; ) {
final boolean hasMoreEntries = in.readBoolean();
// number of entries in this chunk
@@ -812,7 +798,6 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
final Set<K> result = new HashSet<>();
for (; ; ) {
boolean hasMoreEntries = in.readBoolean();
// number of entries in the chunk
@@ -850,7 +835,6 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
* @return
*/
private long writeEventAnSkip(EventId event) {
long sizeLocation = writeEvent(event);
// skips for the transaction id
@@ -911,7 +895,6 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
}
private Bytes blockingFetch(long sizeLocation) {
try {
long startTime = System.currentTimeMillis();
return blockingFetchThrowable(sizeLocation, this.config.timeoutMs(),
@@ -926,7 +909,6 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
}
private Bytes blockingFetch0(long sizeLocation, final long transactionId, long startTime) {
try {
return blockingFetchThrowable(sizeLocation, this.config.timeoutMs(), transactionId, startTime);
} catch (IOException e) {
@@ -943,7 +925,6 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
final long timeoutTime = startTime + timeOutMs;
for (; ; ) {
if (clientChannel == null)
clientChannel = lazyConnect(config.timeoutMs(), config.remoteAddress());
@@ -969,7 +950,6 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
}
private Bytes blockingFetchReadOnly(long timeoutTime, final long transactionId) {
try {
return blockingFetch(timeoutTime, transactionId);
} catch (IOException e) {
@@ -989,7 +969,6 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
final int requiredSize = size + SIZE_OF_SIZE;
if (bytes.capacity() < requiredSize) {
long pos = bytes.position();
long limit = bytes.position();
bytes.position(limit);
@@ -1034,7 +1013,6 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
private Bytes receive(int requiredNumberOfBytes, long timeoutTime) throws IOException {
while (bytes.buffer().position() < requiredNumberOfBytes) {
assert requiredNumberOfBytes <= bytes.capacity();
int len = clientChannel.read(bytes.buffer());
src/main/java/net/openhft/chronicle/map/TcpReplicator.java
-40
View File
@@ -140,7 +140,6 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
}
while (selector.isOpen()) {
registerPendingRegistrations();
final int nSelectedKeys = select();
@@ -157,7 +156,6 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
opWriteUpdater.applyUpdates();
if (useJavaNIOSelectionKeys) {
// use the standard java nio selector
if (nSelectedKeys == 0)
@@ -216,7 +214,6 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
}
private void processKey(long approxTime, SelectionKey key) {
try {
if (!key.isValid())
@@ -291,7 +288,6 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
*/
void heartBeatMonitor(long approxTime) {
for (int i = activeKeys.nextSetBit(0); i >= 0; i = activeKeys.nextSetBit(i + 1)) {
try {
final SelectionKey key = selectionKeysStore[i];
@@ -332,12 +328,10 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
*/
private void sendHeartbeatIfRequired(final long approxTime,
@NotNull final SelectionKey key) {
final Attached attachment = (Attached) key.attachment();
if (attachment.isHandShakingComplete() && attachment.entryWriter.lastSentTime +
heartBeatIntervalMillis < approxTime) {
attachment.entryWriter.lastSentTime = approxTime;
attachment.entryWriter.writeHeartbeatToBuffer();
@@ -531,7 +525,6 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
socketChannel.register(selector, OP_READ | OP_WRITE, attached);
if (attached.remoteIdentifier == Byte.MIN_VALUE) {
final byte remoteIdentifier = reader.identifierFromBuffer();
if (remoteIdentifier == STATELESS_CLIENT) {
@@ -589,7 +582,6 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
}
if (!attached.hasRemoteHeartbeatInterval) {
final long value = reader.remoteHeartbeatIntervalFromBuffer();
if (value == Long.MIN_VALUE)
@@ -628,7 +620,6 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
final Attached attached = (Attached) key.attachment();
if (attached.entryWriter.isWorkIncomplete()) {
final boolean completed = attached.entryWriter.doWork();
if (completed)
@@ -668,7 +659,6 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
int len = attached.entryReader.readSocketToBuffer(socketChannel);
if (len == -1) {
if (replicationConfig.autoReconnectedUponDroppedConnection()) {
AbstractConnector connector = attached.connector;
if (connector != null)
@@ -856,7 +846,6 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
addPendingRegistration(new Runnable() {
@Override
public void run() {
final Attached attached = new Attached();
attached.connector = ClientConnector.this;
@@ -926,7 +915,6 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
*/
@Override
public void onChange() {
if (remoteIdentifier != Byte.MIN_VALUE)
TcpReplicator.this.opWriteUpdater.set(remoteIdentifier);
}
@@ -995,11 +983,9 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
final boolean handShakingComplete = attached.isHandShakingComplete();
for (; ; ) {
final boolean wasDataRead = modificationIterator.nextEntry(entryCallback, 0);
if (!wasDataRead) {
// if we have no more data to write to the socket then we will
// un-register OP_WRITE on the selector, until more data becomes available
if (in.position() == 0 && handShakingComplete)
@@ -1059,7 +1045,6 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
* localHeartbeatInterval
*/
private void writeHeartbeatToBuffer() {
// denotes the state - 0 for a heartbeat
in.writeByte(HEARTBEAT.ordinal());
@@ -1124,7 +1109,6 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
}
void resizeBuffer(long size) {
assert size < Integer.MAX_VALUE;
if (size < in.capacity())
@@ -1177,12 +1161,10 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
*/
void entriesFromBuffer(Attached attached, SelectionKey key) throws InterruptedException, IOException {
for (; ; ) {
out.limit(in.position());
// its set to MIN_VALUE when it should be read again
if (state == NOT_SET) {
if (out.remaining() < SIZE_OF_SIZE + 1)
return;
@@ -1217,7 +1199,6 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
if (isStateless) {
if (statelessServerConnector == null) {
LOG.error("", new IllegalArgumentException("received an event " +
"from a stateless map, stateless maps are not " +
"currently supported when using Chronicle Channels"));
@@ -1233,7 +1214,6 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
// fill out the write buffer, so this data will be send when the buffer
// is no longer full, and as such is treated as future work
if (futureWork != null) {
try { // we will complete what we can for now
boolean isComplete = futureWork.doWork(attached.entryWriter.in);
if (!isComplete)
@@ -1260,7 +1240,6 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
* compacts the buffer and updates the {@code in} and {@code out} accordingly
*/
private void compactBuffer() {
// the maxEntrySizeBytes used here may not be the maximum size of the entry in its serialized form
// however, its only use as an indication that the buffer is becoming full and should be compacted
// the buffer can be compacted at any time
@@ -1335,12 +1314,10 @@ class StatelessServerConnector<K, V> {
Work processStatelessEvent(final byte eventId,
@NotNull final Bytes writer,
@NotNull final ByteBufferBytes reader) {
final StatelessChronicleMap.EventId event = VALUES[eventId];
// these methods don't return a result
switch (event) {
case KEY_SET:
return keySet(reader, writer);
@@ -1365,7 +1342,6 @@ class StatelessServerConnector<K, V> {
// these methods return a result
switch (event) {
case LONG_SIZE:
return longSize(reader, writer, sizeLocation);
@@ -1490,7 +1466,6 @@ class StatelessServerConnector<K, V> {
}
private Work longSize(Bytes reader, Bytes writer, final long sizeLocation) {
try {
writer.writeLong(map.longSize());
} catch (Throwable e) {
@@ -1501,7 +1476,6 @@ class StatelessServerConnector<K, V> {
}
private Work hashCode(Bytes reader, Bytes writer, final long sizeLocation) {
try {
writer.writeInt(map.hashCode());
} catch (Throwable e) {
@@ -1513,7 +1487,6 @@ class StatelessServerConnector<K, V> {
}
private Work toString(Bytes reader, Bytes writer, final long sizeLocation) {
final String str;
final long remaining = writer.remaining();
@@ -1536,7 +1509,6 @@ class StatelessServerConnector<K, V> {
}
private Work sendException(Bytes writer, long sizeLocation, Throwable e) {
// move the position to ignore any bytes written so far
writer.position(sizeLocation + HEADER_SIZE);
@@ -1547,7 +1519,6 @@ class StatelessServerConnector<K, V> {
}
private Work isEmpty(Bytes reader, Bytes writer, final long sizeLocation) {
try {
writer.writeBoolean(map.isEmpty());
} catch (Throwable e) {
@@ -1679,7 +1650,6 @@ class StatelessServerConnector<K, V> {
}
private Work putAll(Bytes reader) {
if (MAP_SUPPORTS_BYTES) {
map.putAll(reader);
} else {
@@ -1690,7 +1660,6 @@ class StatelessServerConnector<K, V> {
}
private Work clear(Bytes reader, Bytes writer, final long sizeLocation) {
try {
map.clear();
} catch (Throwable e) {
@@ -1702,7 +1671,6 @@ class StatelessServerConnector<K, V> {
}
private Work values(Bytes reader, Bytes writer) {
final long transactionId = reader.readLong();
Collection<V> values;
@@ -1717,7 +1685,6 @@ class StatelessServerConnector<K, V> {
// this allows us to write more data than the buffer will allow
return new Work() {
@Override
public boolean doWork(Bytes out) {
final long sizeLocation = header(out, transactionId);
@@ -1727,7 +1694,6 @@ class StatelessServerConnector<K, V> {
int count = 0;
while (iterator.hasNext()) {
// we've filled up the buffer, so lets give another channel a chance to send
// some data, we don't know the max key size, we will use the entrySize instead
if (out.remaining() <= maxEntrySizeBytes) {
@@ -1761,7 +1727,6 @@ class StatelessServerConnector<K, V> {
// this allows us to write more data than the buffer will allow
return new Work() {
@Override
public boolean doWork(Bytes out) {
final long sizeLocation = header(out, transactionId);
@@ -1771,7 +1736,6 @@ class StatelessServerConnector<K, V> {
int count = 0;
while (iterator.hasNext()) {
// we've filled up the buffer, so lets give another channel a chance to send
// some data, we don't know the max key size, we will use the entrySize instead
if (out.remaining() <= maxEntrySizeBytes) {
@@ -1791,7 +1755,6 @@ class StatelessServerConnector<K, V> {
}
private Work entrySet(final Bytes reader, Bytes writer) {
final long transactionId = reader.readLong();
final Set<Map.Entry<K, V>> entries;
@@ -1806,7 +1769,6 @@ class StatelessServerConnector<K, V> {
// this allows us to write more data than the buffer will allow
return new Work() {
@Override
public boolean doWork(Bytes out) {
if (out.remaining() <= maxEntrySizeBytes)
@@ -1821,11 +1783,9 @@ class StatelessServerConnector<K, V> {
int count = 0;
while (iterator.hasNext()) {
// we've filled up the buffer, so lets give another channel a chance to send
// some data, we don't know the max key size, we will use the entrySize instead
if (out.remaining() <= maxEntrySizeBytes) {
writeHeader(out, sizeLocation, count, true);
return false;
}
src/main/java/net/openhft/chronicle/map/UdpChannelReplicator.java
-5
View File
@@ -109,7 +109,6 @@ class UdpChannelReplicator extends AbstractChannelReplicator implements Replica.
try {
while (selector.isOpen()) {
registerPendingRegistrations();
// this may block for a long time, upon return the
@@ -127,7 +126,6 @@ class UdpChannelReplicator extends AbstractChannelReplicator implements Replica.
final Set<SelectionKey> selectionKeys = selector.selectedKeys();
for (final SelectionKey key : selectionKeys) {
try {
if (key.isReadable()) {
@@ -207,7 +205,6 @@ class UdpChannelReplicator extends AbstractChannelReplicator implements Replica.
}
private void enableWrites() {
if (writeChannel == null)
return;
@@ -223,7 +220,6 @@ class UdpChannelReplicator extends AbstractChannelReplicator implements Replica.
}
void disableWrites() {
if (writeChannel == null)
return;
@@ -272,7 +268,6 @@ class UdpChannelReplicator extends AbstractChannelReplicator implements Replica.
addPendingRegistration(new Runnable() {
@Override
public void run() {
try {
server.register(selector, OP_WRITE);
writeChannel = server;
src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java
-9
View File
@@ -571,7 +571,6 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
public Future<V> getLater(@NotNull K key) {
final V v = get(key);
return new Future<V>() {
@Override
public boolean cancel(boolean mayInterruptIfRunning) {
return false;
@@ -604,7 +603,6 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
public Future<V> putLater(@NotNull K key, @NotNull V value) {
final V v = put(key, value);
return new Future<V>() {
@Override
public boolean cancel(boolean mayInterruptIfRunning) {
return false;
@@ -637,7 +635,6 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
public Future<V> removeLater(@NotNull K key) {
final V v = remove(key);
return new Future<V>() {
@Override
public boolean cancel(boolean mayInterruptIfRunning) {
return false;
@@ -714,12 +711,10 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
private void registerConverter(XStream xstream) {
final Converter converter = new Converter() {
final Bytes buffer = XStreamConverter.this.buffer;
@Override
public boolean canConvert(Class aClass) {
if (EntrySet.class.isAssignableFrom(aClass)
|| WriteThroughEntry.class.isAssignableFrom(aClass))
return true;
@@ -738,9 +733,7 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
@Override
public void marshal(Object o, HierarchicalStreamWriter writer, MarshallingContext
marshallingContext) {
if (WriteThroughEntry.class.isAssignableFrom(o.getClass())) {
final SimpleEntry e = (SimpleEntry) o;
writer.startNode("chronicle-key");
@@ -760,7 +753,6 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
writer.endNode();
} else if (EntrySet.class
.isAssignableFrom(o.getClass())) {
for (Entry e : (EntrySet) o) {
writer.startNode("chronicle-entry");
marshallingContext.convertAnother(e);
@@ -824,7 +816,6 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
private <E> E get(HierarchicalStreamReader reader,
UnmarshallingContext unmarshallingContext,
Class<E> clazz) {
if (reader.getAttributeCount() > 0) {
final String type = reader.getAttribute("type");
try {