Skip to content

[SPARK-7251][Spark Core] Perform sequential scan when iterating over entries in BytesToBytesMap #5836

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
wants to merge 5 commits into from
Closed

[SPARK-7251][Spark Core] Perform sequential scan when iterating over entries in BytesToBytesMap #5836

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
a620b94
add new iter
May 1, 2015
5957d11
add new iter
May 1, 2015
3e1a3ec
update
May 4, 2015
851d34a
update
May 4, 2015
baaefc5
Fix empty map bug.
May 5, 2015
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
43 changes: 37 additions & 6 deletions unsafe/src/main/java/org/apache/spark/unsafe/map/BytesToBytesMap.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -173,18 +173,35 @@ public BytesToBytesMap(
public Iterator<Location> iterator() {
return new Iterator<Location>() {

private int nextPos = bitset.nextSetBit(0);
private int cur = 0;

private int pageCur = 0;

private MemoryBlock currentPage;

private long addr;
Copy link
Contributor

@JoshRosen JoshRosen May 7, 2015

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is a confusing name; we shouldn't trade off brevity for clarity like this.


@Override
public boolean hasNext() {
return nextPos != -1;
return cur != size;
}

@Override
public Location next() {
final int pos = nextPos;
nextPos = bitset.nextSetBit(nextPos + 1);
return loc.with(pos, 0, true);
if (currentPage == null) {
currentPage = dataPages.get(pageCur++);
Copy link
Contributor

@JoshRosen JoshRosen May 7, 2015

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We should never use inline increment operations like this. When I see this, I have to take time to remember whether get() is called with the value before the increment or after. I don't think that it makes sense to trade off brevity for clarity like this; please move the increment to a separate line.

addr = currentPage.getBaseOffset();
}
long keySize = PlatformDependent.UNSAFE.getLong(memoryManager.getPage(addr), addr);
Copy link
Contributor

@JoshRosen JoshRosen May 7, 2015

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think you're calling getPage() with the wrong units: getPage expects to receive an address that was encoded by the task memory manager, but in the previous if case you've assigned an offset within the page to addr.

This call isn't necessary, though, since we happen to have the currentPage already.

if (keySize == -1L) {
Copy link
Contributor

@JoshRosen JoshRosen May 7, 2015

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

-1 should be lifted into a static constant named END_OF_BLOCK or something to clarify the fact that it's used as an end of block marker.

Copy link
Contributor

@JoshRosen JoshRosen May 13, 2015

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This branch was never taken in the unit tests but was exercised by my stress tester, which explains why I was able to find a bug even though the tests passed. This is why I'm loving IntelliJ's code coverage metrics for this Tungsten stuff: insisting on 100% line coverage really helps to prevent bugs like this.

currentPage = dataPages.get(pageCur++);
Copy link
Contributor

@JoshRosen JoshRosen May 7, 2015

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Same here; do not ever use prefix / postfix increments when passing arguments to calculations.

addr = currentPage.getBaseOffset();
}
loc.with(addr, true);
addr += keySize + 8;
addr += PlatformDependent.UNSAFE.getLong(memoryManager.getPage(addr), addr) + 8;
Copy link
Contributor

@JoshRosen JoshRosen May 7, 2015

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This line is confusing; can the first half of this addition be derived from a call to loc?

cur++;
return loc;
}

@Override
Expand Down Expand Up @@ -291,6 +308,14 @@ Location with(int pos, int keyHashcode, boolean isDefined) {
return this;
}

Location with(long fullKeyAddress, boolean isDefined) {
Copy link
Contributor

@JoshRosen JoshRosen May 7, 2015

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This method is only ever called when isDefined is true, so I think we can just omit that argument.

this.isDefined = isDefined;
if (isDefined) {
updateAddressesAndSizes(fullKeyAddress);
}
return this;
}

/**
* Returns true if the key is defined at this position, and false otherwise.
*/
Expand Down Expand Up @@ -380,7 +405,13 @@ public void putNewKey(
bitset.set(pos);

// If there's not enough space in the current page, allocate a new page:
if (currentDataPage == null || PAGE_SIZE_BYTES - pageCursor < requiredSize) {
if (currentDataPage == null || PAGE_SIZE_BYTES - pageCursor - 8 < requiredSize) {
Copy link
Contributor

@JoshRosen JoshRosen May 7, 2015

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

There's an assertion on line 403 that needs to be updated to reserve 8 extra bytes for the final length.

if (currentDataPage != null) {
Copy link
Contributor

@JoshRosen JoshRosen May 7, 2015

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The fact that we have a check for currentDataPage == null in the outer if loop and currentDataPage != null here implies that the outer loop's first or branch is useless, since the body of the loop will never be executed if it holds.

Copy link
Contributor

@JoshRosen JoshRosen May 14, 2015

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Well, I guess we'll still end up allocating a new block. I'll add a comment to make this more clear.

final Object pageBaseObject = currentDataPage.getBaseObject();
final long pageBaseOffset = currentDataPage.getBaseOffset();
final long lengthOffsetInPage = pageBaseOffset + pageCursor;
PlatformDependent.UNSAFE.putLong(pageBaseObject, lengthOffsetInPage, -1L);
}
MemoryBlock newPage = memoryManager.allocatePage(PAGE_SIZE_BYTES);
dataPages.add(newPage);
pageCursor = 0;
Expand Down
35 changes: 25 additions & 10 deletions unsafe/src/test/java/org/apache/spark/unsafe/map/AbstractBytesToBytesMapSuite.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -167,14 +167,25 @@ public void iteratorTest() throws Exception {
final BytesToBytesMap.Location loc =
map.lookup(value, PlatformDependent.LONG_ARRAY_OFFSET, 8);
Assert.assertFalse(loc.isDefined());
loc.putNewKey(
value,
PlatformDependent.LONG_ARRAY_OFFSET,
8,
value,
PlatformDependent.LONG_ARRAY_OFFSET,
8
);
if (i % 5 == 0) {
loc.putNewKey(
value,
PlatformDependent.LONG_ARRAY_OFFSET,
0,
value,
PlatformDependent.LONG_ARRAY_OFFSET,
8
);
} else {
loc.putNewKey(
value,
PlatformDependent.LONG_ARRAY_OFFSET,
8,
value,
PlatformDependent.LONG_ARRAY_OFFSET,
8
);
}
}
final java.util.BitSet valuesSeen = new java.util.BitSet(size);
final Iterator<BytesToBytesMap.Location> iter = map.iterator();
Expand All @@ -183,11 +194,15 @@ public void iteratorTest() throws Exception {
Assert.assertTrue(loc.isDefined());
final MemoryLocation keyAddress = loc.getKeyAddress();
final MemoryLocation valueAddress = loc.getValueAddress();
final long key = PlatformDependent.UNSAFE.getLong(
final long keyLength = loc.getKeyLength();
final long key = PlatformDependent.UNSAFE.getLong(
keyAddress.getBaseObject(), keyAddress.getBaseOffset());
final long value = PlatformDependent.UNSAFE.getLong(
valueAddress.getBaseObject(), valueAddress.getBaseOffset());
Assert.assertEquals(key, value);
if (value % 5 == 0)
Assert.assertEquals(keyLength, 0);
else
Assert.assertEquals(key, value);
valuesSeen.set((int) value);
}
Assert.assertEquals(size, valuesSeen.cardinality());
Expand Down