You are viewing a plain text version of this content. The canonical link for it is here.
Posted to pr@cassandra.apache.org by "mike-tr-adamson (via GitHub)" <gi...@apache.org> on 2023/07/13 16:07:29 UTC
[GitHub] [cassandra] mike-tr-adamson opened a new pull request, #2487: Fix concurrency in bbtree reader by cloning state
mike-tr-adamson opened a new pull request, #2487:
URL: https://github.com/apache/cassandra/pull/2487
This fixes a concurrency issue with the `BlockBalancedTreeReader`. We incorrectly made the `TraversalState` a member of the `BlockBalancedTreeWalker` such that the same instance of `TraversalState` was used for all queries. This didn't work because the `TraversalState` isn't threadsafe.
This fix separates the `TraversalState` from the `BlockBalancedTreeWalker` and creates a new `TraversalState` for each usage. This is a lightweight operation because the actual packed index is still being read when the walker is created. The `TraversalState` now only handles the mutable state of the traversal.
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
[GitHub] [cassandra] maedhroz commented on a diff in pull request #2487: Fix concurrency in bbtree reader by cloning state
Posted by "maedhroz (via GitHub)" <gi...@apache.org>.
maedhroz commented on code in PR #2487:
URL: https://github.com/apache/cassandra/pull/2487#discussion_r1262966107
##########
src/java/org/apache/cassandra/index/sai/disk/v1/bbtree/BlockBalancedTreeWalker.java:
##########
@@ -46,7 +48,15 @@
public class BlockBalancedTreeWalker implements Closeable
{
final FileHandle treeIndexFile;
- final TraversalState state;
+ final int bytesPerValue;
+ final int numLeaves;
+ final int treeDepth;
+ final byte[] minPackedValue;
+ final byte[] maxPackedValue;
+ final long valueCount;
+ final int maxPointsInLeafNode;
Review Comment:
nit: `maxValuesInLeafNode`?
(can change on commit)
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
[GitHub] [cassandra] maedhroz commented on a diff in pull request #2487: Fix concurrency in bbtree reader by cloning state
Posted by "maedhroz (via GitHub)" <gi...@apache.org>.
maedhroz commented on code in PR #2487:
URL: https://github.com/apache/cassandra/pull/2487#discussion_r1264075164
##########
test/unit/org/apache/cassandra/index/sai/disk/v1/bbtree/BlockBalancedTreeReaderTest.java:
##########
@@ -239,38 +236,39 @@ public void testConcurrentIntersectionsOnSameReader() throws Exception
final BlockBalancedTreeRamBuffer buffer = new BlockBalancedTreeRamBuffer(Integer.BYTES);
byte[] scratch = new byte[4];
- for (int docID = 0; docID < numRows; docID++)
+ for (int rowID = 0; rowID < numRows; rowID++)
{
- NumericUtils.intToSortableBytes(docID, scratch, 0);
- buffer.add(docID, scratch);
+ NumericUtils.intToSortableBytes(rowID, scratch, 0);
+ buffer.add(rowID, scratch);
}
- final BlockBalancedTreeReader reader = finishAndOpenReader(4, buffer);
-
- Expression expression = new Expression(indexContext);
- expression.add(Operator.GT, Int32Type.instance.decompose(444));
- expression.add(Operator.LT, Int32Type.instance.decompose(555));
-
- int concurrency = 100;
-
- ExecutorService executor = Executors.newFixedThreadPool(concurrency);
- List<Future<?>> results = new ArrayList<>();
- for (int thread = 0; thread < concurrency; thread++)
+ try (BlockBalancedTreeReader reader = finishAndOpenReader(4, buffer))
{
- results.add(executor.submit(() -> assertIntersection(reader, expression)));
+ int concurrency = 100;
+
+ ExecutorService executor = Executors.newFixedThreadPool(concurrency);
+ List<Future<?>> results = new ArrayList<>();
+ for (int thread = 0; thread < concurrency; thread++)
+ {
+ results.add(executor.submit(() -> assertRange(reader, 445, 555)));
+ }
+ FBUtilities.waitOnFutures(results);
+ executor.shutdown();
}
- FBUtilities.waitOnFutures(results);
- executor.shutdown();
}
- private void assertIntersection(BlockBalancedTreeReader reader, Expression expression)
+ private void assertRange(BlockBalancedTreeReader reader, long lowerBound, long upperBound)
{
+ Expression expression = new Expression(indexContext);
+ expression.add(Operator.GT, Int32Type.instance.decompose(444));
+ expression.add(Operator.LT, Int32Type.instance.decompose(555));
+
try
{
PostingList intersection = performIntersection(reader, BlockBalancedTreeQueries.balancedTreeQueryFrom(expression, 4));
assertNotNull(intersection);
assertEquals(110, intersection.size());
Review Comment:
nit: `upperBound - lowerBound` instead of hard-coding 110?
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
[GitHub] [cassandra] maedhroz commented on a diff in pull request #2487: Fix concurrency in bbtree reader by cloning state
Posted by "maedhroz (via GitHub)" <gi...@apache.org>.
maedhroz commented on code in PR #2487:
URL: https://github.com/apache/cassandra/pull/2487#discussion_r1262969321
##########
src/java/org/apache/cassandra/index/sai/disk/v1/bbtree/BlockBalancedTreeWalker.java:
##########
@@ -170,60 +242,18 @@ final static class TraversalState
@VisibleForTesting
int maxLevel;
- TraversalState(DataInput dataInput) throws IOException
+ private TraversalState()
{
- maxPointsInLeafNode = dataInput.readVInt();
- bytesPerValue = dataInput.readVInt();
-
- // Read index:
- numLeaves = dataInput.readVInt();
- assert numLeaves > 0;
- treeDepth = dataInput.readVInt();
- minPackedValue = new byte[bytesPerValue];
- maxPackedValue = new byte[bytesPerValue];
-
- dataInput.readBytes(minPackedValue, 0, bytesPerValue);
- dataInput.readBytes(maxPackedValue, 0, bytesPerValue);
-
- if (ByteArrayUtil.compareUnsigned(minPackedValue, 0, maxPackedValue, 0, bytesPerValue) > 0)
- {
- String message = String.format("Min packed value %s is > max packed value %s.",
- new BytesRef(minPackedValue), new BytesRef(maxPackedValue));
- throw new CorruptIndexException(message, dataInput);
- }
-
- valueCount = dataInput.readVLong();
-
- int numBytes = dataInput.readVInt();
- byte[] packedIndex = new byte[numBytes];
- dataInput.readBytes(packedIndex, 0, numBytes);
-
nodeID = 1;
level = 0;
leafBlockFPStack = new long[treeDepth];
leftNodePositions = new int[treeDepth];
rightNodePositions = new int[treeDepth];
splitValuesStack = new byte[treeDepth][];
-
- memoryUsage = ObjectSizes.sizeOfArray(packedIndex) +
- ObjectSizes.sizeOfArray(minPackedValue) +
- ObjectSizes.sizeOfArray(maxPackedValue) +
- ObjectSizes.sizeOfArray(leafBlockFPStack) +
- ObjectSizes.sizeOfArray(leftNodePositions) +
- ObjectSizes.sizeOfArray(rightNodePositions) +
- ObjectSizes.sizeOfArray(splitValuesStack) * bytesPerValue;
Review Comment:
So we're no longer counting the `TraversalState` items in `memoryUsage` in the walker, but that's okay because the bits in state are transient/query-bound anyway?
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
[GitHub] [cassandra] maedhroz commented on a diff in pull request #2487: Fix concurrency in bbtree reader by cloning state
Posted by "maedhroz (via GitHub)" <gi...@apache.org>.
maedhroz commented on code in PR #2487:
URL: https://github.com/apache/cassandra/pull/2487#discussion_r1262963910
##########
test/unit/org/apache/cassandra/index/sai/disk/v1/bbtree/BlockBalancedTreeReaderTest.java:
##########
@@ -222,6 +231,55 @@ public void testResourcesReleaseWhenQueryDoesntMatchAnything() throws Exception
assertNull(intersection);
}
+ @Test
+ public void testConcurrentIntersectionsOnSameReader() throws Exception
+ {
+ int numRows = 1000;
+
+ final BlockBalancedTreeRamBuffer buffer = new BlockBalancedTreeRamBuffer(Integer.BYTES);
+
+ byte[] scratch = new byte[4];
+ for (int docID = 0; docID < numRows; docID++)
+ {
+ NumericUtils.intToSortableBytes(docID, scratch, 0);
+ buffer.add(docID, scratch);
+ }
+
+ final BlockBalancedTreeReader reader = finishAndOpenReader(4, buffer);
+
+ Expression expression = new Expression(indexContext);
+ expression.add(Operator.GT, Int32Type.instance.decompose(444));
+ expression.add(Operator.LT, Int32Type.instance.decompose(555));
+
+ int concurrency = 100;
+
+ ExecutorService executor = Executors.newFixedThreadPool(concurrency);
+ List<Future<?>> results = new ArrayList<>();
+ for (int thread = 0; thread < concurrency; thread++)
+ {
+ results.add(executor.submit(() -> assertIntersection(reader, expression)));
+ }
+ FBUtilities.waitOnFutures(results);
+ executor.shutdown();
+ }
+
+ private void assertIntersection(BlockBalancedTreeReader reader, Expression expression)
Review Comment:
nit: I might clean this up to be like `assertRange()` or something where we pass 445 and 555 as inclusive and exclusive bounds, etc.
(can fix on commit)
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
[GitHub] [cassandra] mike-tr-adamson commented on a diff in pull request #2487: Fix concurrency in bbtree reader by cloning state
Posted by "mike-tr-adamson (via GitHub)" <gi...@apache.org>.
mike-tr-adamson commented on code in PR #2487:
URL: https://github.com/apache/cassandra/pull/2487#discussion_r1263528563
##########
src/java/org/apache/cassandra/index/sai/disk/v1/bbtree/BlockBalancedTreeWalker.java:
##########
@@ -170,60 +242,18 @@ final static class TraversalState
@VisibleForTesting
int maxLevel;
- TraversalState(DataInput dataInput) throws IOException
+ private TraversalState()
{
- maxPointsInLeafNode = dataInput.readVInt();
- bytesPerValue = dataInput.readVInt();
-
- // Read index:
- numLeaves = dataInput.readVInt();
- assert numLeaves > 0;
- treeDepth = dataInput.readVInt();
- minPackedValue = new byte[bytesPerValue];
- maxPackedValue = new byte[bytesPerValue];
-
- dataInput.readBytes(minPackedValue, 0, bytesPerValue);
- dataInput.readBytes(maxPackedValue, 0, bytesPerValue);
-
- if (ByteArrayUtil.compareUnsigned(minPackedValue, 0, maxPackedValue, 0, bytesPerValue) > 0)
- {
- String message = String.format("Min packed value %s is > max packed value %s.",
- new BytesRef(minPackedValue), new BytesRef(maxPackedValue));
- throw new CorruptIndexException(message, dataInput);
- }
-
- valueCount = dataInput.readVLong();
-
- int numBytes = dataInput.readVInt();
- byte[] packedIndex = new byte[numBytes];
- dataInput.readBytes(packedIndex, 0, numBytes);
-
nodeID = 1;
level = 0;
leafBlockFPStack = new long[treeDepth];
leftNodePositions = new int[treeDepth];
rightNodePositions = new int[treeDepth];
splitValuesStack = new byte[treeDepth][];
-
- memoryUsage = ObjectSizes.sizeOfArray(packedIndex) +
- ObjectSizes.sizeOfArray(minPackedValue) +
- ObjectSizes.sizeOfArray(maxPackedValue) +
- ObjectSizes.sizeOfArray(leafBlockFPStack) +
- ObjectSizes.sizeOfArray(leftNodePositions) +
- ObjectSizes.sizeOfArray(rightNodePositions) +
- ObjectSizes.sizeOfArray(splitValuesStack) * bytesPerValue;
Review Comment:
Exactly that. We are only meant to report the memory used by the long-term objects, which in this case is the `packedIndex`.
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
[GitHub] [cassandra] adelapena commented on a diff in pull request #2487: Fix concurrency in bbtree reader by cloning state
Posted by "adelapena (via GitHub)" <gi...@apache.org>.
adelapena commented on code in PR #2487:
URL: https://github.com/apache/cassandra/pull/2487#discussion_r1263640840
##########
test/unit/org/apache/cassandra/index/sai/disk/v1/bbtree/BlockBalancedTreeReaderTest.java:
##########
@@ -162,66 +166,118 @@ public void testSameValuesInLeaf() throws Exception
final BlockBalancedTreeRamBuffer buffer = new BlockBalancedTreeRamBuffer(Integer.BYTES);
byte[] scratch = new byte[4];
- for (int docID = 0; docID < 10; docID++)
+ for (int rowID = 0; rowID < 10; rowID++)
{
- NumericUtils.intToSortableBytes(docID, scratch, 0);
- buffer.add(docID, scratch);
+ NumericUtils.intToSortableBytes(rowID, scratch, 0);
+ buffer.add(rowID, scratch);
}
- for (int docID = 10; docID < 20; docID++)
+ for (int rowID = 10; rowID < 20; rowID++)
{
NumericUtils.intToSortableBytes(10, scratch, 0);
- buffer.add(docID, scratch);
+ buffer.add(rowID, scratch);
}
- for (int docID = 20; docID < 30; docID++)
+ for (int rowID = 20; rowID < 30; rowID++)
{
- NumericUtils.intToSortableBytes(docID, scratch, 0);
- buffer.add(docID, scratch);
+ NumericUtils.intToSortableBytes(rowID, scratch, 0);
+ buffer.add(rowID, scratch);
}
- final BlockBalancedTreeReader reader = finishAndOpenReader(5, buffer);
-
- PostingList postingList = performIntersection(reader, buildQuery(8, 15));
-
- assertEquals(8, postingList.nextPosting());
- assertEquals(9, postingList.nextPosting());
- assertEquals(10, postingList.nextPosting());
- assertEquals(11, postingList.nextPosting());
- assertEquals(12, postingList.nextPosting());
- assertEquals(13, postingList.nextPosting());
- assertEquals(14, postingList.nextPosting());
- assertEquals(15, postingList.nextPosting());
- assertEquals(16, postingList.nextPosting());
- assertEquals(17, postingList.nextPosting());
- assertEquals(18, postingList.nextPosting());
- assertEquals(19, postingList.nextPosting());
- assertEquals(PostingList.END_OF_STREAM, postingList.nextPosting());
+ try (BlockBalancedTreeReader reader = finishAndOpenReader(5, buffer))
+ {
+ PostingList postingList = performIntersection(reader, buildQuery(8, 15));
+
+ assertEquals(8, postingList.nextPosting());
+ assertEquals(9, postingList.nextPosting());
+ assertEquals(10, postingList.nextPosting());
+ assertEquals(11, postingList.nextPosting());
+ assertEquals(12, postingList.nextPosting());
+ assertEquals(13, postingList.nextPosting());
+ assertEquals(14, postingList.nextPosting());
+ assertEquals(15, postingList.nextPosting());
+ assertEquals(16, postingList.nextPosting());
+ assertEquals(17, postingList.nextPosting());
+ assertEquals(18, postingList.nextPosting());
+ assertEquals(19, postingList.nextPosting());
+ assertEquals(PostingList.END_OF_STREAM, postingList.nextPosting());
+ }
}
@Test
public void testResourcesReleaseWhenQueryDoesntMatchAnything() throws Exception
{
final BlockBalancedTreeRamBuffer buffer = new BlockBalancedTreeRamBuffer(Integer.BYTES);
byte[] scratch = new byte[4];
- for (int docID = 0; docID < 1000; docID++)
+ for (int rowID = 0; rowID < 1000; rowID++)
{
- NumericUtils.intToSortableBytes(docID, scratch, 0);
- buffer.add(docID, scratch);
+ NumericUtils.intToSortableBytes(rowID, scratch, 0);
+ buffer.add(rowID, scratch);
}
// add a gap between 1000 and 1100
- for (int docID = 1000; docID < 2000; docID++)
+ for (int rowID = 1000; rowID < 2000; rowID++)
{
- NumericUtils.intToSortableBytes(docID + 100, scratch, 0);
- buffer.add(docID, scratch);
+ NumericUtils.intToSortableBytes(rowID + 100, scratch, 0);
+ buffer.add(rowID, scratch);
}
- final BlockBalancedTreeReader reader = finishAndOpenReader(50, buffer);
+ try (BlockBalancedTreeReader reader = finishAndOpenReader(50, buffer))
+ {
+ final PostingList intersection = performIntersection(reader, buildQuery(1017, 1096));
+ assertNull(intersection);
+ }
+ }
- final PostingList intersection = performIntersection(reader, buildQuery(1017, 1096));
- assertNull(intersection);
+ @Test
+ public void testConcurrentIntersectionsOnSameReader() throws Exception
+ {
+ int numRows = 1000;
+
+ final BlockBalancedTreeRamBuffer buffer = new BlockBalancedTreeRamBuffer(Integer.BYTES);
+
+ byte[] scratch = new byte[4];
+ for (int rowID = 0; rowID < numRows; rowID++)
+ {
+ NumericUtils.intToSortableBytes(rowID, scratch, 0);
+ buffer.add(rowID, scratch);
+ }
+
+ try (BlockBalancedTreeReader reader = finishAndOpenReader(4, buffer))
+ {
+ int concurrency = 100;
+
+ ExecutorService executor = Executors.newFixedThreadPool(concurrency);
+ List<Future<?>> results = new ArrayList<>();
+ for (int thread = 0; thread < concurrency; thread++)
+ {
+ results.add(executor.submit(() -> assertRange(reader, 445, 555)));
+ }
+ FBUtilities.waitOnFutures(results);
+ executor.shutdown();
+ }
+ }
+
+ private void assertRange(BlockBalancedTreeReader reader, long lowerBound, long upperBound)
+ {
+ Expression expression = new Expression(indexContext);
+ expression.add(Operator.GT, Int32Type.instance.decompose(444));
+ expression.add(Operator.LT, Int32Type.instance.decompose(555));
+
+ try
+ {
+ PostingList intersection = performIntersection(reader, BlockBalancedTreeQueries.balancedTreeQueryFrom(expression, 4));
+ assertNotNull(intersection);
+ assertEquals(110, intersection.size());
+ for (long posting = lowerBound; posting < upperBound; posting++)
+ assertEquals(posting, intersection.nextPosting());
+ }
+ catch (IOException e)
+ {
+ throw Throwables.unchecked(e);
+ }
}
+
Review Comment:
Nit: unneeded blank line
##########
test/unit/org/apache/cassandra/index/sai/disk/v1/bbtree/BlockBalancedTreeReaderTest.java:
##########
@@ -162,66 +166,118 @@ public void testSameValuesInLeaf() throws Exception
final BlockBalancedTreeRamBuffer buffer = new BlockBalancedTreeRamBuffer(Integer.BYTES);
byte[] scratch = new byte[4];
- for (int docID = 0; docID < 10; docID++)
+ for (int rowID = 0; rowID < 10; rowID++)
{
- NumericUtils.intToSortableBytes(docID, scratch, 0);
- buffer.add(docID, scratch);
+ NumericUtils.intToSortableBytes(rowID, scratch, 0);
+ buffer.add(rowID, scratch);
}
- for (int docID = 10; docID < 20; docID++)
+ for (int rowID = 10; rowID < 20; rowID++)
{
NumericUtils.intToSortableBytes(10, scratch, 0);
- buffer.add(docID, scratch);
+ buffer.add(rowID, scratch);
}
- for (int docID = 20; docID < 30; docID++)
+ for (int rowID = 20; rowID < 30; rowID++)
{
- NumericUtils.intToSortableBytes(docID, scratch, 0);
- buffer.add(docID, scratch);
+ NumericUtils.intToSortableBytes(rowID, scratch, 0);
+ buffer.add(rowID, scratch);
}
- final BlockBalancedTreeReader reader = finishAndOpenReader(5, buffer);
-
- PostingList postingList = performIntersection(reader, buildQuery(8, 15));
-
- assertEquals(8, postingList.nextPosting());
- assertEquals(9, postingList.nextPosting());
- assertEquals(10, postingList.nextPosting());
- assertEquals(11, postingList.nextPosting());
- assertEquals(12, postingList.nextPosting());
- assertEquals(13, postingList.nextPosting());
- assertEquals(14, postingList.nextPosting());
- assertEquals(15, postingList.nextPosting());
- assertEquals(16, postingList.nextPosting());
- assertEquals(17, postingList.nextPosting());
- assertEquals(18, postingList.nextPosting());
- assertEquals(19, postingList.nextPosting());
- assertEquals(PostingList.END_OF_STREAM, postingList.nextPosting());
+ try (BlockBalancedTreeReader reader = finishAndOpenReader(5, buffer))
+ {
+ PostingList postingList = performIntersection(reader, buildQuery(8, 15));
+
+ assertEquals(8, postingList.nextPosting());
+ assertEquals(9, postingList.nextPosting());
+ assertEquals(10, postingList.nextPosting());
+ assertEquals(11, postingList.nextPosting());
+ assertEquals(12, postingList.nextPosting());
+ assertEquals(13, postingList.nextPosting());
+ assertEquals(14, postingList.nextPosting());
+ assertEquals(15, postingList.nextPosting());
+ assertEquals(16, postingList.nextPosting());
+ assertEquals(17, postingList.nextPosting());
+ assertEquals(18, postingList.nextPosting());
+ assertEquals(19, postingList.nextPosting());
+ assertEquals(PostingList.END_OF_STREAM, postingList.nextPosting());
+ }
}
@Test
public void testResourcesReleaseWhenQueryDoesntMatchAnything() throws Exception
{
final BlockBalancedTreeRamBuffer buffer = new BlockBalancedTreeRamBuffer(Integer.BYTES);
byte[] scratch = new byte[4];
- for (int docID = 0; docID < 1000; docID++)
+ for (int rowID = 0; rowID < 1000; rowID++)
{
- NumericUtils.intToSortableBytes(docID, scratch, 0);
- buffer.add(docID, scratch);
+ NumericUtils.intToSortableBytes(rowID, scratch, 0);
+ buffer.add(rowID, scratch);
}
// add a gap between 1000 and 1100
- for (int docID = 1000; docID < 2000; docID++)
+ for (int rowID = 1000; rowID < 2000; rowID++)
{
- NumericUtils.intToSortableBytes(docID + 100, scratch, 0);
- buffer.add(docID, scratch);
+ NumericUtils.intToSortableBytes(rowID + 100, scratch, 0);
+ buffer.add(rowID, scratch);
}
- final BlockBalancedTreeReader reader = finishAndOpenReader(50, buffer);
+ try (BlockBalancedTreeReader reader = finishAndOpenReader(50, buffer))
+ {
+ final PostingList intersection = performIntersection(reader, buildQuery(1017, 1096));
+ assertNull(intersection);
+ }
+ }
- final PostingList intersection = performIntersection(reader, buildQuery(1017, 1096));
- assertNull(intersection);
+ @Test
+ public void testConcurrentIntersectionsOnSameReader() throws Exception
+ {
+ int numRows = 1000;
+
+ final BlockBalancedTreeRamBuffer buffer = new BlockBalancedTreeRamBuffer(Integer.BYTES);
+
+ byte[] scratch = new byte[4];
+ for (int rowID = 0; rowID < numRows; rowID++)
+ {
+ NumericUtils.intToSortableBytes(rowID, scratch, 0);
+ buffer.add(rowID, scratch);
+ }
+
+ try (BlockBalancedTreeReader reader = finishAndOpenReader(4, buffer))
+ {
+ int concurrency = 100;
+
+ ExecutorService executor = Executors.newFixedThreadPool(concurrency);
+ List<Future<?>> results = new ArrayList<>();
+ for (int thread = 0; thread < concurrency; thread++)
+ {
+ results.add(executor.submit(() -> assertRange(reader, 445, 555)));
+ }
+ FBUtilities.waitOnFutures(results);
+ executor.shutdown();
+ }
+ }
+
+ private void assertRange(BlockBalancedTreeReader reader, long lowerBound, long upperBound)
Review Comment:
Nit: add `@SuppressWarnings("SameParameterValue")
`
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
[GitHub] [cassandra] maedhroz closed pull request #2487: Fix concurrency in bbtree reader by cloning state
Posted by "maedhroz (via GitHub)" <gi...@apache.org>.
maedhroz closed pull request #2487: Fix concurrency in bbtree reader by cloning state
URL: https://github.com/apache/cassandra/pull/2487
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
[GitHub] [cassandra] maedhroz commented on a diff in pull request #2487: Fix concurrency in bbtree reader by cloning state
Posted by "maedhroz (via GitHub)" <gi...@apache.org>.
maedhroz commented on code in PR #2487:
URL: https://github.com/apache/cassandra/pull/2487#discussion_r1262948389
##########
test/unit/org/apache/cassandra/index/sai/disk/v1/bbtree/BlockBalancedTreeReaderTest.java:
##########
@@ -222,6 +231,55 @@ public void testResourcesReleaseWhenQueryDoesntMatchAnything() throws Exception
assertNull(intersection);
}
+ @Test
+ public void testConcurrentIntersectionsOnSameReader() throws Exception
+ {
+ int numRows = 1000;
+
+ final BlockBalancedTreeRamBuffer buffer = new BlockBalancedTreeRamBuffer(Integer.BYTES);
+
+ byte[] scratch = new byte[4];
+ for (int docID = 0; docID < numRows; docID++)
+ {
+ NumericUtils.intToSortableBytes(docID, scratch, 0);
+ buffer.add(docID, scratch);
+ }
+
+ final BlockBalancedTreeReader reader = finishAndOpenReader(4, buffer);
Review Comment:
nit: Little hygiene thing, but I guess we should make sure we close the reader here.
(I can fix on commit.)
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
[GitHub] [cassandra] maedhroz commented on pull request #2487: Fix concurrency in bbtree reader by cloning state
Posted by "maedhroz (via GitHub)" <gi...@apache.org>.
maedhroz commented on PR #2487:
URL: https://github.com/apache/cassandra/pull/2487#issuecomment-1638727504
Committed as https://github.com/apache/cassandra/commit/83419e074d131a68112470fc8761c3edafb5e675
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
[GitHub] [cassandra] maedhroz commented on a diff in pull request #2487: Fix concurrency in bbtree reader by cloning state
Posted by "maedhroz (via GitHub)" <gi...@apache.org>.
maedhroz commented on code in PR #2487:
URL: https://github.com/apache/cassandra/pull/2487#discussion_r1262925279
##########
src/java/org/apache/cassandra/index/sai/disk/v1/bbtree/BlockBalancedTreeWalker.java:
##########
@@ -67,9 +105,52 @@ public class BlockBalancedTreeWalker implements Closeable
}
}
+ @VisibleForTesting
+ public BlockBalancedTreeWalker(DataInput indexInput, long treeIndexRoot) throws IOException
+ {
+ treeIndexFile = null;
+
+ indexInput.skipBytes(treeIndexRoot);
+
+ maxPointsInLeafNode = indexInput.readVInt();
+ bytesPerValue = indexInput.readVInt();
+
+ // Read index:
+ numLeaves = indexInput.readVInt();
+ assert numLeaves > 0;
+ treeDepth = indexInput.readVInt();
+ minPackedValue = new byte[bytesPerValue];
+ maxPackedValue = new byte[bytesPerValue];
+
+ indexInput.readBytes(minPackedValue, 0, bytesPerValue);
+ indexInput.readBytes(maxPackedValue, 0, bytesPerValue);
+
+ if (ByteArrayUtil.compareUnsigned(minPackedValue, 0, maxPackedValue, 0, bytesPerValue) > 0)
+ {
+ String message = String.format("Min packed value %s is > max packed value %s.",
+ new BytesRef(minPackedValue), new BytesRef(maxPackedValue));
+ throw new CorruptIndexException(message, indexInput);
+ }
+
+ valueCount = indexInput.readVLong();
+
+ int numBytes = indexInput.readVInt();
+ packedIndex = new byte[numBytes];
+ indexInput.readBytes(packedIndex, 0, numBytes);
+
+ memoryUsage = ObjectSizes.sizeOfArray(packedIndex) +
+ ObjectSizes.sizeOfArray(minPackedValue) +
+ ObjectSizes.sizeOfArray(maxPackedValue);
+ }
+
public long memoryUsage()
{
- return state.memoryUsage;
+ return memoryUsage;
+ }
+
+ public TraversalState createTraversalState()
Review Comment:
nit: Maybe `newTraversalState()`?
(can just change on commit if you like it)
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
[GitHub] [cassandra] maedhroz commented on a diff in pull request #2487: Fix concurrency in bbtree reader by cloning state
Posted by "maedhroz (via GitHub)" <gi...@apache.org>.
maedhroz commented on code in PR #2487:
URL: https://github.com/apache/cassandra/pull/2487#discussion_r1262937867
##########
test/unit/org/apache/cassandra/index/sai/disk/v1/bbtree/BlockBalancedTreeReaderTest.java:
##########
@@ -222,6 +231,55 @@ public void testResourcesReleaseWhenQueryDoesntMatchAnything() throws Exception
assertNull(intersection);
}
+ @Test
+ public void testConcurrentIntersectionsOnSameReader() throws Exception
+ {
+ int numRows = 1000;
+
+ final BlockBalancedTreeRamBuffer buffer = new BlockBalancedTreeRamBuffer(Integer.BYTES);
+
+ byte[] scratch = new byte[4];
+ for (int docID = 0; docID < numRows; docID++)
Review Comment:
nit: `rowID`
(can just fix on commit)
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org