We tried to use the igite semaphore to control access to resources in a cluster. At low concurrency viz. < 10 threads the semaphore seems to work as documented. However when we have 100’s of threads it the semaphore.tryAcquire(1, 100, TimeUnit.MILLISECONDS); starts taking a lot more than 100 sec. The expectation is that it will fail in 100ms. Actually we have seen it take 15, 20 or even 30 sec.

The semaphore.tryAcquire(1, 100, TimeUnit.MILLISECONDS) does not work as expected or documented at high conucrrency even when enough permits are available.

The following code is simplified representation of the production code.

public class IgniteSemaphoreTest {

private static final int THREAD_POOL_SIZE = 100 ;

private static IgniteSemaphore semaphore = null ;

/**
 * This method can be used instead of semaphore.tryIgnite 
 * This improves the performance drastically. 
 * @param permits
 * @param timeout
 * @param tu
 * @return
 */
private static synchronized boolean tryAcquire(int permits, int timeout, TimeUnit tu) {
    return semaphore.tryAcquire(permits, timeout, tu);
}


public static void main(String[] args) throws Exception {
    // Preparing IgniteConfiguration using Java APIs
    IgniteConfiguration cfg = new IgniteConfiguration();

    // The node will be started as a client node.
    cfg.setClientMode(true);
    cfg.setPeerClassLoadingEnabled(true);
    cfg.setDeploymentMode(DeploymentMode.SHARED);
    // Classes of custom Java logic will be transferred over the wire from this app.
    CacheConfiguration<String, SomeFancyClass> cacheConfig = new CacheConfiguration<String, SomeFancyClass>("processortest");
    cacheConfig.setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL);
    cfg.setCacheConfiguration(cacheConfig);
    // Setting up an IP Finder to ensure the client can locate the servers.
    TcpDiscoveryMulticastIpFinder ipFinder = new TcpDiscoveryMulticastIpFinder();
    ipFinder.setAddresses(Collections.singletonList("127.0.0.1:47500"));
    cfg.setDiscoverySpi(new TcpDiscoverySpi().setIpFinder(ipFinder));

    // Starting the node
    Ignite ignite = Ignition.start(cfg);

    semaphore = ignite.semaphore("mySema12", // Distributed semaphore name.
        100, // Number of permits.
        true, // Release acquired permits if node, that owned them, left topology.
        true // Create if it doesn't exist.
    );

    ExecutorService ser = Executors.newFixedThreadPool(THREAD_POOL_SIZE);
    int JOB_COUNT = 1000 ;
    long [] time_taken = new long[JOB_COUNT];
    for (int i = 0; i < JOB_COUNT; i++) {
        final int index= i ;
        ser.submit(() -> {
            boolean acquired = false;
            try {
                long tm1 = System.currentTimeMillis();
                acquired = semaphore.tryAcquire(1, 100, TimeUnit.MILLISECONDS);                        
                long tm2 = System.currentTimeMillis();
                long diff = tm2 - tm1;
                if (acquired) {
                    System.out.println("Acquired in " + diff);
                } else {
                    System.out.println("Not acquired in " + diff);
                }
                Thread.sleep((int)(Math.random()*1000));
                time_taken[index] = diff ;
            } catch (Exception e) {

            } finally {
                if (acquired) {
                    semaphore.release();
                }
            }

        });

    }

    ser.shutdown();

    while (!ser.awaitTermination(1, TimeUnit.SECONDS)) {
        // nothing to do
    }

    Arrays.sort(time_taken);
    System.out.println("P95 time = "+ percentile(time_taken, 95));
    System.out.println("semaphore released");

    System.out.println(">> Compute task is executed, check for output on the server nodes.");
    try {
        Thread.sleep(60000);
    } catch (Exception e) {
        e.printStackTrace();
    }
    // Disconnect from the cluster.
    ignite.close();
}

public static long percentile(long [] latencies, double percentile) {
    int index = (int) Math.ceil(percentile / 100.0 * latencies.length);
    return latencies[index];
}

}