[jboss-jira] [JBoss JIRA] Commented: (JGRP-357) Unicast messages to unused port result in infinitely retransmission of message
Bela Ban (JIRA)
jira-events at jboss.com
Mon Nov 13 03:32:41 EST 2006
[ http://jira.jboss.com/jira/browse/JGRP-357?page=comments#action_12346758 ]
Bela Ban commented on JGRP-357:
-------------------------------
Suggested fix:
After some investigation (see previous thread about our problem) we have some
questions / remarks about the UNICAST protocol and retransmissions / message
sending to other (unknown) members:
- For some reason (we are still investigating why) our app starts sending (a)
message(s) to an unknown member (IP/port) on a certain channel. The UNICAST
protocol however does not check whether this is not a member in the current
view (current_members list) even when use_gms is true but, instead, check whether
the destination is in the 'previous_members' list and then ignores the message,
see the following code snippet from UNICAST.java:
public void down(Event evt) {
switch (evt.getType()) {
case Event.MSG: // Add UnicastHeader, add to AckSenderWindow and
pass down
Message msg=(Message) evt.getArg();
Object dst=msg.getDest();
/* only handle unicast messages */
if (dst == null || ((Address) dst).isMulticastAddress()) {
break;
}
> if(previous_members.contains(dst)) {
> if(trace)
> log.trace("discarding message to " + dst + " as this
member left the group," +
> " previous_members=" + previous_members);
> return;
}
Why doesn't UNICAST check if the destination is in the current VIEW (current_members
list) and then returns if it isn't?
Maybe a check should be added whether the destination dst still has a matching
connection and delete it.
- Sending the message to the unknown member obviously fails and then retransmission
starts. This will not stop because we will *never* receive a view change in
which the *unknown* member is not present anymore, so, the previous_members
list will never contain the *unknown* member and thus the retransmission will
never stop!
We noticed the following piece of code that would have prevented the retransmissions
but was commented out (also in UNCAST.java):
/** Called by AckSenderWindow to resend messages for which no ACK has been received
yet */
public void retransmit(long seqno, Message msg) {
Object dst=msg.getDest();
> > // bela Dec 23 2002:
> > // this will remove a member on a MERGE request, e.g. A and B merge:
when A sends the unicast
> > // request to B and there's a retransmit(), B will be removed !
> >
> > // if(use_gms && !members.contains(dst)
&& !prev_members.contains(dst)) {
> > //
> > // if(warn) log.warn("UNICAST.retransmit()", "seqno="
+ seqno + ": dest " + dst +
> > // " is not member any longer; removing
entry !");
> >
> > // synchronized(connections) {
> > // removeConnection(dst);
> > // }
> > // return;
> > // }
if(trace)
log.trace("[" + local_addr + "] --> XMIT(" + dst + ": #" + seqno
+ ')');
if(Global.copy)
passDown(new Event(Event.MSG, msg.copy()));
else
passDown(new Event(Event.MSG, msg));
num_xmit_requests_received++;
}
should this piece of code be updated and re-inserted? Of course, it should be
updated to allow MERGE request (and perhaps other messages) to be sent!
We would also like an option in UNICAST that will allow us to specify a max
number of retransmits for 1 message, just to be sure the retransmissions will
stop *sometime*. In our case, the network was flooded with retransmissions and
we wan't to prevent this *whatever happens*.
> Unicast messages to unused port result in infinitely retransmission of message
> ------------------------------------------------------------------------------
>
> Key: JGRP-357
> URL: http://jira.jboss.com/jira/browse/JGRP-357
> Project: JGroups
> Issue Type: Bug
> Affects Versions: 2.4
> Reporter: Vincent Hartsteen
> Assigned To: Bela Ban
> Fix For: 2.5
>
>
> In JGroups 2.4 (but probably this also happens in previous versions, did not verify this) we noticed that it is possible to send a message to a host:port that is not part of the group. This member might have been a member of the group but does not need to have been. This results in the retransmission of the message. However this retransmission never seems to end. The attached program demonstrates this.
> Use the properties file to assign a host-address and port the program will send messages to.
> Properties file:
> ===========
> flood.target.address : 10.220.242.33
> flood.target.port : 2715
> flood.topic : flood.the.network
> flood.stack : UDP(mcast_addr=228.1.2.123;mcast_port=19302;ip_ttl=32;loopback=false;\
> mcast_recv_buf_size=1000000;max_bundle_size=64000;max_bundle_timeout=100;use_incoming_packet_handler=false;\
> use_outgoing_packet_handler=true;ucast_send_buf_size=1000000;ip_ttl=32;enable_bundling=true/):\
> PING(timeout=6000;num_initial_members=8;down_thread=false):\
> MERGE2(max_interval=300000;min_interval=60000;down_thread=false):\
> FD(timeout=30000;max_tries=5;shun=true):\
> pbcast.NAKACK(gc_lag=300;retransmit_timeout=3000,30000,60000,180000;use_mcast_xmit=false;max_xmit_size=8192):\
> UNICAST(timeout=3000,30000,60000,180000):\
> pbcast.STABLE(desired_avg_gossip=300000;stability_delay=30000;max_bytes=65536;down_thread=false):\
> pbcast.GMS(join_timeout=10000;join_retry_timeout=5000;shun=true;print_local_addr=true;down_thread=false):\
> FC(max_credits=64000;min_credits=8000;down_thread=false):\
> FRAG2(frag_size=8000;down_thread=false;up_thread=false):\
> COMPRESS(down_thread=false;min_size=256;compression_level=9;up_thread=true):\
> pbcast.STATE_TRANSFER(down_thread=false;up_thread=false)
> Program:
> =======
> public class Flooder implements MembershipListener, RequestHandler, RspCollector {
> private static Log log = LogFactory.getLog(Flooder.class);
> private static int counter = 0;
> private Channel channel = null;
> private MessageDispatcher dispatcher = null;
> private Timer floodTimer = null;
> private InetAddress targetAddr;
> private int targetPort;
> private String topic;
> private String stack;
> public Flooder() throws IOException, ChannelException {
> Properties properties = new Properties();
> InputStream in = ClassLoader.getSystemClassLoader().getResourceAsStream("flooder.properties");
> properties.load(in);
> in.close();
>
> targetAddr = InetAddress.getByName(properties.getProperty("flood.target.address").trim());
> targetPort = Integer.parseInt(properties.getProperty("flood.target.port"));
> topic = properties.getProperty("flood.topic");
> stack = properties.getProperty("flood.stack");
> initialize();
> }
> public void initialize() throws ChannelException {
> channel = new JChannel(stack);
> channel.setOpt(Channel.BLOCK, Boolean.TRUE);
> channel.setOpt(Channel.GET_STATE_EVENTS, Boolean.TRUE);
> channel.setOpt(Channel.AUTO_RECONNECT, Boolean.TRUE);
> channel.setOpt(Channel.AUTO_GETSTATE, Boolean.TRUE);
> channel.connect(topic);
> dispatcher = new MessageDispatcher(channel, null, this, this);
> log.trace("Connected to topic " + topic + " w/ stack: " + ((JChannel)channel).getProtocolStack().printProtocolSpec(true));
> }
> public void startFlooding() {
> if (floodTimer != null) floodTimer.cancel();
> floodTimer = new Timer();
> floodTimer.schedule(new TimerTask() {
> @Override
> public void run() {
> Address address = new IpAddress(targetAddr, targetPort);
> Message msg = new Message(address, null, ("Flooooooooood#" + ++counter + "#").getBytes());
> Vector<Address> victor = new Vector<Address>(1);
> victor.add(address);
> dispatcher.castMessage(victor, System.currentTimeMillis(), msg, Flooder.this);
> log.trace("Sent Flooding message #" + counter + " to " + address + " @ " + new Date());
> }
> }, 1000, 10000);
> }
> public void block() {
> log.trace("BLOCK");
> }
> public void suspect(Address addr) {
> log.trace("SUSPECT: " + addr);
> }
> public void viewAccepted(View v) {
> log.trace("VIEW ACCEPTED: " + v);
> }
> public Object handle(Message msg) {
> log.trace("MESSAGE RECEIVED: " + msg);
> return "HANDLED";
> }
> public void receiveResponse(Message m) {
> log.trace("RECEIVE RESPONSE: " + m + " hdr=" + m.getHeaders());
> }
> public void viewChange(View v) {
> log.trace("VIEW CHANGE: " + v);
> }
>
> public static void main(String[] args) {
> try {
> Flooder flooder = new Flooder();
> flooder.startFlooding();
> } catch (Exception ex) {
> log.error("Exception occurred: ", ex);
> }
> }
> }
--
This message is automatically generated by JIRA.
-
If you think it was sent incorrectly contact one of the administrators: http://jira.jboss.com/jira/secure/Administrators.jspa
-
For more information on JIRA, see: http://www.atlassian.com/software/jira
More information about the jboss-jira
mailing list