Composite message additions

This change borrows the optimization from BatchMessage to not serialize dest and src for each message This also fixes incorrect size being reported for BatchMessage and in various places where Message.writeToNoAddrs is used.
belaban · Nov 15, 2024 · 8e62e7b · 8e62e7b
1 parent feb2c51
commit 8e62e7b
Show file tree

Hide file tree

Showing 14 changed files with 43 additions and 34 deletions.
diff --git a/src/org/jgroups/BaseMessage.java b/src/org/jgroups/BaseMessage.java
@@ -238,6 +238,19 @@ public int size() {
 
         retval+=Global.SHORT_SIZE;  // number of headers
         retval+=Headers.marshalledSize(this.headers);
+        retval+=payloadSize();
+        return retval;
+    }
+
+    public int sizeNoAddrs(Address src) {
+        int retval=Global.BYTE_SIZE // leading byte
+          + Global.SHORT_SIZE;      // flags
+        if(sender != null && !sender.equals(src))
+            retval+=Util.size(sender);
+
+        retval+=Global.SHORT_SIZE;  // number of headers
+        retval+=Headers.marshalledSize(this.headers);
+        retval+=payloadSize();
         return retval;
     }
 
@@ -274,7 +287,7 @@ public void writeTo(DataOutput out) throws IOException {
     public void writeToNoAddrs(Address src, DataOutput out) throws IOException {
         byte leading=0;
 
-        boolean write_src_addr=src == null || sender != null && !sender.equals(src);
+        boolean write_src_addr=sender != null && !sender.equals(src);
 
         if(write_src_addr)
             leading=Util.setFlag(leading, SRC_SET);
@@ -341,5 +354,5 @@ protected static Header[] createHeaders(int size) {
         return size > 0? new Header[size] : new Header[Util.DEFAULT_HEADERS];
     }
 
-
+    protected abstract int payloadSize();
 }
diff --git a/src/org/jgroups/BatchMessage.java b/src/org/jgroups/BatchMessage.java
@@ -118,12 +118,12 @@ public String toString() {
         return String.format("%s, %d message(s)", super.toString(), getNumberOfMessages());
     }
 
-    public int size() {
-        int retval=super.size() + Global.INT_SIZE;
+    @Override protected int payloadSize() {
+        int retval=Global.INT_SIZE; // count
         retval+=Util.size(orig_src);
         if(msgs != null) {
             for(int i=0; i < index; i++)
-                retval+=msgs[i].size() + Global.SHORT_SIZE; // type
+                retval+=msgs[i].sizeNoAddrs(getSrc()) + Global.SHORT_SIZE; // type
         }
         return retval;
     }
@@ -140,7 +140,7 @@ public void writePayload(DataOutput out) throws IOException {
             for(int i=0; i < index; i++) {
                 Message msg=msgs[i];
                 out.writeShort(msg.getType());
-                msg.writeToNoAddrs(this.src(), out);
+                msg.writeToNoAddrs(getSrc(), out);
             }
         }
     }

diff --git a/src/org/jgroups/BytesMessage.java b/src/org/jgroups/BytesMessage.java
@@ -227,11 +227,6 @@ public <T extends Object> T getObject(ClassLoader loader) {
     }
 
 
-    public int size() {
-        return super.size() +sizeOfPayload();
-    }
-
-
     /**
      * Copies the byte array. If offset and length are used (to refer to another array), the copy will contain only
      * the subset that offset and length point to, copying the subset into the new copy.<p/>
@@ -245,7 +240,7 @@ public int size() {
         return copy;
     }
 
-    protected int sizeOfPayload() {
+    @Override protected int payloadSize() {
         int retval=Global.INT_SIZE; // length
         if(array != null)
             retval+=length;         // number of bytes in the array

diff --git a/src/org/jgroups/CompositeMessage.java b/src/org/jgroups/CompositeMessage.java
@@ -3,7 +3,6 @@
 
 
 import org.jgroups.util.ByteArray;
-
 import java.io.DataInput;
 import java.io.DataOutput;
 import java.io.IOException;
@@ -118,11 +117,11 @@ public String toString() {
         return String.format("%s, %d message(s)", super.toString(), getNumberOfMessages());
     }
 
-    public int size() {
-        int retval=super.size() + Global.INT_SIZE; // length
+    @Override protected int payloadSize() {
+        int retval=Global.INT_SIZE; // count
         if(msgs != null) {
             for(int i=0; i < index; i++)
-                retval+=msgs[i].size() + Global.SHORT_SIZE; // type
+                retval+=msgs[i].sizeNoAddrs(getSrc()) + Global.SHORT_SIZE; // type
         }
         return retval;
     }
@@ -142,8 +141,7 @@ public void writePayload(DataOutput out) throws IOException {
             for(int i=0; i < index; i++) {
                 Message msg=msgs[i];
                 out.writeShort(msg.getType());
-                // msg.writeToNoAddrs(src(), out);
-                msg.writeTo(out);
+                msg.writeToNoAddrs(getSrc(), out);
             }
         }
     }
@@ -156,7 +154,7 @@ public void readPayload(DataInput in) throws IOException, ClassNotFoundException
                 short type=in.readShort();
                 Message msg=MessageFactory.create(type).setDest(getDest());
                 if(msg.getSrc() == null)
-                    msg.setSrc(src());
+                    msg.setSrc(getSrc());
                 msg.readFrom(in);
                 msgs[i]=msg;
             }

diff --git a/src/org/jgroups/EmptyMessage.java b/src/org/jgroups/EmptyMessage.java
@@ -40,4 +40,6 @@ public void                  writePayload(DataOutput out) throws IOException {
     public void                  readPayload(DataInput in) throws IOException, ClassNotFoundException {
         // no payload to read
     }
+
+    protected int                payloadSize() { return 0; }
 }
diff --git a/src/org/jgroups/FragmentedMessage.java b/src/org/jgroups/FragmentedMessage.java
@@ -35,7 +35,7 @@ public FragmentedMessage(Message original_msg, int off, int len) {
     public boolean           hasArray()           {return false;}
     public boolean           hasPayload()         {return true;}
     public Supplier<Message> create()             {return FragmentedMessage::new;}
-    protected int            sizeOfPayload()      {return Global.INT_SIZE + length;}
+    protected int            payloadSize()        {return Global.INT_SIZE + length;}
 
     @Override
     protected Message copyPayload(Message copy) {

diff --git a/src/org/jgroups/LongMessage.java b/src/org/jgroups/LongMessage.java
@@ -60,8 +60,8 @@ public void readPayload(DataInput in) throws IOException, ClassNotFoundException
         value=Bits.readLongCompressed(in);
     }
 
-    public int size() {
-        return super.size() + Bits.size(value);
+    @Override protected int payloadSize() {
+        return Bits.size(value);
     }
 
     public String toString() {

diff --git a/src/org/jgroups/Message.java b/src/org/jgroups/Message.java
@@ -204,6 +204,9 @@ default Message      setFlag(short flag, boolean transient_flags) {
      */
     int                  size();
 
+    /** Returns the exact size of the marshalled message without destination (and possibly source) address */
+    int                  sizeNoAddrs(Address src);
+
     /** Writes the message to an output stream excluding the destination (and possibly source) address */
     void                 writeToNoAddrs(Address src, DataOutput out) throws IOException;
 

diff --git a/src/org/jgroups/NioMessage.java b/src/org/jgroups/NioMessage.java
@@ -179,8 +179,6 @@ public NioMessage copy(boolean copy_payload, boolean copy_headers) {
 
     /* ----------------------------------- Interface Streamable  ------------------------------- */
 
-    public int size() {return super.size() +sizeOfPayload();}
-
     public String toString() {
         return String.format("%s %s", super.toString(), use_direct_memory_for_allocations? "(direct)" : "");
     }
@@ -191,7 +189,7 @@ public String toString() {
         return copy;
     }
 
-    protected int sizeOfPayload() {
+    @Override protected int payloadSize() {
         return Global.INT_SIZE + getLength() + Global.BYTE_SIZE; // for use_direct_memory_for_allocations
     }
 

diff --git a/src/org/jgroups/ObjectMessage.java b/src/org/jgroups/ObjectMessage.java
@@ -67,7 +67,7 @@ public ObjectMessage(Address dest, SizeStreamable obj) {
     public boolean           hasPayload()                         {return obj != null;}
     public boolean           hasArray()                           {return false;}
     public int               getOffset()                          {return 0;}
-    public int               getLength()                          {return obj != null? objSize() : 0;}
+    public int               getLength()                          {return obj != null? payloadSize() : 0;}
     public byte[]            getArray()                           {throw new UnsupportedOperationException();}
     public ObjectMessage     setArray(byte[] b, int off, int len) {throw new UnsupportedOperationException();}
     public ObjectMessage     setArray(ByteArray buf)              {throw new UnsupportedOperationException();}
@@ -132,10 +132,6 @@ public <T extends Object> T getObject() {
         return isWrapped() || obj instanceof ObjectWrapperPrimitive? ((ObjectWrapperPrimitive)obj).getObject() : (T)obj;
     }
 
-    public int size() {
-        return super.size() + objSize();
-    }
-
     public void writePayload(DataOutput out) throws IOException {
         Util.writeGenericStreamable(obj, out);
     }
@@ -157,7 +153,7 @@ public String toString() {
         return super.toString() + String.format(", obj: %s", obj);
     }
 
-    protected int objSize() {
+    @Override protected int payloadSize() {
         return Util.size(obj);
     }
 }
diff --git a/src/org/jgroups/protocols/RingBufferBundler.java b/src/org/jgroups/protocols/RingBufferBundler.java
@@ -204,7 +204,7 @@ protected int marshalMessagesToSameDestination(Address dest, Message[] buf, int
             if(msg != null && Objects.equals(dest, msg.getDest())) {
                 if(list != null)
                     list.add(msg);
-                int size=msg.size() + Global.SHORT_SIZE;
+                int size=msg.sizeNoAddrs(msg.getSrc()) + Global.SHORT_SIZE;
                 if(bytes + size > max_bundle_size)
                     break;
                 bytes+=size;

diff --git a/src/org/jgroups/protocols/RingBufferBundlerLockless.java b/src/org/jgroups/protocols/RingBufferBundlerLockless.java
@@ -222,7 +222,7 @@ protected int marshalMessagesToSameDestination(Address dest, Message[] buf,
         while(available_msgs > 0) {
             Message msg=buf[start_index];
             if(msg != null && Objects.equals(dest, msg.getDest())) {
-                int msg_size=msg.size() + Global.SHORT_SIZE;;
+                int msg_size=msg.sizeNoAddrs(msg.getSrc()) + Global.SHORT_SIZE;;
                 if(bytes + msg_size > max_bundle_size)
                     break;
                 bytes+=msg_size;

diff --git a/src/org/jgroups/protocols/RingBufferBundlerLockless2.java b/src/org/jgroups/protocols/RingBufferBundlerLockless2.java
@@ -207,7 +207,7 @@ protected int marshalMessagesToSameDestination(Address dest, Message[] buf, fina
         for(int i=start_index; i != end_index; i=increment(i)) {
             Message msg=buf[i];
             if(msg != null && msg != NULL_MSG && Objects.equals(dest, msg.getDest())) {
-                int msg_size=msg.size() + Global.SHORT_SIZE;
+                int msg_size=msg.sizeNoAddrs(msg.getSrc()) + Global.SHORT_SIZE;
                 if(bytes + msg_size > max_bundle_size)
                     break;
                 bytes+=msg_size;

diff --git a/tests/junit-functional/org/jgroups/tests/MessageFactoryTest.java b/tests/junit-functional/org/jgroups/tests/MessageFactoryTest.java
@@ -87,5 +87,9 @@ public void writePayload(DataOutput out) throws IOException {
 
         public void readPayload(DataInput in) throws IOException, ClassNotFoundException {
         }
+
+        protected int payloadSize() {
+            return 0;
+        }
     }
 }