bundles/org.simantics.graph.db/src/org/simantics/graph/db/StreamingTransferableGraphFileReader.java

   1 package org.simantics.graph.db;\r
   2 \r
   3 import java.io.DataInput;\r
   4 import java.io.DataInputStream;\r
   5 import java.io.File;\r
   6 import java.io.IOException;\r
   7 import java.io.InputStream;\r
   8 import java.nio.ByteBuffer;\r
   9 import java.nio.channels.ReadableByteChannel;\r
  10 import java.util.ArrayList;\r
  11 import java.util.Arrays;\r
  12 import java.util.List;\r
  13 import java.util.TreeMap;\r
  14 \r
  15 import org.simantics.databoard.Bindings;\r
  16 import org.simantics.databoard.binding.Binding;\r
  17 import org.simantics.databoard.binding.mutable.Variant;\r
  18 import org.simantics.databoard.container.DataContainer;\r
  19 import org.simantics.databoard.container.DataContainers;\r
  20 import org.simantics.databoard.serialization.Serializer;\r
  21 import org.simantics.databoard.type.Datatype;\r
  22 import org.simantics.db.ReadGraph;\r
  23 import org.simantics.graph.db.TransferableGraphSource.TransferableGraphSourceValueProcedure;\r
  24 import org.simantics.graph.representation.ByteFileReader;\r
  25 import org.simantics.graph.representation.Extensions;\r
  26 import org.simantics.graph.representation.External;\r
  27 import org.simantics.graph.representation.Identity;\r
  28 import org.simantics.graph.representation.Internal;\r
  29 import org.simantics.graph.representation.Root;\r
  30 import org.simantics.graph.representation.Value;\r
  31 \r
  32 \r
  33 final public class StreamingTransferableGraphFileReader extends ByteFileReader {\r
  34 \r
  35         final static class InputChannel implements ReadableByteChannel {\r
  36 \r
  37                 final private InputStream stream;\r
  38                 \r
  39                 public InputChannel(InputStream stream) {\r
  40                         this.stream = stream;\r
  41                 }\r
  42                 \r
  43                 @Override\r
  44                 public boolean isOpen() {\r
  45                         return true;\r
  46                 }\r
  47 \r
  48                 @Override\r
  49                 public void close() throws IOException {\r
  50                 }\r
  51 \r
  52                 @Override\r
  53                 public int read(ByteBuffer dst) throws IOException {\r
  54                         int pos = dst.position();\r
  55                         int limit = dst.limit();\r
  56                         int i=stream.read(dst.array(), pos, limit-pos);\r
  57                         return i;\r
  58                 }\r
  59                 \r
  60         }\r
  61         \r
  62         private static boolean init = true;\r
  63         \r
  64         final private static int SIZE = 1<<18;\r
  65         \r
  66         public StreamingTransferableGraphFileReader(File file) throws Exception {\r
  67                 super(file, SIZE);\r
  68                 if(init) {\r
  69                         init=false;\r
  70                         @SuppressWarnings("resource")\r
  71                         StreamingTransferableGraphFileReader r = new StreamingTransferableGraphFileReader(file, 0);\r
  72                         for(int i=0;i<40000;i++) r.readTG();\r
  73                 }\r
  74         }\r
  75         \r
  76         public StreamingTransferableGraphFileReader(InputStream stream) throws Exception {\r
  77                 super(null, new InputChannel(stream), SIZE);\r
  78                 if(init) {\r
  79                         init=false;\r
  80                         @SuppressWarnings("resource")\r
  81                         StreamingTransferableGraphFileReader r = new StreamingTransferableGraphFileReader(stream, 0);\r
  82                         for(int i=0;i<40000;i++) r.readTG();\r
  83                 }\r
  84         }\r
  85 \r
  86         public StreamingTransferableGraphFileReader(ReadableByteChannel channel) throws Exception {\r
  87                 super(null, channel, SIZE);\r
  88                 if(init) {\r
  89                         init=false;\r
  90                         @SuppressWarnings("resource")\r
  91                         StreamingTransferableGraphFileReader r = new StreamingTransferableGraphFileReader(channel, 0);\r
  92                         for(int i=0;i<40000;i++) r.readTG();\r
  93                 }\r
  94         }\r
  95 \r
  96         public StreamingTransferableGraphFileReader(ReadableByteChannel channel, int size) throws IOException {\r
  97                 super(null, channel, SIZE);\r
  98         }\r
  99 \r
 100         public StreamingTransferableGraphFileReader(InputStream stream, int size) throws IOException {\r
 101                 super(null, new InputChannel(stream), size);\r
 102         }\r
 103 \r
 104         public StreamingTransferableGraphFileReader(File file, int size) throws IOException {\r
 105                 super(file, size);\r
 106         }\r
 107 \r
 108         class FileTransferableGraphSource implements TransferableGraphSource {\r
 109 \r
 110                 InputStream in = new InputStream() {\r
 111 \r
 112             @Override\r
 113             public int read() throws IOException {\r
 114                 return getByte();\r
 115             }\r
 116             \r
 117             @Override\r
 118             public int read(byte[] b) throws IOException {\r
 119                 // FIXME not correctly implemented                \r
 120                 System.arraycopy(safeBytes(b.length), 0, b, 0, b.length);\r
 121                 return b.length;\r
 122             }\r
 123             \r
 124             @Override\r
 125             public int read(byte[] b, int off, int len) throws IOException {\r
 126                 for(int i=0;i<len;i++)\r
 127                     b[off++] = (byte)getByte();\r
 128                 return len;\r
 129             }\r
 130                     \r
 131                 };\r
 132                 DataInputStream dis = new DataInputStream(in);\r
 133                 \r
 134                 DataContainer header;\r
 135                 Extensions extensions;\r
 136                 int resourceCount;\r
 137                 \r
 138                 private int identities;\r
 139                 private int stmLength;\r
 140                 private int valueLength;\r
 141                 \r
 142                 public FileTransferableGraphSource() throws Exception {\r
 143                     init();\r
 144                 }\r
 145                 \r
 146                 private void init() throws Exception {\r
 147 \r
 148             // Header\r
 149             header = DataContainers.readHeader(dis);\r
 150             \r
 151             // Content variant data type\r
 152             Bindings.getSerializerUnchecked(Datatype.class).deserialize((DataInput)dis);\r
 153 \r
 154             resourceCount = safeInt();\r
 155             \r
 156             List<Object> idcontext = new ArrayList<Object>(); \r
 157             extensions = (Extensions)Bindings.getSerializerUnchecked(Extensions.class).deserialize((DataInput)dis, idcontext);\r
 158                     \r
 159                 }\r
 160                 \r
 161         @Override\r
 162         public void reset() throws Exception {\r
 163             StreamingTransferableGraphFileReader.this.reset();\r
 164             throw new UnsupportedOperationException();\r
 165         }\r
 166         \r
 167                 @Override\r
 168                 public DataContainer getHeader() throws Exception {\r
 169                     return header;\r
 170                 }\r
 171                 \r
 172                 @Override\r
 173                 public int getResourceCount() throws Exception {\r
 174                         return resourceCount;\r
 175                 }\r
 176 \r
 177                 @Override\r
 178                 public int getIdentityCount() throws Exception {\r
 179                         identities = safeInt();\r
 180                         return identities;\r
 181                 }\r
 182 \r
 183                 @Override\r
 184                 public int getStatementCount() throws Exception {\r
 185                         stmLength = safeInt();\r
 186                         return stmLength;\r
 187                 }\r
 188 \r
 189                 @Override\r
 190                 public int getValueCount() throws Exception {\r
 191                         valueLength = safeInt();\r
 192                         return valueLength;\r
 193                 }\r
 194 \r
 195                 @Override\r
 196                 public void forStatements(ReadGraph graph, TransferableGraphSourceProcedure<int[]> procedure) throws Exception {\r
 197                         \r
 198                         int[] value = new int[4];\r
 199 \r
 200                         for(int stmIndex=0;stmIndex<stmLength;) {\r
 201                                 \r
 202                                 value[stmIndex & 3] = safeInt();\r
 203                                 stmIndex++;\r
 204                                 if((stmIndex & 3) == 0) procedure.execute(value);\r
 205                                 \r
 206                                 // Cached bytes \r
 207                                 int avail = (SIZE-byteIndex) >> 2;\r
 208                                 int allowed = Math.min(stmLength-stmIndex, avail);\r
 209                                 for(int index = byteIndex, i=0;i<allowed;i++) {\r
 210                                         value[stmIndex & 3] = ((bytes[index++]&0xff)<<24) | ((bytes[index++]&0xff)<<16) | ((bytes[index++]&0xff)<<8) | ((bytes[index++]&0xff));\r
 211                                         stmIndex++;\r
 212                                         if((stmIndex & 3) == 0) procedure.execute(value);\r
 213 //                                      statements[stmIndex++] = ((bytes[index++]&0xff)<<24) | ((bytes[index++]&0xff)<<16) | ((bytes[index++]&0xff)<<8) | ((bytes[index++]&0xff));                              \r
 214                                 }\r
 215                                 byteIndex += allowed<<2;\r
 216                                 \r
 217                         }\r
 218                         \r
 219                 }\r
 220 \r
 221                 @Override\r
 222                 public void forIdentities(ReadGraph graph, TransferableGraphSourceProcedure<Identity> procedure) throws Exception {\r
 223                         \r
 224                         for(int i=0;i<identities;i++) {\r
 225                                 \r
 226                                 int rid = safeInt();\r
 227                                 byte type = bytes[byteIndex++];\r
 228                                 // External\r
 229                                 if(type == 1) {\r
 230                                         \r
 231                                         int parent = safeInt();\r
 232                                         int nameLen = bytes[byteIndex++]&0xff;\r
 233                                         \r
 234                                         if(byteIndex+nameLen < SIZE) {\r
 235                                                 procedure.execute(new Identity(rid, new External(parent, utf(bytes, byteIndex, byteIndex + nameLen))));\r
 236                                                 byteIndex += nameLen;\r
 237                                         } else {\r
 238                                                 procedure.execute(new Identity(rid, new External(parent, utf(safeBytes(nameLen), 0, nameLen))));\r
 239                                         }\r
 240                                         \r
 241                                 } \r
 242                                 // Internal\r
 243                                 else if(type == 3) {\r
 244                                         \r
 245                                         int parent = safeInt();\r
 246                                         int nameLen = bytes[byteIndex++]&0xff;\r
 247                                         if(byteIndex+nameLen < SIZE) {\r
 248                                                 procedure.execute(new Identity(rid, new Internal(parent, utf(bytes, byteIndex, byteIndex + nameLen))));\r
 249                                                 byteIndex += nameLen;\r
 250                                         } else {\r
 251                                                 procedure.execute(new Identity(rid, new Internal(parent, utf(safeBytes(nameLen), 0, nameLen))));\r
 252                                         }\r
 253                                         \r
 254                                 }\r
 255                                 // Root\r
 256                                 else if(type == 0) {\r
 257                                         \r
 258                                         int nameLen = bytes[byteIndex++]&0xff;\r
 259                                         String name = utf(safeBytes(nameLen), 0, nameLen);\r
 260                                         int nameLen2 = bytes[byteIndex++]&0xff;\r
 261                                         String rType = utf(safeBytes(nameLen2), 0, nameLen2);\r
 262                                         procedure.execute(new Identity(rid, new Root(name, rType)));\r
 263 \r
 264                                 } else if(type == 2) {\r
 265                                         throw new UnsupportedOperationException();\r
 266                                 }\r
 267 \r
 268                         }\r
 269                         \r
 270                 }\r
 271 \r
 272                 @Override\r
 273                 public void forValues(ReadGraph graph, TransferableGraphSourceProcedure<Value> procedure) throws Exception {\r
 274                         \r
 275                         Serializer variantSerializer = Bindings.getSerializerUnchecked(Bindings.VARIANT);\r
 276                         \r
 277                         List<Object> idcontext = new ArrayList<>(); \r
 278                         \r
 279                         for(int i=0;i<valueLength;i++) {\r
 280                                 int resource = safeInt();\r
 281                                 Variant value = (Variant)variantSerializer\r
 282                                         .deserialize((DataInput)dis, idcontext);\r
 283                                 procedure.execute(new Value(resource, value));\r
 284                         }\r
 285                         \r
 286                         \r
 287                 }\r
 288                 \r
 289                 @Override\r
 290                 public void forValues2(ReadGraph graph, TransferableGraphSourceValueProcedure procedure) throws Exception {\r
 291 \r
 292                     Binding datatypeBinding = Bindings.getBinding(Datatype.class);\r
 293             Serializer datatypeSerializer = Bindings.getSerializerUnchecked(datatypeBinding);\r
 294             \r
 295             List<Object> idContext = new ArrayList<>(); \r
 296             \r
 297             for(int i=0;i<valueLength;i++) {\r
 298                 int resource = safeInt();\r
 299                 idContext.clear();\r
 300                 Datatype type = (Datatype)datatypeSerializer.deserialize((DataInput)dis, idContext);\r
 301                 procedure.execute(resource, type, dis);\r
 302             }\r
 303                     \r
 304                 }\r
 305 \r
 306                 @Override\r
 307                 public TreeMap<String, Variant> getExtensions() {\r
 308                         return extensions.map;\r
 309                 }\r
 310 \r
 311                 @Override\r
 312                 public void close() {\r
 313                 }\r
 314         }\r
 315         \r
 316         public TransferableGraphSource readTG() throws Exception {\r
 317 \r
 318                 if(getSize() == 0) return null;\r
 319                 \r
 320                 return new FileTransferableGraphSource();\r
 321 \r
 322         }\r
 323         \r
 324     public static void main(String[] args) {\r
 325 \r
 326         try {\r
 327 \r
 328             File file = new File("c:/work/Model.apros");\r
 329             StreamingTransferableGraphFileReader reader = new StreamingTransferableGraphFileReader(file, SIZE);\r
 330             reader = new StreamingTransferableGraphFileReader(file);\r
 331             long s = System.nanoTime();\r
 332             TransferableGraphSource tgs = reader.readTG();\r
 333             int ids = tgs.getIdentityCount();\r
 334             System.out.println("identity count " + ids);\r
 335 //            tgs.forIdentities(null, id -> { /*System.out.println("Identity: " + id);*/ });\r
 336             tgs.forIdentities(null, id -> { System.out.println("Identity: " + id); });\r
 337             int stats = tgs.getStatementCount();\r
 338             System.out.println("statement count " + stats/4 + " (" + stats + ")");\r
 339 //            tgs.forStatements(null, id -> { /*System.out.println(Arrays.toString(id));*/ });\r
 340             tgs.forStatements(null, id -> { System.out.println(Arrays.toString(id)); });\r
 341             int values = tgs.getValueCount();\r
 342             System.out.println("value count " + values);\r
 343             int[] valueNo = {0};\r
 344             tgs.forValues2(null, new TransferableGraphSourceValueProcedure() {\r
 345                 @Override\r
 346                 public void rawCopy(int resource, int length, DataInput input) throws Exception {\r
 347                     System.out.println("value " + (valueNo[0]++) + ": rawCopy("+ resource + ", " + length + ", " + input + ")");\r
 348                     for (int i = 0; i < length; ++i)\r
 349                         input.readByte();\r
 350                 }\r
 351                 @Override\r
 352                 public void execute(int resource, Datatype type, DataInput input) throws Exception {\r
 353                     Object value = Bindings.getSerializer(Bindings.getBinding(type)).deserialize(input);\r
 354                     System.out.println("value " + (valueNo[0]++) + ": execute("+ resource + ", " + type.toSingleLineString() + ", " + input + "): " + value);\r
 355                 }\r
 356             });\r
 357             long d = System.nanoTime() - s;\r
 358             System.err.println("Duration=" + 1e-9*d + "s.");\r
 359         } catch (Throwable t) {\r
 360             t.printStackTrace();\r
 361         }\r
 362     }\r
 363 \r
 364 }\r