Chronicle Map, testing tcp/udp replication in the same host

[Daniel López]

Hi there,

I'm evaluating Chronicle Map for one of our projects and I've been able to share data in the same host using the file mechanism. Cool. Now I wanted to test if I could set up tcp/udp replication correctly and even though processes in the same host share data through the mapped file, is there any way I can "force" them to share data through the network even if in the same host?

I have tried not setting the file in the ChronicleMapBuilder and using a different identifier, but I still don't see updates in one map reflected in the other. Should that work? Or is ChronicleMap too smart? :)

This is just for quick & easy testing before I start testing in multiple machines, but just in case I'm doing something wrong and this is supposed to work...

Cheers,

D.

[Rob Austin]

yes this should work see the following test case (replicating between 3 nodes ), which uses different ports ( for each instance of the same map aka node) and local host

net.openhft.chronicle.map.TCPSocketReplication3wayTest

under https://github.com/OpenHFT/Chronicle-Map/tree/2.1/src/test/java

--
You received this message because you are subscribed to the Google Groups "Chronicle" group.
To unsubscribe from this group and stop receiving emails from it, send an email to java-chronicl...@googlegroups.com.
For more options, visit https://groups.google.com/d/optout.

[Ben Cotton]

> Or is ChronicleMap too smart? :)

I would bet this is the reason.  :-)

curious that you would (ever?) want udp/tcp as a transport when ChronicleMap gives you IPC over /dev/shm as a same host transport.  IPC is 1,000x faster than OSI loopback (over udp/tcp).

[Daniel López]

Well, this is just the initial test before testing the processes in different hosts, I'm not going to really use it like that. But I'm using a Spring bean factory I created that sets all the parameters, to share several maps among 3 different apps, so I'm modiyfing things to find the easiest way to configure it and I thought I would test it first locally, if possible.

So, no, not planning to use it like that in production :).

D.

[Daniel López]

All right,

I'll check tomorrow if there's something that I'm missing.I'm using a ReplicationHub with different identifier and different port in the same tcpTransportAndNetwork for each process (jvm), but with the same channels for the equivalent maps... I'm setting up a TcpTransportAndNetworkConfig configuration where everybody has everybody else as endpoint, as there is no "order", that should not be a problem, right?

Time to debug the logs and see if they really find each other.

Thanks,

D.

[Roman Leventov]

On Mon, Jul 13, 2015 at 7:08 PM, Daniel López <d.lo...@gmail.com> wrote:

 I'm setting up a TcpTransportAndNetworkConfig configuration where everybody has everybody else as endpoint, as there is no "order", that should not be a problem, right?

I think so. 

[Daniel López]

No luck, so far.

This is what I'm doing:

-----------

// Build a replication hub for all Maps

            TcpTransportAndNetworkConfig tcpTransportAndNetworkConfig =

                TcpTransportAndNetworkConfig.of(clusterPort, addresses).heartBeatInterval(1, TimeUnit.SECONDS);

            replicationHub = ReplicationHub.builder().tcpTransportAndNetwork(tcpTransportAndNetworkConfig).createWithId(identifier);

            log.info("ReplicationHub listening at {} created with identifier {}", clusterPort, identifier);

            for (InetSocketAddress address : addresses) {

                log.info("ReplicationHub will connect to {}", address);

            }

----------

// Create a map with a different channel for each replicated map I want

ChronicleMapBuilder<K, V> chronicleMapBuilder = ChronicleMapBuilder.of(keyClass, valueClass).entries(maxEntries).averageValueSize(averageValueSize);

chronicleMapBuilder.instance().replicatedViaChannel(replicationHub.createChannel(channel));

chronicleMapBuilder.removeReturnsNull(true);

chronicleMapBuilder.putReturnsNull(true);

ChronicleMap<K, V> map = chronicleMapBuilder.create();

----------

When I start the this code in one app, I get this in the log:

...

20150714 10:39:51 INFO -  - ChronicleMapsConfiguration - ReplicationHub listening at 3806 created with identifier 3

20150714 10:39:51 INFO -  - ChronicleMapsConfiguration - ReplicationHub will connect to localhost/127.0.0.1:3807

...

then I start the other app and I get this other log, using exactly the same code (same .jar) but different configuration:

...

20150714 10:40:08 INFO -  - ChronicleMapsConfiguration - ReplicationHub listening at 3807 created with identifier 2

20150714 10:40:08 INFO -  - ChronicleMapsConfiguration - ReplicationHub will connect to localhost/127.0.0.1:3806

...

I then update the values in the maps in one of the apps, and check the sizes in both places. One app maps are full of data, the other app maps are empty.

I tried to get some logs but even setting the log level to TRACE, I did not see any log coming from Chronicle. If instead of setting the replication hub, I set the same identifier in both apps and use createPersistedTo(file), both apps see each other's updates fine.

Am I missing something?

Cheers,

D.

[Rob Austin]

try this see net.openhft.chronicle.map.TCPSocketReplication3VoidValueTest

 

package net.openhft.chronicle.map;

import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.Closeable;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.Set;

import static net.openhft.chronicle.map.Builder.newMapVoid;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;

/**
 * Test  ReplicatedChronicleMap where the Replicated is over a TCP Socket
 *
 * @author Rob Austin.
 */

public class TCPSocketReplication3VoidValueTest {

    private ChronicleMap<Integer, Void> map1;
    private ChronicleMap<Integer, Void> map2;
    private ChronicleMap<Integer, Void> map3;

    @Before
    public void setup() throws IOException {
        map1 = newMapVoid((byte) 1, 8036, new InetSocketAddress("localhost", 8037),
                new InetSocketAddress("localhost", 8039));
        map2 = newMapVoid((byte) 2, 8037, new InetSocketAddress("localhost", 8039));
        map3 = newMapVoid((byte) 3, 8039);
    }

    @After
    public void tearDown() throws InterruptedException {
        for (final Closeable closeable : new Closeable[]{map1, map2, map3}) {
            try {
                closeable.close();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        System.gc();
    }

    Set<Thread> threads;

    @Before
    public void sampleThreads() {
        threads = Thread.getAllStackTraces().keySet();
    }

    @After
    public void checkThreadsShutdown() {
        StatelessClientTest.checkThreadsShutdown(threads);
    }

    @Test
    public void test3() throws IOException, InterruptedException {

        assertEquals(null, map3.put(5, null));

        // allow time for the recompilation to resolve
        assertTrue(waitTillEqual(15000));

        assertEquals(map3, map2);
        assertTrue(!map1.isEmpty());
    }

    @Test
    public void test() throws IOException, InterruptedException {

        assertEquals(null, map1.put(1, null));
        assertEquals(null, map1.put(2, null));

        assertEquals(null, map2.put(5, null));
        assertEquals(null, map2.put(6, null));

        map1.remove(2);
        map2.remove(3);
        map1.remove(3);
        map2.put(5, null);

        // allow time for the recompilation to resolve
        assertTrue(waitTillEqual(5000));

        assertEquals(map3, map3);
        assertTrue(!map1.isEmpty());
    }

    @Test
    public void testClear() throws IOException, InterruptedException {

        assertEquals(null, map1.put(1, null));
        assertEquals(null, map1.put(2, null));

        assertEquals(null, map2.put(5, null));
        assertEquals(null, map2.put(6, null));

        map1.clear();

        map2.put(5, null);

        // allow time for the recompilation to resolve
        assertTrue("test timed out", waitTillEqual(15000));

        assertEquals(map3, map3);
        assertTrue(!map1.isEmpty());
    }

    private boolean waitTillEqual(final int timeOutMs) throws InterruptedException {
        int t = 0;
        for (; t < timeOutMs; t++) {
            if (map1.equals(map2) &&
                    map1.equals(map3) &&
                    map2.equals(map3))
                return true;
            Thread.sleep(1);
        }
        return false;
    }

}

public static <T extends ChronicleMap<Integer, Void>> T newMapVoid(
        final byte identifier,
        final int serverPort,
        final InetSocketAddress... endpoints) throws IOException {
    return (T) newTcpSocketShmBuilder(Integer.class, Void.class,
            identifier, serverPort, endpoints).create();
}

public static <K, V> ChronicleMapBuilder<K, V> newTcpSocketShmBuilder(
        Class<K> kClass, Class<V> vClass,
        final byte identifier,
        final int serverPort,
        final InetSocketAddress... endpoints) throws IOException {
    TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig.of(serverPort, Arrays.asList(endpoints))
            .heartBeatInterval(1L, TimeUnit.SECONDS).autoReconnectedUponDroppedConnection(true);
    return ChronicleMapBuilder.of(kClass, vClass)
            .entries(SIZE)
            .replication(identifier, tcpConfig);
}

[Daniel López]

It seems putting nulls as values is causing an issue?

        assertEquals(null, map1.put(1, null));

-> 

java.lang.NullPointerException

at net.openhft.chronicle.map.VanillaChronicleMap.checkValue(VanillaChronicleMap.java:199)

at net.openhft.chronicle.map.VanillaChronicleMap.put(VanillaChronicleMap.java:460)

at tst.TCPSocketReplication3VoidValueTest.test(TCPSocketReplication3VoidValueTest.java:80)

at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)

at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)

at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)

at java.lang.reflect.Method.invoke(Method.java:483)

at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50)

at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)

at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47)

at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)

at org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26)

at org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27)

at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:325)

at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:78)

at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:57)

at org.junit.runners.ParentRunner$3.run(ParentRunner.java:290)

at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:71)

at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:288)

at org.junit.runners.ParentRunner.access$000(ParentRunner.java:58)

at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:268)

at org.junit.runners.ParentRunner.run(ParentRunner.java:363)

at org.eclipse.jdt.internal.junit4.runner.JUnit4TestReference.run(JUnit4TestReference.java:50)

at org.eclipse.jdt.internal.junit.runner.TestExecution.run(TestExecution.java:38)

at org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.runTests(RemoteTestRunner.java:459)

at org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.runTests(RemoteTestRunner.java:675)

at org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.run(RemoteTestRunner.java:382)

at org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.main(RemoteTestRunner.java:192)

[Rob Austin]

I don’t think values can be null..

[Rob Austin]

having said that I see the test is using NULL values, - I will have to check.

[Rob Austin]

Which version of chronicle map are you using ?

[Daniel López]

Thanks. I did a couple of changes and I've managed to get it to connect, if just use one map and use directly the tcpTransportAndNetworkConfig.

So, if I set it like that:

...

chronicleMapBuilder.replication(identifier, tcpTransportAndNetworkConfig);

chronicleMapBuilder.create();

...

instead of 

...

ReplicationHub replicationHub = ReplicationHub.builder().tcpTransportAndNetwork(tcpTransportAndNetworkConfig).createWithId(identifier);

chronicleMapBuilder.instance().replicatedViaChannel(replicationHub.createChannel(channel));

chronicleMapBuilder.create();

Then, I can see the replication working, no other changes to the code.

Moreover, in the first case, as soon as the first application starts I can see the TcpReplicator trying to connect to the other app, which has not been started yet:

...

20150714 11:41:41 DEBUG-  - TcpReplicator - 

java.net.ConnectException: Connection refused: no further information

...

On the other hand, when I create the Map with the replication hub, I see no message from the replication hub. Maybe I'm missing a piece to "start" the replication when using the replication hub?

S!

D.

[Daniel López]

3.1.3-alpha

I'm going to try with latest stable to verify. I had forgotten it was the alpha.

S!

D.

[Rob Austin]

yes- I suggest you use the last released version see - http://search.maven.org/#artifactdetails%7Cnet.openhft%7Cchronicle-map%7C2.1.17%7Cjar

rather than the alpha.

Rob

[Daniel López]

Unfortunately, using 2.1.17 yields the same results.

...

chronicleMapBuilder.replication(identifier, tcpTransportAndNetworkConfig);

chronicleMapBuilder.create();

...

works

...

ReplicationHub replicationHub = ReplicationHub.builder().tcpTransportAndNetwork(tcpTransportAndNetworkConfig).createWithId(identifier);

chronicleMapBuilder.instance().replicatedViaChannel(replicationHub.createChannel(channel));

chronicleMapBuilder.create();

...

Does not work.

S!

D.

[Daniel López]

I found it.

chronicleMapBuilder.instance().replicatedViaChannel(replicationHub.createChannel(channel)).create();

is not the same as 

chronicleMapBuilder.instance().replicatedViaChannel(replicationHub.createChannel(channel));

chronicleMapBuilder.create();

The second way does not get the map to be replicated. A subtle change, easy to miss, IMHO.

S!

D.

[Rob Austin]

Thanks for the feedback - we will try to improve this API in the alpha.

[Roman Leventov]

Yes, instance() returns an immutable configuration object, that returns an independent copy with the specific configuration changed, on each call.

Before instance(), ChronicleMapBuilder is mutable.