shishkovilja commented on code in PR #11512:
URL: https://github.com/apache/ignite/pull/11512#discussion_r1830444381
##########
modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxWithKeyContentionSelfTest.java:
##########
@@ -216,120 +198,66 @@ public void
testOptimisticRepeatableReadCheckContentionTxMetricNear() throws Exc
* @param isolation Isolation level.
* @throws Exception If failed.
*/
- private void runKeyCollisionsMetric(TransactionConcurrency concurrency,
TransactionIsolation isolation) throws Exception {
+ private void runKeyCollisionsMetric(TransactionConcurrency concurrency,
TransactionIsolation isolation, boolean nearCashe)
Review Comment:
```suggestion
private void runKeyCollisionsMetric(TransactionConcurrency concurrency,
TransactionIsolation isolation, boolean nearCache)
```
##########
modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxWithKeyContentionSelfTest.java:
##########
@@ -62,9 +57,6 @@ public class TxWithKeyContentionSelfTest extends
GridCommonAbstractTest {
/** Client flag. */
private boolean client;
Review Comment:
Let's remove this field, because we can just call `startClientGrid()` in
test.
##########
modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxWithKeyContentionSelfTest.java:
##########
@@ -216,120 +198,66 @@ public void
testOptimisticRepeatableReadCheckContentionTxMetricNear() throws Exc
* @param isolation Isolation level.
* @throws Exception If failed.
*/
- private void runKeyCollisionsMetric(TransactionConcurrency concurrency,
TransactionIsolation isolation) throws Exception {
+ private void runKeyCollisionsMetric(TransactionConcurrency concurrency,
TransactionIsolation isolation, boolean nearCashe)
+ throws Exception {
Ignite ig = startGridsMultiThreaded(3);
int contCnt = (int)U.staticField(IgniteTxManager.class,
"COLLISIONS_QUEUE_THRESHOLD") * 5;
- CountDownLatch txLatch = new CountDownLatch(contCnt);
-
ig.cluster().state(ClusterState.ACTIVE);
client = true;
Ignite cl = startGrid();
- IgniteTransactions cliTxMgr = cl.transactions();
-
- IgniteCache<Integer, Integer> cache = ig.cache(DEFAULT_CACHE_NAME);
-
- IgniteCache<Integer, Integer> cache0 = cl.cache(DEFAULT_CACHE_NAME);
+ IgniteCache<Integer, Integer> cache =
ig.getOrCreateCache(getCacheConfiguration(nearCashe));
Review Comment:
```suggestion
IgniteCache<Integer, Integer> cache =
ig.createCache(getCacheConfiguration(nearCache));
```
##########
modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxWithKeyContentionSelfTest.java:
##########
@@ -216,120 +198,66 @@ public void
testOptimisticRepeatableReadCheckContentionTxMetricNear() throws Exc
* @param isolation Isolation level.
* @throws Exception If failed.
*/
- private void runKeyCollisionsMetric(TransactionConcurrency concurrency,
TransactionIsolation isolation) throws Exception {
+ private void runKeyCollisionsMetric(TransactionConcurrency concurrency,
TransactionIsolation isolation, boolean nearCashe)
+ throws Exception {
Ignite ig = startGridsMultiThreaded(3);
int contCnt = (int)U.staticField(IgniteTxManager.class,
"COLLISIONS_QUEUE_THRESHOLD") * 5;
- CountDownLatch txLatch = new CountDownLatch(contCnt);
-
ig.cluster().state(ClusterState.ACTIVE);
client = true;
Ignite cl = startGrid();
- IgniteTransactions cliTxMgr = cl.transactions();
-
- IgniteCache<Integer, Integer> cache = ig.cache(DEFAULT_CACHE_NAME);
-
- IgniteCache<Integer, Integer> cache0 = cl.cache(DEFAULT_CACHE_NAME);
+ IgniteCache<Integer, Integer> cache =
ig.getOrCreateCache(getCacheConfiguration(nearCashe));
final Integer keyId = primaryKey(cache);
- CountDownLatch blockOnce = new CountDownLatch(1);
-
- for (Ignite ig0 : G.allGrids()) {
- if (ig0.configuration().isClientMode())
- continue;
-
- TestRecordingCommunicationSpi commSpi0 =
-
(TestRecordingCommunicationSpi)ig0.configuration().getCommunicationSpi();
-
- commSpi0.blockMessages(new IgniteBiPredicate<ClusterNode,
Message>() {
- @Override public boolean apply(ClusterNode node, Message msg) {
- if (msg instanceof GridNearTxFinishResponse &&
blockOnce.getCount() > 0) {
- blockOnce.countDown();
-
- return true;
- }
-
- return false;
- }
- });
- }
-
- IgniteInternalFuture f = GridTestUtils.runAsync(() -> {
- try (Transaction tx = cliTxMgr.txStart(concurrency, isolation)) {
- cache0.put(keyId, 0);
- tx.commit();
- }
- });
-
- blockOnce.await();
+ IgniteTransactions transactions = cl.transactions();
- GridCompoundFuture<?, ?> finishFut = new GridCompoundFuture<>();
+ AtomicBoolean doTest = new AtomicBoolean(true);
- for (int i = 0; i < contCnt; ++i) {
- IgniteInternalFuture f0 = GridTestUtils.runAsync(() -> {
- try (Transaction tx = cliTxMgr.txStart(concurrency,
isolation)) {
- cache0.put(keyId, 0);
+ IgniteInternalFuture f0 = GridTestUtils.runMultiThreadedAsync(() -> {
+ while (doTest.get()) {
+ try (Transaction tx = transactions.txStart(concurrency,
isolation)) {
+ cache.put(keyId, 0);
Review Comment:
`cache` is an instance of server node cache, `transactions` is an instance
of client node transactions. This block is not working as you expected.
##########
modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxWithKeyContentionSelfTest.java:
##########
@@ -216,120 +198,66 @@ public void
testOptimisticRepeatableReadCheckContentionTxMetricNear() throws Exc
* @param isolation Isolation level.
* @throws Exception If failed.
*/
- private void runKeyCollisionsMetric(TransactionConcurrency concurrency,
TransactionIsolation isolation) throws Exception {
+ private void runKeyCollisionsMetric(TransactionConcurrency concurrency,
TransactionIsolation isolation, boolean nearCashe)
+ throws Exception {
Ignite ig = startGridsMultiThreaded(3);
int contCnt = (int)U.staticField(IgniteTxManager.class,
"COLLISIONS_QUEUE_THRESHOLD") * 5;
- CountDownLatch txLatch = new CountDownLatch(contCnt);
-
ig.cluster().state(ClusterState.ACTIVE);
client = true;
Ignite cl = startGrid();
- IgniteTransactions cliTxMgr = cl.transactions();
-
- IgniteCache<Integer, Integer> cache = ig.cache(DEFAULT_CACHE_NAME);
-
- IgniteCache<Integer, Integer> cache0 = cl.cache(DEFAULT_CACHE_NAME);
+ IgniteCache<Integer, Integer> cache =
ig.getOrCreateCache(getCacheConfiguration(nearCashe));
final Integer keyId = primaryKey(cache);
- CountDownLatch blockOnce = new CountDownLatch(1);
-
- for (Ignite ig0 : G.allGrids()) {
- if (ig0.configuration().isClientMode())
- continue;
-
- TestRecordingCommunicationSpi commSpi0 =
-
(TestRecordingCommunicationSpi)ig0.configuration().getCommunicationSpi();
-
- commSpi0.blockMessages(new IgniteBiPredicate<ClusterNode,
Message>() {
- @Override public boolean apply(ClusterNode node, Message msg) {
- if (msg instanceof GridNearTxFinishResponse &&
blockOnce.getCount() > 0) {
- blockOnce.countDown();
-
- return true;
- }
-
- return false;
- }
- });
- }
-
- IgniteInternalFuture f = GridTestUtils.runAsync(() -> {
- try (Transaction tx = cliTxMgr.txStart(concurrency, isolation)) {
- cache0.put(keyId, 0);
- tx.commit();
- }
- });
-
- blockOnce.await();
+ IgniteTransactions transactions = cl.transactions();
- GridCompoundFuture<?, ?> finishFut = new GridCompoundFuture<>();
+ AtomicBoolean doTest = new AtomicBoolean(true);
- for (int i = 0; i < contCnt; ++i) {
- IgniteInternalFuture f0 = GridTestUtils.runAsync(() -> {
- try (Transaction tx = cliTxMgr.txStart(concurrency,
isolation)) {
- cache0.put(keyId, 0);
+ IgniteInternalFuture f0 = GridTestUtils.runMultiThreadedAsync(() -> {
+ while (doTest.get()) {
+ try (Transaction tx = transactions.txStart(concurrency,
isolation)) {
+ cache.put(keyId, 0);
tx.commit();
-
- txLatch.countDown();
}
- });
-
- finishFut.add(f0);
- }
-
- finishFut.markInitialized();
-
- for (Ignite ig0 : G.allGrids()) {
- TestRecordingCommunicationSpi commSpi0 =
-
(TestRecordingCommunicationSpi)ig0.configuration().getCommunicationSpi();
-
- if (ig0.configuration().isClientMode())
- continue;
-
- commSpi0.stopBlock();
- }
+ }
+ }, contCnt, "threadName");
IgniteTxManager srvTxMgr =
((IgniteEx)ig).context().cache().context().tm();
- assertTrue(GridTestUtils.waitForCondition(new GridAbsPredicate() {
- @Override public boolean apply() {
- try {
- U.invoke(IgniteTxManager.class, srvTxMgr,
"collectTxCollisionsInfo");
- }
- catch (IgniteCheckedException e) {
- fail(e.toString());
- }
-
- CacheMetrics metrics =
ig.cache(DEFAULT_CACHE_NAME).localMetrics();
+ try {
+ assertTrue(GridTestUtils.waitForCondition(new GridAbsPredicate() {
+ @Override public boolean apply() {
+ try {
+ U.invoke(IgniteTxManager.class, srvTxMgr,
"collectTxCollisionsInfo");
+ }
+ catch (IgniteCheckedException e) {
Review Comment:
We do not need this try-catch block.
##########
modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxWithKeyContentionSelfTest.java:
##########
@@ -216,120 +198,66 @@ public void
testOptimisticRepeatableReadCheckContentionTxMetricNear() throws Exc
* @param isolation Isolation level.
* @throws Exception If failed.
*/
- private void runKeyCollisionsMetric(TransactionConcurrency concurrency,
TransactionIsolation isolation) throws Exception {
+ private void runKeyCollisionsMetric(TransactionConcurrency concurrency,
TransactionIsolation isolation, boolean nearCashe)
+ throws Exception {
Ignite ig = startGridsMultiThreaded(3);
int contCnt = (int)U.staticField(IgniteTxManager.class,
"COLLISIONS_QUEUE_THRESHOLD") * 5;
- CountDownLatch txLatch = new CountDownLatch(contCnt);
-
ig.cluster().state(ClusterState.ACTIVE);
client = true;
Ignite cl = startGrid();
- IgniteTransactions cliTxMgr = cl.transactions();
-
- IgniteCache<Integer, Integer> cache = ig.cache(DEFAULT_CACHE_NAME);
-
- IgniteCache<Integer, Integer> cache0 = cl.cache(DEFAULT_CACHE_NAME);
+ IgniteCache<Integer, Integer> cache =
ig.getOrCreateCache(getCacheConfiguration(nearCashe));
final Integer keyId = primaryKey(cache);
- CountDownLatch blockOnce = new CountDownLatch(1);
-
- for (Ignite ig0 : G.allGrids()) {
- if (ig0.configuration().isClientMode())
- continue;
-
- TestRecordingCommunicationSpi commSpi0 =
-
(TestRecordingCommunicationSpi)ig0.configuration().getCommunicationSpi();
-
- commSpi0.blockMessages(new IgniteBiPredicate<ClusterNode,
Message>() {
- @Override public boolean apply(ClusterNode node, Message msg) {
- if (msg instanceof GridNearTxFinishResponse &&
blockOnce.getCount() > 0) {
- blockOnce.countDown();
-
- return true;
- }
-
- return false;
- }
- });
- }
-
- IgniteInternalFuture f = GridTestUtils.runAsync(() -> {
- try (Transaction tx = cliTxMgr.txStart(concurrency, isolation)) {
- cache0.put(keyId, 0);
- tx.commit();
- }
- });
-
- blockOnce.await();
+ IgniteTransactions transactions = cl.transactions();
- GridCompoundFuture<?, ?> finishFut = new GridCompoundFuture<>();
+ AtomicBoolean doTest = new AtomicBoolean(true);
- for (int i = 0; i < contCnt; ++i) {
- IgniteInternalFuture f0 = GridTestUtils.runAsync(() -> {
- try (Transaction tx = cliTxMgr.txStart(concurrency,
isolation)) {
- cache0.put(keyId, 0);
+ IgniteInternalFuture f0 = GridTestUtils.runMultiThreadedAsync(() -> {
+ while (doTest.get()) {
+ try (Transaction tx = transactions.txStart(concurrency,
isolation)) {
+ cache.put(keyId, 0);
tx.commit();
-
- txLatch.countDown();
}
- });
-
- finishFut.add(f0);
- }
-
- finishFut.markInitialized();
-
- for (Ignite ig0 : G.allGrids()) {
- TestRecordingCommunicationSpi commSpi0 =
-
(TestRecordingCommunicationSpi)ig0.configuration().getCommunicationSpi();
-
- if (ig0.configuration().isClientMode())
- continue;
-
- commSpi0.stopBlock();
- }
+ }
+ }, contCnt, "threadName");
Review Comment:
Move arguments to a separate lines.
##########
modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxWithKeyContentionSelfTest.java:
##########
@@ -216,120 +198,66 @@ public void
testOptimisticRepeatableReadCheckContentionTxMetricNear() throws Exc
* @param isolation Isolation level.
* @throws Exception If failed.
*/
- private void runKeyCollisionsMetric(TransactionConcurrency concurrency,
TransactionIsolation isolation) throws Exception {
+ private void runKeyCollisionsMetric(TransactionConcurrency concurrency,
TransactionIsolation isolation, boolean nearCashe)
+ throws Exception {
Ignite ig = startGridsMultiThreaded(3);
int contCnt = (int)U.staticField(IgniteTxManager.class,
"COLLISIONS_QUEUE_THRESHOLD") * 5;
- CountDownLatch txLatch = new CountDownLatch(contCnt);
-
ig.cluster().state(ClusterState.ACTIVE);
client = true;
Ignite cl = startGrid();
- IgniteTransactions cliTxMgr = cl.transactions();
-
- IgniteCache<Integer, Integer> cache = ig.cache(DEFAULT_CACHE_NAME);
-
- IgniteCache<Integer, Integer> cache0 = cl.cache(DEFAULT_CACHE_NAME);
+ IgniteCache<Integer, Integer> cache =
ig.getOrCreateCache(getCacheConfiguration(nearCashe));
final Integer keyId = primaryKey(cache);
- CountDownLatch blockOnce = new CountDownLatch(1);
-
- for (Ignite ig0 : G.allGrids()) {
- if (ig0.configuration().isClientMode())
- continue;
-
- TestRecordingCommunicationSpi commSpi0 =
-
(TestRecordingCommunicationSpi)ig0.configuration().getCommunicationSpi();
-
- commSpi0.blockMessages(new IgniteBiPredicate<ClusterNode,
Message>() {
- @Override public boolean apply(ClusterNode node, Message msg) {
- if (msg instanceof GridNearTxFinishResponse &&
blockOnce.getCount() > 0) {
- blockOnce.countDown();
-
- return true;
- }
-
- return false;
- }
- });
- }
-
- IgniteInternalFuture f = GridTestUtils.runAsync(() -> {
- try (Transaction tx = cliTxMgr.txStart(concurrency, isolation)) {
- cache0.put(keyId, 0);
- tx.commit();
- }
- });
-
- blockOnce.await();
+ IgniteTransactions transactions = cl.transactions();
- GridCompoundFuture<?, ?> finishFut = new GridCompoundFuture<>();
+ AtomicBoolean doTest = new AtomicBoolean(true);
- for (int i = 0; i < contCnt; ++i) {
- IgniteInternalFuture f0 = GridTestUtils.runAsync(() -> {
- try (Transaction tx = cliTxMgr.txStart(concurrency,
isolation)) {
- cache0.put(keyId, 0);
+ IgniteInternalFuture f0 = GridTestUtils.runMultiThreadedAsync(() -> {
+ while (doTest.get()) {
+ try (Transaction tx = transactions.txStart(concurrency,
isolation)) {
+ cache.put(keyId, 0);
tx.commit();
-
- txLatch.countDown();
}
- });
-
- finishFut.add(f0);
- }
-
- finishFut.markInitialized();
-
- for (Ignite ig0 : G.allGrids()) {
- TestRecordingCommunicationSpi commSpi0 =
-
(TestRecordingCommunicationSpi)ig0.configuration().getCommunicationSpi();
-
- if (ig0.configuration().isClientMode())
- continue;
-
- commSpi0.stopBlock();
- }
+ }
+ }, contCnt, "threadName");
IgniteTxManager srvTxMgr =
((IgniteEx)ig).context().cache().context().tm();
- assertTrue(GridTestUtils.waitForCondition(new GridAbsPredicate() {
- @Override public boolean apply() {
- try {
- U.invoke(IgniteTxManager.class, srvTxMgr,
"collectTxCollisionsInfo");
- }
- catch (IgniteCheckedException e) {
- fail(e.toString());
- }
-
- CacheMetrics metrics =
ig.cache(DEFAULT_CACHE_NAME).localMetrics();
+ try {
+ assertTrue(GridTestUtils.waitForCondition(new GridAbsPredicate() {
+ @Override public boolean apply() {
+ try {
+ U.invoke(IgniteTxManager.class, srvTxMgr,
"collectTxCollisionsInfo");
+ }
+ catch (IgniteCheckedException e) {
+ fail(e.toString());
+ }
- String coll1 = metrics.getTxKeyCollisions();
+ CacheMetrics metrics =
ig.cache(DEFAULT_CACHE_NAME).localMetrics();
- if (!coll1.isEmpty()) {
- String coll2 = metrics.getTxKeyCollisions();
+ String coll = metrics.getTxKeyCollisions();
- // check idempotent
- assertEquals(coll1, coll2);
+ if (!coll.isEmpty()) {
+ assertTrue(coll.contains("queueSize"));
- assertTrue(coll1.contains("queueSize"));
+ return true;
+ }
- return true;
- }
- else
return false;
- }
- }, 10_000));
-
- f.get();
-
- finishFut.get();
+ }
+ }, getTestTimeout()));
Review Comment:
Move arguments to a separate lines.
--
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: notifications-unsubscr...@ignite.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
