Undo ref-counting (https://issues.redhat.com/browse/JGRP-2948)

Author: belaban

src/org/jgroups/JChannel.java

@@ -549,13 +549,7 @@ public Object down(Event evt) {
     }
 
     public Object down(Message msg) {
-        msg.incr();
-        try {
-            return prot_stack.down(msg);
-        }
-        finally {
-            msg.decr();
-        }
+        return msg != null? prot_stack.down(msg) : null;
     }
 
     /**

src/org/jgroups/Message.java

@@ -16,7 +16,7 @@
  * @author Bela Ban
  * @since  5.0
  */
-public interface Message extends SizeStreamable, Constructable<Message>, Refcountable<Message> {
+public interface Message extends SizeStreamable, Constructable<Message> {
 
     // The type of the message. Cannot be an enum, as users can register additional types
     short BYTES_MSG        = 0,

src/org/jgroups/Refcountable.java

@@ -208,12 +208,8 @@ protected void handleAck(Address sender, long ack) {
             return;
         }
         long old_min=rc[0], new_min=rc[1];
-        if(new_min > old_min) {
-
-            buf.forEach(buf.low()+1, buf.hd(), DECR, false);
-
+        if(new_min > old_min)
             buf.purge(new_min); // unblocks senders waiting for space to become available
-        }
     }
 
     @Override

src/org/jgroups/RefcountedBytesMessage.java

@@ -123,11 +123,6 @@ public abstract class ReliableMulticast extends Protocol implements DiagnosticsH
         return hdr == null || hdr.getType() != NakAckHeader.MSG? -1 : hdr.getSeqno();
     };
     protected final Predicate<Message> HAS_HEADER=m -> m != null && m.getHeader(id) != null;
-    protected static final Buffer.Visitor<Message> DECR=(seqno, msg) -> {
-        if(msg != null)
-            msg.decr();
-        return true;
-    };
 
 
     @ManagedAttribute(description="Number of retransmit requests received",type=SCALAR)
@@ -519,10 +514,8 @@ public Object down(Message msg) {
             if(dont_loopback_set && needToSendAck(send_entry, 1))
                 handleAck(local_addr, win.highestDelivered()); // https://issues.redhat.com/browse/JGRP-2829
         }
-        else {
-            msg.decr();
+        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
     }
@@ -714,7 +707,6 @@ protected boolean send(Message msg, Buffer<Message> win, boolean dont_loopback_s
             msg.putHeader(this.id, NakAckHeader.createMessageHeader(msg_id));
             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
-            msg.incr();
             down_prot.down(msg); // if this fails, since msg is in sent_msgs, it can be retransmitted
             return true;
         }
@@ -1234,7 +1226,6 @@ protected void stable(Digest digest) {
 
             // delete *delivered* msgs that are stable (all messages with seqnos <= seqno)
             if(hd >= 0 && win != null) {
-                win.forEach(win.low()+1, win.hd(), DECR, false);
                 log.trace("%s: deleting msgs <= %s from %s", local_addr, hd, member);
                 win.purge(hd);
             }

src/org/jgroups/RefcountedNioMessage.java

@@ -184,12 +184,6 @@ public abstract class ReliableUnicast extends Protocol implements AgeOutCache.Ha
 
     protected static final BiConsumer<MessageBatch,Message> BATCH_ACCUMULATOR=MessageBatch::add;
 
-    protected static final Buffer.Visitor<Message> DECR=(seqno, msg) -> {
-        if(msg != null)
-            msg.decr();
-        return true;
-    };
-
     protected abstract Buffer<Message> createBuffer(long initial_seqno);
     protected abstract boolean         needToSendAck(Entry e, int num_acks);
 
@@ -687,10 +681,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 {
-            msg.decr();
+        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()
     }
 
@@ -1036,7 +1028,6 @@ protected void handleAckReceived(Address sender, long seqno, short conn_id, int
 
         Buffer<Message> win=entry != null? entry.buf : null;
         if(win != null && entry.updateLastTimestamp(timestamp)) {
-            win.forEach(win.low()+1, seqno, DECR, false);
             win.purge(seqno, true); // removes all messages <= seqno (forced purge)
             num_acks_received.increment();
         }
@@ -1110,9 +1101,6 @@ protected boolean send(Message msg, SenderEntry entry, boolean dont_loopback_set
             msg.putHeader(this.id,UnicastHeader.createDataHeader(seqno, send_conn_id,seqno == DEFAULT_FIRST_SEQNO));
             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
-            boolean is_loopback=msg.dest() != null && msg.dest().equals(local_addr);
-            if(!dont_loopback_set && !is_loopback)
-                msg.incr();
             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);

src/org/jgroups/protocols/NAKACK4.java

@@ -196,12 +196,6 @@ public class UNICAST3 extends Protocol implements AgeOutCache.Handler<Address> {
 
     protected static final BiConsumer<MessageBatch,Message> BATCH_ACCUMULATOR=MessageBatch::add;
 
-    protected static final Table.Visitor<Message> DECR=(seqno, msg, row, col) -> {
-        if(msg != null)
-            msg.decr();
-        return true;
-    };
-
     /** Used for testing only! */
     public Table<Message> getSendWindow(Address target) {
         SenderEntry entry=send_table.get(target);
@@ -713,12 +707,10 @@ public Object down(Message msg) {
             return up_prot.up(msg);
         }
         SenderEntry entry=getSenderEntry(dst);
-        boolean is_loopback=dst.equals(local_addr), dont_loopback_set=msg.isFlagSet(DONT_LOOPBACK) && is_loopback;
+        boolean dont_loopback_set=msg.isFlagSet(DONT_LOOPBACK) && dst.equals(local_addr);
         short send_conn_id=entry.connId();
         long seqno=entry.sent_msgs_seqno.getAndIncrement();
         long sleep=10;
-        if(!dont_loopback_set && !is_loopback)
-            msg.incr();
         do {
             try {
                 msg.putHeader(this.id,UnicastHeader3.createDataHeader(seqno,send_conn_id,seqno == DEFAULT_FIRST_SEQNO));
@@ -1107,7 +1099,6 @@ protected void handleAckReceived(Address sender, long seqno, short conn_id, int
 
         Table<Message> win=entry != null? entry.msgs : null;
         if(win != null && entry.updateLastTimestamp(timestamp)) {
-            win.forEach(win.getLow()+1, seqno, DECR);
             win.purge(seqno, true); // removes all messages <= seqno (forced purge)
             num_acks_received.increment();
         }

src/org/jgroups/protocols/ReliableMulticast.java

@@ -162,12 +162,6 @@ public class NAKACK2 extends Protocol implements DiagnosticsHandler.ProbeHandler
         return hdr == null || hdr.getType() != NakAckHeader2.MSG? -1 : hdr.getSeqno();
     };
     protected final Predicate<Message> HAS_HEADER=m -> m != null && m.getHeader(id) != null;
-    protected static final Table.Visitor<Message> DECR=(seqno, msg, row, col) -> {
-        if(msg != null)
-            msg.decr();
-        return true;
-    };
-
 
     @ManagedAttribute(description="Number of retransmit requests received",type=AttributeType.SCALAR)
     protected final LongAdder xmit_reqs_received=new LongAdder();
@@ -798,7 +792,6 @@ protected void send(Message msg) {
         Table<Message> buf=local_xmit_table;
         if(buf == null && (buf=local_xmit_table=xmit_table.get(local_addr)) == null) // discard message if there is no entry for local_addr
             return;
-        msg.incr();
         if(msg.getSrc() == null)
             msg.setSrc(local_addr); // this needs to be done so we can check whether the message sender is the local_addr
 
@@ -1309,7 +1302,6 @@ protected void stable(Digest digest) {
 
             // delete *delivered* msgs that are stable (all messages with seqnos <= seqno)
             if(hd >= 0 && buf != null) {
-                buf.forEach(buf.getLow()+1, hd, DECR);
                 log.trace("%s: deleting msgs <= %s from %s", local_addr, hd, member);
                 buf.purge(hd);
             }