X-Git-Url: https://gerrit.simantics.org/r/gitweb?a=blobdiff_plain;f=bundles%2Forg.simantics.graph.db%2Fsrc%2Forg%2Fsimantics%2Fgraph%2Fdb%2FStreamingTransferableGraphFileReader.java;fp=bundles%2Forg.simantics.graph.db%2Fsrc%2Forg%2Fsimantics%2Fgraph%2Fdb%2FStreamingTransferableGraphFileReader.java;h=89c0620d53cd1de2df0ed7f113e162230de9f4d5;hb=969bd23cab98a79ca9101af33334000879fb60c5;hp=0000000000000000000000000000000000000000;hpb=866dba5cd5a3929bbeae85991796acb212338a08;p=simantics%2Fplatform.git

diff --git a/bundles/org.simantics.graph.db/src/org/simantics/graph/db/StreamingTransferableGraphFileReader.java b/bundles/org.simantics.graph.db/src/org/simantics/graph/db/StreamingTransferableGraphFileReader.java
new file mode 100644
index 000000000..89c0620d5
--- /dev/null
+++ b/bundles/org.simantics.graph.db/src/org/simantics/graph/db/StreamingTransferableGraphFileReader.java
@@ -0,0 +1,364 @@
+package org.simantics.graph.db;
+
+import java.io.DataInput;
+import java.io.DataInputStream;
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.nio.channels.ReadableByteChannel;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.TreeMap;
+
+import org.simantics.databoard.Bindings;
+import org.simantics.databoard.binding.Binding;
+import org.simantics.databoard.binding.mutable.Variant;
+import org.simantics.databoard.container.DataContainer;
+import org.simantics.databoard.container.DataContainers;
+import org.simantics.databoard.serialization.Serializer;
+import org.simantics.databoard.type.Datatype;
+import org.simantics.db.ReadGraph;
+import org.simantics.graph.db.TransferableGraphSource.TransferableGraphSourceValueProcedure;
+import org.simantics.graph.representation.ByteFileReader;
+import org.simantics.graph.representation.Extensions;
+import org.simantics.graph.representation.External;
+import org.simantics.graph.representation.Identity;
+import org.simantics.graph.representation.Internal;
+import org.simantics.graph.representation.Root;
+import org.simantics.graph.representation.Value;
+
+
+final public class StreamingTransferableGraphFileReader extends ByteFileReader {
+
+	final static class InputChannel implements ReadableByteChannel {
+
+		final private InputStream stream;
+		
+		public InputChannel(InputStream stream) {
+			this.stream = stream;
+		}
+		
+		@Override
+		public boolean isOpen() {
+			return true;
+		}
+
+		@Override
+		public void close() throws IOException {
+		}
+
+		@Override
+		public int read(ByteBuffer dst) throws IOException {
+			int pos = dst.position();
+			int limit = dst.limit();
+			int i=stream.read(dst.array(), pos, limit-pos);
+			return i;
+		}
+		
+	}
+	
+	private static boolean init = true;
+	
+	final private static int SIZE = 1<<18;
+	
+	public StreamingTransferableGraphFileReader(File file) throws Exception {
+		super(file, SIZE);
+		if(init) {
+			init=false;
+			@SuppressWarnings("resource")
+			StreamingTransferableGraphFileReader r = new StreamingTransferableGraphFileReader(file, 0);
+			for(int i=0;i<40000;i++) r.readTG();
+		}
+	}
+	
+	public StreamingTransferableGraphFileReader(InputStream stream) throws Exception {
+		super(null, new InputChannel(stream), SIZE);
+		if(init) {
+			init=false;
+			@SuppressWarnings("resource")
+			StreamingTransferableGraphFileReader r = new StreamingTransferableGraphFileReader(stream, 0);
+			for(int i=0;i<40000;i++) r.readTG();
+		}
+	}
+
+	public StreamingTransferableGraphFileReader(ReadableByteChannel channel) throws Exception {
+		super(null, channel, SIZE);
+		if(init) {
+			init=false;
+			@SuppressWarnings("resource")
+			StreamingTransferableGraphFileReader r = new StreamingTransferableGraphFileReader(channel, 0);
+			for(int i=0;i<40000;i++) r.readTG();
+		}
+	}
+
+	public StreamingTransferableGraphFileReader(ReadableByteChannel channel, int size) throws IOException {
+		super(null, channel, SIZE);
+	}
+
+	public StreamingTransferableGraphFileReader(InputStream stream, int size) throws IOException {
+		super(null, new InputChannel(stream), size);
+	}
+
+	public StreamingTransferableGraphFileReader(File file, int size) throws IOException {
+		super(file, size);
+	}
+
+	class FileTransferableGraphSource implements TransferableGraphSource {
+
+		InputStream in = new InputStream() {
+
+            @Override
+            public int read() throws IOException {
+                return getByte();
+            }
+            
+            @Override
+            public int read(byte[] b) throws IOException {
+                // FIXME not correctly implemented                
+                System.arraycopy(safeBytes(b.length), 0, b, 0, b.length);
+                return b.length;
+            }
+            
+            @Override
+            public int read(byte[] b, int off, int len) throws IOException {
+                for(int i=0;i<len;i++)
+                    b[off++] = (byte)getByte();
+                return len;
+            }
+		    
+		};
+		DataInputStream dis = new DataInputStream(in);
+		
+		DataContainer header;
+		Extensions extensions;
+		int resourceCount;
+		
+		private int identities;
+		private int stmLength;
+		private int valueLength;
+		
+		public FileTransferableGraphSource() throws Exception {
+		    init();
+		}
+		
+		private void init() throws Exception {
+
+            // Header
+            header = DataContainers.readHeader(dis);
+            
+            // Content variant data type
+            Bindings.getSerializerUnchecked(Datatype.class).deserialize((DataInput)dis);
+
+            resourceCount = safeInt();
+            
+            List<Object> idcontext = new ArrayList<Object>(); 
+            extensions = (Extensions)Bindings.getSerializerUnchecked(Extensions.class).deserialize((DataInput)dis, idcontext);
+		    
+		}
+		
+        @Override
+        public void reset() throws Exception {
+            StreamingTransferableGraphFileReader.this.reset();
+            throw new UnsupportedOperationException();
+        }
+        
+		@Override
+		public DataContainer getHeader() throws Exception {
+		    return header;
+		}
+		
+		@Override
+		public int getResourceCount() throws Exception {
+			return resourceCount;
+		}
+
+		@Override
+		public int getIdentityCount() throws Exception {
+			identities = safeInt();
+			return identities;
+		}
+
+		@Override
+		public int getStatementCount() throws Exception {
+			stmLength = safeInt();
+			return stmLength;
+		}
+
+		@Override
+		public int getValueCount() throws Exception {
+			valueLength = safeInt();
+			return valueLength;
+		}
+
+		@Override
+		public void forStatements(ReadGraph graph, TransferableGraphSourceProcedure<int[]> procedure) throws Exception {
+			
+			int[] value = new int[4];
+
+			for(int stmIndex=0;stmIndex<stmLength;) {
+				
+				value[stmIndex & 3] = safeInt();
+				stmIndex++;
+				if((stmIndex & 3) == 0) procedure.execute(value);
+				
+				// Cached bytes 
+				int avail = (SIZE-byteIndex) >> 2;
+				int allowed = Math.min(stmLength-stmIndex, avail);
+				for(int index = byteIndex, i=0;i<allowed;i++) {
+					value[stmIndex & 3] = ((bytes[index++]&0xff)<<24) | ((bytes[index++]&0xff)<<16) | ((bytes[index++]&0xff)<<8) | ((bytes[index++]&0xff));
+					stmIndex++;
+					if((stmIndex & 3) == 0) procedure.execute(value);
+//					statements[stmIndex++] = ((bytes[index++]&0xff)<<24) | ((bytes[index++]&0xff)<<16) | ((bytes[index++]&0xff)<<8) | ((bytes[index++]&0xff));    				
+				}
+				byteIndex += allowed<<2;
+				
+			}
+			
+		}
+
+		@Override
+		public void forIdentities(ReadGraph graph, TransferableGraphSourceProcedure<Identity> procedure) throws Exception {
+			
+			for(int i=0;i<identities;i++) {
+				
+				int rid = safeInt();
+				byte type = bytes[byteIndex++];
+				// External
+				if(type == 1) {
+					
+					int parent = safeInt();
+					int nameLen = bytes[byteIndex++]&0xff;
+					
+					if(byteIndex+nameLen < SIZE) {
+						procedure.execute(new Identity(rid, new External(parent, utf(bytes, byteIndex, byteIndex + nameLen))));
+						byteIndex += nameLen;
+					} else {
+						procedure.execute(new Identity(rid, new External(parent, utf(safeBytes(nameLen), 0, nameLen))));
+					}
+					
+				} 
+				// Internal
+				else if(type == 3) {
+					
+					int parent = safeInt();
+					int nameLen = bytes[byteIndex++]&0xff;
+					if(byteIndex+nameLen < SIZE) {
+						procedure.execute(new Identity(rid, new Internal(parent, utf(bytes, byteIndex, byteIndex + nameLen))));
+						byteIndex += nameLen;
+					} else {
+						procedure.execute(new Identity(rid, new Internal(parent, utf(safeBytes(nameLen), 0, nameLen))));
+					}
+					
+				}
+				// Root
+				else if(type == 0) {
+					
+					int nameLen = bytes[byteIndex++]&0xff;
+					String name = utf(safeBytes(nameLen), 0, nameLen);
+					int nameLen2 = bytes[byteIndex++]&0xff;
+					String rType = utf(safeBytes(nameLen2), 0, nameLen2);
+					procedure.execute(new Identity(rid, new Root(name, rType)));
+
+				} else if(type == 2) {
+					throw new UnsupportedOperationException();
+				}
+
+			}
+			
+		}
+
+		@Override
+		public void forValues(ReadGraph graph, TransferableGraphSourceProcedure<Value> procedure) throws Exception {
+			
+			Serializer variantSerializer = Bindings.getSerializerUnchecked(Bindings.VARIANT);
+			
+			List<Object> idcontext = new ArrayList<>(); 
+			
+			for(int i=0;i<valueLength;i++) {
+				int resource = safeInt();
+				Variant value = (Variant)variantSerializer
+				        .deserialize((DataInput)dis, idcontext);
+				procedure.execute(new Value(resource, value));
+			}
+			
+			
+		}
+		
+		@Override
+		public void forValues2(ReadGraph graph, TransferableGraphSourceValueProcedure procedure) throws Exception {
+
+		    Binding datatypeBinding = Bindings.getBinding(Datatype.class);
+            Serializer datatypeSerializer = Bindings.getSerializerUnchecked(datatypeBinding);
+            
+            List<Object> idContext = new ArrayList<>(); 
+            
+            for(int i=0;i<valueLength;i++) {
+                int resource = safeInt();
+                idContext.clear();
+                Datatype type = (Datatype)datatypeSerializer.deserialize((DataInput)dis, idContext);
+                procedure.execute(resource, type, dis);
+            }
+		    
+		}
+
+		@Override
+		public TreeMap<String, Variant> getExtensions() {
+			return extensions.map;
+		}
+
+		@Override
+		public void close() {
+		}
+	}
+	
+	public TransferableGraphSource readTG() throws Exception {
+
+		if(getSize() == 0) return null;
+		
+		return new FileTransferableGraphSource();
+
+	}
+	
+    public static void main(String[] args) {
+
+        try {
+
+            File file = new File("c:/work/Model.apros");
+            StreamingTransferableGraphFileReader reader = new StreamingTransferableGraphFileReader(file, SIZE);
+            reader = new StreamingTransferableGraphFileReader(file);
+            long s = System.nanoTime();
+            TransferableGraphSource tgs = reader.readTG();
+            int ids = tgs.getIdentityCount();
+            System.out.println("identity count " + ids);
+//            tgs.forIdentities(null, id -> { /*System.out.println("Identity: " + id);*/ });
+            tgs.forIdentities(null, id -> { System.out.println("Identity: " + id); });
+            int stats = tgs.getStatementCount();
+            System.out.println("statement count " + stats/4 + " (" + stats + ")");
+//            tgs.forStatements(null, id -> { /*System.out.println(Arrays.toString(id));*/ });
+            tgs.forStatements(null, id -> { System.out.println(Arrays.toString(id)); });
+            int values = tgs.getValueCount();
+            System.out.println("value count " + values);
+            int[] valueNo = {0};
+            tgs.forValues2(null, new TransferableGraphSourceValueProcedure() {
+                @Override
+                public void rawCopy(int resource, int length, DataInput input) throws Exception {
+                    System.out.println("value " + (valueNo[0]++) + ": rawCopy("+ resource + ", " + length + ", " + input + ")");
+                    for (int i = 0; i < length; ++i)
+                        input.readByte();
+                }
+                @Override
+                public void execute(int resource, Datatype type, DataInput input) throws Exception {
+                    Object value = Bindings.getSerializer(Bindings.getBinding(type)).deserialize(input);
+                    System.out.println("value " + (valueNo[0]++) + ": execute("+ resource + ", " + type.toSingleLineString() + ", " + input + "): " + value);
+                }
+            });
+            long d = System.nanoTime() - s;
+            System.err.println("Duration=" + 1e-9*d + "s.");
+        } catch (Throwable t) {
+            t.printStackTrace();
+        }
+    }
+
+}