Views and ForkChannels

[snyderra]

Using ForkChannels with a standard JChannel.

JChannel jchannel = JChannel("udp.xml");
JChannel forkChannel = new ForkChannel(jChannel, "fork_stack","forkcluster01",true, ProtocolStack.Position.ABOVE, TP.class);

In my case I have multiple, but for this example let's consider just one.

Upon initial creation and connection, the view_accepted is called for each channel/cluster. However, once client A and B are established and C joins, A and B will have view_accepted called on the main channel but not called on the forkChannel receiver. C will have view_accepted call for both channels.

Is this expected behavior? Is this configurable to have the views updated on all channels?

[belaban]

Do you have a sample program? Note that FORK is not intended to be added directly above the transport, but it is usually the top protocol in a given stack!

[snyderra]

What I am wanting is view_accepted to be called when a ForkChannel connects and disconnects. This can be dropped next to the ForkChannelTest.java and run with mvn -Dtest=ForkChannelViewAcceptedTest test
Also moved to the top of the stack as suggested.

package org.jgroups.tests;

import org.jgroups.*;
import org.jgroups.fork.ForkChannel;
import org.jgroups.stack.ProtocolStack;
import org.jgroups.util.FastArray;
import org.testng.annotations.Test;

import static org.testng.AssertJUnit.*;

import java.util.logging.Logger;
import java.util.List;
import java.util.logging.Level;

public class ForkChannelViewAcceptedTest {
    protected static final String CLUSTER = "ForkChannelTest";
    private static final Logger logger = Logger.getLogger(ForkChannelTest.class.getName());

    private static class MyViewReceiver<T> extends org.jgroups.util.MyReceiver<T> {
       protected final List<View>            viewList=new FastArray<>(128);

       @Override
        public void viewAccepted(View view) {
            super.viewAccepted(view);
            lock.lock();
            try {
                viewList.add(view);
            }
            finally {
                lock.unlock();
            }
        }

        public List<View> getViews() {
            return viewList;
        }
        @Override
        public MyViewReceiver<T> reset() {
            super.reset();
            viewList.clear();
            return this;
        }
    }

    @Test
    public void testViewAccepted() throws Exception {

        JChannel aa = null;
        JChannel bb = null;
        MyViewReceiver<Message> raa = (MyViewReceiver) new MyViewReceiver<Message>().rawMsgs(true).verbose(true).name(CLUSTER + "AA");
        MyViewReceiver<Message> rbb = (MyViewReceiver) new MyViewReceiver<Message>().rawMsgs(true).verbose(true).name(CLUSTER + "BB");
        MyViewReceiver<Message> raafc = (MyViewReceiver) new MyViewReceiver<Message>().rawMsgs(true).verbose(true).name(CLUSTER + "AAFC");
        MyViewReceiver<Message> rbbfc = (MyViewReceiver) new MyViewReceiver<Message>().rawMsgs(true).verbose(true).name(CLUSTER + "BBFC");


        try {
            //setup initial network
            aa = new JChannel("udp.xml").name("AA");
            bb = new JChannel("udp.xml").name("BB");
            aa.setReceiver(raa);
            bb.setReceiver(rbb);
            aa.connect(CLUSTER);
            bb.connect(CLUSTER);
            while (aa.isConnected() == false || bb.isConnected() == false) {
                Thread.sleep(100);
                logger.log(Level.INFO, CLUSTER + " not connected yet");
            }

            //setup forkchannels
            ForkChannel aafc = new ForkChannel(aa, "fork_test",CLUSTER+"-fc",true, ProtocolStack.Position.ABOVE, aa.getProtocolStack().getTopProtocol().getClass());
            ForkChannel bbfc = new ForkChannel(bb, "fork_test",CLUSTER+"-fc",true, ProtocolStack.Position.ABOVE, bb.getProtocolStack().getTopProtocol().getClass());
            aafc.setReceiver(raafc);
            bbfc.setReceiver(rbbfc);

            // Connect the first fork channel
            aafc.connect(CLUSTER+"-fc");
            while (aafc.isConnected() == false) {
                Thread.sleep(100);
                logger.log(Level.INFO, CLUSTER + " not connected yet");
            }
            // Expecting 1 view update for the one fork channel connected
            assertEquals(1,raafc.viewList.size());
            // Expecting the view to contain only one member, it self as bbfc has not connected yet
            assertEquals(1,raafc.viewList.get(0).size());

            assertEquals(0,rbbfc.viewList.size());

            // Connect the second fork channel
            bbfc.connect(CLUSTER+"-fc");

            // Wait for the fork channels to connect
            while (aafc.isConnected() == false || bbfc.isConnected() == false) {
                Thread.sleep(100);
                logger.log(Level.INFO, CLUSTER + " not connected yet");
            }

            System.out.println("Closing fork channel bbfc");
            bbfc.close();
            while (bbfc.isConnected()) {
                Thread.sleep(100);
                logger.log(Level.INFO, "waiting for bbfc to close");
            }
            Thread.sleep(1000);
            System.out.println("aafc.view" + aafc.getView());

            // When bbfc is closed, the view should be updated
            //    expected at least two view updates. one for bbfc connect and one for bbfc close
            assertTrue(raafc.viewList.size() >= 2);
            //    after bbfc is closed, the view should contain only one member, it self.
            assertEquals(1,raafc.viewList.get(raafc.viewList.size()-1).size());

        } finally {
            System.out.println("Closing channels");
            if (aa != null)
                aa.close();
            if (bb != null)
                bb.close();
            while (aa.isConnected() || bb.isConnected()) {
                Thread.sleep(100);
                logger.log(Level.INFO, "waiting for channels to close");
            }
        }
    }
}

[snyderra]

Also tried inserting at every position in the stack and did not see any difference in the calls to view_accepted.

[belaban]

Hi Rob
you have wrong assumptions about ForkChannels: a ForkChannel always has the view of its underlying main channel. When a ForkChannel is connected, disconnected or closed, this will not affect the underlying main channel!

Below's the changed program. Note that it fails at the end when you close the second ForkChannel. You expect that this will change trigger a view change, but it doesn't. A view change will only be triggered when you close one of the main channels (AA or BB).

See ForkChannelTest as an example.

public class ForkChannelViewAcceptedTest {
protected static final String CLUSTER = "ForkChannelTest";
private static final Log logger = LogFactory.getLog(ForkChannelTest.class);

private static class MyViewReceiver<T> extends org.jgroups.util.MyReceiver<T> {
   protected final List<View> viewList=new FastArray<>(128);

   @Override
    public void viewAccepted(View view) {
        super.viewAccepted(view);
        lock.lock();
        try {
            viewList.add(view);
        }
        finally {
            lock.unlock();
        }
    }

    public List<View> getViews() {
        return viewList;
    }
    @Override
    public MyViewReceiver<T> reset() {
        super.reset();
        viewList.clear();
        return this;
    }
}

@Test
public void testViewAccepted() throws Exception {

    JChannel aa = null;
    JChannel bb = null;
    MyViewReceiver<Message> raa = (MyViewReceiver) new MyViewReceiver<Message>().rawMsgs(true).verbose(true).name(CLUSTER + "AA");
    MyViewReceiver<Message> rbb = (MyViewReceiver) new MyViewReceiver<Message>().rawMsgs(true).verbose(true).name(CLUSTER + "BB");
    MyViewReceiver<Message> raafc = (MyViewReceiver) new MyViewReceiver<Message>().rawMsgs(true).verbose(true).name(CLUSTER + "AAFC");
    MyViewReceiver<Message> rbbfc = (MyViewReceiver) new MyViewReceiver<Message>().rawMsgs(true).verbose(true).name(CLUSTER + "BBFC");


    try {
        //setup initial network
        aa = new JChannel("udp.xml").name("AA").setReceiver(raa);
        bb = new JChannel("udp.xml").name("BB").setReceiver(rbb);

        //setup forkchannels
        ForkChannel aafc = new ForkChannel(aa, "fork_test", CLUSTER+"-fc", true, ABOVE, aa.getProtocolStack().getTopProtocol().getClass());
        ForkChannel bbfc = new ForkChannel(bb, "fork_test",CLUSTER+"-fc",true, ABOVE, bb.getProtocolStack().getTopProtocol().getClass());
        aafc.setReceiver(raafc);
        bbfc.setReceiver(rbbfc);

        aa.connect(CLUSTER);

        // Connect the first fork channel
        aafc.connect(CLUSTER+"-fc");
        while (aafc.isConnected() == false) {
            Thread.sleep(100);
            logger.info(CLUSTER + " not connected yet");
        }
        // Expecting 1 view update for the one fork channel connected
        assert 1 ==raafc.viewList.size();
        // Expecting the view to contain only one member, it self as bbfc has not connected yet
        assert 1 == raafc.viewList.get(0).size();

        assert 0 == rbbfc.viewList.size();

        bb.connect(CLUSTER);
        // Connect the second fork channel
        bbfc.connect(CLUSTER+"-fc");

        // Wait for the fork channels to connect
        while (aafc.isConnected() == false || bbfc.isConnected() == false) {
            Thread.sleep(100);
            logger.info(CLUSTER + " not connected yet");
        }

        System.out.println("Closing fork channel bbfc");
        bbfc.close();
        while (bbfc.isConnected()) {
            Thread.sleep(100);
            logger.info("waiting for bbfc to close");
        }
        Thread.sleep(1000);
        System.out.println("aafc.view" + aafc.getView());

        // When bbfc is closed, the view should be updated
        //    expected at least two view updates. one for bbfc connect and one for bbfc close
        assert raafc.viewList.size() >= 2;
        //    after bbfc is closed, the view should contain only one member, it self.
        assert 1 == raafc.viewList.get(raafc.viewList.size()-1).size();

    } finally {
        System.out.println("Closing channels");
        if (aa != null)
            aa.close();
        if (bb != null)
            bb.close();
        while (aa.isConnected() || bb.isConnected()) {
            Thread.sleep(100);
            logger.info("waiting for channels to close");
        }
    }
}

}

[snyderra]

Thanks for the clarification.

I am using this on android devices for message passing. The current implementation uses the default multicast config with the listening port changed. The android devices are hitting resource limits with 6 devices on the network with 10 channels each as the owner of the channel, so each device has 60 connections max. The thought is that each device can subscribe or unsubscribe as needed. This way each device only gets the messages it needs. I am triggering the initial messages off of view_accepted updates.

Moving to ForkChannels took care of the resource issue on android, but I no longer have a way to tell if the remote is still subscribed.

As Jgroups is extremely configurable is there something that I am missing? Another configuration that would share the underlying connection yet provide membership views for each channel?

Thanks for your help, its appreciated.

[belaban]

You could use only a single channel on each device and do the multiplexing/demultiplexing yourself, e.g. based on a 'topic'. This is actually what ForkChannel provides: the fork-channel-id serves as the muxer/demuxer. So a message is sent to every device, and ForkChannels which don't match the topic drop the message.
If you want to know the members subscribed to a given topic, you could consult each FORK instance, walk through the fork-stacks and then the fork-channels.
Or - alternatively - you could send out a TOPIC-SUBSCRIBE/UNSUBSCRIBE message and maintain the cache of fork channels on each device yourself. Note, however, that this would entail performing some kind of 'state transfer' of this cache to new members. Or have all members send presence messages to a newly joined member.

I'm not sure if I understood your requirements correctly...