Github user kiszk commented on a diff in the pull request:
https://github.com/apache/spark/pull/19222#discussion_r179391228
--- Diff:
common/unsafe/src/main/java/org/apache/spark/unsafe/memory/MemoryBlock.java ---
@@ -45,38 +45,164 @@
*/
public static final int FREED_IN_ALLOCATOR_PAGE_NUMBER = -3;
- private final long length;
+ @Nullable
+ protected Object obj;
+
+ protected long offset;
+
+ protected long length;
/**
* Optional page number; used when this MemoryBlock represents a page
allocated by a
- * TaskMemoryManager. This field is public so that it can be modified by
the TaskMemoryManager,
- * which lives in a different package.
+ * TaskMemoryManager. This field can be updated using setPageNumber
method so that
+ * this can be modified by the TaskMemoryManager, which lives in a
different package.
*/
- public int pageNumber = NO_PAGE_NUMBER;
+ private int pageNumber = NO_PAGE_NUMBER;
- public MemoryBlock(@Nullable Object obj, long offset, long length) {
- super(obj, offset);
+ protected MemoryBlock(@Nullable Object obj, long offset, long length) {
+ if (offset < 0 || length < 0) {
+ throw new IllegalArgumentException(
+ "Length " + length + " and offset " + offset + "must be
non-negative");
+ }
+ this.obj = obj;
+ this.offset = offset;
this.length = length;
}
+ protected MemoryBlock() {
+ this(null, 0, 0);
+ }
+
+ public final Object getBaseObject() {
+ return obj;
+ }
+
+ public final long getBaseOffset() {
+ return offset;
+ }
+
+ public void resetObjAndOffset() {
+ this.obj = null;
+ this.offset = 0;
+ }
+
/**
* Returns the size of the memory block.
*/
- public long size() {
+ public final long size() {
return length;
}
- /**
- * Creates a memory block pointing to the memory used by the long array.
- */
- public static MemoryBlock fromLongArray(final long[] array) {
- return new MemoryBlock(array, Platform.LONG_ARRAY_OFFSET, array.length
* 8L);
+ public final void setPageNumber(int pageNum) {
+ pageNumber = pageNum;
+ }
+
+ public final int getPageNumber() {
+ return pageNumber;
}
/**
* Fills the memory block with the specified byte value.
*/
- public void fill(byte value) {
+ public final void fill(byte value) {
Platform.setMemory(obj, offset, length, value);
}
+
+ /**
+ * Instantiate MemoryBlock for given object type with new offset
+ */
+ public final static MemoryBlock allocateFromObject(Object obj, long
offset, long length) {
+ MemoryBlock mb = null;
+ if (obj instanceof byte[]) {
+ byte[] array = (byte[])obj;
+ mb = new ByteArrayMemoryBlock(array, offset, length);
+ } else if (obj instanceof long[]) {
+ long[] array = (long[])obj;
+ mb = new OnHeapMemoryBlock(array, offset, length);
+ } else if (obj == null) {
+ // we assume that to pass null pointer means off-heap
+ mb = new OffHeapMemoryBlock(offset, length);
+ } else {
+ throw new UnsupportedOperationException(
+ "Instantiate MemoryBlock for type " + obj.getClass() + " is not
supported now");
+ }
+ return mb;
+ }
+
+ /**
+ * Just instantiate the same type of MemoryBlock with the new absolute
offset and size. The data
--- End diff --
First of all, this API does not allow users to *expand* a memory block due
to `checkSubBlockRange()`. This check does not accept `offset < 0` or `offset +
size > length`.
I have to update this assert as `this.offset + offset + size > length`.
Also, I will add *subset* in the comment, and add some tests in
`MemoryBlockSuite`
```
public MemoryBlock subBlock(long offset, long size) {
checkSubBlockRange(offset, size);
return new OnHeapMemoryBlock(array, this.offset + offset, size);
}
```
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
