Merge branch 'cfredri4-revert-876'

Author: belaban

src/org/jgroups/protocols/ReliableMulticast.java

@@ -105,9 +105,6 @@ public abstract class ReliableMulticast extends Protocol implements DiagnosticsH
     @ManagedAttribute(description="Number of messages received",type=SCALAR)
     protected final LongAdder num_messages_received=new LongAdder();
 
-    @ManagedAttribute(description="Number of sends dropped",type=SCALAR)
-    protected final LongAdder num_sends_dropped=new LongAdder();
-
     protected static final Message DUMMY_OOB_MSG=new EmptyMessage().setFlag(OOB);
 
     // Accepts messages which are (1) non-null, (2) no DUMMY_OOB_MSGs and (3) not OOB_DELIVERED
@@ -242,7 +239,6 @@ public ReliableMulticast clearCachedBatches() {
     public ReliableMulticast sendsCanBlock(boolean s)                 {this.sends_can_block=s; return this;}
     public long              getNumMessagesSent()                     {return num_messages_sent.sum();}
     public long              getNumMessagesReceived()                 {return num_messages_received.sum();}
-    public long              getNumSendsDropped()                     {return num_sends_dropped.sum();}
     public boolean           reuseMessageBatches()                    {return reuse_message_batches;}
     public ReliableMulticast reuseMessageBatches(boolean b)           {this.reuse_message_batches=b; return this;}
     public boolean           sendAtomically()                         {return send_atomically;}
@@ -356,7 +352,6 @@ protected Entry sendEntry() {
     public void resetStats() {
         num_messages_sent.reset();
         num_messages_received.reset();
-        num_sends_dropped.reset();
         xmit_reqs_received.reset();
         xmit_reqs_sent.reset();
         xmit_rsps_received.reset();
@@ -514,17 +509,14 @@ public Object down(Message msg) {
             msg.setSrc(local_addr); // this needs to be done so we can check whether the message sender is the local_addr
         boolean dont_loopback_set=msg.isFlagSet(DONT_LOOPBACK);
         Buffer<Message> win=send_entry.buf();
-        boolean sent=send(msg, win, dont_loopback_set);
-        last_seqno_resender.skipNext();
-        if(sent) {
+        if(send(msg, win, dont_loopback_set)) {
             num_messages_sent.increment();
             if(dont_loopback_set && needToSendAck(send_entry, 1))
                 handleAck(local_addr, win.highestDelivered()); // https://issues.redhat.com/browse/JGRP-2829
         }
-        else {
-            num_sends_dropped.increment();
-            log.warn("%s: discarded message due to full send buffer, message: %s", local_addr, msg);
-        }
+        else
+            log.trace("%s: dropped message due to closed send buffer, message: %s", local_addr, msg);
+        last_seqno_resender.skipNext();
         return null;    // don't pass down the stack
     }
 
@@ -710,10 +702,10 @@ protected boolean send(Message msg, Buffer<Message> win, boolean dont_loopback_s
             if(is_trace)
                 log.trace("%s --> [all]: #%d", local_addr, msg_id);
             msg.putHeader(this.id, NakAckHeader.createMessageHeader(msg_id));
-            boolean added=addToSendBuffer(win, msg_id, msg, dont_loopback_set? remove_filter : null);
-            if(added)
-                down_prot.down(msg); // if this fails, since msg is in sent_msgs, it can be retransmitted
-            return added;
+            if(!addToSendBuffer(win, msg_id, msg, dont_loopback_set? remove_filter : null))
+                return false; // e.g. message already present in send buffer, or buffer is closed
+            down_prot.down(msg); // if this fails, since msg is in sent_msgs, it can be retransmitted
+            return true;
         }
         finally {
             if(lock != null)

src/org/jgroups/protocols/ReliableUnicast.java

@@ -101,8 +101,6 @@ public abstract class ReliableUnicast extends Protocol implements AgeOutCache.Ha
     protected final LongAdder num_msgs_sent=new LongAdder();
     @ManagedAttribute(description="Number of message received",type=SCALAR)
     protected final LongAdder num_msgs_received=new LongAdder();
-    @ManagedAttribute(description="Number of sends dropped",type=SCALAR)
-    protected final LongAdder num_sends_dropped=new LongAdder();
     @ManagedAttribute(description="Number of acks sent",type=SCALAR)
     protected final LongAdder num_acks_sent=new LongAdder();
     @ManagedAttribute(description="Number of acks received",type=SCALAR)
@@ -287,7 +285,6 @@ public ReliableUnicast trimCachedBatches() {
     public long getNumMessagesReceived() {return num_msgs_received.sum();}
 
 
-    public long getNumSendsDropped()     {return num_sends_dropped.sum();}
     public long getNumAcksSent()         {return num_acks_sent.sum();}
     public long getNumAcksReceived()     {return num_acks_received.sum();}
     public long getNumXmits()            {return num_xmits.sum();}
@@ -376,7 +373,7 @@ public String printSendWindowMessages() {
 
     public void resetStats() {
         avg_delivery_batch_size.clear();
-        Stream.of(num_msgs_sent, num_msgs_received, num_sends_dropped, num_acks_sent, num_acks_received, num_xmits,
+        Stream.of(num_msgs_sent, num_msgs_received, num_acks_sent, num_acks_received, num_xmits,
                 xmit_reqs_received, xmit_reqs_sent, xmit_rsps_sent, num_loopbacks).forEach(LongAdder::reset);
         send_table.values().stream().map(e -> e.buf).forEach(Buffer::resetStats);
         recv_table.values().stream().map(e -> e.buf).forEach(Buffer::resetStats);
@@ -683,10 +680,8 @@ public Object down(Message msg) {
         boolean dont_loopback_set=msg.isFlagSet(DONT_LOOPBACK) && dst.equals(local_addr);
         if(send(msg, entry, dont_loopback_set))
             num_msgs_sent.increment();
-        else {
-            num_sends_dropped.increment();
-            log.warn("%s: discarded message due to full send buffer, message: %s", local_addr, msg);
-        }
+        else
+            log.trace("%s: dropped message due to closed send buffer, message: %s", local_addr, msg);
         return null; // the message was already sent down the stack in send()
     }
 
@@ -1092,9 +1087,8 @@ protected boolean send(Message msg, SenderEntry entry, boolean dont_loopback_set
         try {
             seqno=entry.seqno.getAndIncrement();
             msg.putHeader(this.id,UnicastHeader.createDataHeader(seqno, send_conn_id,seqno == DEFAULT_FIRST_SEQNO));
-            boolean added=addToSendBuffer(buf, seqno, msg, dont_loopback_set? remove_filter : null);
-            if(!added) // e.g. message already present in send buffer
-                return false;
+            if(!addToSendBuffer(buf, seqno, msg, dont_loopback_set? remove_filter : null))
+                return false; // e.g. message already present in send buffer, or buffer is closed
             down_prot.down(msg); // if this fails, since msg is in sent_msgs, it can be retransmitted
             if(entry.state() == State.CLOSING)
                 entry.state(State.OPEN);