viphadke commented on issue #11421:
URL: https://github.com/apache/ignite/issues/11421#issuecomment-2210202895
Here is the source
package ignitetest;
import java.io.IOException;
import java.io.Serializable;
import java.util.Arrays;
import java.util.Collections;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import javax.cache.Cache;
import javax.cache.processor.EntryProcessorException;
import javax.cache.processor.MutableEntry;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.IgniteLock;
import org.apache.ignite.IgniteSemaphore;
import org.apache.ignite.Ignition;
import org.apache.ignite.cache.CacheAtomicityMode;
import org.apache.ignite.cache.CacheEntryProcessor;
import org.apache.ignite.cache.query.QueryCursor;
import org.apache.ignite.cache.query.ScanQuery;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.DeploymentMode;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.lang.IgniteRunnable;
import org.apache.ignite.resources.IgniteInstanceResource;
import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
import
org.apache.ignite.spi.discovery.tcp.ipfinder.multicast.TcpDiscoveryMulticastIpFinder;
import com.thingworx.cache.values.Tuple2;
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];
}
}
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
