diff --git a/pom.xml b/pom.xml index 9bcc6461517..4e4cff5be70 100644 --- a/pom.xml +++ b/pom.xml @@ -519,4 +519,23 @@ + + + + jdk9plus + + + !1.8 + + + + javax.annotation + javax.annotation-api + 1.3.2 + provided + + + + + diff --git a/src/main/java/org/tikv/common/columnar/TiBlockColumnVector.java b/src/main/java/org/tikv/common/columnar/TiBlockColumnVector.java index beff8234f8f..0341447d5c8 100644 --- a/src/main/java/org/tikv/common/columnar/TiBlockColumnVector.java +++ b/src/main/java/org/tikv/common/columnar/TiBlockColumnVector.java @@ -15,12 +15,15 @@ package org.tikv.common.columnar; -import static org.tikv.common.util.MemoryUtil.EMPTY_BYTE_BUFFER_DIRECT; +import static org.tikv.common.util.MemoryUtil.EMPTY_BYTE_BUFFER; import java.math.BigDecimal; import java.nio.ByteBuffer; import java.nio.charset.StandardCharsets; -import java.sql.Timestamp; +import java.time.Instant; +import java.time.ZoneId; +import java.time.ZonedDateTime; +import java.util.Objects; import org.joda.time.LocalDate; import org.tikv.common.columnar.datatypes.CHType; import org.tikv.common.types.AbstractDateTimeType; @@ -29,36 +32,34 @@ import org.tikv.common.util.MemoryUtil; public class TiBlockColumnVector extends TiColumnVector { - long offsetsAddr; - ByteBuffer offsets; - long nullMapAddr; - ByteBuffer nullMap; - long dataAddr; - ByteBuffer data; - private int fixedLength; + private final ByteBuffer offsets; + private final ByteBuffer nullMap; + private final ByteBuffer data; + private final int fixedLength; public TiBlockColumnVector(CHType type, ByteBuffer data, int numOfRows, int fixedLength) { super(type.toDataType(), numOfRows); - this.data = data; - this.dataAddr = MemoryUtil.getAddress(data); - fillEmptyNullMap(); - fillEmptyOffsets(); + this.data = Objects.requireNonNull(data); + this.nullMap = null; + this.offsets = null; this.fixedLength = fixedLength; } public TiBlockColumnVector(CHType type) { super(type.toDataType(), 0); + this.data = EMPTY_BYTE_BUFFER; + this.nullMap = null; + this.offsets = null; + this.fixedLength = -1; } public TiBlockColumnVector( CHType type, ByteBuffer nullMap, ByteBuffer data, int numOfRows, int fixedLength) { // chType -> data type super(type.toDataType(), numOfRows); - this.nullMap = nullMap; - this.nullMapAddr = MemoryUtil.getAddress(nullMap); - this.data = data; - this.dataAddr = MemoryUtil.getAddress(data); - fillEmptyOffsets(); + this.nullMap = Objects.requireNonNull(nullMap); + this.data = Objects.requireNonNull(data); + this.offsets = null; this.fixedLength = fixedLength; } @@ -67,25 +68,12 @@ public TiBlockColumnVector( CHType type, ByteBuffer nullMap, ByteBuffer offsets, ByteBuffer data, int numOfRows) { // chType -> data type super(type.toDataType(), numOfRows); - this.offsets = offsets; - this.offsetsAddr = MemoryUtil.getAddress(offsets); - this.nullMap = nullMap; - this.nullMapAddr = MemoryUtil.getAddress(nullMap); - this.data = data; - this.dataAddr = MemoryUtil.getAddress(data); + this.offsets = Objects.requireNonNull(offsets); + this.nullMap = nullMap; // may be null + this.data = Objects.requireNonNull(data); this.fixedLength = -1; } - private void fillEmptyNullMap() { - this.nullMap = EMPTY_BYTE_BUFFER_DIRECT; - this.nullMapAddr = MemoryUtil.getAddress(this.nullMap); - } - - private void fillEmptyOffsets() { - this.offsets = EMPTY_BYTE_BUFFER_DIRECT; - this.offsetsAddr = MemoryUtil.getAddress(this.offsets); - } - /** * Cleans up memory for this column vector. The column vector is not usable after this. * @@ -93,42 +81,37 @@ private void fillEmptyOffsets() { * in-memory and we don't expect any exception to happen during closing. */ @Override - public void close() { - if (dataAddr != 0) { - MemoryUtil.free(data); - } - - if (offsetsAddr != 0) { - MemoryUtil.free(offsets); - } - - if (nullMapAddr != 0) { - MemoryUtil.free(nullMap); - } - dataAddr = 0; - offsetsAddr = 0; - nullMapAddr = 0; - } + public void close() {} /** Returns true if this column vector contains any null values. */ @Override public boolean hasNull() { - return nullMap == null; + if (nullMap != null) { + byte[] array = nullMap.array(); + for (byte b : array) { + if (b != 0) return true; + } + } + return false; } /** Returns the number of nulls in this column vector. */ @Override public int numNulls() { - throw new UnsupportedOperationException("numNulls is not supported for TiBlockColumnVector"); + int n = 0; + if (nullMap != null) { + byte[] array = nullMap.array(); + for (byte b : array) { + if (b != 0) n++; + } + } + return n; } /** Returns whether the value at rowId is NULL. */ @Override public boolean isNullAt(int rowId) { - if (nullMap == EMPTY_BYTE_BUFFER_DIRECT) { - return false; - } - return MemoryUtil.getByte(nullMapAddr + rowId) != 0; + return nullMap != null && nullMap.get(rowId) != 0; } /** @@ -146,7 +129,7 @@ public boolean getBoolean(int rowId) { */ @Override public byte getByte(int rowId) { - return MemoryUtil.getByte(dataAddr + rowId); + return data.get(rowId); } /** @@ -155,7 +138,7 @@ public byte getByte(int rowId) { */ @Override public short getShort(int rowId) { - return MemoryUtil.getShort(dataAddr + (rowId << 1)); + return data.getShort(rowId << 1); } /** @@ -167,11 +150,12 @@ public int getInt(int rowId) { if (type instanceof DateType) { return (int) getTime(rowId); } - return MemoryUtil.getInt(dataAddr + (rowId << 2)); + return data.getInt(rowId << 2); } + // returns microseconds since epoch - fields are interpreted in current default timezone private long getDateTime(int rowId) { - long v = MemoryUtil.getLong(dataAddr + (rowId << 3)); + long v = data.getLong(rowId << 3); long ymdhms = v >>> 24; long ymd = ymdhms >>> 17; int day = (int) (ymd & ((1 << 5) - 1)); @@ -184,13 +168,15 @@ private long getDateTime(int rowId) { int minute = (hms >>> 6) & ((1 << 6) - 1); int hour = hms >>> 12; int microsec = (int) (v % (1 << 24)); - Timestamp ts = - new Timestamp(year - 1900, month - 1, day, hour, minute, second, microsec * 1000); - return ts.getTime() / 1000 * 1000000 + ts.getNanos() / 1000; + ZonedDateTime zdt = + ZonedDateTime.of( + year, month, day, hour, minute, second, microsec * 1000, ZoneId.systemDefault()); + Instant instant = zdt.toInstant(); + return instant.getEpochSecond() * 1000_000L + instant.getNano() / 1000; } private long getTime(int rowId) { - long v = MemoryUtil.getLong(dataAddr + (rowId << 3)); + long v = data.getLong(rowId << 3); long ymd = v >>> 41; long ym = ymd >>> 5; int year = (int) (ym / 13); @@ -215,7 +201,7 @@ public long getLong(int rowId) { } else if (fixedLength == 4) { return getInt(rowId); } else if (fixedLength == 8) { - return MemoryUtil.getLong(dataAddr + (rowId * fixedLength)); + return data.getLong(rowId << 3); } throw new UnsupportedOperationException( String.format("getting long with fixed length %d", fixedLength)); @@ -227,7 +213,7 @@ public long getLong(int rowId) { */ @Override public float getFloat(int rowId) { - return MemoryUtil.getFloat(dataAddr + (rowId * fixedLength)); + return data.getFloat(rowId * fixedLength); } /** @@ -236,7 +222,7 @@ public float getFloat(int rowId) { */ @Override public double getDouble(int rowId) { - return MemoryUtil.getDouble(dataAddr + (rowId * fixedLength)); + return data.getDouble(rowId * fixedLength); } /** @@ -244,28 +230,24 @@ public double getDouble(int rowId) { */ @Override public BigDecimal getDecimal(int rowId, int precision, int scale) { - long rowIdAddr = rowId * fixedLength + dataAddr; if (fixedLength == 4) { - return MemoryUtil.getDecimal32(rowIdAddr, scale); + return MemoryUtil.getDecimal32(data, rowId << 2, scale); } else if (fixedLength == 8) { - return MemoryUtil.getDecimal64(rowIdAddr, scale); + return MemoryUtil.getDecimal64(data, rowId << 3, scale); } else if (fixedLength == 16) { - return MemoryUtil.getDecimal128(rowIdAddr, scale); + return MemoryUtil.getDecimal128(data, rowId << 4, scale); } else { - return MemoryUtil.getDecimal256(rowIdAddr, scale); + return MemoryUtil.getDecimal256(data, rowId * fixedLength, scale); } } private long offsetAt(int i) { - return i == 0 ? 0 : MemoryUtil.getLong(offsetsAddr + ((i - 1) << 3)); + return i == 0 ? 0L : offsets.getLong((i - 1) << 3); } public int sizeAt(int i) { return (int) - (i == 0 - ? MemoryUtil.getLong(offsetsAddr) - : MemoryUtil.getLong(offsetsAddr + (i << 3)) - - MemoryUtil.getLong(offsetsAddr + ((i - 1) << 3))); + (i == 0 ? offsets.getLong(0) : offsets.getLong(i << 3) - offsets.getLong((i - 1) << 3)); } /** @@ -278,13 +260,13 @@ public String getUTF8String(int rowId) { // FixedString case if (fixedLength != -1) { byte[] chars = new byte[fixedLength]; - MemoryUtil.getBytes((int) (dataAddr + fixedLength * rowId), chars, 0, fixedLength); + data.get(chars, rowId * fixedLength, fixedLength); return new String(chars); } else { - long offset = (dataAddr + offsetAt(rowId)); + int offset = (int) offsetAt(rowId); int numBytes = sizeAt(rowId) - 1; byte[] chars = new byte[numBytes]; - MemoryUtil.getBytes(offset, chars, 0, numBytes); + data.get(chars, offset, numBytes); return new String(chars, StandardCharsets.UTF_8); } } @@ -295,10 +277,10 @@ public String getUTF8String(int rowId) { @Override public byte[] getBinary(int rowId) { if (type.equals(BytesType.BLOB) || type.equals(BytesType.TINY_BLOB)) { - long offset = (dataAddr + offsetAt(rowId)); + int offset = (int) offsetAt(rowId); int numBytes = sizeAt(rowId) - 1; byte[] ret = new byte[numBytes]; - MemoryUtil.getBytes(offset, ret, 0, numBytes); + data.get(ret, offset, numBytes); return ret; } else { throw new UnsupportedOperationException( diff --git a/src/main/java/org/tikv/common/columnar/datatypes/AutoGrowByteBuffer.java b/src/main/java/org/tikv/common/columnar/datatypes/AutoGrowByteBuffer.java index 70e59643323..7975d2cd25e 100644 --- a/src/main/java/org/tikv/common/columnar/datatypes/AutoGrowByteBuffer.java +++ b/src/main/java/org/tikv/common/columnar/datatypes/AutoGrowByteBuffer.java @@ -20,12 +20,10 @@ import org.tikv.common.util.MemoryUtil; public class AutoGrowByteBuffer { - private final ByteBuffer initBuf; private ByteBuffer buf; public AutoGrowByteBuffer(ByteBuffer initBuf) { initBuf.clear(); - this.initBuf = initBuf; this.buf = initBuf; } @@ -39,22 +37,12 @@ public ByteBuffer getByteBuffer() { private void beforeIncrease(int inc) { int minCap = buf.position() + inc; - if (minCap > buf.capacity()) { - int newCap = buf.capacity(); + int newCap = buf.capacity(); + if (minCap > newCap) { do { newCap = newCap << 1; } while (minCap > newCap); - - ByteBuffer newBuf = MemoryUtil.allocateDirect(newCap); - MemoryUtil.copyMemory( - MemoryUtil.getAddress(buf), MemoryUtil.getAddress(newBuf), buf.position()); - newBuf.position(buf.position()); - - if (buf != initBuf) { - MemoryUtil.free(buf); - } - - buf = newBuf; + buf = MemoryUtil.copyOf(buf, newCap); } } diff --git a/src/main/java/org/tikv/common/columnar/datatypes/CHType.java b/src/main/java/org/tikv/common/columnar/datatypes/CHType.java index 83ff5cd7802..3ec85c498d1 100644 --- a/src/main/java/org/tikv/common/columnar/datatypes/CHType.java +++ b/src/main/java/org/tikv/common/columnar/datatypes/CHType.java @@ -15,7 +15,7 @@ package org.tikv.common.columnar.datatypes; -import static org.tikv.common.util.MemoryUtil.allocateDirect; +import static org.tikv.common.util.MemoryUtil.allocate; import java.nio.ByteBuffer; import org.tikv.common.codec.CodecDataInput; @@ -41,7 +41,7 @@ public void setNullable(boolean nullable) { protected ByteBuffer decodeNullMap(CodecDataInput cdi, int size) { // read size * uint8 from cdi - ByteBuffer buffer = allocateDirect(size); + ByteBuffer buffer = allocate(size); MemoryUtil.readFully(buffer, cdi, size); buffer.clear(); return buffer; @@ -63,13 +63,13 @@ public TiBlockColumnVector decode(CodecDataInput cdi, int size) { } if (isNullable()) { ByteBuffer nullMap = decodeNullMap(cdi, size); - ByteBuffer buffer = allocateDirect(bufferSize(size)); + ByteBuffer data = allocate(bufferSize(size)); // read bytes from cdi to buffer(off-heap) - MemoryUtil.readFully(buffer, cdi, bufferSize(size)); - buffer.clear(); - return new TiBlockColumnVector(this, nullMap, buffer, size, length); + MemoryUtil.readFully(data, cdi, bufferSize(size)); + data.clear(); + return new TiBlockColumnVector(this, nullMap, data, size, length); } else { - ByteBuffer buffer = allocateDirect(bufferSize(size)); + ByteBuffer buffer = allocate(bufferSize(size)); MemoryUtil.readFully(buffer, cdi, bufferSize(size)); buffer.clear(); return new TiBlockColumnVector(this, buffer, size, length); diff --git a/src/main/java/org/tikv/common/columnar/datatypes/CHTypeString.java b/src/main/java/org/tikv/common/columnar/datatypes/CHTypeString.java index 9a91afbb531..bbaa542530a 100644 --- a/src/main/java/org/tikv/common/columnar/datatypes/CHTypeString.java +++ b/src/main/java/org/tikv/common/columnar/datatypes/CHTypeString.java @@ -15,8 +15,7 @@ package org.tikv.common.columnar.datatypes; -import static org.tikv.common.util.MemoryUtil.EMPTY_BYTE_BUFFER_DIRECT; -import static org.tikv.common.util.MemoryUtil.allocateDirect; +import static org.tikv.common.util.MemoryUtil.allocate; import com.google.common.base.Preconditions; import java.nio.ByteBuffer; @@ -31,7 +30,7 @@ public class CHTypeString extends CHType { // Use to prevent frequently reallocate the chars buffer. // ClickHouse does not pass a total length at the beginning, so sad... private static final ThreadLocal initBuffer = - ThreadLocal.withInitial(() -> allocateDirect(102400)); + ThreadLocal.withInitial(() -> allocate(102400)); public CHTypeString() { this.length = -1; @@ -57,12 +56,12 @@ public TiBlockColumnVector decode(CodecDataInput cdi, int size) { if (isNullable()) { nullMap = decodeNullMap(cdi, size); } else { - nullMap = EMPTY_BYTE_BUFFER_DIRECT; + nullMap = null; } - ByteBuffer offsets = allocateDirect(size << 3); - ByteBuffer initCharsBuf = initBuffer.get(); - AutoGrowByteBuffer autoGrowCharsBuf = new AutoGrowByteBuffer(initCharsBuf); + ByteBuffer offsets = allocate(size << 3); + ByteBuffer initDataBuf = initBuffer.get(); + AutoGrowByteBuffer autoGrowDataBuf = new AutoGrowByteBuffer(initDataBuf); int offset = 0; for (int i = 0; i < size; i++) { @@ -71,20 +70,18 @@ public TiBlockColumnVector decode(CodecDataInput cdi, int size) { offset += valueSize + 1; offsets.putLong(offset); - autoGrowCharsBuf.put(cdi, valueSize); - autoGrowCharsBuf.putByte((byte) 0); // terminating zero byte + autoGrowDataBuf.put(cdi, valueSize); + autoGrowDataBuf.putByte((byte) 0); // terminating zero byte } - Preconditions.checkState(offset == autoGrowCharsBuf.dataSize()); + Preconditions.checkState(offset == autoGrowDataBuf.dataSize()); - ByteBuffer chars = autoGrowCharsBuf.getByteBuffer(); - if (chars == initCharsBuf) { + ByteBuffer data = autoGrowDataBuf.getByteBuffer(); + if (data == initDataBuf) { // Copy out. - ByteBuffer newChars = allocateDirect(offset); - MemoryUtil.copyMemory(MemoryUtil.getAddress(chars), MemoryUtil.getAddress(newChars), offset); - chars = newChars; + data = MemoryUtil.copyOf(data, offset); } - return new TiBlockColumnVector(this, nullMap, offsets, chars, size); + return new TiBlockColumnVector(this, nullMap, offsets, data, size); } } diff --git a/src/main/java/org/tikv/common/util/MemoryUtil.java b/src/main/java/org/tikv/common/util/MemoryUtil.java index 5c5021cca0e..6a5c99c1bb4 100644 --- a/src/main/java/org/tikv/common/util/MemoryUtil.java +++ b/src/main/java/org/tikv/common/util/MemoryUtil.java @@ -16,270 +16,46 @@ package org.tikv.common.util; import com.google.common.primitives.UnsignedLong; -import com.sun.management.OperatingSystemMXBean; -import java.lang.management.ManagementFactory; -import java.lang.reflect.Field; import java.math.BigDecimal; import java.math.BigInteger; -import java.nio.Buffer; import java.nio.ByteBuffer; import java.nio.ByteOrder; -import java.nio.channels.FileChannel; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; +import java.util.Arrays; import org.tikv.common.codec.CodecDataInput; -import sun.misc.Cleaner; -import sun.misc.Unsafe; -import sun.nio.ch.DirectBuffer; -// Copied from io.indexr.util.MemoryUtil.java with some modifications. +// Copied from io.indexr.util.MemoryUtil.java with lots of modifications. public class MemoryUtil { - public static final ByteBuffer EMPTY_BYTE_BUFFER_DIRECT = allocateDirect(0); - public static final Unsafe unsafe; - private static final Logger logger = LoggerFactory.getLogger(MemoryUtil.class); - private static final long UNSAFE_COPY_THRESHOLD = 1024 * 1024L; // copied from java.nio.Bits - private static final Class DIRECT_BYTE_BUFFER_CLASS; - private static final long DIRECT_BYTE_BUFFER_ADDRESS_OFFSET; - private static final long DIRECT_BYTE_BUFFER_CAPACITY_OFFSET; - private static final long DIRECT_BYTE_BUFFER_LIMIT_OFFSET; - private static final long DIRECT_BYTE_BUFFER_POSITION_OFFSET; - private static final long DIRECT_BYTE_BUFFER_ATTACHMENT_OFFSET; - private static final long DIRECT_BYTE_BUFFER_CLEANER; - private static final Class BYTE_BUFFER_CLASS; - private static final long BYTE_BUFFER_OFFSET_OFFSET; - private static final long BYTE_BUFFER_HB_OFFSET; - private static final long BYTE_ARRAY_BASE_OFFSET; + public static final ByteBuffer EMPTY_BYTE_BUFFER = allocate(0); - private static final long STRING_VALUE_OFFSET; - - private static final boolean BIG_ENDIAN = ByteOrder.nativeOrder().equals(ByteOrder.BIG_ENDIAN); - private static int PAGE_SIZE = -1; - - private static boolean HDFS_READ_HACK_ENABLE; - private static long BlockReaderLocal_verifyChecksum; - private static long BlockReaderLocal_dataIn; - private static long DFSInputStream_verifyChecksum; - private static long DFSInputStream_blockReader; - - static { - // We support all in fact. - // if (BIG_ENDIAN) { - // throw new RuntimeException("We only support littel endian platform!"); - // } - try { - Field field = Unsafe.class.getDeclaredField("theUnsafe"); - field.setAccessible(true); - unsafe = (Unsafe) field.get(null); - - Class clazz = ByteBuffer.allocateDirect(0).getClass(); - DIRECT_BYTE_BUFFER_ADDRESS_OFFSET = - unsafe.objectFieldOffset(Buffer.class.getDeclaredField("address")); - DIRECT_BYTE_BUFFER_CAPACITY_OFFSET = - unsafe.objectFieldOffset(Buffer.class.getDeclaredField("capacity")); - DIRECT_BYTE_BUFFER_LIMIT_OFFSET = - unsafe.objectFieldOffset(Buffer.class.getDeclaredField("limit")); - DIRECT_BYTE_BUFFER_POSITION_OFFSET = - unsafe.objectFieldOffset(Buffer.class.getDeclaredField("position")); - DIRECT_BYTE_BUFFER_ATTACHMENT_OFFSET = - unsafe.objectFieldOffset(clazz.getDeclaredField("att")); - DIRECT_BYTE_BUFFER_CLEANER = unsafe.objectFieldOffset(clazz.getDeclaredField("cleaner")); - DIRECT_BYTE_BUFFER_CLASS = clazz; - - clazz = ByteBuffer.allocate(0).getClass(); - BYTE_BUFFER_OFFSET_OFFSET = - unsafe.objectFieldOffset(ByteBuffer.class.getDeclaredField("offset")); - BYTE_BUFFER_HB_OFFSET = unsafe.objectFieldOffset(ByteBuffer.class.getDeclaredField("hb")); - BYTE_BUFFER_CLASS = clazz; - - BYTE_ARRAY_BASE_OFFSET = unsafe.arrayBaseOffset(byte[].class); - - STRING_VALUE_OFFSET = - MemoryUtil.unsafe.objectFieldOffset(String.class.getDeclaredField("value")); - - try { - BlockReaderLocal_verifyChecksum = - unsafe.objectFieldOffset( - Class.forName("org.apache.hadoop.hdfs.BlockReaderLocal") - .getDeclaredField("verifyChecksum")); - BlockReaderLocal_dataIn = - unsafe.objectFieldOffset( - Class.forName("org.apache.hadoop.hdfs.BlockReaderLocal") - .getDeclaredField("dataIn")); - - DFSInputStream_verifyChecksum = - unsafe.objectFieldOffset( - Class.forName("org.apache.hadoop.hdfs.DFSInputStream") - .getDeclaredField("verifyChecksum")); - DFSInputStream_blockReader = - unsafe.objectFieldOffset( - Class.forName("org.apache.hadoop.hdfs.DFSInputStream") - .getDeclaredField("blockReader")); - - HDFS_READ_HACK_ENABLE = true; - } catch (Exception e) { - HDFS_READ_HACK_ENABLE = false; - logger.warn("hdfs read hack is off because of error: {}", e.getCause()); - } - } catch (Exception e) { - throw new AssertionError(e); - } - } - - public static boolean getBlockReaderLocal_verifyChecksum(Object br) { - return unsafe.getBoolean(br, BlockReaderLocal_verifyChecksum); - } - - public static void setBlockReaderLocal_verifyChecksum(Object br, boolean value) { - unsafe.putBoolean(br, BlockReaderLocal_verifyChecksum, value); - } - - public static FileChannel getBlockReaderLocal_dataIn(Object br) { - return (FileChannel) unsafe.getObject(br, BlockReaderLocal_dataIn); - } - - public static int pageSize() { - if (PAGE_SIZE == -1) { - PAGE_SIZE = unsafe.pageSize(); - } - return PAGE_SIZE; - } - - public static void setMemory(long addr, long size, byte v) { - unsafe.setMemory(addr, size, v); - } - - public static void setMemory(Object base, long offset, long size, byte v) { - unsafe.setMemory(base, offset, size, v); - } - - public static long getAddress(ByteBuffer buffer) { - assert buffer.getClass() == DIRECT_BYTE_BUFFER_CLASS; - return unsafe.getLong(buffer, DIRECT_BYTE_BUFFER_ADDRESS_OFFSET); - } - - public static int getCap(ByteBuffer buffer) { - assert buffer.getClass() == DIRECT_BYTE_BUFFER_CLASS; - return unsafe.getInt(buffer, DIRECT_BYTE_BUFFER_CAPACITY_OFFSET); - } - - public static long allocate(long size) { - // return Native.malloc(size); - return unsafe.allocateMemory(size); - } - - public static void free(long addr) { - // Native.free(addr); - unsafe.freeMemory(addr); - } - - /** Good manner to free a buffer before forget it. */ - public static void free(ByteBuffer buffer) { - if (buffer.isDirect()) { - Cleaner cleaner = ((DirectBuffer) buffer).cleaner(); - if (cleaner != null) { - cleaner.clean(); - } - } - } - - public static void setByte(long address, byte b) { - unsafe.putByte(address, b); - } - - public static void setShort(long address, short s) { - unsafe.putShort(address, s); - } - - public static void setInt(long address, int l) { - unsafe.putInt(address, l); - } - - public static void setDecimal(long address, BigDecimal v, int scale) { - BigDecimal bigDec = v; - BigInteger bigInt = bigDec.scaleByPowerOfTen(scale).toBigInteger(); - byte[] arr = bigInt.toByteArray(); - for (int i = 0; i < arr.length; i++) { - unsafe.putByte(address + i, arr[arr.length - 1 - i]); - } - } - - public static void setDecimal256(long address, BigDecimal v, int scale) { - BigDecimal bigDec = v; - BigInteger bigInt = bigDec.scaleByPowerOfTen(scale).toBigInteger(); - int sign = 0; - if (bigInt.signum() < 0) { - sign = 1; - bigInt = bigInt.abs(); - } - byte[] arr = bigInt.toByteArray(); - if (arr.length > 32) { - throw new RuntimeException("The inserting decimal is out of range: " + v.toString()); - } - int limbs = arr.length / 8; - if (arr.length % 8 > 0) { - limbs++; - } - - for (int i = 0; i < arr.length; i++) { - unsafe.putByte(address + i, arr[arr.length - 1 - i]); - } - unsafe.putShort(address + 32, (short) limbs); - unsafe.putShort(address + 34, (short) sign); - } - - public static void setLong(long address, long l) { - unsafe.putLong(address, l); - } - - public static void setFloat(long address, float v) { - unsafe.putFloat(address, v); - } - - public static void setDouble(long address, double v) { - unsafe.putDouble(address, v); - } - - public static byte getByte(long address) { - return unsafe.getByte(address); - } - - public static short getShort(long address) { - return (short) (unsafe.getShort(address) & 0xffff); - } - - public static int getInt(long address) { - return unsafe.getInt(address); - } - - public static BigDecimal getDecimal32(long address, int scale) { - int n = getInt(address); + public static BigDecimal getDecimal32(ByteBuffer data, int offset, int scale) { + int n = data.getInt(offset); BigInteger dec = BigInteger.valueOf(n); return new BigDecimal(dec, scale); } - public static BigDecimal getDecimal64(long address, int scale) { - long n = getLong(address); + public static BigDecimal getDecimal64(ByteBuffer data, int offset, int scale) { + long n = data.getLong(offset); BigInteger dec = BigInteger.valueOf(n); return new BigDecimal(dec, scale); } - public static BigDecimal getDecimal128(long address, int scale) { - UnsignedLong n0 = UnsignedLong.fromLongBits(getLong(address)); - long n1 = getLong(address + 8); + public static BigDecimal getDecimal128(ByteBuffer data, int offset, int scale) { + UnsignedLong n0 = UnsignedLong.fromLongBits(data.getLong(offset)); + long n1 = data.getLong(offset + 8); BigInteger dec = BigInteger.valueOf(n1); dec = dec.shiftLeft(64).add(n0.bigIntegerValue()); return new BigDecimal(dec, scale); } - public static BigDecimal getDecimal256(long address, int scale) { - int limbs = getShort(address + 32); + public static BigDecimal getDecimal256(ByteBuffer data, int offset, int scale) { + int limbs = data.getShort(offset + 32); BigInteger dec = BigInteger.ZERO; for (int i = limbs - 1; i >= 0; i--) { - UnsignedLong d = UnsignedLong.fromLongBits(unsafe.getLong(address + i * 8)); + UnsignedLong d = UnsignedLong.fromLongBits(data.getLong(offset + i * 8)); dec = dec.shiftLeft(64).add(d.bigIntegerValue()); } - int sign = unsafe.getByte(address + 34); + int sign = data.get(offset + 34); BigDecimal result = new BigDecimal(dec, scale); if (sign > 0) { return result.negate(); @@ -287,297 +63,25 @@ public static BigDecimal getDecimal256(long address, int scale) { return result; } - public static long getLong(long address) { - return unsafe.getLong(address); - } - - public static float getFloat(long address) { - return unsafe.getFloat(address); - } - - public static double getDouble(long address) { - return unsafe.getDouble(address); - } - - public static ByteBuffer getByteBuffer(long address, int length, boolean autoFree) { - ByteBuffer instance = getHollowDirectByteBuffer(); - if (autoFree) { - Cleaner cleaner = Cleaner.create(instance, new Deallocator(address)); - setByteBuffer(instance, address, length, cleaner); - } else { - setByteBuffer(instance, address, length, null); - } - instance.order(ByteOrder.nativeOrder()); - return instance; - } - - public static ByteBuffer getHollowDirectByteBuffer() { - ByteBuffer instance; - try { - instance = (ByteBuffer) unsafe.allocateInstance(DIRECT_BYTE_BUFFER_CLASS); - } catch (InstantiationException e) { - throw new AssertionError(e); - } - instance.order(ByteOrder.nativeOrder()); - return instance; - } - - public static void setByteBuffer(ByteBuffer instance, long address, int length, Cleaner cleaner) { - unsafe.putLong(instance, DIRECT_BYTE_BUFFER_ADDRESS_OFFSET, address); - unsafe.putInt(instance, DIRECT_BYTE_BUFFER_CAPACITY_OFFSET, length); - unsafe.putInt(instance, DIRECT_BYTE_BUFFER_LIMIT_OFFSET, length); - if (cleaner != null) { - unsafe.putObject(instance, DIRECT_BYTE_BUFFER_CLEANER, cleaner); - } - } - - public static Object getAttachment(ByteBuffer instance) { - assert instance.getClass() == DIRECT_BYTE_BUFFER_CLASS; - return unsafe.getObject(instance, DIRECT_BYTE_BUFFER_ATTACHMENT_OFFSET); - } - - public static void setAttachment(ByteBuffer instance, Object next) { - assert instance.getClass() == DIRECT_BYTE_BUFFER_CLASS; - unsafe.putObject(instance, DIRECT_BYTE_BUFFER_ATTACHMENT_OFFSET, next); - } - - public static ByteBuffer duplicateDirectByteBuffer(ByteBuffer source, ByteBuffer hollowBuffer) { - assert source.getClass() == DIRECT_BYTE_BUFFER_CLASS; - unsafe.putLong( - hollowBuffer, - DIRECT_BYTE_BUFFER_ADDRESS_OFFSET, - unsafe.getLong(source, DIRECT_BYTE_BUFFER_ADDRESS_OFFSET)); - unsafe.putInt( - hollowBuffer, - DIRECT_BYTE_BUFFER_POSITION_OFFSET, - unsafe.getInt(source, DIRECT_BYTE_BUFFER_POSITION_OFFSET)); - unsafe.putInt( - hollowBuffer, - DIRECT_BYTE_BUFFER_LIMIT_OFFSET, - unsafe.getInt(source, DIRECT_BYTE_BUFFER_LIMIT_OFFSET)); - unsafe.putInt( - hollowBuffer, - DIRECT_BYTE_BUFFER_CAPACITY_OFFSET, - unsafe.getInt(source, DIRECT_BYTE_BUFFER_CAPACITY_OFFSET)); - return hollowBuffer; - } - - public static ByteBuffer duplicateDirectByteBuffer(ByteBuffer source) { - return duplicateDirectByteBuffer(source, getHollowDirectByteBuffer()); - } - - public static long getLongByByte(long address) { - if (BIG_ENDIAN) { - return (((long) unsafe.getByte(address)) << 56) - | (((long) unsafe.getByte(address + 1) & 0xff) << 48) - | (((long) unsafe.getByte(address + 2) & 0xff) << 40) - | (((long) unsafe.getByte(address + 3) & 0xff) << 32) - | (((long) unsafe.getByte(address + 4) & 0xff) << 24) - | (((long) unsafe.getByte(address + 5) & 0xff) << 16) - | (((long) unsafe.getByte(address + 6) & 0xff) << 8) - | (((long) unsafe.getByte(address + 7) & 0xff)); - } else { - return (((long) unsafe.getByte(address + 7)) << 56) - | (((long) unsafe.getByte(address + 6) & 0xff) << 48) - | (((long) unsafe.getByte(address + 5) & 0xff) << 40) - | (((long) unsafe.getByte(address + 4) & 0xff) << 32) - | (((long) unsafe.getByte(address + 3) & 0xff) << 24) - | (((long) unsafe.getByte(address + 2) & 0xff) << 16) - | (((long) unsafe.getByte(address + 1) & 0xff) << 8) - | (((long) unsafe.getByte(address) & 0xff)); - } - } - - public static int getIntByByte(long address) { - if (BIG_ENDIAN) { - return (((int) unsafe.getByte(address)) << 24) - | (((int) unsafe.getByte(address + 1) & 0xff) << 16) - | (((int) unsafe.getByte(address + 2) & 0xff) << 8) - | (((int) unsafe.getByte(address + 3) & 0xff)); - } else { - return (((int) unsafe.getByte(address + 3)) << 24) - | (((int) unsafe.getByte(address + 2) & 0xff) << 16) - | (((int) unsafe.getByte(address + 1) & 0xff) << 8) - | (((int) unsafe.getByte(address) & 0xff)); - } - } - - public static int getShortByByte(long address) { - if (BIG_ENDIAN) { - return (((int) unsafe.getByte(address)) << 8) | (((int) unsafe.getByte(address + 1) & 0xff)); - } else { - return (((int) unsafe.getByte(address + 1)) << 8) | (((int) unsafe.getByte(address) & 0xff)); - } - } - - public static void putLongByByte(long address, long value) { - if (BIG_ENDIAN) { - unsafe.putByte(address, (byte) (value >> 56)); - unsafe.putByte(address + 1, (byte) (value >> 48)); - unsafe.putByte(address + 2, (byte) (value >> 40)); - unsafe.putByte(address + 3, (byte) (value >> 32)); - unsafe.putByte(address + 4, (byte) (value >> 24)); - unsafe.putByte(address + 5, (byte) (value >> 16)); - unsafe.putByte(address + 6, (byte) (value >> 8)); - unsafe.putByte(address + 7, (byte) (value)); - } else { - unsafe.putByte(address + 7, (byte) (value >> 56)); - unsafe.putByte(address + 6, (byte) (value >> 48)); - unsafe.putByte(address + 5, (byte) (value >> 40)); - unsafe.putByte(address + 4, (byte) (value >> 32)); - unsafe.putByte(address + 3, (byte) (value >> 24)); - unsafe.putByte(address + 2, (byte) (value >> 16)); - unsafe.putByte(address + 1, (byte) (value >> 8)); - unsafe.putByte(address, (byte) (value)); - } - } - - public static void putIntByByte(long address, int value) { - if (BIG_ENDIAN) { - unsafe.putByte(address, (byte) (value >> 24)); - unsafe.putByte(address + 1, (byte) (value >> 16)); - unsafe.putByte(address + 2, (byte) (value >> 8)); - unsafe.putByte(address + 3, (byte) (value)); - } else { - unsafe.putByte(address + 3, (byte) (value >> 24)); - unsafe.putByte(address + 2, (byte) (value >> 16)); - unsafe.putByte(address + 1, (byte) (value >> 8)); - unsafe.putByte(address, (byte) (value)); - } - } - - public static void setBytes(long address, ByteBuffer buffer) { - int start = buffer.position(); - int count = buffer.limit() - start; - if (count == 0) return; - - if (buffer.isDirect()) setBytes(((DirectBuffer) buffer).address() + start, address, count); - else setBytes(address, buffer.array(), buffer.arrayOffset() + start, count); - } - - /** - * Transfers objCount bytes from buffer to Memory - * - * @param address start offset in the memory - * @param buffer the data buffer - * @param bufferOffset start offset of the buffer - * @param count number of bytes to transfer - */ - public static void setBytes(long address, byte[] buffer, int bufferOffset, int count) { - assert buffer != null; - assert !(bufferOffset < 0 || count < 0 || bufferOffset + count > buffer.length); - setBytes(buffer, bufferOffset, address, count); - } - - public static void setBytes(long src, long trg, long count) { - while (count > 0) { - long size = Math.min(count, UNSAFE_COPY_THRESHOLD); - unsafe.copyMemory(src, trg, size); - count -= size; - src += size; - trg += size; - } - } - - public static void setBytes(byte[] src, int offset, long trg, long count) { - while (count > 0) { - long size = (count > UNSAFE_COPY_THRESHOLD) ? UNSAFE_COPY_THRESHOLD : count; - unsafe.copyMemory(src, BYTE_ARRAY_BASE_OFFSET + offset, null, trg, size); - count -= size; - offset += size; - trg += size; - } - } - - /** - * Transfers objCount bytes from Memory starting at memoryOffset to buffer starting at - * bufferOffset - * - * @param address start offset in the memory - * @param buffer the data buffer - * @param bufferOffset start offset of the buffer - * @param count number of bytes to transfer - */ - public static void getBytes(long address, byte[] buffer, int bufferOffset, int count) { - if (buffer == null) throw new NullPointerException(); - else if (bufferOffset < 0 || count < 0 || count > buffer.length - bufferOffset) - throw new IndexOutOfBoundsException(); - else if (count == 0) return; - - unsafe.copyMemory(null, address, buffer, BYTE_ARRAY_BASE_OFFSET + bufferOffset, count); - } - - public static char[] getStringValue(String str) { - return (char[]) MemoryUtil.unsafe.getObject(str, STRING_VALUE_OFFSET); - } - - public static void copyMemory(long fromAddr, long toAddr, int count) { - copyMemory(null, fromAddr, null, toAddr, count); - } - - public static void copyMemory( - 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) { - while (length > 0) { - long size = Math.min(length, UNSAFE_COPY_THRESHOLD); - unsafe.copyMemory(src, srcOffset, dst, dstOffset, size); - length -= size; - srcOffset += size; - dstOffset += size; - } - } else { - srcOffset += length; - dstOffset += length; - while (length > 0) { - long size = Math.min(length, UNSAFE_COPY_THRESHOLD); - srcOffset -= size; - dstOffset -= size; - unsafe.copyMemory(src, srcOffset, dst, dstOffset, size); - length -= size; - } - } - } - - public static long getTotalPhysicalMemorySize() { - return ((OperatingSystemMXBean) ManagementFactory.getOperatingSystemMXBean()) - .getTotalPhysicalMemorySize(); - } - - public static ByteBuffer allocateDirect(int cap) { - ByteBuffer bb = ByteBuffer.allocateDirect(cap); + public static ByteBuffer allocate(int cap) { + ByteBuffer bb = ByteBuffer.allocate(cap); // It make operation faster, but kill the cross platform ability. bb.order(ByteOrder.nativeOrder()); return bb; } - public static void readFully(ByteBuffer dst, CodecDataInput cdi, int length) { - // read bytes from cdi to buffer(off-heap) - long disAddr = MemoryUtil.getAddress(dst); - long bufPos = dst.position(); - for (int i = 0; i < length && !cdi.eof(); i++) { - byte b = cdi.readByte(); - MemoryUtil.setByte(disAddr + bufPos + i, b); - } - dst.position((int) (bufPos + length)); + public static ByteBuffer copyOf(ByteBuffer buf, int newCap) { + ByteBuffer newBuf = ByteBuffer.wrap(Arrays.copyOf(buf.array(), newCap)); + newBuf.position(buf.position()); + return newBuf; } - private static class Deallocator implements Runnable { - private long address; - - private Deallocator(long address) { - assert (address != 0); - this.address = address; - } - - @Override - public void run() { - if (address == 0) { - return; - } - free(address); - address = 0; + public static void readFully(ByteBuffer dst, CodecDataInput cdi, int length) { + int remaining; + for (remaining = length; remaining > 0 && !cdi.eof(); remaining--) { + byte b = cdi.readByte(); + dst.put(b); } + dst.position(dst.position() + remaining); } }