applications/forwardingrules-sync/src/test/java/org/opendaylight/openflowplugin/applications/frsync/util/SemaphoreKeeperTest.java

   1 /**
   2  * Copyright (c) 2016 Cisco Systems, Inc. and others. All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8
   9 package org.opendaylight.openflowplugin.applications.frsync.util;
  10
  11 import java.util.concurrent.ConcurrentHashMap;
  12 import java.util.concurrent.ConcurrentMap;
  13 import java.util.concurrent.ExecutorService;
  14 import java.util.concurrent.LinkedBlockingQueue;
  15 import java.util.concurrent.Semaphore;
  16 import java.util.concurrent.ThreadPoolExecutor;
  17 import java.util.concurrent.TimeUnit;
  18
  19 import org.junit.Assert;
  20 import org.junit.Before;
  21 import org.junit.Test;
  22 import org.opendaylight.openflowplugin.applications.frsync.SemaphoreKeeper;
  23 import org.slf4j.Logger;
  24 import org.slf4j.LoggerFactory;
  25
  26 /**
  27  * Test for {@link SemaphoreKeeperGuavaImpl}.
  28  */
  29 public class SemaphoreKeeperTest {
  30     private static final Logger LOG = LoggerFactory.getLogger(SemaphoreKeeperTest.class);
  31     private SemaphoreKeeperGuavaImpl<String> semaphoreKeeper;
  32     final String key = "11";
  33
  34     @Before
  35     public void setUp() throws Exception {
  36         semaphoreKeeper = new SemaphoreKeeperGuavaImpl(1, true);
  37     }
  38
  39     @Test
  40     public void testSummonGuard() throws Exception {
  41         Semaphore semaphore1 = semaphoreKeeper.summonGuard(key);
  42         final int g1FingerPrint = semaphore1.hashCode();
  43         Semaphore semaphore2 = semaphoreKeeper.summonGuard(key);
  44         final int g2FingerPrint = semaphore2.hashCode();
  45
  46         Assert.assertSame(semaphore1, semaphore2);
  47         Assert.assertEquals(1, semaphore1.availablePermits());
  48
  49         semaphore1.acquire();
  50         semaphore1.release();
  51         Assert.assertEquals(1, semaphore1.availablePermits());
  52         semaphore1 = null;
  53         System.gc();
  54
  55         semaphore2.acquire();
  56         semaphore2.release();
  57         Assert.assertEquals(1, semaphore2.availablePermits());
  58         semaphore2 = null;
  59         Assert.assertEquals(g1FingerPrint, g2FingerPrint);
  60
  61         System.gc();
  62         final Semaphore semaphore3 = semaphoreKeeper.summonGuard(key);
  63         Assert.assertNotEquals(g1FingerPrint, semaphore3.hashCode());
  64     }
  65
  66     @Test
  67     public void testReleaseGuard() throws Exception {
  68         for (int total = 1; total <= 10; total++) {
  69             LOG.info("test run: {}", total);
  70             final Worker task = new Worker(semaphoreKeeper, key);
  71
  72             final ExecutorService executorService = new ThreadPoolExecutor(5, 5,
  73                     0L, TimeUnit.MILLISECONDS,
  74                     new LinkedBlockingQueue<Runnable>()) {
  75                 @Override
  76                 protected void afterExecute(final Runnable r, final Throwable t) {
  77                     super.afterExecute(r, t);
  78                     if (t != null) {
  79                         LOG.error("pool thread crashed", t);
  80                     }
  81                 }
  82             };
  83
  84             final int steps = 10;
  85             for (int i = 0; i < steps; i++) {
  86                 executorService.submit(task);
  87             }
  88             Thread.sleep(50L);
  89             LOG.info("STARTING new serie");
  90             System.gc();
  91
  92             for (int i = 0; i < steps; i++) {
  93                 executorService.submit(task);
  94             }
  95             Thread.sleep(1000L);
  96             System.gc();
  97
  98             executorService.shutdown();
  99             final boolean terminated = executorService.awaitTermination(10, TimeUnit.SECONDS);
 100             if (!terminated) {
 101                 LOG.warn("pool stuck, forcing termination");
 102                 executorService.shutdownNow();
 103                 Assert.fail("pool failed to finish gracefully");
 104             }
 105
 106             final int counterSize = task.getCounterSize();
 107             LOG.info("final counter = {}", counterSize);
 108             Assert.assertEquals(20, counterSize);
 109         }
 110     }
 111
 112     private static class Worker implements Runnable {
 113         private final SemaphoreKeeper<String> keeper;
 114         private final String key;
 115         private ConcurrentMap<Integer, Integer> counter = new ConcurrentHashMap<>();
 116         private volatile int index = 0;
 117
 118         public Worker(SemaphoreKeeper<String> keeper, final String key) {
 119             this.keeper = keeper;
 120             this.key = key;
 121         }
 122
 123         @Override
 124         public void run() {
 125             try {
 126                 final Semaphore guard = keeper.summonGuard(key);
 127                 Thread.sleep(2L);
 128                 guard.acquire();
 129                 counter.putIfAbsent(index, 0);
 130                 counter.put(index, counter.get(index) + 1);
 131                 LOG.debug("queue: {} [{}] - {}", guard.getQueueLength(), guard.hashCode(), counter.size());
 132                 index++;
 133                 guard.release();
 134             } catch (Exception e) {
 135                 LOG.warn("acquiring failed.. ", e);
 136             }
 137         }
 138
 139         public int getCounterSize() {
 140             return counter.size();
 141         }
 142     }
 143 }