Fixed - methods belongs to transactional objects get blocked at high concurrency. #1459

redisson/src/main/java/org/redisson/RedissonLock.java

@@ -575,12 +575,11 @@ public class RedissonLock extends RedissonExpirable implements RLock {
                     return;
                 }
 
-                // waiting for message
                 final RedissonLockEntry entry = getEntry(currentThreadId);
-                synchronized (entry) {
                 if (entry.getLatch().tryAcquire()) {
                     lockAsync(leaseTime, unit, subscribeFuture, result, currentThreadId);
                 } else {
+                    // waiting for message
                     final AtomicReference<Timeout> futureRef = new AtomicReference<Timeout>();
                     final Runnable listener = new Runnable() {
                         @Override
@@ -598,18 +597,15 @@ public class RedissonLock extends RedissonExpirable implements RLock {
                         Timeout scheduledFuture = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {
                             @Override
                             public void run(Timeout timeout) throws Exception {
-                                    synchronized (entry) {
                                 if (entry.removeListener(listener)) {
                                     lockAsync(leaseTime, unit, subscribeFuture, result, currentThreadId);
                                 }
                             }
-                                }
                         }, ttl, TimeUnit.MILLISECONDS);
                         futureRef.set(scheduledFuture);
                     }
                 }
             }
-            }
         });
     }
 
@@ -768,7 +764,6 @@ public class RedissonLock extends RedissonExpirable implements RLock {
                 // waiting for message
                 final long current = System.currentTimeMillis();
                 final RedissonLockEntry entry = getEntry(currentThreadId);
-                synchronized (entry) {
                 if (entry.getLatch().tryAcquire()) {
                     tryLockAsync(time, leaseTime, unit, subscribeFuture, result, currentThreadId);
                 } else {
@@ -798,7 +793,6 @@ public class RedissonLock extends RedissonExpirable implements RLock {
                         Timeout scheduledFuture = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {
                             @Override
                             public void run(Timeout timeout) throws Exception {
-                                    synchronized (entry) {
                                 if (entry.removeListener(listener)) {
                                     long elapsed = System.currentTimeMillis() - current;
                                     time.addAndGet(-elapsed);
@@ -806,13 +800,11 @@ public class RedissonLock extends RedissonExpirable implements RLock {
                                     tryLockAsync(time, leaseTime, unit, subscribeFuture, result, currentThreadId);
                                 }
                             }
-                                }
                         }, t, TimeUnit.MILLISECONDS);
                         futureRef.set(scheduledFuture);
                     }
                 }
             }
-            }
         });
     }
 

redisson/src/main/java/org/redisson/RedissonPermitExpirableSemaphore.java

@@ -213,7 +213,6 @@ public class RedissonPermitExpirableSemaphore extends RedissonExpirable implemen
                 // waiting for message
                 final long current = System.currentTimeMillis();
                 final RedissonLockEntry entry = getEntry();
-                synchronized (entry) {
                 if (entry.getLatch().tryAcquire()) {
                     tryAcquireAsync(time, permits, subscribeFuture, result, ttl, timeUnit);
                 } else {
@@ -266,7 +265,6 @@ public class RedissonPermitExpirableSemaphore extends RedissonExpirable implemen
                                 return;
                             }
 
-                                synchronized (entry) {
                             if (entry.removeListener(listener)) {
                                 long elapsed = System.currentTimeMillis() - current;
                                 time.addAndGet(-elapsed);
@@ -274,12 +272,10 @@ public class RedissonPermitExpirableSemaphore extends RedissonExpirable implemen
                                 tryAcquireAsync(time, permits, subscribeFuture, result, ttl, timeUnit);
                             }
                         }
-                            }
                     }, t, TimeUnit.MILLISECONDS);
                     waitTimeoutFutureRef.set(waitTimeoutFuture);
                 }
             }
-            }
         });
         
     }
@@ -318,7 +314,6 @@ public class RedissonPermitExpirableSemaphore extends RedissonExpirable implemen
                 }
 
                 final RedissonLockEntry entry = getEntry();
-                synchronized (entry) {
                 if (entry.getLatch().tryAcquire(permits)) {
                     acquireAsync(permits, subscribeFuture, result, ttl, timeUnit);
                 } else {
@@ -347,7 +342,6 @@ public class RedissonPermitExpirableSemaphore extends RedissonExpirable implemen
                     entry.addListener(listener);
                 }
             }
-            }
         });
     }
 

redisson/src/main/java/org/redisson/RedissonSemaphore.java

@@ -180,7 +180,6 @@ public class RedissonSemaphore extends RedissonExpirable implements RSemaphore {
                 // waiting for message
                 final long current = System.currentTimeMillis();
                 final RedissonLockEntry entry = getEntry();
-                synchronized (entry) {
                 if (entry.getLatch().tryAcquire()) {
                     tryAcquireAsync(time, permits, subscribeFuture, result);
                 } else {
@@ -207,7 +206,6 @@ public class RedissonSemaphore extends RedissonExpirable implements RSemaphore {
                         Timeout scheduledFuture = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {
                             @Override
                             public void run(Timeout timeout) throws Exception {
-                                    synchronized (entry) {
                                 if (entry.removeListener(listener)) {
                                     long elapsed = System.currentTimeMillis() - current;
                                     time.addAndGet(-elapsed);
@@ -215,13 +213,11 @@ public class RedissonSemaphore extends RedissonExpirable implements RSemaphore {
                                     tryAcquireAsync(time, permits, subscribeFuture, result);
                                 }
                             }
-                                }
                         }, t, TimeUnit.MILLISECONDS);
                         futureRef.set(scheduledFuture);
                     }
                 }
             }
-            }
         });
         
     }
@@ -251,7 +247,6 @@ public class RedissonSemaphore extends RedissonExpirable implements RSemaphore {
                 }
                 
                 final RedissonLockEntry entry = getEntry();
-                synchronized (entry) {
                 if (entry.getLatch().tryAcquire(permits)) {
                     acquireAsync(permits, subscribeFuture, result);
                 } else {
@@ -264,7 +259,6 @@ public class RedissonSemaphore extends RedissonExpirable implements RSemaphore {
                     entry.addListener(listener);
                 }
             }
-            }
         });
     }
 

redisson/src/main/java/org/redisson/pubsub/LockPubSub.java

@@ -35,12 +35,8 @@ public class LockPubSub extends PublishSubscribe<RedissonLockEntry> {
     @Override
     protected void onMessage(RedissonLockEntry value, Long message) {
         if (message.equals(unlockMessage)) {
-            while (true) {
             Runnable runnableToExecute = value.getListeners().poll();
-                if (runnableToExecute == null) {
+            if (runnableToExecute != null) {
-                    break;
-                }
-                
                 runnableToExecute.run();
             }
 

redisson/src/main/java/org/redisson/pubsub/SemaphorePubSub.java

@@ -32,12 +32,8 @@ public class SemaphorePubSub extends PublishSubscribe<RedissonLockEntry> {
 
     @Override
     protected void onMessage(RedissonLockEntry value, Long message) {
-        while (true) {
         Runnable runnableToExecute = value.getListeners().poll();
-            if (runnableToExecute == null) {
+        if (runnableToExecute != null) {
-                break;
-            }
-            
             runnableToExecute.run();
         }
         

redisson/src/test/java/org/redisson/transaction/RedissonBaseTransactionalMapTest.java

@@ -22,8 +22,8 @@ public abstract class RedissonBaseTransactionalMapTest extends BaseTest {
     
     @Test
     public void testFastPut() throws InterruptedException {
-        ExecutorService executor = Executors.newFixedThreadPool(16);
+        ExecutorService executor = Executors.newFixedThreadPool(200);
-        for (int i = 0; i < 500; i++) {
+        for (int i = 0; i < 2000; i++) {
             executor.submit(() -> {
                 for (int j = 0; j < 100; j++) {
                     RTransaction t = redisson.createTransaction(TransactionOptions.defaults());
@@ -35,7 +35,7 @@ public abstract class RedissonBaseTransactionalMapTest extends BaseTest {
         }
         
         executor.shutdown();
-        assertThat(executor.awaitTermination(1, TimeUnit.MINUTES)).isTrue();
+        assertThat(executor.awaitTermination(2, TimeUnit.MINUTES)).isTrue();
     }