change blockid in pixels-cache from file name to real block id

Author: bianhq

.gitignore

@@ -21,4 +21,6 @@ package-info.java
 *.txt
 test30G_pixels/
 *.csv
-resources/*.xml
+resources/*.xml
+*.so
+.vscode

cpp/Makefile

@@ -0,0 +1,7 @@
+all: JNITest
+
+JNITest:
+	$(CXX) -shared -fPIC JNITest.cc -o JNITest.so -I/usr/lib/jvm/java-1.8.0-openjdk-amd64/include/ -I/usr/lib/jvm/java-8-openjdk-amd64/include/linux/
+
+clean:
+	rm -rf ./JNITest.so

pixels-cache/src/main/java/cn/edu/ruc/iir/pixels/cache/CacheLogger.java

@@ -0,0 +1,86 @@
+package cn.edu.ruc.iir.pixels.cache;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.util.Random;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+/**
+ * Created at: 19-5-12
+ * Author: hank
+ */
+public class CacheLogger implements Runnable
+{
+    private static final Logger logger = LogManager.getLogger(CacheLogger.class);
+
+    private ConcurrentLinkedQueue<Long> searchLatency = new ConcurrentLinkedQueue<>();
+    private ConcurrentLinkedQueue<Long> readLatency = new ConcurrentLinkedQueue<>();
+    private volatile boolean shutdown = false;
+
+    public void setShutdown()
+    {
+        this.shutdown = true;
+    }
+
+    public void addSearchLatency(long latency)
+    {
+        this.searchLatency.add(latency);
+    }
+
+    public void addReadLatency(long latency)
+    {
+        this.readLatency.add(latency);
+    }
+
+    @Override
+    public void run()
+    {
+        while (this.shutdown == false)
+        {
+            try
+            {
+                Thread.sleep(5000);
+            } catch (InterruptedException e)
+            {
+                e.printStackTrace();
+            }
+            Object[] searches = searchLatency.toArray();
+            searchLatency.clear();
+            Object[] reads = readLatency.toArray();
+            readLatency.clear();
+
+            long sum = 0;
+            for (Object latency : searches)
+            {
+                sum += (Long) latency;
+            }
+            logger.info("avg search latency: (" + (sum*1.0/searches.length) + ") ns");
+
+            sum = 0;
+            for (Object latency : reads)
+            {
+                sum += (Long) latency;
+            }
+            logger.info("avg read latency: (" + (sum*1.0/reads.length) + ") ns");
+
+        }
+    }
+
+    public static void main(String[] args) throws InterruptedException
+    {
+        CacheLogger cacheLogger = new CacheLogger();
+        Thread thread = new Thread(cacheLogger);
+        thread.start();
+        Random random = new Random();
+
+        for (int i = 0; i < 1000; ++i)
+        {
+            cacheLogger.addReadLatency(random.nextInt(100));
+            cacheLogger.addSearchLatency(random.nextInt(200));
+            Thread.sleep(100);
+        }
+
+        cacheLogger.setShutdown();
+    }
+}

pixels-cache/src/main/java/cn/edu/ruc/iir/pixels/cache/NativePixelsCacheReader.java

@@ -0,0 +1,29 @@
+package cn.edu.ruc.iir.pixels.cache;
+
+/**
+ * Created at: 19-5-11
+ * Author: hank
+ */
+public class NativePixelsCacheReader
+{
+    private String indexFileLocation;
+    private long indexFileSize;
+    private String cacheFileLocation;
+    private long cacheFileSize;
+
+    public NativePixelsCacheReader(String indexFileLocation, long indexFileSize,
+                                   String cacheFileLocation, long cacheFileSize)
+    {
+        this.indexFileLocation = indexFileLocation;
+        this.indexFileSize = indexFileSize;
+        this.cacheFileLocation = cacheFileLocation;
+        this.cacheFileSize = cacheFileSize;
+    }
+
+    static
+    {
+        System.loadLibrary("lib_pixels.so");
+    }
+
+    private native byte[] getFromCache();
+}

pixels-cache/src/main/java/cn/edu/ruc/iir/pixels/cache/PixelsCacheKey.java

@@ -1,10 +1,8 @@
 package cn.edu.ruc.iir.pixels.cache;
 
 import cn.edu.ruc.iir.pixels.common.utils.Constants;
-import org.apache.commons.compress.utils.CharsetNames;
 
 import java.nio.ByteBuffer;
-import java.nio.charset.Charset;
 import java.util.Arrays;
 import java.util.Objects;
 
@@ -19,18 +17,18 @@ public class PixelsCacheKey
 {
     private final int SIZE = 2 * Short.BYTES + Constants.MAX_BLOCK_ID_LEN;
     private final ByteBuffer keyBuffer = ByteBuffer.allocate(SIZE);
-    private String blockId;  // TODO: it could be better to use long as the type of blockid
+    private long blockId;  // TODO: it could be better to use long as the type of blockid
     private short rowGroupId;
     private short columnId;
 
-    public PixelsCacheKey(String blockId, short rowGroupId, short columnId)
+    public PixelsCacheKey(long blockId, short rowGroupId, short columnId)
     {
         this.blockId = blockId;
         this.rowGroupId = rowGroupId;
         this.columnId = columnId;
     }
 
-    public String getBlockId()
+    public long getBlockId()
     {
         return blockId;
     }
@@ -53,7 +51,7 @@ public byte[] getBytes()
         // And we'd better use long (int64) for block id, instead of a string file name.
         // Fixed key length (12 bytes) should be more efficient. I noticed that block ids in hdfs-2.7.3 looks
         // like a sequence number, not really random.
-        keyBuffer.put(blockId.getBytes(Charset.forName(CharsetNames.UTF_8)));
+        keyBuffer.putLong(blockId);
         keyBuffer.putShort(rowGroupId);
         keyBuffer.putShort(columnId);
         return Arrays.copyOfRange(keyBuffer.array(), 0, keyBuffer.position());

pixels-cache/src/main/java/cn/edu/ruc/iir/pixels/cache/PixelsCacheReader.java

@@ -66,7 +66,7 @@ public static PixelsCacheReader.Builder newBuilder()
      * @param columnId column id
      * @return columnlet content
      * */
-    public byte[] get(String blockId, short rowGroupId, short columnId)
+    public byte[] get(long blockId, short rowGroupId, short columnId)
     {
         byte[] content = new byte[0];
         // check rw flag, if not readable, return empty bytes
@@ -110,7 +110,7 @@ public byte[] get(String blockId, short rowGroupId, short columnId)
      * This interface is only used by TESTS, DO NOT USE.
      * It will be removed soon!
      * */
-    public PixelsCacheIdx search(String blockId, short rowGroupId, short columnId)
+    public PixelsCacheIdx search(long blockId, short rowGroupId, short columnId)
     {
         PixelsCacheKey cacheKey = new PixelsCacheKey(blockId, rowGroupId, columnId);
         byte[] cacheKeyBytes = cacheKey.getBytes();

pixels-cache/src/main/java/cn/edu/ruc/iir/pixels/cache/PixelsCacheWriter.java

@@ -1,5 +1,6 @@
 package cn.edu.ruc.iir.pixels.cache;
 
+import cn.edu.ruc.iir.pixels.common.exception.FSException;
 import cn.edu.ruc.iir.pixels.common.metadata.domain.Compact;
 import cn.edu.ruc.iir.pixels.common.metadata.domain.Layout;
 import cn.edu.ruc.iir.pixels.common.utils.Constants;
@@ -172,6 +173,10 @@ public int updateAll(int version, Layout layout)
         catch (IOException e) {
             e.printStackTrace();
             return -1;
+        } catch (FSException e)
+        {
+            e.printStackTrace();
+            return -1;
         }
     }
 
@@ -196,7 +201,7 @@ public void flush()
     }
 
     private int internalUpdate(int version, Layout layout, String[] files)
-            throws IOException
+            throws IOException, FSException
     {
         int status = 0;
         // get the new caching layout
@@ -242,7 +247,7 @@ private int internalUpdate(int version, Layout layout, String[] files)
                     break outer_loop;
                 }
                 else {
-                    radix.put(new PixelsCacheKey(file, rowGroupId, columnId), // TODO: it is not a good idea to use filename as block id.
+                    radix.put(new PixelsCacheKey(pixelsPhysicalReader.getCurrentBlockId(), rowGroupId, columnId), // TODO: it is not a good idea to use filename as block id.
                               new PixelsCacheIdx(cacheOffset, physicalLens[i]));
                     byte[] columnlet = pixelsPhysicalReader.read(physicalOffsets[i], physicalLens[i]);
                     cacheFile.putBytes(cacheOffset, columnlet);

pixels-cache/src/main/java/cn/edu/ruc/iir/pixels/cache/PixelsPhysicalReader.java

@@ -1,5 +1,6 @@
 package cn.edu.ruc.iir.pixels.cache;
 
+import cn.edu.ruc.iir.pixels.common.exception.FSException;
 import cn.edu.ruc.iir.pixels.common.physical.PhysicalReader;
 import cn.edu.ruc.iir.pixels.common.physical.PhysicalReaderUtil;
 import cn.edu.ruc.iir.pixels.core.PixelsProto;
@@ -38,6 +39,7 @@ private PixelsProto.FileTail readFileTail()
                 return PixelsProto.FileTail.parseFrom(fileTailBuffer);
             }
             catch (IOException e) {
+                // TODO: deal with this exception or throw it
                 e.printStackTrace();
             }
         }
@@ -66,4 +68,9 @@ public byte[] read(long offset, int length)
 
         return content;
     }
+
+    public long getCurrentBlockId() throws FSException
+    {
+        return physicalReader.getCurrentBlockId();
+    }
 }

pixels-cache/src/test/c++/Makefile

@@ -0,0 +1,7 @@
+all: test_jni
+
+test_jni:
+	$(CXX) -shared -fPIC $@.cc -o $@.so -I/usr/lib/jvm/java-1.8.0-openjdk-amd64/include/ -I/usr/lib/jvm/java-8-openjdk-amd64/include/linux/
+
+clean:
+	rm -rf ./TestJni.so

pixels-cache/src/test/c++/test_jni.cc

@@ -0,0 +1,26 @@
+#include <iostream>
+#include "test_jni.h"
+using namespace std;
+
+static int sum = 0;
+
+JNIEXPORT void JNICALL Java_cn_edu_ruc_iir_pixels_cache_TestJni_sayHello(JNIEnv *env, jclass cls)
+{
+    cout << "Hello Native\n" << endl;
+}
+
+JNIEXPORT jint JNICALL Java_cn_edu_ruc_iir_pixels_cache_TestJni_add
+  (JNIEnv *env, jclass cls, jint a)
+{
+    sum += a;
+    return sum;
+}
+
+JNIEXPORT jbyteArray JNICALL Java_cn_edu_ruc_iir_pixels_cache_TestJni_echo
+  (JNIEnv *env, jclass cls, jbyteArray bytes, jint length)
+{
+    jbyte* data = env->GetByteArrayElements(bytes, 0);
+    jbyteArray res = env->NewByteArray(length);
+    env->SetByteArrayRegion(res, 0, length, data);
+    return res;
+}