Skip to content

[SPARK-10399][CORE][SQL] Introduce OffHeapMemoryBlock to hold DirectByteBuffer, refactor spark.unsafe.Platform #11494

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 4 commits into from
Closed

[SPARK-10399][CORE][SQL] Introduce OffHeapMemoryBlock to hold DirectByteBuffer, refactor spark.unsafe.Platform #11494

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
8465f2f
new classes
yzotov Mar 3, 2016
e0619db
changes to unsafe subsystem
yzotov Mar 3, 2016
f9f2776
changes to spark-core
yzotov Mar 3, 2016
4a4e62e
spark.unsafe.Platform interface changed, BufferHolder is replaced by …
yzotov Mar 3, 2016
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
153 changes: 124 additions & 29 deletions common/unsafe/src/main/java/org/apache/spark/unsafe/Platform.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@

import java.lang.reflect.Field;

import org.apache.spark.unsafe.memory.MemoryBlock;
import sun.misc.Unsafe;

public final class Platform {
Expand All @@ -37,68 +38,92 @@ public final class Platform {

public static final int DOUBLE_ARRAY_OFFSET;

public static int getInt(Object object, long offset) {
public static int getInt(MemoryBlock object, long offset) {
return _UNSAFE.getInt(object.getBaseObject(), offset);
}

public static int getInt(byte[] object, long offset) {
return _UNSAFE.getInt(object, offset);
}

public static void putInt(Object object, long offset, int value) {
public static void putInt(MemoryBlock object, long offset, int value) {
_UNSAFE.putInt(object.getBaseObject(), offset, value);
}

public static void putInt(byte[] object, long offset, int value) {
_UNSAFE.putInt(object, offset, value);
}

public static boolean getBoolean(Object object, long offset) {
return _UNSAFE.getBoolean(object, offset);
public static boolean getBoolean(MemoryBlock object, long offset) {
return _UNSAFE.getBoolean(object.getBaseObject(), offset);
}

public static void putBoolean(Object object, long offset, boolean value) {
_UNSAFE.putBoolean(object, offset, value);
public static void putBoolean(MemoryBlock object, long offset, boolean value) {
_UNSAFE.putBoolean(object.getBaseObject(), offset, value);
}

public static byte getByte(Object object, long offset) {
public static byte getByte(MemoryBlock object, long offset) {
return _UNSAFE.getByte(object.getBaseObject(), offset);
}

public static byte getByte(byte[] object, long offset) {
return _UNSAFE.getByte(object, offset);
}

public static void putByte(Object object, long offset, byte value) {
_UNSAFE.putByte(object, offset, value);
public static void putByte(MemoryBlock object, long offset, byte value) {
_UNSAFE.putByte(object.getBaseObject(), offset, value);
}

public static short getShort(MemoryBlock object, long offset) {
return _UNSAFE.getShort(object.getBaseObject(), offset);
}

public static short getShort(Object object, long offset) {
return _UNSAFE.getShort(object, offset);
public static void putShort(MemoryBlock object, long offset, short value) {
_UNSAFE.putShort(object.getBaseObject(), offset, value);
}

public static void putShort(Object object, long offset, short value) {
_UNSAFE.putShort(object, offset, value);
public static long getLong(MemoryBlock object, long offset) {
return _UNSAFE.getLong(object.getBaseObject(), offset);
}

public static long getLong(Object object, long offset) {
public static long getLong(byte[] object, long offset) {
return _UNSAFE.getLong(object, offset);
}

public static void putLong(Object object, long offset, long value) {
_UNSAFE.putLong(object, offset, value);
public static void putLong(MemoryBlock object, long offset, long value) {
_UNSAFE.putLong(object.getBaseObject(), offset, value);
}

public static float getFloat(MemoryBlock object, long offset) {
return _UNSAFE.getFloat(object.getBaseObject(), offset);
}

public static float getFloat(Object object, long offset) {
public static float getFloat(byte[] object, long offset) {
return _UNSAFE.getFloat(object, offset);
}

public static void putFloat(Object object, long offset, float value) {
_UNSAFE.putFloat(object, offset, value);
public static void putFloat(MemoryBlock object, long offset, float value) {
_UNSAFE.putFloat(object.getBaseObject(), offset, value);
}

public static double getDouble(Object object, long offset) {
public static double getDouble(MemoryBlock object, long offset) {
return _UNSAFE.getDouble(object.getBaseObject(), offset);
}

public static double getDouble(byte[] object, long offset) {
return _UNSAFE.getDouble(object, offset);
}

public static void putDouble(Object object, long offset, double value) {
_UNSAFE.putDouble(object, offset, value);
public static void putDouble(MemoryBlock object, long offset, double value) {
_UNSAFE.putDouble(object.getBaseObject(), offset, value);
}

public static Object getObjectVolatile(Object object, long offset) {
return _UNSAFE.getObjectVolatile(object, offset);
public static Object getObjectVolatile(MemoryBlock object, long offset) {
return _UNSAFE.getObjectVolatile(object.getBaseObject(), offset);
}

public static void putObjectVolatile(Object object, long offset, Object value) {
_UNSAFE.putObjectVolatile(object, offset, value);
public static void putObjectVolatile(MemoryBlock object, long offset, Object value) {
_UNSAFE.putObjectVolatile(object.getBaseObject(), offset, value);
}

public static long allocateMemory(long size) {
Expand All @@ -111,7 +136,7 @@ public static void freeMemory(long address) {

public static long reallocateMemory(long address, long oldSize, long newSize) {
long newMemory = _UNSAFE.allocateMemory(newSize);
copyMemory(null, address, null, newMemory, oldSize);
copyMemory0(null, address, null, newMemory, oldSize);
freeMemory(address);
return newMemory;
}
Expand All @@ -120,8 +145,8 @@ public static void setMemory(long address, byte value, long size) {
_UNSAFE.setMemory(address, size, value);
}

public static void copyMemory(
Object src, long srcOffset, Object dst, long dstOffset, long length) {
static void copyMemory0(
Object src, long srcOffset, Object dst, long dstOffset, long length) {
// Check if dstOffset is before or after srcOffset to determine if we should copy
// forward or backwards. This is necessary in case src and dst overlap.
if (dstOffset < srcOffset) {
Expand All @@ -146,6 +171,76 @@ public static void copyMemory(
}
}

public static void copyMemory(
MemoryBlock src, long srcOffset, MemoryBlock dst, long dstOffset, long length) {
Platform.copyMemory0(src.getBaseObject(), srcOffset, dst.getBaseObject(), dstOffset, length);
}

public static void copyMemory(
byte[] src, long srcOffset, MemoryBlock dst, long dstOffset, long length) {
Platform.copyMemory0(src, srcOffset, dst.getBaseObject(), dstOffset, length);
}

public static void copyMemory(
short[] src, long srcOffset, MemoryBlock dst, long dstOffset, long length) {
Platform.copyMemory0(src, srcOffset, dst.getBaseObject(), dstOffset, length);
}

public static void copyMemory(
int[] src, long srcOffset, MemoryBlock dst, long dstOffset, long length) {
Platform.copyMemory0(src, srcOffset, dst.getBaseObject(), dstOffset, length);
}

public static void copyMemory(
long[] src, long srcOffset, MemoryBlock dst, long dstOffset, long length) {
Platform.copyMemory0(src, srcOffset, dst.getBaseObject(), dstOffset, length);
}

public static void copyMemory(
float[] src, long srcOffset, MemoryBlock dst, long dstOffset, long length) {
Platform.copyMemory0(src, srcOffset, dst.getBaseObject(), dstOffset, length);
}

public static void copyMemory(
double[] src, long srcOffset, MemoryBlock dst, long dstOffset, long length) {
Platform.copyMemory0(src, srcOffset, dst.getBaseObject(), dstOffset, length);
}

public static void copyMemory(
MemoryBlock src, long srcOffset, byte[] dst, long dstOffset, long length) {
Platform.copyMemory0(src.getBaseObject(), srcOffset, dst, dstOffset, length);
}

public static void copyMemory(
byte[] src, long srcOffset, byte[] dst, long dstOffset, long length) {
Platform.copyMemory0(src, srcOffset, dst, dstOffset, length);
}

public static void copyMemory(
byte[] src, long srcOffset, short[] dst, long dstOffset, long length) {
Platform.copyMemory0(src, srcOffset, dst, dstOffset, length);
}

public static void copyMemory(
byte[] src, long srcOffset, int[] dst, long dstOffset, long length) {
Platform.copyMemory0(src, srcOffset, dst, dstOffset, length);
}

public static void copyMemory(
byte[] src, long srcOffset, long[] dst, long dstOffset, long length) {
Platform.copyMemory0(src, srcOffset, dst, dstOffset, length);
}

public static void copyMemory(
byte[] src, long srcOffset, float[] dst, long dstOffset, long length) {
Platform.copyMemory0(src, srcOffset, dst, dstOffset, length);
}

public static void copyMemory(
byte[] src, long srcOffset, double[] dst, long dstOffset, long length) {
Platform.copyMemory0(src, srcOffset, dst, dstOffset, length);
}

/**
* Raises an exception bypassing compiler checks for checked exceptions.
*/
Expand Down
23 changes: 22 additions & 1 deletion common/unsafe/src/main/java/org/apache/spark/unsafe/array/ByteArrayMethods.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,8 @@
package org.apache.spark.unsafe.array;

import org.apache.spark.unsafe.Platform;
import org.apache.spark.unsafe.memory.ByteArrayMemoryBlock;
import org.apache.spark.unsafe.memory.MemoryBlock;

public class ByteArrayMethods {

Expand Down Expand Up @@ -45,7 +47,7 @@ public static int roundNumberOfBytesToNearestWord(int numBytes) {
* @return true if the arrays are equal, false otherwise
*/
public static boolean arrayEquals(
Object leftBase, long leftOffset, Object rightBase, long rightOffset, final long length) {
MemoryBlock leftBase, long leftOffset, MemoryBlock rightBase, long rightOffset, final long length) {
int i = 0;
while (i <= length - 8) {
if (Platform.getLong(leftBase, leftOffset + i) !=
Expand All @@ -63,4 +65,23 @@ public static boolean arrayEquals(
}
return true;
}

public static boolean arrayEquals(
byte[] leftBase, long leftOffset, MemoryBlock rightBase, long rightOffset, final long length) {
ByteArrayMemoryBlock bleft = ByteArrayMemoryBlock.fromByteArray(leftBase);
return ByteArrayMethods.arrayEquals(bleft, leftOffset, rightBase, rightOffset, length);
}

public static boolean arrayEquals(
MemoryBlock leftBase, long leftOffset, byte[] rightBase, long rightOffset, final long length) {
ByteArrayMemoryBlock bright = ByteArrayMemoryBlock.fromByteArray(rightBase);
return ByteArrayMethods.arrayEquals(leftBase, leftOffset, bright, rightOffset, length);
}

public static boolean arrayEquals(
byte[] leftBase, long leftOffset, byte[] rightBase, long rightOffset, final long length) {
ByteArrayMemoryBlock bleft = ByteArrayMemoryBlock.fromByteArray(leftBase);
ByteArrayMemoryBlock bright = ByteArrayMemoryBlock.fromByteArray(rightBase);
return ByteArrayMethods.arrayEquals(bleft, leftOffset, bright, rightOffset, length);
}
}
12 changes: 3 additions & 9 deletions common/unsafe/src/main/java/org/apache/spark/unsafe/array/LongArray.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,15 +33,13 @@ public final class LongArray {
private static final long WIDTH = 8;

private final MemoryBlock memory;
private final Object baseObj;
private final long baseOffset;

private final long length;

public LongArray(MemoryBlock memory) {
assert memory.size() < (long) Integer.MAX_VALUE * 8: "Array size > 4 billion elements";
this.memory = memory;
this.baseObj = memory.getBaseObject();
this.baseOffset = memory.getBaseOffset();
this.length = memory.size() / WIDTH;
}
Expand All @@ -50,10 +48,6 @@ public MemoryBlock memoryBlock() {
return memory;
}

public Object getBaseObject() {
return baseObj;
}

public long getBaseOffset() {
return baseOffset;
}
Expand All @@ -70,7 +64,7 @@ public long size() {
*/
public void zeroOut() {
for (long off = baseOffset; off < baseOffset + length * WIDTH; off += WIDTH) {
Platform.putLong(baseObj, off, 0);
Platform.putLong(memory, off, 0);
}
}

Expand All @@ -80,7 +74,7 @@ public void zeroOut() {
public void set(int index, long value) {
assert index >= 0 : "index (" + index + ") should >= 0";
assert index < length : "index (" + index + ") should < length (" + length + ")";
Platform.putLong(baseObj, baseOffset + index * WIDTH, value);
Platform.putLong(memory, baseOffset + index * WIDTH, value);
}

/**
Expand All @@ -89,6 +83,6 @@ public void set(int index, long value) {
public long get(int index) {
assert index >= 0 : "index (" + index + ") should >= 0";
assert index < length : "index (" + index + ") should < length (" + length + ")";
return Platform.getLong(baseObj, baseOffset + index * WIDTH);
return Platform.getLong(memory, baseOffset + index * WIDTH);
}
}
11 changes: 6 additions & 5 deletions common/unsafe/src/main/java/org/apache/spark/unsafe/bitset/BitSetMethods.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
package org.apache.spark.unsafe.bitset;

import org.apache.spark.unsafe.Platform;
import org.apache.spark.unsafe.memory.MemoryBlock;

/**
* Methods for working with fixed-size uncompressed bitsets.
Expand All @@ -37,7 +38,7 @@ private BitSetMethods() {
/**
* Sets the bit at the specified index to {@code true}.
*/
public static void set(Object baseObject, long baseOffset, int index) {
public static void set(MemoryBlock baseObject, long baseOffset, int index) {
assert index >= 0 : "index (" + index + ") should >= 0";
final long mask = 1L << (index & 0x3f); // mod 64 and shift
final long wordOffset = baseOffset + (index >> 6) * WORD_SIZE;
Expand All @@ -48,7 +49,7 @@ public static void set(Object baseObject, long baseOffset, int index) {
/**
* Sets the bit at the specified index to {@code false}.
*/
public static void unset(Object baseObject, long baseOffset, int index) {
public static void unset(MemoryBlock baseObject, long baseOffset, int index) {
assert index >= 0 : "index (" + index + ") should >= 0";
final long mask = 1L << (index & 0x3f); // mod 64 and shift
final long wordOffset = baseOffset + (index >> 6) * WORD_SIZE;
Expand All @@ -59,7 +60,7 @@ public static void unset(Object baseObject, long baseOffset, int index) {
/**
* Returns {@code true} if the bit is set at the specified index.
*/
public static boolean isSet(Object baseObject, long baseOffset, int index) {
public static boolean isSet(MemoryBlock baseObject, long baseOffset, int index) {
assert index >= 0 : "index (" + index + ") should >= 0";
final long mask = 1L << (index & 0x3f); // mod 64 and shift
final long wordOffset = baseOffset + (index >> 6) * WORD_SIZE;
Expand All @@ -70,7 +71,7 @@ public static boolean isSet(Object baseObject, long baseOffset, int index) {
/**
* Returns {@code true} if any bit is set.
*/
public static boolean anySet(Object baseObject, long baseOffset, long bitSetWidthInWords) {
public static boolean anySet(MemoryBlock baseObject, long baseOffset, long bitSetWidthInWords) {
long addr = baseOffset;
for (int i = 0; i < bitSetWidthInWords; i++, addr += WORD_SIZE) {
if (Platform.getLong(baseObject, addr) != 0) {
Expand Down Expand Up @@ -98,7 +99,7 @@ public static boolean anySet(Object baseObject, long baseOffset, long bitSetWidt
* @return the index of the next set bit, or -1 if there is no such bit
*/
public static int nextSetBit(
Object baseObject,
MemoryBlock baseObject,
long baseOffset,
int fromIndex,
int bitsetSizeInWords) {
Expand Down
Loading