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 branch '2.1' of github.com:OpenHFT/Chronicle-Map into 2.1

Browse Source
This commit is contained in:
Rob Austin
2015-02-02 19:44:19 +00:00
parent c56994f9c9 a19cd2060b
commit 357803e0df
16 changed files with 1251 additions and 1059 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java
+69 -49
View File
@@ -55,7 +55,6 @@ abstract class AbstractChannelReplicator implements Closeable {
public static final int SIZE_OF_SIZE = 4;
public static final int SIZE_OF_TRANSACTION_ID = 8;
public static final int SIZE_OF_TIME_SHIFT = 2;
private static final Logger LOG = LoggerFactory.getLogger(AbstractChannelReplicator.class);
@@ -67,6 +66,15 @@ abstract class AbstractChannelReplicator implements Closeable {
final CloseablesManager closeables = new CloseablesManager();
final Selector selector = openSelector(closeables);
private final ExecutorService executorService;
private final ExecutorService connectExecutorService = Executors.newCachedThreadPool(
new NamedThreadFactory("TCP-connect", true) {
@Override
public Thread newThread(@net.openhft.lang.model.constraints.NotNull Runnable r) {
return lastThread = super.newThread(r);
}
});
private volatile Thread lastThread;
private Throwable startedHere;
private Future<?> future;
@@ -228,10 +236,31 @@ abstract class AbstractChannelReplicator implements Closeable {
@Override
public void close() {
if (Thread.interrupted())
LOG.warn("Already interrupted");
long start = System.currentTimeMillis();
closeResources();
isClosed = true;
connectExecutorService.shutdown();
try {
if (!connectExecutorService.awaitTermination(5, TimeUnit.MILLISECONDS)) {
connectExecutorService.shutdownNow();
connectExecutorService.awaitTermination(10, TimeUnit.SECONDS);
}
} catch (InterruptedException e) {
LOG.error("", e);
}
try {
closeResources();
if (!executorService.awaitTermination(5, TimeUnit.MILLISECONDS)) {
executorService.shutdownNow();
executorService.awaitTermination(10, TimeUnit.SECONDS);
}
} catch (InterruptedException e) {
LOG.error("", e);
}
try {
if (future != null)
@@ -261,18 +290,6 @@ abstract class AbstractChannelReplicator implements Closeable {
isClosed = true;
executorService.shutdown();
try {
executorService.awaitTermination(5, TimeUnit.MILLISECONDS);
} catch (Exception e) {
executorService.shutdownNow();
try {
executorService.awaitTermination(10, TimeUnit.SECONDS);
} catch (InterruptedException e1) {
LOG.error("", e);
}
}
closeables.closeQuietly();
if (segmentState != null)
segmentState.close();
@@ -593,50 +610,53 @@ 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 {
if (reconnectionInterval > 0)
Thread.sleep(reconnectionInterval);
socketChannel = doConnect();
try {
connectExecutorService.submit(new Runnable() {
public void run() {
SelectableChannel socketChannel = null;
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
if (reconnectionInterval > 0)
Thread.sleep(reconnectionInterval);
socketChannel = doConnect();
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);
return;
}
AbstractConnector.this.socketChannel = socketChannel;
} catch (Exception e) {
closeQuietly(socketChannel);
return;
LOG.debug("", e);
}
AbstractConnector.this.socketChannel = socketChannel;
} catch (Exception e) {
closeQuietly(socketChannel);
LOG.debug("", e);
}
}
private void closeQuietly(SelectableChannel socketChannel) {
if (socketChannel == null)
return;
try {
socketChannel.close();
} catch (Exception e1) {
// do nothing
}
}
});
thread.setName(name);
thread.setDaemon(true);
thread.start();
});
} catch (Exception e) {
if (!isClosed)
LOG.error("", e);
}
}
private void closeQuietly(SelectableChannel socketChannel) {
if (socketChannel == null)
return;
try {
socketChannel.close();
} catch (Exception e1) {
// do nothing
}
}
public void setSuccessfullyConnected() {
connectionAttempts = 0;
}
}
}
src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java
+1 -1
View File
@@ -103,7 +103,7 @@ public final class ChronicleMapBuilder<K, V> implements Cloneable,
ChronicleHashBuilder<K, ChronicleMap<K, V>, ChronicleMapBuilder<K, V>>,
MapBuilder<ChronicleMapBuilder<K, V>>, Serializable {
static final byte UDP_REPLICATION_MODIFICATION_ITERATOR_ID = (byte) 128;
static final byte UDP_REPLICATION_MODIFICATION_ITERATOR_ID = (byte) 127;
private static final int DEFAULT_KEY_OR_VALUE_SIZE = 120;
private static final long DEFAULT_ENTRIES = 1 << 20;
private static final int MAX_SEGMENTS = (1 << 30);
src/main/java/net/openhft/chronicle/map/ReplicatedChronicleMap.java
+3 -2
View File
@@ -29,6 +29,7 @@ import net.openhft.chronicle.hash.serialization.internal.MetaBytesWriter;
import net.openhft.lang.Maths;
import net.openhft.lang.collection.ATSDirectBitSet;
import net.openhft.lang.io.Bytes;
import net.openhft.lang.io.CheckedBytes;
import net.openhft.lang.io.MultiStoreBytes;
import net.openhft.lang.io.NativeBytesI;
import net.openhft.lang.threadlocal.ThreadLocalCopies;
@@ -188,7 +189,7 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
Bytes modDelBytes = ms.bytes(offset, assignedModIterBitSetSizeInBytes()).zeroOut();
offset += assignedModIterBitSetSizeInBytes();
startOfModificationIterators = offset;
modIterSet = new ATSDirectBitSet(modDelBytes);
modIterSet = new ATSDirectBitSet(new CheckedBytes(modDelBytes));
}
/**
@@ -232,7 +233,7 @@ final class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? supe
}
@Override
V put1(@NotNull K key, V value, boolean replaceIfPresent) {
V put1(K key, V value, boolean replaceIfPresent) {
return put0(key, value, replaceIfPresent, localIdentifier, currentTime());
}
src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java
+17 -5
View File
@@ -37,6 +37,7 @@ import java.lang.reflect.Field;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.nio.channels.AsynchronousCloseException;
import java.nio.channels.SocketChannel;
import java.util.*;
import java.util.concurrent.atomic.AtomicLong;
@@ -81,7 +82,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
private final InetSocketAddress remoteAddress;
private final long timeoutMs;
private final int tcpBufferSize;
private Class<K> kClass;
private Class<V> vClass;
private boolean putReturnsNull;
@@ -376,7 +377,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
// read a single byte back
while (this.connectionOutBuffer.position() <= 0) {
int read = clientChannel.read(this.connectionOutBuffer);// the remote identifier
if (read ==-1)
if (read == -1)
throw new IOException("server conncetion closed");
checkTimeout(timeoutTime);
}
@@ -1133,6 +1134,9 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
try {
return blockingFetchThrowable(timeoutTime, transactionId);
} catch (AsynchronousCloseException e) {
LOG.error("name=" + name, e);
throw new RuntimeException(e);
} catch (IOException e) {
close();
throw new IORuntimeException(e);
@@ -1169,8 +1173,12 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
} else
inBytes.limit(inBytes.capacity());
// block until we have received all the bytes in this chunk
receiveBytesFromSocket(remainingBytes, timeoutTime);
try {
// block until we have received all the bytes in this chunk
receiveBytesFromSocket(remainingBytes, timeoutTime);
}catch (java.nio.channels.AsynchronousCloseException e ){
return inBytes;
}
final boolean isException = inBytes.readBoolean();
@@ -1308,6 +1316,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
@SuppressWarnings("UnusedReturnValue")
private Bytes receiveBytesFromSocket(int requiredNumberOfBytes, long timeoutTime) throws IOException {
// assert !outBytesLock.isHeldByCurrentThread();
assert inBytesLock.isHeldByCurrentThread();
@@ -1315,11 +1324,14 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
inBytes.limit(requiredNumberOfBytes);
inBytes.buffer().position(0);
inBytes.buffer().limit(requiredNumberOfBytes);
SocketChannel clientChannel0 = clientChannel;
if (clientChannel0 == null)
return inBytes;
while (inBytes.buffer().remaining() > 0) {
assert requiredNumberOfBytes <= inBytes.capacity();
int len = clientChannel.read(inBytes.buffer());
int len = clientChannel0.read(inBytes.buffer());
if (len == -1)
throw new IORuntimeException("Disconnection to server");
src/main/java/net/openhft/chronicle/map/VanillaChronicleMap.java
+9 -2
View File
@@ -234,6 +234,9 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
// todo remove the CheckedBytesStore before the release
this.ms = new CheckedBytesStore(bytesStore);
// this.ms = bytesStore;
onHeaderCreated();
long offset = getHeaderSize();
@@ -530,11 +533,15 @@ class VanillaChronicleMap<K, KI, MKI extends MetaBytesInterop<K, ? super KI>,
}
@Override
public final V putIfAbsent(@net.openhft.lang.model.constraints.NotNull K key, V value) {
public final V putIfAbsent(K key, V value) {
if (key == null)
throw new NullPointerException();
return put1(key, value, false);
}
V put1(@net.openhft.lang.model.constraints.NotNull K key, V value, boolean replaceIfPresent) {
V put1(K key, V value, boolean replaceIfPresent) {
if (key == null)
throw new NullPointerException();
checkKey(key);
checkValue(value);
ThreadLocalCopies copies = keyInteropProvider.getCopies(null);
src/test/java/net/openhft/chronicle/map/ChronicleMapTest.java
+1088 -940
View File
File diff suppressed because it is too large Load Diff
src/test/java/net/openhft/chronicle/map/DemoChronicleMapTest.java
+3 -1
View File
@@ -3,7 +3,6 @@ package net.openhft.chronicle.map;
import net.openhft.lang.model.DataValueClasses;
import net.openhft.lang.model.constraints.MaxSize;
import net.openhft.lang.values.IntValue;
import org.junit.Ignore;
import org.junit.Test;
import java.io.File;
@@ -51,6 +50,8 @@ public class DemoChronicleMapTest {
assertNotNull(v);
}
}
file.delete();
}
@Test
@@ -93,6 +94,7 @@ public class DemoChronicleMapTest {
assertNotNull(v);
}
}
file.delete();
}
}
src/test/java/net/openhft/chronicle/map/EntryCountMapTest.java
+20 -7
View File
@@ -190,17 +190,26 @@ public class EntryCountMapTest {
Random random = new Random(seed);
int counter = minSize + random.nextInt(9999 + maxSize);
final int stride = 1 + random.nextInt(100);
int maxKeySize = "key:".length() +
final int maxKeySize = "key:".length() +
(int) round(log10(moreThanMaxSize(maxSize) * stride + counter)) + 1;
final ChronicleMap<CharSequence, LongValue> map =
getSharedMap(minSize, segments, maxKeySize);
return es.submit(new Callable<Void>() {
@Override
public Void call() throws Exception {
testEntriesMaxSize0(segments, minSize, maxSize, seed, stride, map);
return null;
File f = null;
try (final ChronicleMap<CharSequence, LongValue> map =
getSharedMap(minSize, segments, maxKeySize)) {
f = map.file();
testEntriesMaxSize0(segments, minSize, maxSize, seed, stride, map);
} finally {
if (f != null && f.exists())
f.delete();
return null;
}
}
});
}
void testEntriesMaxSize(int segments, int minSize, int maxSize, int seed) throws IOException {
@@ -210,8 +219,12 @@ public class EntryCountMapTest {
int stride = 1 + random.nextInt(100);
int maxKeySize = "key:".length() +
(int) round(log10(moreThanMaxSize(maxSize) * stride + counter)) + 1;
ChronicleMap<CharSequence, LongValue> map = getSharedMap(minSize, segments, maxKeySize);
testEntriesMaxSize0(segments, minSize, maxSize, counter, stride, map);
File file;
try (ChronicleMap<CharSequence, LongValue> map = getSharedMap(minSize, segments, maxKeySize)) {
file = map.file();
testEntriesMaxSize0(segments, minSize, maxSize, counter, stride, map);
}
file.delete();
}
void testEntriesMaxSize0(int segments, int minSize, int maxSize, int counter, int stride,
src/test/java/net/openhft/chronicle/map/LargeEntriesTest.java
+30 -30
View File
@@ -41,7 +41,6 @@ import static org.junit.Assert.*;
*/
public class LargeEntriesTest {
@Ignore("Issues with TC")
@Test
public void testLargeStrings() throws ExecutionException, InterruptedException, IOException {
final int ENTRIES = 250;
@@ -49,7 +48,7 @@ public class LargeEntriesTest {
File file = File.createTempFile("largeEntries" + System.currentTimeMillis(), ".deleteme");
file.deleteOnExit();
final ChronicleMap<String, String> map = ChronicleMapBuilder
try (final ChronicleMap<String, String> map = ChronicleMapBuilder
.of(String.class, String.class)
.valueMarshaller(SnappyStringMarshaller.INSTANCE)
.actualSegments(1) // to force an error.
@@ -57,37 +56,38 @@ public class LargeEntriesTest {
.averageKeySize(10)
.averageValueSize(ENTRY_SIZE)
.putReturnsNull(true)
.createPersistedTo(file);
{
warmUpCompression(ENTRY_SIZE);
int threads = 4; //Runtime.getRuntime().availableProcessors();
ExecutorService es = Executors.newFixedThreadPool(threads);
final int block = ENTRIES / threads;
for (int i = 0; i < 3; i++) {
long start = System.currentTimeMillis();
List<Future<?>> futureList = new ArrayList<>();
for (int t = 0; t < threads; t++) {
final int finalT = t;
futureList.add(es.submit(new Runnable() {
@Override
public void run() {
exerciseLargeStrings(map, finalT * block, finalT * block + block,
ENTRY_SIZE);
}
}));
.createPersistedTo(file)) {
{
warmUpCompression(ENTRY_SIZE);
int threads = 4; //Runtime.getRuntime().availableProcessors();
ExecutorService es = Executors.newFixedThreadPool(threads);
final int block = ENTRIES / threads;
for (int i = 0; i < 3; i++) {
long start = System.currentTimeMillis();
List<Future<?>> futureList = new ArrayList<>();
for (int t = 0; t < threads; t++) {
final int finalT = t;
futureList.add(es.submit(new Runnable() {
@Override
public void run() {
exerciseLargeStrings(map, finalT * block, finalT * block + block,
ENTRY_SIZE);
}
}));
}
for (Future<?> future : futureList) {
future.get();
}
long time = System.currentTimeMillis() - start;
long operations = 3;
System.out.printf("Put/Get rate was %.1f MB/s%n",
operations * ENTRIES * ENTRY_SIZE / 1e6 / (time / 1e3));
}
for (Future<?> future : futureList) {
future.get();
}
long time = System.currentTimeMillis() - start;
long operations = 3;
System.out.printf("Put/Get rate was %.1f MB/s%n",
operations * ENTRIES * ENTRY_SIZE / 1e6 / (time / 1e3));
es.shutdown();
}
es.shutdown();
if (es.isTerminated())
map.close();
}
file.delete();
}
private void warmUpCompression(int entrySize) {
src/test/java/net/openhft/chronicle/map/StatelessClientTest.java
+1 -3
View File
@@ -24,7 +24,6 @@ import net.openhft.lang.io.ByteBufferBytes;
import net.openhft.lang.io.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -151,8 +150,7 @@ public class StatelessClientTest {
}
}
// TODO ignore why?
@Ignore
@Test(timeout = 10000)
public void testBufferOverFlowPutAllAndEntrySet() throws IOException, InterruptedException {
int port = s_port++;
src/test/java/net/openhft/chronicle/map/TCPSocketReplicationBootStrapTest.java
+5 -1
View File
@@ -55,8 +55,9 @@ public class TCPSocketReplicationBootStrapTest {
ChronicleMapBuilder<Integer, CharSequence> map2aBuilder =
newTcpSocketShmBuilder(Integer.class, CharSequence.class,
(byte) 2, 8092, new InetSocketAddress("localhost", 8091));
File persistenceFile = getPersistenceFile();
final ChronicleMap<Integer, CharSequence> map2a =
map2aBuilder.createPersistedTo(getPersistenceFile());
map2aBuilder.createPersistedTo(persistenceFile);
map2a.put(10, "EXAMPLE-10"); // this will be the last time that map1 go an update from map2
long lastModificationTime;
@@ -85,6 +86,9 @@ public class TCPSocketReplicationBootStrapTest {
// add data into it
waitTillEqual(5000);
assertEquals("ADDED WHEN DISCONNECTED TO MAP1", map1.get(11));
map2.file().delete();
persistenceFile.delete();
}
@Test
src/test/java/net/openhft/chronicle/map/TCPSocketReplicationIntValueTest.java
+3 -3
View File
@@ -24,7 +24,6 @@ import net.openhft.lang.model.DataValueClasses;
import net.openhft.lang.values.IntValue;
import org.junit.After;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Test;
import java.io.Closeable;
@@ -34,7 +33,6 @@ import java.nio.ByteBuffer;
import java.util.Random;
import java.util.Set;
import static net.openhft.chronicle.map.Builder.getPersistenceFile;
import static net.openhft.chronicle.map.Builder.newTcpSocketShmBuilder;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
@@ -59,7 +57,7 @@ public class TCPSocketReplicationIntValueTest {
((Byteable) value).bytes(new ByteBufferBytes(ByteBuffer.allocateDirect(4)), 0);
map1Builder = newTcpSocketShmBuilder(IntValue.class, CharSequence.class,
(byte) 1, s_port, new InetSocketAddress("localhost", s_port + 1));
map1 = map1Builder.createPersistedTo(getPersistenceFile());
map1 = map1Builder.create();
map2 = newTcpSocketShmBuilder(IntValue.class, CharSequence.class,
(byte) 2, s_port + 1)
.entries(Builder.SIZE)
@@ -72,7 +70,9 @@ public class TCPSocketReplicationIntValueTest {
for (final Closeable closeable : new Closeable[]{map1, map2}) {
try {
closeable.close();
} catch (Exception e) {
e.printStackTrace();
}
src/test/java/net/openhft/chronicle/map/TimeBasedReplicationTest.java
-7
View File
@@ -25,7 +25,6 @@ import org.junit.Before;
import org.junit.Test;
import org.mockito.Mockito;
import java.io.File;
import java.io.IOException;
import java.util.Set;
import java.util.concurrent.TimeUnit;
@@ -39,12 +38,6 @@ public class TimeBasedReplicationTest extends JSR166TestCase {
public static final byte IDENTIFIER = 1;
private static File getPersistenceFile() {
String TMP = System.getProperty("java.io.tmpdir");
File file = new File(TMP + "/chm-test" + System.nanoTime());
file.deleteOnExit();
return file;
}
Set<Thread> threads;
src/test/java/net/openhft/chronicle/map/UDPSocketReplicationTest.java
+1 -1
View File
@@ -78,7 +78,7 @@ public class UDPSocketReplicationTest {
}
@Test
@Ignore
@Ignore("just used for manual testing")
public void testBufferOverflow() throws IOException, InterruptedException {
for (int i = 0; i < 1024; i++) {
src/test/java/net/openhft/chronicle/map/jsr166/map/MapCheck.java
-6
View File
@@ -111,12 +111,6 @@ public class MapCheck {
serTest(size);
}
private static File getPersistenceFile() {
String TMP = System.getProperty("java.io.tmpdir");
File file = new File(TMP + "/chm-test" + System.nanoTime());
file.deleteOnExit();
return file;
}
static Map newMap() {
try {
src/test/java/net/openhft/chronicle/map/jsr166/map/StatelessChronicleMapTest.java
+1 -1
View File
@@ -308,7 +308,7 @@ public class StatelessChronicleMapTest extends JSR166TestCase {
static int s_port = 11050;
@Test(timeout = 30000)
@Ignore
@Ignore("performance test")
public void testByteArrayPerf() throws IOException {
ByteBuffer key = ByteBuffer.allocate(8);
ByteBuffer value = ByteBuffer.allocate(50);