DatasetInputAdapter implements InputAdapter { @Override public ITuple pullDataItem(int timeout) { ITuple i = dataset.consumeData_zerocopy(iTuple); return i; } DatasetInputAdapter(WorkerConfig wc, int streamId, Dataset dataset); @Override int getStreamId(); @Override short returnType(); @Override DataStoreType getDataStoreType(); @Override ITuple pullDataItem(int timeout); ITuple _pullDataItem(int timeout); @Override List<ITuple> pullDataItems(int timeout);  }
@Test public void test() { Properties p = new Properties(); p.setProperty(WorkerConfig.WORKER_IP, ""); p.setProperty(WorkerConfig.MASTER_IP, ""); p.setProperty(WorkerConfig.PROPERTIES_FILE, ""); WorkerConfig wc = new WorkerConfig(p); DataReferenceManager drm = DataReferenceManager.makeDataReferenceManager(wc); Schema schema = SchemaBuilder.getInstance().newField(Type.INT, "uid").newField(Type.LONG, "ts").build(); DataStore dataStore = new DataStore(schema, DataStoreType.IN_MEMORY); InetAddress whatever = null; try { whatever = InetAddress.getLocalHost(); } catch (UnknownHostException e) { e.printStackTrace(); } ControlEndPoint endPoint = new ControlEndPoint(0, whatever.getHostAddress(), 0); DataReference dr = DataReference.makeManagedDataReference(dataStore, endPoint, ServeMode.STORE); OBuffer ob = drm.manageNewDataReference(dr); int uid = 1; long ts = 1; byte[] ot1 = OTuple.create(schema, new String[]{"uid", "ts"}, new Object[]{uid,ts}); uid = 2; ts = 2; byte[] ot2 = OTuple.create(schema, new String[]{"uid", "ts"}, new Object[]{uid,ts}); uid = 3; ts = 3; byte[] ot3 = OTuple.create(schema, new String[]{"uid", "ts"}, new Object[]{uid,ts}); ob.write(ot1, null); ob.write(ot2, null); ob.write(ot3, null); IBuffer ib = drm.getInputBufferFor(dr); DatasetInputAdapter dia = new DatasetInputAdapter(wc, 0, ((Dataset)ib)); ITuple iTuple = dia.pullDataItem(100); int uid1 = iTuple.getInt("uid"); long ts1 = iTuple.getLong("ts"); System.out.println("1- UID: " + uid1 + ", 2- TS: " + ts1); iTuple = dia.pullDataItem(100); int uid2 = iTuple.getInt("uid"); long ts2 = iTuple.getLong("ts"); System.out.println("2- UID: " + uid2 + ", 2- TS: " + ts2); iTuple = dia.pullDataItem(100); int uid3 = iTuple.getInt("uid"); long ts3 = iTuple.getLong("ts"); System.out.println("3- UID: " + uid3 + ", 2- TS: " + ts3); }
@Test public void testMultipleBuffers() { Properties p = new Properties(); p.setProperty(WorkerConfig.WORKER_IP, ""); p.setProperty(WorkerConfig.MASTER_IP, ""); p.setProperty(WorkerConfig.PROPERTIES_FILE, ""); WorkerConfig wc = new WorkerConfig(p); DataReferenceManager drm = DataReferenceManager.makeDataReferenceManager(wc); Schema schema = SchemaBuilder.getInstance().newField(Type.LONG, "uid").newField(Type.LONG, "ts").build(); DataStore dataStore = new DataStore(schema, DataStoreType.IN_MEMORY); InetAddress whatever = null; try { whatever = InetAddress.getLocalHost(); } catch (UnknownHostException e) { e.printStackTrace(); } ControlEndPoint endPoint = new ControlEndPoint(0, whatever.getHostAddress(), 0); DataReference dr = DataReference.makeManagedDataReference(dataStore, endPoint, ServeMode.STORE); OBuffer ob = drm.manageNewDataReference(dr); int numTuplesToWrite = 500; long uid = 1; long ts = 1; for(int i = 0; i < numTuplesToWrite; i++) { byte[] ot1 = OTuple.create(schema, new String[]{"uid", "ts"}, new Object[]{uid,ts}); ob.write(ot1, null); uid++; ts++; } IBuffer ib = drm.getInputBufferFor(dr); DatasetInputAdapter dia = new DatasetInputAdapter(wc, 0, ((Dataset)ib)); for(int i = 0; i < numTuplesToWrite; i++) { ITuple iTuple = dia.pullDataItem(100); long uid1 = iTuple.getInt("uid"); long ts1 = iTuple.getLong("ts"); System.out.println("1- UID: " + uid1 + ", 2- TS: " + ts1); } }
@Test public void testWriteReadMicrobenchmark() { Properties p = new Properties(); p.setProperty(WorkerConfig.WORKER_IP, ""); p.setProperty(WorkerConfig.MASTER_IP, ""); p.setProperty(WorkerConfig.PROPERTIES_FILE, ""); WorkerConfig wc = new WorkerConfig(p); DataReferenceManager drm = DataReferenceManager.makeDataReferenceManager(wc); Schema schema = SchemaBuilder.getInstance().newField(Type.LONG, "uid").newField(Type.LONG, "ts").build(); DataStore dataStore = new DataStore(schema, DataStoreType.IN_MEMORY); InetAddress whatever = null; try { whatever = InetAddress.getLocalHost(); } catch (UnknownHostException e) { e.printStackTrace(); } ControlEndPoint endPoint = new ControlEndPoint(0, whatever.getHostAddress(), 0); DataReference dr = DataReference.makeManagedDataReference(dataStore, endPoint, ServeMode.STORE); OBuffer ob = drm.manageNewDataReference(dr); int numTuplesToWrite = 3000000; long uid = 1; long ts = 1; long startW = System.currentTimeMillis(); for(int i = 0; i < numTuplesToWrite; i++) { byte[] ot1 = OTuple.create(schema, new String[]{"uid", "ts"}, new Object[]{uid,ts}); ob.write(ot1, null); uid++; ts++; } long endW = System.currentTimeMillis(); IBuffer ib = drm.getInputBufferFor(dr); DatasetInputAdapter dia = new DatasetInputAdapter(wc, 0, ((Dataset)ib)); long startR = System.currentTimeMillis(); for(int i = 0; i < numTuplesToWrite; i++) { ITuple iTuple = dia.pullDataItem(100); long uid1 = iTuple.getInt("uid"); long ts1 = iTuple.getLong("ts"); } long endR = System.currentTimeMillis(); System.out.println("Time to write: " + (endW - startW)); System.out.println("Time to read: " + (endR - startR)); }
@Test public void testWriteReadMicrobenchmarkIsolated() { Properties p = new Properties(); p.setProperty(WorkerConfig.WORKER_IP, ""); p.setProperty(WorkerConfig.MASTER_IP, ""); p.setProperty(WorkerConfig.PROPERTIES_FILE, ""); WorkerConfig wc = new WorkerConfig(p); DataReferenceManager drm = DataReferenceManager.makeDataReferenceManager(wc); Schema schema = SchemaBuilder.getInstance().newField(Type.LONG, "uid").newField(Type.LONG, "ts").build(); DataStore dataStore = new DataStore(schema, DataStoreType.IN_MEMORY); InetAddress whatever = null; try { whatever = InetAddress.getLocalHost(); } catch (UnknownHostException e) { e.printStackTrace(); } ControlEndPoint endPoint = new ControlEndPoint(0, whatever.getHostAddress(), 0); DataReference dr = DataReference.makeManagedDataReference(dataStore, endPoint, ServeMode.STORE); OBuffer ob = drm.manageNewDataReference(dr); int numTuplesToWrite = 3000000; long uid = 1; long ts = 1; byte[] ot1 = OTuple.create(schema, new String[]{"uid", "ts"}, new Object[]{uid,ts}); long startW = System.currentTimeMillis(); for(int i = 0; i < numTuplesToWrite; i++) { ob.write(ot1, null); } long endW = System.currentTimeMillis(); IBuffer ib = drm.getInputBufferFor(dr); DatasetInputAdapter dia = new DatasetInputAdapter(wc, 0, ((Dataset)ib)); long startR = System.currentTimeMillis(); for(int i = 0; i < numTuplesToWrite; i++) { ITuple iTuple = dia.pullDataItem(100); } long endR = System.currentTimeMillis(); System.out.println("(Isolation) Time to write: " + (endW - startW)); System.out.println("(Isolation) Time to read: " + (endR - startR)); }