From 51b4712932003190185fce04baad86c0b9dd29e8 Mon Sep 17 00:00:00 2001
From: fhanik <fhanik@13f79535-47bb-0310-9956-ffa450edef68>
Date: Thu, 17 May 2007 13:45:15 +0000
Subject: [PATCH] Add locks instead of synchronized statements to avoid issues
 between receiving and sending

git-svn-id: https://svn.apache.org/repos/asf/tomcat/tc6.0.x/trunk@538920 13f79535-47bb-0310-9956-ffa450edef68
---
 .../group/interceptors/OrderInterceptor.java       | 94 +++++++++++++++-------
 .../test/interceptors/TestOrderInterceptor.java    | 40 ++++++++-
 2 files changed, 103 insertions(+), 31 deletions(-)

diff --git a/java/org/apache/catalina/tribes/group/interceptors/OrderInterceptor.java b/java/org/apache/catalina/tribes/group/interceptors/OrderInterceptor.java
index 34d88523d..9bd1a7583 100644
--- a/java/org/apache/catalina/tribes/group/interceptors/OrderInterceptor.java
+++ b/java/org/apache/catalina/tribes/group/interceptors/OrderInterceptor.java
@@ -25,6 +25,7 @@ import org.apache.catalina.tribes.group.ChannelInterceptorBase;
 import org.apache.catalina.tribes.group.InterceptorPayload;
 import org.apache.catalina.tribes.io.XByteBuffer;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
 
 
 
@@ -59,36 +60,49 @@ public class OrderInterceptor extends ChannelInterceptorBase {
     private long expire = 3000;
     private boolean forwardExpired = true;
     private int maxQueue = Integer.MAX_VALUE;
+    
+    ReentrantReadWriteLock inLock = new ReentrantReadWriteLock(true);
+    ReentrantReadWriteLock outLock= new ReentrantReadWriteLock(true);
 
-    public synchronized void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload) throws ChannelException {
+    public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload) throws ChannelException {
         if ( !okToProcess(msg.getOptions()) ) {
             super.sendMessage(destination, msg, payload);
             return;
         }
-        for ( int i=0; i<destination.length; i++ ) {
-            int nr = incCounter(destination[i]);
-            //reduce byte copy
-            msg.getMessage().append(nr);
-            try {
-                getNext().sendMessage(new Member[] {destination[i]}, msg, payload);
-            }finally {
-                msg.getMessage().trim(4);
+        try {
+            outLock.writeLock().lock();
+            for ( int i=0; i<destination.length; i++ ) {
+                int nr = incCounter(destination[i]);
+                //reduce byte copy
+                msg.getMessage().append(nr);
+                try {
+                    getNext().sendMessage(new Member[] {destination[i]}, msg, payload);
+                }finally {
+                    msg.getMessage().trim(4);
+                }
             }
+        }finally {
+            outLock.writeLock().unlock();
         }
     }
 
-    public synchronized void messageReceived(ChannelMessage msg) {
+    public void messageReceived(ChannelMessage msg) {
         if ( !okToProcess(msg.getOptions()) ) {
             super.messageReceived(msg);
             return;
         }
-        int msgnr = XByteBuffer.toInt(msg.getMessage().getBytesDirect(),msg.getMessage().getLength()-4);
-        msg.getMessage().trim(4);
-        MessageOrder order = new MessageOrder(msgnr,(ChannelMessage)msg.deepclone());
-        if ( processIncoming(order) ) processLeftOvers(msg.getAddress(),false);
-    }
+        try {
+            inLock.writeLock().lock();
+            int msgnr = XByteBuffer.toInt(msg.getMessage().getBytesDirect(),msg.getMessage().getLength()-4);
+            msg.getMessage().trim(4);
+            MessageOrder order = new MessageOrder(msgnr,(ChannelMessage)msg.deepclone());
+            if ( processIncoming(order) ) processLeftOvers(msg.getAddress(),false);
     
-    public void processLeftOvers(Member member, boolean force) {
+        }finally {
+            inLock.writeLock().unlock();
+        }
+    }
+    protected void processLeftOvers(Member member, boolean force) {
         MessageOrder tmp = (MessageOrder)incoming.get(member);
         if ( force ) {
             Counter cnt = getInCounter(member);
@@ -101,7 +115,7 @@ public class OrderInterceptor extends ChannelInterceptorBase {
      * @param order MessageOrder
      * @return boolean - true if a message expired and was processed
      */
-    public boolean processIncoming(MessageOrder order) {
+    protected boolean processIncoming(MessageOrder order) {
         boolean result = false;
         Member member = order.getMessage().getAddress();
         Counter cnt = getInCounter(member);
@@ -147,28 +161,50 @@ public class OrderInterceptor extends ChannelInterceptorBase {
         return result;
     }
     
-    public synchronized void memberAdded(Member member) {
+    public void memberAdded(Member member) {
+        //reset counters
+        try {
+            inLock.writeLock().lock();
+            getInCounter(member);
+        }finally {
+            inLock.writeLock().unlock();
+        }
+        try {
+            outLock.writeLock().lock();
+            getOutCounter(member);
+        }finally {
+            outLock.writeLock().unlock();
+        }
         //notify upwards
-        getInCounter(member);
-        getOutCounter(member);
         super.memberAdded(member);
     }
 
-    public synchronized void memberDisappeared(Member member) {
-        //notify upwards
-        outcounter.remove(member);
-        incounter.remove(member);
+    public void memberDisappeared(Member member) {
+        //reset counters
+        try {
+            inLock.writeLock().lock();
+            incounter.remove(member);
+        }finally {
+            inLock.writeLock().unlock();
+        }
+        try {
+            outLock.writeLock().lock();
+            outcounter.remove(member);
+        }finally {
+            outLock.writeLock().unlock();
+        }
         //clear the remaining queue
         processLeftOvers(member,true);
+        //notify upwards
         super.memberDisappeared(member);
     }
     
-    public int incCounter(Member mbr) { 
+    protected int incCounter(Member mbr) { 
         Counter cnt = getOutCounter(mbr);
         return cnt.inc();
     }
     
-    public Counter getInCounter(Member mbr) {
+    protected Counter getInCounter(Member mbr) {
         Counter cnt = (Counter)incounter.get(mbr);
         if ( cnt == null ) {
             cnt = new Counter();
@@ -178,7 +214,7 @@ public class OrderInterceptor extends ChannelInterceptorBase {
         return cnt;
     }
 
-    public Counter getOutCounter(Member mbr) {
+    protected Counter getOutCounter(Member mbr) {
         Counter cnt = (Counter)outcounter.get(mbr);
         if ( cnt == null ) {
             cnt = new Counter();
@@ -187,7 +223,7 @@ public class OrderInterceptor extends ChannelInterceptorBase {
         return cnt;
     }
 
-    public static class Counter {
+    protected static class Counter {
         private AtomicInteger value = new AtomicInteger(0);
         
         public int getCounter() {
@@ -203,7 +239,7 @@ public class OrderInterceptor extends ChannelInterceptorBase {
         }
     }
     
-    public static class MessageOrder {
+    protected static class MessageOrder {
         private long received = System.currentTimeMillis();
         private MessageOrder next;
         private int msgNr;
diff --git a/test/org/apache/catalina/tribes/test/interceptors/TestOrderInterceptor.java b/test/org/apache/catalina/tribes/test/interceptors/TestOrderInterceptor.java
index cd4f99dab..3f4ff0e7a 100644
--- a/test/org/apache/catalina/tribes/test/interceptors/TestOrderInterceptor.java
+++ b/test/org/apache/catalina/tribes/test/interceptors/TestOrderInterceptor.java
@@ -31,6 +31,7 @@ import org.apache.catalina.tribes.group.ChannelInterceptorBase;
 import org.apache.catalina.tribes.ChannelMessage;
 import org.apache.catalina.tribes.group.InterceptorPayload;
 import org.apache.catalina.tribes.ChannelException;
+import java.util.concurrent.atomic.AtomicInteger;
 
 public class TestOrderInterceptor extends TestCase {
 
@@ -76,8 +77,9 @@ public class TestOrderInterceptor extends TestCase {
     
     public void testOrder1() throws Exception {
         Member[] dest = channels[0].getMembers();
+        final AtomicInteger value = new AtomicInteger(0);
         for ( int i=0; i<100; i++ ) {
-            channels[0].send(dest,new Integer(i),0);
+            channels[0].send(dest,new Integer(value.getAndAdd(1)),0);
         }
         Thread.sleep(5000);
         for ( int i=0; i<test.length; i++ ) {
@@ -85,6 +87,40 @@ public class TestOrderInterceptor extends TestCase {
         }
     }
     
+    public void testOrder2() throws Exception {
+        final Member[] dest = channels[0].getMembers();
+        final AtomicInteger value = new AtomicInteger(0);
+        Runnable run = new Runnable() {
+            public void run() {
+                for (int i = 0; i < 100; i++) {
+                    try {
+                        synchronized (channels[0]) {
+                            channels[0].send(dest, new Integer(value.getAndAdd(1)), 0);
+                        }
+                    }catch ( Exception x ) {
+                        x.printStackTrace();
+                        assertEquals(true,false);
+                    }
+                }
+            }
+        };
+        Thread[] threads = new Thread[5];
+        for (int i=0;i<threads.length;i++) {
+            threads[i] = new Thread(run);
+        }
+        for (int i=0;i<threads.length;i++) {
+            threads[i].start();
+        }
+        for (int i=0;i<threads.length;i++) {
+            threads[i].join();
+        }
+        Thread.sleep(5000);
+        for ( int i=0; i<test.length; i++ ) {
+            super.assertEquals(false,test[i].fail);
+        }
+    }
+
+
     protected void tearDown() throws Exception {
         System.out.println("tearDown");
         super.tearDown();
@@ -112,7 +148,7 @@ public class TestOrderInterceptor extends TestCase {
             Integer i = (Integer)msg;
             if ( i.intValue() != cnt ) fail = true;
             else cnt++;
-            System.out.println("Listener["+id+"] Message received:"+i+" Count:"+total);
+            System.out.println("Listener["+id+"] Message received:"+i+" Count:"+total+" Fail:"+fail);
 
         }
 
-- 
2.11.0