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

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