[jboss-jira] [JBoss JIRA] (JGRP-1426) Group unable to accept new members: FLUSH stuck after MERGE_RSP timed out

[ https://issues.jboss.org/browse/JGRP-1426?page=com.atlassian.jira.plugin.... ] David Hotham edited comment on JGRP-1426 at 2/22/12 7:30 AM: ------------------------------------------------------------- Hi, I decided that I should write a standalone repro for this. I can't see how to attach files to this issue so I'm just going to paste the code below. To repro: - I'm using loopback addresses 10.239.0.1 through 10.239.0.4 on my PC, and these are hardcoded in. You'll want to tweak to match whatever network you use - run instances 10.239.0.1 through 10.239.0.3 directly; these will just sit around - run instance 10.239.0.4 in a wrapper (I'll include the python code that I've been using) which - sits waiting for the file AOK.txt to be written - when it is written, kills instance 10.239.0.4 and starts a new one. Sooner or later you should see that D gets stuck outside the group, with one of the other members repeatedly logging "GMS flush by coordinator failed". I usually see it happen after about 10 or 15 loops. Here's the java code: {noformat} import org.jgroups.*; import org.jgroups.protocols.*; import org.jgroups.protocols.pbcast.*; import org.jgroups.stack.*; import java.io.File; import java.io.IOException; import java.net.InetAddress; import java.util.ArrayList; import java.util.List; public class Test extends ReceiverAdapter { String localAddress; Boolean sendMessage = false; public Test(String address) { localAddress = address; } // Parameter is an IP address, expected to be one of 10.239.0.1 - 4. public static void main(String[] args) throws Exception { Test test = new Test(args[0]); test.start(); } public void start() throws Exception { // Set up the stack, and channel. InetAddress inetAddr = InetAddress.getByName(localAddress); IpAddress member_a = new IpAddress("10.239.0.1", 60000); IpAddress member_b = new IpAddress("10.239.0.2", 60000); IpAddress member_c = new IpAddress("10.239.0.3", 60000); IpAddress member_d = new IpAddress("10.239.0.4", 60000); List<IpAddress> initialHosts = new ArrayList<IpAddress>(); initialHosts.add(member_a); initialHosts.add(member_b); initialHosts.add(member_c); initialHosts.add(member_d); JChannel ch=new JChannel(false); ProtocolStack stack=new ProtocolStack(); ch.setProtocolStack(stack); ch.setReceiver(this); stack.addProtocol((new TCP()) .setValue("bind_port", 60000) .setValue("bind_addr", inetAddr)); TCPPING tcpping = new TCPPING(); tcpping.setInitialHosts(initialHosts); stack.addProtocol(tcpping) .addProtocol(new MERGE3()) .addProtocol(new FD()) .addProtocol(new VERIFY_SUSPECT()) .addProtocol((new NAKACK2()) .setValue("use_mcast_xmit", false)) .addProtocol(new UNICAST2()) .addProtocol(new STABLE()) .addProtocol(new GMS()) .addProtocol(new UFC()) .addProtocol(new MFC()) .addProtocol(new FRAG2()) .addProtocol(new SEQUENCER()) .addProtocol(new FLUSH()); for (Protocol protocol: stack.getProtocols()) { protocol.setProtocolStack(stack); } stack.init(); // Connect to the cluster. ch.setName(localAddress); ch.connect("TestCluster"); // Loop round, waiting to be told to send a message. // Not thread safe - we might not send as many messages as we 'should' - but good enough. Boolean done = false; while (!done) { if (sendMessage) { sendMessage = false; ch.send(null, "This is a broadcast message from " + localAddress); } Thread.sleep(100); } } public void viewAccepted(View new_view) { System.out.println("view: " + new_view); // When D sees that the view has reached four members it creates the file AOK.txt, at which point we expect to // kill it. if (localAddress.equals("10.239.0.4") && (new_view.size() == 4)) { File f = new File("/tmp/AOK.txt"); try { f.createNewFile(); } catch (IOException e) { System.out.println("couldn't create AOK.txt"); } } // Changes of view trigger message sending. sendMessage = true; } public void receive(Message msg) { System.out.println("<< " + msg.getObject() + " [" + msg.getSrc() + "]"); } } {noformat} And the python wrapper: {noformat} import os import random import subprocess import time class TestWrapper: def __init__(self): self.subProc = None def startMemberD(self): try: self.subProc = subprocess.Popen("java -jar target/Test.jar 10.239.0.4") except OSError, e: print e def killMemberD(self): try: self.subProc.kill() except: os.kill(self.subProc.pid) self.subProc = None def start(self): counter = 0 while counter < 1000: counter += 1 print "Starting member D on loop %d" % counter self.startMemberD() # Wait for D to signal that all is well while not os.path.exists("c:\\downloads\\AOK.txt"): time.sleep(1) # Kill D, wait a bit. os.remove("/tmp/AOK.txt") time.sleep(2) self.killMemberD() delay = random.randrange(0, 10) print "Sleeping for %d seconds" % delay time.sleep(delay) wrapper = TestWrapper() wrapper.start() {noformat} was (Author: dimbleby): Hi, I decided that I should write a standalone repro for this. I can't see how to attach files to this issue so I'm just going to paste the code below. To repro: - I'm using loopback addresses 10.239.0.1 through 10.239.0.4 on my PC, and these are hardcoded in. You'll want to tweak to match whatever network you use - run instances 10.239.0.1 through 10.239.0.3 directly; these will just sit around - run instance 10.239.0.4 in a wrapper (I'll include the python code that I've been using) which - sits waiting for the file AOK.txt to be written - when it is written, kills instance 10.239.0.4 and starts a new one. Sooner or later you should see that D gets stuck outside the group, with one of the other members repeatedly logging "GMS flush by coordinator failed". I usually see it happen after about 10 or 15 loops. Here's the java code: {noformat} import org.jgroups.*; import org.jgroups.protocols.*; import org.jgroups.protocols.pbcast.*; import org.jgroups.stack.*; import java.io.File; import java.io.IOException; import java.net.InetAddress; import java.util.ArrayList; import java.util.List; public class Test extends ReceiverAdapter { String localAddress; Boolean sendMessage = false; public Test(String address) { localAddress = address; } // Parameter is an IP address, expected to be one of 10.239.0.1 - 4. public static void main(String[] args) throws Exception { Test test = new Test(args[0]); test.start(); } public void start() throws Exception { // Set up the stack, and channel. InetAddress inetAddr = InetAddress.getByName(localAddress); IpAddress member_a = new IpAddress("10.239.0.1", 60000); IpAddress member_b = new IpAddress("10.239.0.2", 60000); IpAddress member_c = new IpAddress("10.239.0.3", 60000); IpAddress member_d = new IpAddress("10.239.0.4", 60000); List<IpAddress> initialHosts = new ArrayList<IpAddress>(); initialHosts.add(member_a); initialHosts.add(member_b); initialHosts.add(member_c); initialHosts.add(member_d); JChannel ch=new JChannel(false); ProtocolStack stack=new ProtocolStack(); ch.setProtocolStack(stack); ch.setReceiver(this); stack.addProtocol((new TCP()) .setValue("bind_port", 60000) .setValue("bind_addr", inetAddr)); TCPPING tcpping = new TCPPING(); tcpping.setInitialHosts(initialHosts); stack.addProtocol(tcpping) .addProtocol(new MERGE3()) .addProtocol(new FD()) .addProtocol(new VERIFY_SUSPECT()) .addProtocol((new NAKACK2()) .setValue("use_mcast_xmit", false)) .addProtocol(new UNICAST2()) .addProtocol(new STABLE()) .addProtocol(new GMS()) .addProtocol(new UFC()) .addProtocol(new MFC()) .addProtocol(new FRAG2()) .addProtocol(new SEQUENCER()) .addProtocol(new FLUSH()); for (Protocol protocol: stack.getProtocols()) { protocol.setProtocolStack(stack); } stack.init(); // Connect to the cluster. ch.setName(localAddress); ch.connect("TestCluster"); // Loop round, waiting to be told to send a message. // Not thread safe - we might not send as many messages as we 'should' - but good enough. Boolean done = false; while (!done) { if (sendMessage) { sendMessage = false; ch.send(null, "This is a broadcast message from " + localAddress); } Thread.sleep(100); } } public void viewAccepted(View new_view) { System.out.println("view: " + new_view); // When D sees that the view has reached four members it creates the file AOK.txt, at which point we expect to // kill it. if (localAddress.equals("10.239.0.4") && (new_view.size() == 4)) { File f = new File("/tmp/AOK.txt"); try { f.createNewFile(); } catch (IOException e) { System.out.println("couldn't create AOK.txt"); } } // Changes of view trigger message sending. sendMessage = true; } public void receive(Message msg) { System.out.println("<< " + msg.getObject() + " [" + msg.getSrc() + "]"); } } And the python wrapper: import os import random import subprocess import time class TestWrapper: def __init__(self): self.subProc = None def startMemberD(self): try: self.subProc = subprocess.Popen("java -jar target/Test.jar 10.239.0.4") except OSError, e: print e def killMemberD(self): try: self.subProc.kill() except: os.kill(self.subProc.pid) self.subProc = None def start(self): counter = 0 while counter < 1000: counter += 1 print "Starting member D on loop %d" % counter self.startMemberD() # Wait for D to signal that all is well while not os.path.exists("c:\\downloads\\AOK.txt"): time.sleep(1) # Kill D, wait a bit. os.remove("/tmp/AOK.txt") time.sleep(2) self.killMemberD() delay = random.randrange(0, 10) print "Sleeping for %d seconds" % delay time.sleep(delay) wrapper = TestWrapper() wrapper.start() {noformat} -- This message is automatically generated by JIRA. If you think it was sent incorrectly, please contact your JIRA administrators: https://issues.jboss.org/secure/ContactAdministrators!default.jspa For more information on JIRA, see: http://www.atlassian.com/software/jira