Concurrency Experiments.

common/src/main/java/org/red5/server/net/rtmp/event/Aggregate.java

@@ -36,11 +36,6 @@ public class Aggregate extends BaseEvent implements IoConstants, IStreamData<Agg
 
     private static Logger log = LoggerFactory.getLogger(Aggregate.class);
 
-    /**
-     * Data
-     */
-    protected IoBuffer data;
-
     /**
      * Data type
      */

common/src/main/java/org/red5/server/net/rtmp/event/AudioData.java

@@ -31,8 +31,6 @@ public class AudioData extends BaseEvent implements IStreamData<AudioData>, IStr
 
     private static final long serialVersionUID = -4102940670913999407L;
 
-    protected IoBuffer data;
-
     /**
      * Data type
      */
@@ -253,6 +251,19 @@ public void writeExternal(ObjectOutput out) throws IOException {
         }
     }
 
+    @Override
+    public BaseEvent forkedDuplicate() {
+
+        AudioData fork = new AudioData(super.concurrentDataCopy());
+        fork.setTimestamp(this.timestamp);
+        if (header != null) {
+            fork.setHeader(header.clone());
+        }
+        fork.setSource(this.getSource());
+        fork.setSourceType(this.getSourceType());
+        return fork;
+    }
+
     /**
      * Duplicate this message / event.
      *

common/src/main/java/org/red5/server/net/rtmp/event/BaseEvent.java

@@ -13,6 +13,7 @@
 import java.io.ObjectOutput;
 import java.util.concurrent.atomic.AtomicInteger;
 
+import org.apache.mina.core.buffer.IoBuffer;
 import org.red5.server.api.event.IEventListener;
 import org.red5.server.net.rtmp.message.Constants;
 import org.red5.server.net.rtmp.message.Header;
@@ -33,11 +34,20 @@ public abstract class BaseEvent implements Constants, IRTMPEvent, Externalizable
     // (2) make it aspect oriented
     private static final boolean allocationDebugging = false;
 
+    protected AtomicInteger forkRefs = new AtomicInteger();
+
     /**
      * Event type
      */
     private Type type;
 
+    /**
+     * Multi-threaded copy-able array.
+     */
+    protected ForkableData forkableData;
+
+    protected IoBuffer data;
+
     /**
      * Source type
      */
@@ -86,6 +96,41 @@ public BaseEvent(Type type) {
         this(type, null);
     }
 
+    /**
+     * Creates a raw byte array which is used to allow concurrent non-blocking copying of this event.
+     * Owning thread should call addForkReference for each thread making a copy and copying threads should call removeForkReference.
+     * @return
+     */
+    public boolean prepareForkedDuplication() {
+        if (data != null && data.rewind().hasRemaining()) {
+            byte[] bytes = new byte[data.remaining()];
+            data.get(bytes);
+            data.rewind();
+            forkableData = new ForkableData(bytes);
+            return true;
+        }
+        return false;
+    }
+
+    protected IoBuffer concurrentDataCopy() {
+        return IoBuffer.wrap(forkableData.rawData).asReadOnlyBuffer();
+    }
+
+    public void addForkReference() {
+        forkRefs.incrementAndGet();
+    }
+
+    public void removeForkReference() {
+        int ref = forkRefs.decrementAndGet();
+        if (ref <= 0) {
+            forkableData = null;
+        }
+    }
+
+    public BaseEvent forkedDuplicate() {
+        return null;
+    };
+
     /**
      * Create new event of given type
      *
@@ -290,4 +335,15 @@ public void writeExternal(ObjectOutput out) throws IOException {
         out.writeInt(timestamp);
     }
 
+    /**
+     * Provides final field of bytes for mounting non-blocking concurrent read access.
+     */
+    private static class ForkableData {
+        final byte[] rawData;
+
+        private ForkableData(byte[] bytes) {
+            rawData = bytes;
+        }
+    }
+
 }

common/src/main/java/org/red5/server/net/rtmp/event/Notify.java

@@ -36,11 +36,6 @@ public class Notify extends BaseEvent implements ICommand, IStreamData<Notify>,
      */
     protected IServiceCall call;
 
-    /**
-     * Event data
-     */
-    protected IoBuffer data;
-
     /**
      * Event data type
      */
@@ -273,6 +268,21 @@ public void readExternal(ObjectInput in) throws IOException, ClassNotFoundExcept
         }
     }
 
+    @Override
+    public BaseEvent forkedDuplicate() {
+
+        Notify fork = new Notify(super.concurrentDataCopy());
+        fork.setTimestamp(this.timestamp);
+        fork.setAction(action);
+        fork.setCall(call);
+        if (header != null) {
+            fork.setHeader(header.clone());
+        }
+        fork.setSource(this.getSource());
+        fork.setSourceType(this.getSourceType());
+        return fork;
+    }
+
     /** {@inheritDoc} */
     @Override
     public void writeExternal(ObjectOutput out) throws IOException {

common/src/main/java/org/red5/server/net/rtmp/event/Unknown.java

@@ -22,11 +22,6 @@
 public class Unknown extends BaseEvent {
     private static final long serialVersionUID = -1352770037962252975L;
 
-    /**
-     * Event data
-     */
-    protected IoBuffer data;
-
     /**
      * Type of data
      */

common/src/main/java/org/red5/server/net/rtmp/event/VideoData.java

@@ -33,11 +33,6 @@ public class VideoData extends BaseEvent implements IoConstants, IStreamData<Vid
 
     private static final long serialVersionUID = 5538859593815804830L;
 
-    /**
-     * Video data
-     */
-    protected IoBuffer data;
-
     /**
      * Data type
      */
@@ -151,6 +146,8 @@ public void setData(IoBuffer data) {
                 packetType = VideoPacketType.valueOf(data.get());
             }
             data.reset();
+
+            config = (packetType == VideoPacketType.SequenceStart);
         }
 
         this.data = data;
@@ -303,6 +300,19 @@ public void writeExternal(ObjectOutput out) throws IOException {
         }
     }
 
+    @Override
+    public BaseEvent forkedDuplicate() {
+
+        VideoData fork = new VideoData(super.concurrentDataCopy());
+        fork.setTimestamp(this.timestamp);
+        if (header != null) {
+            fork.setHeader(header.clone());
+        }
+        fork.setSource(this.getSource());
+        fork.setSourceType(this.getSourceType());
+        return fork;
+    }
+
     /**
      * Duplicate this message / event.
      *
@@ -342,5 +352,4 @@ public VideoData duplicate() throws IOException, ClassNotFoundException {
     public String toString() {
         return String.format("Video - ts: %s length: %s", getTimestamp(), (data != null ? data.limit() : '0'));
     }
-
 }

pom.xml

@@ -24,7 +24,7 @@
     <name>Red5</name>
     <description>The Red5 server</description>
     <groupId>org.red5</groupId>
-    <version>2.0.18</version>
+    <version>2.0.19</version>
     <url>https://github.com/Red5/red5-server</url>
     <inceptionYear>2005</inceptionYear>
     <organization>