Avihai Berkovitz created IGNITE-2145:
----------------------------------------
Summary: Async cache operations on primary node might block
Key: IGNITE-2145
URL: https://issues.apache.org/jira/browse/IGNITE-2145
Project: Ignite
Issue Type: Bug
Affects Versions: ignite-1.4
Environment: Windows 8.1 64 bit
java version "1.8.0_51"
Java(TM) SE Runtime Environment (build 1.8.0_51-b16)
Java HotSpot(TM) 64-Bit Server VM (build 25.51-b03, mixed mode)
Ignite 1.4.0
Reporter: Avihai Berkovitz
When executing async operations on caches the initial function call is supposed
to return immediately, without regard to cache state, network traffic, etc. If
we reference a cache key on a different node it works. But if the primary node
for a cache key is the same node that initiate the operation everything is
executed synchronously.
This is probably an optimization, but can have side effects. For example, if a
long running CacheProcessor is invoked on this key, every operation will block.
Here is some code to illustrate the problem. Notice that there is a 4 seconds
delay between "Before async op" and "After async op". If you reverse the
isPrimary() condition it won't happen.
{code}
IgniteConfiguration igniteConfiguration = new IgniteConfiguration()
.setFailoverSpi(new AlwaysFailoverSpi())
.setGridLogger(new Slf4jLogger())
.setPeerClassLoadingEnabled(false)
.setDeploymentMode(DeploymentMode.CONTINUOUS);
Ignite ignite1 = Ignition.start(igniteConfiguration);
Ignite ignite2 = Ignition.start(igniteConfiguration.setGridName("2"));
assert ignite1.cluster().nodes().size() == 2;
assert ignite2.cluster().nodes().size() == 2;
CacheConfiguration<String, Integer> cacheConfiguration = new
CacheConfiguration<String, Integer>()
.setName("test")
.setCacheMode(CacheMode.PARTITIONED)
.setAtomicityMode(CacheAtomicityMode.ATOMIC)
.setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC);
IgniteCache<String, Integer> cache1 =
ignite1.getOrCreateCache(cacheConfiguration);
IgniteCache<String, Integer> cache2 =
ignite2.getOrCreateCache(cacheConfiguration);
// Find key mapped to the first node
String key = IntStream.range(0, 100)
.mapToObj(value -> "HI" + value)
.filter(s ->
ignite1.affinity(cacheConfiguration.getName()).isPrimary(ignite1.cluster().localNode(),
s))
.findFirst().get();
// Run a blocked processor from node 2
Thread thread = new Thread() {
@Override
public void run() {
System.out.println("Invoking blocked processor");
cache2.invoke(key, (entry, arguments) -> {
System.out.println("IN INVOKE");
try {
Thread.sleep(5 * 1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
return null;
});
System.out.println("Blocked invoke returned");
}
};
thread.start();
Thread.sleep(1000);
IgniteCache<String, Integer> async1 = cache1.withAsync();
System.out.println("Before async op");
//async1.put(key, 1);
//async1.get(key);
async1.containsKey(key);
System.out.println("After async op");
assert async1.future().isDone();
thread.join();
ignite2.close();
ignite1.close();
{code}
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
