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

Merge remote-tracking branch 'origin/master'

Browse Source
This commit is contained in:
Rob Austin
2014-09-12 12:07:28 +01:00
parent bc48a6aefb d4ec1f2e27
commit a611e041bd
5 changed files with 145 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java
+10 -43
View File
@@ -55,7 +55,7 @@ abstract class AbstractChannelReplicator implements Closeable {
static final int SIZE_OF_SHORT = 2;
private static final Logger LOG = LoggerFactory.getLogger(AbstractChannelReplicator.class);
final Selector selector;
final Set<Closeable> closeables = Collections.synchronizedSet(new LinkedHashSet<Closeable>());
final CloseablesManager closeables = new CloseablesManager();
private final ExecutorService executorService;
private final Queue<Runnable> pendingRegistrations = new ConcurrentLinkedQueue<Runnable>();
@Nullable
@@ -110,28 +110,14 @@ abstract class AbstractChannelReplicator implements Closeable {
@Override
public void close() {
synchronized (this.closeables) {
for (Closeable closeable : this.closeables) {
try {
closeable.close();
} catch (IOException e) {
LOG.error("", e);
}
}
closeables.clear();
}
closeables.closeQuietly();
executorService.shutdownNow();
}
void closeEarlyAndQuietly(SelectableChannel channel) {
try {
if (throttler != null)
throttler.remove(channel);
closeables.remove(channel);
channel.close();
} catch (IOException ex) {
// do nothing
}
if (throttler != null)
throttler.remove(channel);
closeables.closeQuietly(channel);
}
@@ -317,20 +303,15 @@ abstract class AbstractChannelReplicator implements Closeable {
* connectionAttempts}
*/
public final void connectLater() {
try {
if (socketChannel != null)
socketChannel.close();
if (socketChannel != null) {
closeables.closeQuietly(socketChannel);
socketChannel = null;
} catch (IOException e1) {
LOG.error("", e1);
}
final long reconnectionInterval = connectionAttempts * 100;
if (connectionAttempts < 5)
connectionAttempts++;
doConnect(reconnectionInterval);
}
/**
@@ -349,31 +330,17 @@ abstract class AbstractChannelReplicator implements Closeable {
final Thread thread = new Thread(new Runnable() {
public void run() {
SelectableChannel socketChannel = null;
try {
if (reconnectionInterval > 0)
Thread.sleep(reconnectionInterval);
synchronized (closeables) {
socketChannel = doConnect();
if (socketChannel == null)
return;
closeables.add(socketChannel);
AbstractConnector.this.socketChannel = socketChannel;
}
SelectableChannel socketChannel = doConnect();
closeables.add(socketChannel);
AbstractConnector.this.socketChannel = socketChannel;
} catch (Exception e) {
if (socketChannel != null)
try {
socketChannel.close();
AbstractConnector.this.socketChannel = null;
} catch (IOException e1) {
//
}
LOG.error("", e);
}
}
});
src/main/java/net/openhft/chronicle/map/CloseablesManager.java
+87
View File
@@ -0,0 +1,87 @@
/*
* Copyright 2014 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 org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.Closeable;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
final class CloseablesManager implements Closeable {
private static final Logger LOG = LoggerFactory.getLogger(TcpReplicator.class.getName());
private boolean isClosed = false;
private List<Closeable> closeables = new ArrayList<Closeable>();
private void checkState() {
if (isClosed)
throw new IllegalStateException();
}
public synchronized void add(Closeable closeable) {
checkState();
if (closeable == null)
throw new NullPointerException();
closeables.add(closeable);
}
public synchronized void close(Closeable closeable) throws IOException {
checkState();
for (Iterator<Closeable> iterator = closeables.iterator(); iterator.hasNext(); ) {
if (iterator.next() == closeable)
iterator.remove();
}
closeable.close();
}
public synchronized void closeQuietly(Closeable closeable) {
try {
close(closeable);
} catch (IOException e) {
LOG.error("", e);
}
}
@Override
public synchronized void close() throws IOException {
IOException ex = null;
for (Closeable closeable : closeables) {
try {
closeable.close();
} catch (IOException e) {
LOG.error("", e);
ex = e;
}
}
closeables.clear();
isClosed = true;
if (ex != null)
throw ex;
}
public synchronized void closeQuietly() {
try {
close();
} catch (IOException e) {
// do nothing
}
}
}
src/main/java/net/openhft/chronicle/map/ReplicatingCluster.java
+30 -20
View File
@@ -28,6 +28,8 @@ import org.slf4j.LoggerFactory;
import java.io.Closeable;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;
@@ -93,10 +95,12 @@ public final class ReplicatingCluster implements Closeable {
@Override
public boolean hasNext() {
for (int i = (int) chronicleChannelBitSet.nextSetBit(0); i >= 0; i = (int)
chronicleChannelBitSet.nextSetBit(i + 1)) {
final ModificationIterator modificationIterator = chronicleChannels[i]
.acquireModificationIterator(remoteIdentifier, notifier);
// old-style iteration to avoid 1) iterator object creation
// 2) ConcurrentModificationException
for (int i = 0, len = chronicleChannelList.size(); i < len; i++) {
final ModificationIterator modificationIterator =
chronicleChannelList.get(i)
.acquireModificationIterator(remoteIdentifier, notifier);
if (modificationIterator.hasNext())
return true;
}
@@ -106,11 +110,11 @@ public final class ReplicatingCluster implements Closeable {
@Override
public boolean nextEntry(@NotNull EntryCallback callback,
final int na) {
for (int i = (int) chronicleChannelBitSet.nextSetBit(0); i >= 0;
i = (int) chronicleChannelBitSet.nextSetBit(i + 1)) {
final ModificationIterator modificationIterator = chronicleChannels[i]
for (int i = 0, len = chronicleChannelList.size(); i < len; i++) {
Replica chronicleChannel = chronicleChannelList.get(i);
final ModificationIterator modificationIterator = chronicleChannel
.acquireModificationIterator(remoteIdentifier, notifier);
if (modificationIterator.nextEntry(callback, i))
if (modificationIterator.nextEntry(callback, chronicleChannelIds.get(i)))
return true;
}
return false;
@@ -118,9 +122,9 @@ public final class ReplicatingCluster implements Closeable {
@Override
public void dirtyEntries(long fromTimeStamp) {
for (int i = (int) chronicleChannelBitSet.nextSetBit(0); i >= 0;
i = (int) chronicleChannelBitSet.nextSetBit(i + 1)) {
chronicleChannels[i].acquireModificationIterator(remoteIdentifier, notifier)
for (int i = 0, len = chronicleChannelList.size(); i < len; i++) {
chronicleChannelList.get(i)
.acquireModificationIterator(remoteIdentifier, notifier)
.dirtyEntries(fromTimeStamp);
notifier.onChange();
}
@@ -138,8 +142,8 @@ public final class ReplicatingCluster implements Closeable {
@Override
public long lastModificationTime(byte remoteIdentifier) {
long t = 0;
for (int i = (int) chronicleChannelBitSet.previousSetBit(chronicleChannels.length);
i > 0; i = (int) chronicleChannelBitSet.previousSetBit(i - 1)) {
// not including the SystemQueue at index 0
for (int i = 1, len = chronicleChannelList.size(); i < len; i++) {
t = (t == 0) ? chronicleChannels[i].lastModificationTime(remoteIdentifier) :
min(t, chronicleChannels[i].lastModificationTime(remoteIdentifier));
}
@@ -152,8 +156,9 @@ public final class ReplicatingCluster implements Closeable {
}
};
private final int maxEntrySize;
private final DirectBitSet chronicleChannelBitSet;
private final Replica[] chronicleChannels;
private final List<Replica> chronicleChannelList;
private final List<Integer> chronicleChannelIds;
private final EntryExternalizable[] channelEntryExternalizables;
private final AtomicReferenceArray<PayloadProvider> systemModificationIterator =
new AtomicReferenceArray<PayloadProvider>(128);
@@ -169,7 +174,8 @@ public final class ReplicatingCluster implements Closeable {
maxEntrySize = builder.maxEntrySize;
chronicleChannels = new Replica[builder.maxNumberOfChronicles];
channelEntryExternalizables = new EntryExternalizable[builder.maxNumberOfChronicles];
chronicleChannelBitSet = newBitSet(chronicleChannels.length);
chronicleChannelList = new ArrayList<Replica>();
chronicleChannelIds = new ArrayList<Integer>();
MessageHandler systemMessageHandler = new MessageHandler() {
@Override
public void onMessage(Bytes bytes) {
@@ -241,8 +247,9 @@ public final class ReplicatingCluster implements Closeable {
" is already in use.");
}
chronicleChannels[chronicleChannel] = replica;
chronicleChannelList.add(replica);
chronicleChannelIds.add((int) chronicleChannel);
channelEntryExternalizables[chronicleChannel] = entryExternalizable;
chronicleChannelBitSet.set(chronicleChannel);
if (chronicleChannel == 0)
return;
@@ -263,9 +270,11 @@ public final class ReplicatingCluster implements Closeable {
for (AbstractChannelReplicator replicator : replicators) {
replicator.close();
}
for (int i = (int) chronicleChannelBitSet.nextSetBit(0); i > 0;
i = (int) chronicleChannelBitSet.nextSetBit(i + 1)) {
chronicleChannels[i].close();
// indexed reverse loop to avoid ConcurrentModificationException and remove elements from
// the end of the loop for correct iteration
for (int i = chronicleChannelList.size() - 1; i >= 0; i--) {
Replica chronicleChannel = chronicleChannelList.get(i);
chronicleChannel.close();
}
}
@@ -408,7 +417,8 @@ public final class ReplicatingCluster implements Closeable {
@Override
public void close() throws IOException {
chronicleChannelBitSet.clear(chronicleChannel);
chronicleChannelList.remove(chronicleChannels[chronicleChannel]);
chronicleChannelIds.remove(chronicleChannel);
chronicleChannels[chronicleChannel] = null;
channelEntryExternalizables[chronicleChannel] = null;
}
src/main/java/net/openhft/chronicle/map/TcpReplicator.java
+6 -17
View File
@@ -46,7 +46,6 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
private static final Logger LOG = LoggerFactory.getLogger(TcpReplicator.class.getName());
private static final int BUFFER_SIZE = 0x100000; // 1MB
final AtomicBoolean forceBootstrap = new AtomicBoolean();
private final Map<SocketAddress, AbstractConnector> connectorBySocket = new ConcurrentHashMap<SocketAddress, AbstractConnector>();
private final SelectionKey[] selectionKeysStore = new SelectionKey[Byte.MAX_VALUE + 1];
// used to instruct the selector thread to set OP_WRITE on a key correlated by the bit index in the
// bitset
@@ -101,15 +100,11 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
void process() throws IOException {
try {
final Details serverDetails = new Details(serverInetSocketAddress, localIdentifier);
connectorBySocket.put(serverInetSocketAddress, new ServerConnector(serverDetails));
new ServerConnector(serverDetails).connect();
for (InetSocketAddress client : endpoints) {
final Details clientDetails = new Details(client, localIdentifier);
connectorBySocket.put(client, new ClientConnector(clientDetails));
}
for (AbstractConnector connector : connectorBySocket.values()) {
connector.connect();
new ClientConnector(clientDetails).connect();
}
while (selector.isOpen()) {
@@ -277,19 +272,13 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
final SocketChannel channel = (SocketChannel) key.channel();
if (approxTimeOutTime > attached.entryReader.lastHeartBeatReceived + attached.remoteHeartbeatInterval) {
if (approxTimeOutTime >
attached.entryReader.lastHeartBeatReceived + attached.remoteHeartbeatInterval) {
if (LOG.isDebugEnabled())
LOG.debug("lost connection, attempting to reconnect. " +
"missed heartbeat from identifier=" + attached.remoteIdentifier);
try {
channel.socket().close();
channel.close();
activeKeys.clear(attached.remoteIdentifier);
closeables.remove(channel);
} catch (IOException e) {
LOG.debug("", e);
}
activeKeys.clear(attached.remoteIdentifier);
closeables.closeQuietly(channel.socket());
attached.connector.connectLater();
}
}
src/main/java/net/openhft/chronicle/map/UdpReplicator.java
+12 -15
View File
@@ -164,23 +164,20 @@ class UdpReplicator extends AbstractChannelReplicator implements Replica.Modific
final InetSocketAddress hostAddress = new InetSocketAddress(port);
client.configureBlocking(false);
synchronized (closeables) {
if (address.isMulticastAddress()) {
client.setOption(IP_MULTICAST_IF, networkInterface);
client.setOption(SO_REUSEADDR, true);
client.bind(hostAddress);
client.join(address, networkInterface);
if (LOG.isDebugEnabled())
LOG.debug("Connecting via multicast, group=" + address);
} else {
client.bind(hostAddress);
}
if (address.isMulticastAddress()) {
client.setOption(IP_MULTICAST_IF, networkInterface);
client.setOption(SO_REUSEADDR, true);
client.bind(hostAddress);
client.join(address, networkInterface);
if (LOG.isDebugEnabled())
LOG.debug("Listening on port " + port);
closeables.add(client);
LOG.debug("Connecting via multicast, group=" + address);
} else {
client.bind(hostAddress);
}
if (LOG.isDebugEnabled())
LOG.debug("Listening on port " + port);
closeables.add(client);
return client;
}