You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@ignite.apache.org by pragmaticbigdata <am...@gmail.com> on 2016/07/27 13:30:11 UTC
Multi-threaded transactions
I am using ignite 1.6.I am trying to execute two transactions in parallel and
expecting one of them to fail since the same cache keys are updated. I found
that the second transaction doesn't fail and the commit succeeds. Below is
the code I am trying out let me know what is missing.
After starting a server node, below is the code executed on the client node
public void testMultiThreadTransactionsOnSimpleCache() {
CacheConfiguration<Integer, Integer> cfg = new
CacheConfiguration<>();
cfg.setName("IntCache");
cfg.setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL);
try (Ignite ignite = initializeIgniteInClientMode()) {
IgniteCache<Integer, Integer> cache =
ignite.getOrCreateCache(cfg);
ExecutorService executorService =
Executors.newFixedThreadPool(2);
final CountDownLatch latch = new CountDownLatch(2);
executorService.submit(() -> {
logger.info("Starting a thread");
try (Transaction tx =
ignite.transactions().txStart(TransactionConcurrency.OPTIMISTIC,
TransactionIsolation.READ_COMMITTED)) {
for (int i = 0; i < 20; i++) {
cache.put(i, 500 + i);
}
tx.commit();
}
logger.info("Commit done");
latch.countDown();
});
executorService.submit(() -> {
logger.info("Starting a thread");
try (Transaction tx =
ignite.transactions().txStart(TransactionConcurrency.OPTIMISTIC,
TransactionIsolation.READ_COMMITTED)) {
for (int i = 0; i < 20; i++) {
cache.put(i, 600 + i);
}
latch.countDown();
try {
latch.await(); //making sure that the first thread
is done with the commit
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
try {
TimeUnit.SECONDS.sleep(2); //making sure that the
first thread is done with the commit
} catch (InterruptedException e) {
e.printStackTrace();
}
logger.info("Trying a commit");
try {
tx.commit();
} catch (TransactionOptimisticException e) {
logger.error("Transaction Failed", e);
}
}
});
executorService.shutdown();
try {
executorService.awaitTermination(Long.MAX_VALUE,
TimeUnit.NANOSECONDS);
logger.info("All threads complete");
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
logger.info("Cache size : {}", cache.size());
for (Cache.Entry<Integer, Integer> aCache : cache) {
logger.info("Key : {}, Value : {}", aCache.getKey(),
aCache.getValue());
}
}
}
Below are the logs
19:12:55.958 INFO [pool-2-thread-2][c.o.s.i.test.TransactionTest] Starting
a thread
19:12:55.969 INFO [pool-2-thread-1][c.o.s.i.test.TransactionTest] Starting
a thread
19:12:56.194 INFO [pool-2-thread-1][c.o.s.i.test.TransactionTest] Commit
done
19:12:58.194 INFO [pool-2-thread-2][c.o.s.i.test.TransactionTest] Trying a
commit
19:12:58.212 INFO [main][c.o.s.i.test.TransactionTest] All threads complete
19:12:58.290 INFO [main][c.o.s.i.test.TransactionTest] Cache size : 20
19:12:58.319 INFO [main][c.o.s.i.test.TransactionTest] Key : 6, Value : 606
19:12:58.319 INFO [main][c.o.s.i.test.TransactionTest] Key : 12, Value :
612
19:12:58.319 INFO [main][c.o.s.i.test.TransactionTest] Key : 19, Value :
619
19:12:58.319 INFO [main][c.o.s.i.test.TransactionTest] Key : 5, Value : 605
19:12:58.319 INFO [main][c.o.s.i.test.TransactionTest] Key : 13, Value :
613
19:12:58.320 INFO [main][c.o.s.i.test.TransactionTest] Key : 0, Value : 600
19:12:58.320 INFO [main][c.o.s.i.test.TransactionTest] Key : 18, Value :
618
19:12:58.320 INFO [main][c.o.s.i.test.TransactionTest] Key : 7, Value : 607
19:12:58.320 INFO [main][c.o.s.i.test.TransactionTest] Key : 8, Value : 608
19:12:58.320 INFO [main][c.o.s.i.test.TransactionTest] Key : 9, Value : 609
19:12:58.320 INFO [main][c.o.s.i.test.TransactionTest] Key : 2, Value : 602
19:12:58.320 INFO [main][c.o.s.i.test.TransactionTest] Key : 17, Value :
617
19:12:58.321 INFO [main][c.o.s.i.test.TransactionTest] Key : 15, Value :
615
19:12:58.321 INFO [main][c.o.s.i.test.TransactionTest] Key : 16, Value :
616
19:12:58.322 INFO [main][c.o.s.i.test.TransactionTest] Key : 1, Value : 601
19:12:58.322 INFO [main][c.o.s.i.test.TransactionTest] Key : 10, Value :
610
19:12:58.322 INFO [main][c.o.s.i.test.TransactionTest] Key : 14, Value :
614
19:12:58.322 INFO [main][c.o.s.i.test.TransactionTest] Key : 3, Value : 603
19:12:58.322 INFO [main][c.o.s.i.test.TransactionTest] Key : 11, Value :
611
19:12:58.322 INFO [main][c.o.s.i.test.TransactionTest] Key : 4, Value : 604
I am expecting the second transaction to fail and the cache to have 20
entries with values between 500 and 520.
I tried out changing the concurrent mode and the isolation levels too but
that didn't make a difference.
To verify that the transactions are working correctly I tried to
sequentially execute two transactions with one executing commit and another
rollback. That works as expected.
Thanks.
--
View this message in context: http://apache-ignite-users.70518.x6.nabble.com/Multi-threaded-transactions-tp6563.html
Sent from the Apache Ignite Users mailing list archive at Nabble.com.
Re: Multi-threaded transactions
Posted by vkulichenko <va...@gmail.com>.
Hi,
Why do you expect one of the transactions to fail? Main purpose of the
transactions is to guarantee data updates atomicity and consistency and your
result looks correct from this standpoint. If you expect
TransactionOptimisticException, it can happen only in
OPTIMISTIC/SERIALIZABLE mode, but even in this mode Ignite will try its best
not to fail and commit both transactions.
For more information on different transaction modes please refer to this
page: https://apacheignite.readme.io/v1.6/docs/transactions
-Val
--
View this message in context: http://apache-ignite-users.70518.x6.nabble.com/Multi-threaded-transactions-tp6563p6570.html
Sent from the Apache Ignite Users mailing list archive at Nabble.com.