ibessonov commented on code in PR #1697: URL: https://github.com/apache/ignite-3/pull/1697#discussion_r1114124187
########## modules/storage-api/src/main/java/org/apache/ignite/internal/storage/util/ReentrantLockByRowId.java: ########## @@ -0,0 +1,175 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.internal.storage.util; + +import java.util.HashSet; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.locks.ReentrantLock; +import java.util.function.Supplier; +import org.apache.ignite.internal.storage.RowId; + +/** + * {@link ReentrantLock} by row ID. + * + * <p>Allows synchronization of version chain update operations. + */ +public class ReentrantLockByRowId { + private final ThreadLocal<Object> lockedRowIds = new ThreadLocal<>(); + + private final ConcurrentMap<RowId, LockHolder<ReentrantLock>> lockHolderByRowId = new ConcurrentHashMap<>(); + + /** + * Executes the supplier under lock by row ID. + * + * @param <T> Return type. + * @param rowId Row ID. + * @param supplier Supplier to execute under the lock. + * @return Value. + */ + public <T> T inLock(RowId rowId, Supplier<T> supplier) { + acquireLock0(rowId); + + try { + return supplier.get(); + } finally { + releaseLock0(rowId, false); + } + } + + /** + * Acquires the lock by row ID. + * + * @param rowId Row ID. + */ + public void acquireLock(RowId rowId) { + acquireLock0(rowId); + + Object lockedRowIds = this.lockedRowIds.get(); + + if (lockedRowIds == null) { + this.lockedRowIds.set(rowId); + } else if (lockedRowIds.getClass() == RowId.class) { + RowId lockedRowId = (RowId) lockedRowIds; + + if (!lockedRowId.equals(rowId)) { + Set<RowId> rowIds = new HashSet<>(); + + rowIds.add(lockedRowId); + rowIds.add(rowId); + + this.lockedRowIds.set(rowIds); + } + } else { + ((Set<RowId>) lockedRowIds).add(rowId); + } + } + + /** + * Releases the lock by row ID. + * + * @param rowId Row ID. + * @throws IllegalStateException If the lock could not be found by row ID. + * @throws IllegalMonitorStateException If the current thread does not hold this lock. + */ + public void releaseLock(RowId rowId) { + releaseLock0(rowId, false); + + LockHolder<ReentrantLock> lockHolder = lockHolderByRowId.get(rowId); + + if (lockHolder != null && lockHolder.getLock().isHeldByCurrentThread()) { + return; + } + + Object lockedRowIds = this.lockedRowIds.get(); + + assert lockedRowIds != null : rowId; + + if (lockedRowIds.getClass() == RowId.class) { + RowId lockedRowId = (RowId) lockedRowIds; + + assert lockedRowId.equals(rowId) : "rowId=" + rowId + ", lockedRowId=" + lockedRowId; + + this.lockedRowIds.remove(); + } else { + Set<RowId> rowIds = ((Set<RowId>) lockedRowIds); + + boolean remove = rowIds.remove(rowId); + + assert remove : "rowId=" + rowId + ", lockedRowIds=" + rowIds; + + if (rowIds.isEmpty()) { + this.lockedRowIds.remove(); + } + } + } + + /** + * Releases all locks {@link #acquireLock(RowId) acquired} by the current thread if exists. + * + * <p>Order of releasing the locks is not defined, each lock will be released with all re-entries. + */ + public void releaseAllLockByCurrentThread() { + Object lockedRowIds = this.lockedRowIds.get(); + + this.lockedRowIds.remove(); + + if (lockedRowIds == null) { + return; + } else if (lockedRowIds.getClass() == RowId.class) { + releaseLock0(((RowId) lockedRowIds), true); + } else { + ((Set<RowId>) lockedRowIds).forEach(rowId -> releaseLock0(rowId, true)); Review Comment: Why "forEach" and not a loop? I thought that we don't use such constructions in hot code ########## modules/storage-api/src/main/java/org/apache/ignite/internal/storage/util/ReentrantLockByRowId.java: ########## @@ -0,0 +1,175 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.internal.storage.util; + +import java.util.HashSet; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.locks.ReentrantLock; +import java.util.function.Supplier; +import org.apache.ignite.internal.storage.RowId; + +/** + * {@link ReentrantLock} by row ID. + * + * <p>Allows synchronization of version chain update operations. + */ +public class ReentrantLockByRowId { + private final ThreadLocal<Object> lockedRowIds = new ThreadLocal<>(); + + private final ConcurrentMap<RowId, LockHolder<ReentrantLock>> lockHolderByRowId = new ConcurrentHashMap<>(); + + /** + * Executes the supplier under lock by row ID. + * + * @param <T> Return type. + * @param rowId Row ID. + * @param supplier Supplier to execute under the lock. + * @return Value. + */ + public <T> T inLock(RowId rowId, Supplier<T> supplier) { + acquireLock0(rowId); + + try { + return supplier.get(); + } finally { + releaseLock0(rowId, false); + } + } + + /** + * Acquires the lock by row ID. + * + * @param rowId Row ID. + */ + public void acquireLock(RowId rowId) { + acquireLock0(rowId); + + Object lockedRowIds = this.lockedRowIds.get(); + + if (lockedRowIds == null) { + this.lockedRowIds.set(rowId); + } else if (lockedRowIds.getClass() == RowId.class) { Review Comment: Please replace it with `lockedRowIds instanceof RowId`. Please do the same in other method that uses similar construction ########## modules/storage-api/src/test/java/org/apache/ignite/internal/storage/util/ReentrantLockByRowIdTest.java: ########## @@ -0,0 +1,156 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.internal.storage.util; + +import static org.apache.ignite.internal.testframework.IgniteTestUtils.runAsync; +import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willCompleteSuccessfully; +import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willFailFast; +import static org.hamcrest.MatcherAssert.assertThat; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeoutException; +import org.apache.ignite.internal.storage.RowId; +import org.junit.jupiter.api.Test; + +/** + * Class for testing {@link ReentrantLockByRowId}. + */ +public class ReentrantLockByRowIdTest { + @Test + void testSimple() { + ReentrantLockByRowId lockByRowId = new ReentrantLockByRowId(); + + RowId rowId = new RowId(0); + + lockByRowId.acquireLock(rowId); + lockByRowId.releaseLock(rowId); + + assertEquals(1, lockByRowId.inLock(rowId, () -> 1)); + } + + @Test + void testSimpleReEnter() { Review Comment: I see no assertions in this test ########## modules/storage-api/src/main/java/org/apache/ignite/internal/storage/util/ReentrantLockByRowId.java: ########## @@ -0,0 +1,175 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.internal.storage.util; + +import java.util.HashSet; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.locks.ReentrantLock; +import java.util.function.Supplier; +import org.apache.ignite.internal.storage.RowId; + +/** + * {@link ReentrantLock} by row ID. + * + * <p>Allows synchronization of version chain update operations. + */ +public class ReentrantLockByRowId { + private final ThreadLocal<Object> lockedRowIds = new ThreadLocal<>(); + + private final ConcurrentMap<RowId, LockHolder<ReentrantLock>> lockHolderByRowId = new ConcurrentHashMap<>(); + + /** + * Executes the supplier under lock by row ID. + * + * @param <T> Return type. + * @param rowId Row ID. + * @param supplier Supplier to execute under the lock. + * @return Value. + */ + public <T> T inLock(RowId rowId, Supplier<T> supplier) { + acquireLock0(rowId); + + try { + return supplier.get(); + } finally { + releaseLock0(rowId, false); + } + } + + /** + * Acquires the lock by row ID. + * + * @param rowId Row ID. + */ + public void acquireLock(RowId rowId) { + acquireLock0(rowId); + + Object lockedRowIds = this.lockedRowIds.get(); + + if (lockedRowIds == null) { + this.lockedRowIds.set(rowId); + } else if (lockedRowIds.getClass() == RowId.class) { + RowId lockedRowId = (RowId) lockedRowIds; + + if (!lockedRowId.equals(rowId)) { + Set<RowId> rowIds = new HashSet<>(); + + rowIds.add(lockedRowId); + rowIds.add(rowId); + + this.lockedRowIds.set(rowIds); + } + } else { + ((Set<RowId>) lockedRowIds).add(rowId); + } + } + + /** + * Releases the lock by row ID. + * + * @param rowId Row ID. + * @throws IllegalStateException If the lock could not be found by row ID. + * @throws IllegalMonitorStateException If the current thread does not hold this lock. + */ + public void releaseLock(RowId rowId) { + releaseLock0(rowId, false); + + LockHolder<ReentrantLock> lockHolder = lockHolderByRowId.get(rowId); + + if (lockHolder != null && lockHolder.getLock().isHeldByCurrentThread()) { + return; + } + + Object lockedRowIds = this.lockedRowIds.get(); + + assert lockedRowIds != null : rowId; + + if (lockedRowIds.getClass() == RowId.class) { + RowId lockedRowId = (RowId) lockedRowIds; + + assert lockedRowId.equals(rowId) : "rowId=" + rowId + ", lockedRowId=" + lockedRowId; + + this.lockedRowIds.remove(); + } else { + Set<RowId> rowIds = ((Set<RowId>) lockedRowIds); + + boolean remove = rowIds.remove(rowId); + + assert remove : "rowId=" + rowId + ", lockedRowIds=" + rowIds; + + if (rowIds.isEmpty()) { + this.lockedRowIds.remove(); Review Comment: Would it be better to use `set(null)` instead of `remove`? The latter might be noticeably slower ########## modules/storage-api/src/main/java/org/apache/ignite/internal/storage/util/ReentrantLockByRowId.java: ########## @@ -0,0 +1,175 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.internal.storage.util; + +import java.util.HashSet; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.locks.ReentrantLock; +import java.util.function.Supplier; +import org.apache.ignite.internal.storage.RowId; + +/** + * {@link ReentrantLock} by row ID. + * + * <p>Allows synchronization of version chain update operations. + */ +public class ReentrantLockByRowId { + private final ThreadLocal<Object> lockedRowIds = new ThreadLocal<>(); + + private final ConcurrentMap<RowId, LockHolder<ReentrantLock>> lockHolderByRowId = new ConcurrentHashMap<>(); + + /** + * Executes the supplier under lock by row ID. + * + * @param <T> Return type. + * @param rowId Row ID. + * @param supplier Supplier to execute under the lock. + * @return Value. + */ + public <T> T inLock(RowId rowId, Supplier<T> supplier) { + acquireLock0(rowId); + + try { + return supplier.get(); + } finally { + releaseLock0(rowId, false); + } + } + + /** + * Acquires the lock by row ID. + * + * @param rowId Row ID. + */ + public void acquireLock(RowId rowId) { + acquireLock0(rowId); + + Object lockedRowIds = this.lockedRowIds.get(); + + if (lockedRowIds == null) { + this.lockedRowIds.set(rowId); + } else if (lockedRowIds.getClass() == RowId.class) { + RowId lockedRowId = (RowId) lockedRowIds; + + if (!lockedRowId.equals(rowId)) { + Set<RowId> rowIds = new HashSet<>(); + + rowIds.add(lockedRowId); + rowIds.add(rowId); + + this.lockedRowIds.set(rowIds); + } + } else { + ((Set<RowId>) lockedRowIds).add(rowId); + } + } + + /** + * Releases the lock by row ID. + * + * @param rowId Row ID. + * @throws IllegalStateException If the lock could not be found by row ID. + * @throws IllegalMonitorStateException If the current thread does not hold this lock. + */ + public void releaseLock(RowId rowId) { + releaseLock0(rowId, false); + + LockHolder<ReentrantLock> lockHolder = lockHolderByRowId.get(rowId); + + if (lockHolder != null && lockHolder.getLock().isHeldByCurrentThread()) { + return; + } + + Object lockedRowIds = this.lockedRowIds.get(); + + assert lockedRowIds != null : rowId; + + if (lockedRowIds.getClass() == RowId.class) { + RowId lockedRowId = (RowId) lockedRowIds; + + assert lockedRowId.equals(rowId) : "rowId=" + rowId + ", lockedRowId=" + lockedRowId; + + this.lockedRowIds.remove(); + } else { + Set<RowId> rowIds = ((Set<RowId>) lockedRowIds); + + boolean remove = rowIds.remove(rowId); + + assert remove : "rowId=" + rowId + ", lockedRowIds=" + rowIds; + + if (rowIds.isEmpty()) { + this.lockedRowIds.remove(); + } + } + } + + /** + * Releases all locks {@link #acquireLock(RowId) acquired} by the current thread if exists. + * + * <p>Order of releasing the locks is not defined, each lock will be released with all re-entries. + */ + public void releaseAllLockByCurrentThread() { + Object lockedRowIds = this.lockedRowIds.get(); + + this.lockedRowIds.remove(); + + if (lockedRowIds == null) { + return; + } else if (lockedRowIds.getClass() == RowId.class) { + releaseLock0(((RowId) lockedRowIds), true); + } else { + ((Set<RowId>) lockedRowIds).forEach(rowId -> releaseLock0(rowId, true)); + } + } + + private void acquireLock0(RowId rowId) { + LockHolder<ReentrantLock> lockHolder = lockHolderByRowId.compute(rowId, (rowId1, reentrantLockLockHolder) -> { + if (reentrantLockLockHolder == null) { + reentrantLockLockHolder = new LockHolder<>(new ReentrantLock()); + } + + reentrantLockLockHolder.incrementHolders(); + + return reentrantLockLockHolder; + }); + + lockHolder.getLock().lock(); + } + + private void releaseLock0(RowId rowId, boolean untilHoldByCurrentThread) { + LockHolder<ReentrantLock> lockHolder = lockHolderByRowId.get(rowId); + + if (lockHolder == null) { + throw new IllegalStateException("Could not find lock by row ID: " + rowId); + } + + ReentrantLock lock = lockHolder.getLock(); + + do { + lock.unlock(); + + lockHolderByRowId.compute(rowId, (rowId1, reentrantLockLockHolder) -> { + assert reentrantLockLockHolder != null; + + return reentrantLockLockHolder.decrementHolders() ? null : reentrantLockLockHolder; + }); + } while (untilHoldByCurrentThread && lock.getHoldCount() > 0); Review Comment: You don't remove the holder from the map, is this intentional? ########## modules/storage-api/src/main/java/org/apache/ignite/internal/storage/util/ReentrantLockByRowId.java: ########## @@ -0,0 +1,175 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.internal.storage.util; + +import java.util.HashSet; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.locks.ReentrantLock; +import java.util.function.Supplier; +import org.apache.ignite.internal.storage.RowId; + +/** + * {@link ReentrantLock} by row ID. + * + * <p>Allows synchronization of version chain update operations. + */ +public class ReentrantLockByRowId { + private final ThreadLocal<Object> lockedRowIds = new ThreadLocal<>(); + + private final ConcurrentMap<RowId, LockHolder<ReentrantLock>> lockHolderByRowId = new ConcurrentHashMap<>(); + + /** + * Executes the supplier under lock by row ID. + * + * @param <T> Return type. + * @param rowId Row ID. + * @param supplier Supplier to execute under the lock. + * @return Value. + */ + public <T> T inLock(RowId rowId, Supplier<T> supplier) { + acquireLock0(rowId); + + try { + return supplier.get(); + } finally { + releaseLock0(rowId, false); + } + } + + /** + * Acquires the lock by row ID. + * + * @param rowId Row ID. + */ + public void acquireLock(RowId rowId) { + acquireLock0(rowId); + + Object lockedRowIds = this.lockedRowIds.get(); + + if (lockedRowIds == null) { + this.lockedRowIds.set(rowId); + } else if (lockedRowIds.getClass() == RowId.class) { + RowId lockedRowId = (RowId) lockedRowIds; + + if (!lockedRowId.equals(rowId)) { + Set<RowId> rowIds = new HashSet<>(); + + rowIds.add(lockedRowId); + rowIds.add(rowId); + + this.lockedRowIds.set(rowIds); + } + } else { + ((Set<RowId>) lockedRowIds).add(rowId); + } + } + + /** + * Releases the lock by row ID. + * + * @param rowId Row ID. + * @throws IllegalStateException If the lock could not be found by row ID. + * @throws IllegalMonitorStateException If the current thread does not hold this lock. + */ + public void releaseLock(RowId rowId) { + releaseLock0(rowId, false); + + LockHolder<ReentrantLock> lockHolder = lockHolderByRowId.get(rowId); + + if (lockHolder != null && lockHolder.getLock().isHeldByCurrentThread()) { + return; + } + + Object lockedRowIds = this.lockedRowIds.get(); + + assert lockedRowIds != null : rowId; + + if (lockedRowIds.getClass() == RowId.class) { + RowId lockedRowId = (RowId) lockedRowIds; + + assert lockedRowId.equals(rowId) : "rowId=" + rowId + ", lockedRowId=" + lockedRowId; + + this.lockedRowIds.remove(); + } else { + Set<RowId> rowIds = ((Set<RowId>) lockedRowIds); + + boolean remove = rowIds.remove(rowId); + + assert remove : "rowId=" + rowId + ", lockedRowIds=" + rowIds; + + if (rowIds.isEmpty()) { + this.lockedRowIds.remove(); + } + } + } + + /** + * Releases all locks {@link #acquireLock(RowId) acquired} by the current thread if exists. + * + * <p>Order of releasing the locks is not defined, each lock will be released with all re-entries. + */ + public void releaseAllLockByCurrentThread() { + Object lockedRowIds = this.lockedRowIds.get(); + + this.lockedRowIds.remove(); + + if (lockedRowIds == null) { + return; + } else if (lockedRowIds.getClass() == RowId.class) { + releaseLock0(((RowId) lockedRowIds), true); + } else { + ((Set<RowId>) lockedRowIds).forEach(rowId -> releaseLock0(rowId, true)); + } + } + + private void acquireLock0(RowId rowId) { + LockHolder<ReentrantLock> lockHolder = lockHolderByRowId.compute(rowId, (rowId1, reentrantLockLockHolder) -> { + if (reentrantLockLockHolder == null) { + reentrantLockLockHolder = new LockHolder<>(new ReentrantLock()); + } + + reentrantLockLockHolder.incrementHolders(); + + return reentrantLockLockHolder; + }); + + lockHolder.getLock().lock(); + } + + private void releaseLock0(RowId rowId, boolean untilHoldByCurrentThread) { Review Comment: Name `untilHoldByCurrentThread` feels complicated, maybe there's another name that you could give to this parameter. ########## modules/storage-api/src/testFixtures/java/org/apache/ignite/internal/storage/AbstractMvPartitionStorageConcurrencyTest.java: ########## @@ -193,19 +202,39 @@ void testConcurrentGc(AddAndCommit addAndCommit) { for (int i = 0; i < REPEATS; i++) { addAndCommit.perform(this, TABLE_ROW); + addAndCommit.perform(this, TABLE_ROW2); + addAndCommit.perform(this, null); + Collection<byte[]> rows = toRowBytes(TABLE_ROW, TABLE_ROW2); + runRace( - () -> pollForVacuum(HybridTimestamp.MAX_VALUE), - () -> pollForVacuum(HybridTimestamp.MAX_VALUE), - () -> pollForVacuum(HybridTimestamp.MAX_VALUE), - () -> pollForVacuum(HybridTimestamp.MAX_VALUE) + () -> assertRemoveRow(pollForVacuum(HybridTimestamp.MAX_VALUE).binaryRow().bytes(), rows), + () -> assertRemoveRow(pollForVacuum(HybridTimestamp.MAX_VALUE).binaryRow().bytes(), rows) ); + assertNull(pollForVacuum(HybridTimestamp.MAX_VALUE)); + assertNull(storage.closestRowId(ROW_ID)); + + assertThat(rows, empty()); } } + private void assertRemoveRow(byte[] rowBytes, Collection<byte[]> rows) { + assertNotNull(rowBytes); + + byte[] found = rows.stream().filter(bytes -> Arrays.equals(bytes, rowBytes)).findFirst().orElseGet(null); Review Comment: You could have used byte buffers instead, they provide the default "equals" implementation ########## modules/storage-api/src/test/java/org/apache/ignite/internal/storage/util/ReentrantLockByRowIdTest.java: ########## @@ -0,0 +1,156 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.internal.storage.util; + +import static org.apache.ignite.internal.testframework.IgniteTestUtils.runAsync; +import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willCompleteSuccessfully; +import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willFailFast; +import static org.hamcrest.MatcherAssert.assertThat; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeoutException; +import org.apache.ignite.internal.storage.RowId; +import org.junit.jupiter.api.Test; + +/** + * Class for testing {@link ReentrantLockByRowId}. + */ +public class ReentrantLockByRowIdTest { + @Test + void testSimple() { + ReentrantLockByRowId lockByRowId = new ReentrantLockByRowId(); + + RowId rowId = new RowId(0); + + lockByRowId.acquireLock(rowId); + lockByRowId.releaseLock(rowId); + + assertEquals(1, lockByRowId.inLock(rowId, () -> 1)); + } + + @Test + void testSimpleReEnter() { + ReentrantLockByRowId lockByRowId = new ReentrantLockByRowId(); + + RowId rowId = new RowId(0); + + lockByRowId.acquireLock(rowId); + lockByRowId.acquireLock(rowId); + + lockByRowId.inLock(rowId, () -> { + lockByRowId.acquireLock(rowId); + + lockByRowId.releaseLock(rowId); + + return null; + }); + + lockByRowId.releaseLock(rowId); + lockByRowId.releaseLock(rowId); + } + + @Test + void testReleaseError() { + ReentrantLockByRowId lockByRowId = new ReentrantLockByRowId(); + + assertThrows(IllegalStateException.class, () -> lockByRowId.releaseLock(new RowId(0))); + + RowId rowId = new RowId(0); + + assertThat(runAsync(() -> lockByRowId.acquireLock(rowId)), willCompleteSuccessfully()); + + assertThrows(IllegalMonitorStateException.class, () -> lockByRowId.releaseLock(rowId)); + } + + @Test + void testBlockSimple() { + ReentrantLockByRowId lockByRowId = new ReentrantLockByRowId(); + + RowId rowId = new RowId(0); + + lockByRowId.acquireLock(rowId); + lockByRowId.acquireLock(rowId); + + CompletableFuture<?> acquireLockFuture = runAsync(() -> { + lockByRowId.acquireLock(rowId); + lockByRowId.releaseLock(rowId); + }); + + assertThat(acquireLockFuture, willFailFast(TimeoutException.class)); + + lockByRowId.releaseLock(rowId); + + assertThat(acquireLockFuture, willFailFast(TimeoutException.class)); + + lockByRowId.releaseLock(rowId); + + assertThat(acquireLockFuture, willCompleteSuccessfully()); + + lockByRowId.acquireLock(rowId); Review Comment: Why do you re-acquire the lock? Also, why don't you use try/finally blocks in your tests? ########## modules/storage-api/src/test/java/org/apache/ignite/internal/storage/util/ReentrantLockByRowIdTest.java: ########## @@ -0,0 +1,156 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.internal.storage.util; + +import static org.apache.ignite.internal.testframework.IgniteTestUtils.runAsync; +import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willCompleteSuccessfully; +import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willFailFast; +import static org.hamcrest.MatcherAssert.assertThat; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeoutException; +import org.apache.ignite.internal.storage.RowId; +import org.junit.jupiter.api.Test; + +/** + * Class for testing {@link ReentrantLockByRowId}. + */ +public class ReentrantLockByRowIdTest { + @Test + void testSimple() { + ReentrantLockByRowId lockByRowId = new ReentrantLockByRowId(); + + RowId rowId = new RowId(0); + + lockByRowId.acquireLock(rowId); + lockByRowId.releaseLock(rowId); + + assertEquals(1, lockByRowId.inLock(rowId, () -> 1)); + } + + @Test + void testSimpleReEnter() { + ReentrantLockByRowId lockByRowId = new ReentrantLockByRowId(); + + RowId rowId = new RowId(0); + + lockByRowId.acquireLock(rowId); + lockByRowId.acquireLock(rowId); + + lockByRowId.inLock(rowId, () -> { + lockByRowId.acquireLock(rowId); + + lockByRowId.releaseLock(rowId); + + return null; + }); + + lockByRowId.releaseLock(rowId); + lockByRowId.releaseLock(rowId); + } + + @Test + void testReleaseError() { + ReentrantLockByRowId lockByRowId = new ReentrantLockByRowId(); + + assertThrows(IllegalStateException.class, () -> lockByRowId.releaseLock(new RowId(0))); + + RowId rowId = new RowId(0); + + assertThat(runAsync(() -> lockByRowId.acquireLock(rowId)), willCompleteSuccessfully()); + + assertThrows(IllegalMonitorStateException.class, () -> lockByRowId.releaseLock(rowId)); + } + + @Test + void testBlockSimple() { + ReentrantLockByRowId lockByRowId = new ReentrantLockByRowId(); + + RowId rowId = new RowId(0); + + lockByRowId.acquireLock(rowId); + lockByRowId.acquireLock(rowId); + + CompletableFuture<?> acquireLockFuture = runAsync(() -> { + lockByRowId.acquireLock(rowId); + lockByRowId.releaseLock(rowId); + }); + + assertThat(acquireLockFuture, willFailFast(TimeoutException.class)); + + lockByRowId.releaseLock(rowId); + + assertThat(acquireLockFuture, willFailFast(TimeoutException.class)); Review Comment: It's not "willFailFast", it's "wontCompleteWithinShortTimeout" (but I'd prefer something shorter) -- 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]
