bundles/org.simantics.layer0.utils/src/org/simantics/layer0/utils/writer/DelayedGraphWriter.java

   1 /*******************************************************************************
   2  * Copyright (c) 2007, 2010 Association for Decentralized Information Management
   3  * in Industry THTH ry.
   4  * All rights reserved. This program and the accompanying materials
   5  * are made available under the terms of the Eclipse Public License v1.0
   6  * which accompanies this distribution, and is available at
   7  * http://www.eclipse.org/legal/epl-v10.html
   8  *
   9  * Contributors:
  10  *     VTT Technical Research Centre of Finland - initial API and implementation
  11  *******************************************************************************/
  12 package org.simantics.layer0.utils.writer;
  13
  14
  15 import java.io.BufferedInputStream;
  16 import java.io.BufferedOutputStream;
  17 import java.io.File;
  18 import java.io.FileInputStream;
  19 import java.io.FileOutputStream;
  20 import java.io.IOException;
  21 import java.io.ObjectInputStream;
  22 import java.io.ObjectOutputStream;
  23 import java.nio.channels.FileChannel;
  24
  25 import org.eclipse.core.runtime.IProgressMonitor;
  26 import org.simantics.db.ReadGraph;
  27 import org.simantics.db.Resource;
  28 import org.simantics.db.WriteOnlyGraph;
  29 import org.simantics.db.exception.DatabaseException;
  30
  31 import gnu.trove.map.hash.TIntIntHashMap;
  32 import gnu.trove.map.hash.TIntLongHashMap;
  33
  34 public class DelayedGraphWriter extends AbstractDelayedGraphWriter {
  35
  36         protected File file;
  37         protected ObjectOutputStream s;
  38         TIntIntHashMap clusterHints = new TIntIntHashMap();
  39         TIntLongHashMap clusterIds = new TIntLongHashMap();
  40
  41         public DelayedGraphWriter(ReadGraph graph) {
  42                 super(graph);
  43                 try {
  44                         file = File.createTempFile("graph", ".tmp");
  45                         System.out.println("Temp file: " + file.getAbsolutePath());
  46                         s = new ObjectOutputStream(new BufferedOutputStream(new FileOutputStream(file), 1024*1024));
  47 //                      s = new ObjectOutputStream(backup);
  48                         file.deleteOnExit();
  49                 } catch (IOException e) {
  50                         throw new RuntimeException("Opening output stream to temporary file failed", e);
  51                 }
  52         }
  53
  54         private void writeRef(int ref) throws IOException {
  55                 if(ref > 0)
  56                         timestamps.set(ref-1, time++);
  57                 s.writeInt(ref);
  58         }
  59
  60         @Override
  61         public GraphWriter let(Resource p, Resource o) throws DatabaseException {
  62                 assert(p != null);
  63                 assert(o != null);
  64                 try {
  65                         s.writeByte(0);
  66                         writeRef(current);
  67                         writeRef(getPredicateId(p));
  68                         writeRef(getId(o));
  69                 } catch (IOException e) {
  70                         throw new RuntimeException("Writing statement failed.", e);
  71                 }
  72                 return this;
  73         }
  74
  75         @Override
  76         public GraphWriter let(Resource p, Object value, Resource dataType) throws DatabaseException {
  77                 assert(p != null);
  78                 assert(value != null);
  79                 assert(dataType != null);
  80                 try {
  81                         ++internalCount;
  82                         timestamps.add(0);
  83
  84                         s.writeByte(1);
  85                         writeRef(internalCount);
  86                         s.writeUnshared(value);
  87                         writeRef(getId(dataType));
  88
  89                         s.writeByte(0);
  90                         writeRef(current);
  91                         writeRef(getPredicateId(p));
  92                         writeRef(internalCount);
  93                 } catch (IOException e) {
  94                         throw new RuntimeException("Writing statement failed.", e);
  95                 }
  96                 return this;
  97         }
  98
  99         @Override
 100         public GraphWriter let(int clusterHint, Resource p, Object value,
 101                 Resource dataType) throws DatabaseException {
 102                 let(p, value, dataType);
 103                 clusterHints.put(internalCount, clusterHint);
 104                 return this;
 105         }
 106
 107         @Override
 108     public GraphWriter createLiteral(Object value, Resource dataType) {
 109                 assert(value != null);
 110                 assert(dataType != null);
 111         try {
 112             ++internalCount;
 113             timestamps.add(0);
 114
 115             s.writeByte(1);
 116             writeRef(internalCount);
 117             s.writeUnshared(value);
 118             writeRef(getId(dataType));
 119
 120             current = internalCount;
 121         } catch (IOException e) {
 122             throw new RuntimeException("Writing statement failed.", e);
 123         }
 124         return this;
 125     }
 126
 127         @Override
 128         public GraphWriter createLiteral(int clusterHint, Object value,
 129                 Resource dataType) {
 130                 createLiteral(value, dataType);
 131
 132                 clusterHints.put(current, clusterHint);
 133                 return this;
 134         }
 135
 136     @Override
 137     public GraphWriter createInverse(int cluster, Resource r) {
 138         assert(r != null);
 139         create(cluster);
 140         try {
 141             s.writeByte(5);
 142             writeRef(current);
 143             writeRef(getId(r));
 144         } catch (IOException e) {
 145             throw new RuntimeException("Writing statement failed.", e);
 146         }
 147         return this;
 148     }
 149
 150     @Override
 151     public GraphWriter createInverse(Resource r) {
 152         assert(r != null);
 153         create();
 154         try {
 155             s.writeByte(5);
 156             writeRef(current);
 157             writeRef(getId(r));
 158         } catch (IOException e) {
 159             throw new RuntimeException("Writing statement failed.", e);
 160         }
 161         return this;
 162     }
 163
 164     protected Resource getResource(WriteOnlyGraph wg, Resource[] internals, long[] resourceIds, int id) {
 165                 if(id > 0) {
 166                         Resource ret = internals[id-1];
 167                         if(ret == null) {
 168                             throw new Error("Error");
 169 //                              ret = wg.newResource(types.get(id));
 170 //                              internals[id-1] = ret;
 171 //                              resourceIds[id-1] = ret.getResourceId();
 172                         }
 173                         if(timestamps.getQuick(id-1)==time)
 174                                 internals[id-1] = null;
 175                         ++time;
 176                         return ret;
 177                 }
 178                 else if(id < 0) {
 179                         return externals.get(-1-id);
 180                 }
 181                 else
 182                         return null;
 183         }
 184
 185         @Override
 186         public void commit(IProgressMonitor monitor, WriteOnlyGraph wg) throws DatabaseException {
 187                 try {
 188                         if (s != null) {
 189                                 s.close();
 190                                 s = null;
 191                         }
 192                         externalsInv = null;
 193                         monitor.beginTask("Writing database", 100);
 194                         FileInputStream fis = new FileInputStream(file);
 195                         FileChannel fc = fis.getChannel();
 196                         try (ObjectInputStream is = new ObjectInputStream(new BufferedInputStream(fis, 1024*1024))) {
 197                                 commit(monitor, wg, is, fc);
 198                         }
 199                 } catch (IOException e) {
 200                         throw new DatabaseException("Commiting delayed graph writings failed.", e);
 201                 } catch (ClassNotFoundException e) {
 202                         throw new DatabaseException("Commiting delayed graph writings failed.", e);
 203                 } finally {
 204                         file.delete();
 205                         monitor.done();
 206                 }
 207         }
 208
 209         private void commit(IProgressMonitor monitor, WriteOnlyGraph wg, ObjectInputStream is, FileChannel fc) throws DatabaseException, IOException, ClassNotFoundException {
 210                 int lastPercentageDone = 0;
 211                 long fileLength = file.length();
 212                 //System.out.println("size of commit file: " + fileLength);
 213
 214                 Resource[] internals = new Resource[internalCount];
 215                 resourceIds = new long[internalCount];
 216
 217                 int resourceCounter = 0;
 218                 int statementCounter = 0;
 219                 int valueCounter = 0;
 220                 time = 0;
 221                 while(true) {
 222                         switch(is.read()) {
 223                         case -1:
 224                                 System.out.println("clusterIds.size() = " + clusterIds.size());
 225                                 System.out.println("Wrote " + resourceCounter + " resources, " + statementCounter + " statements and " + valueCounter + " values.");
 226                                 return;
 227
 228                         case 0: {
 229                                 int s = is.readInt();
 230                                 int p = is.readInt();
 231                                 int o = is.readInt();
 232                                 Resource rs = getResource(wg, internals, resourceIds, s);
 233                                 Resource rp = getResource(wg, internals, resourceIds, p);
 234                                 Resource ro = getResource(wg, internals, resourceIds, o);
 235                                 Resource rpInv = inverses.get(p);
 236                                 wg.claim(rs, rp, rpInv, ro);
 237                                 statementCounter += 2;
 238                         } break;
 239                         case 1: {
 240                                 int id = is.readInt();
 241                                 Object value = is.readUnshared();
 242                                 int type = is.readInt();
 243
 244                                 Resource r = newResource(wg, internals, id);
 245                                 wg.claim(r, l0.InstanceOf, null, getResource(wg, internals, resourceIds, type));
 246                                 wg.claimValue(r, value);
 247                                 statementCounter ++;
 248                                 resourceCounter ++;
 249                                 valueCounter ++;
 250
 251                         } break;
 252                         case 2: {
 253                                 wg.flushCluster();
 254                         } break;
 255                         case 3: {
 256
 257                                 int s = is.readInt();
 258                                 int t = is.readInt();
 259
 260                                 Resource type = getResource(wg, internals, resourceIds, t);
 261                                 wg.claim(newResource(wg, internals, s), l0.InstanceOf, null, type);
 262                                 statementCounter ++;
 263                                 resourceCounter ++;
 264
 265                         } break;
 266                         case 4: {
 267
 268                                 int s = is.readInt();
 269                                 newResource(wg, internals, s);
 270                                 resourceCounter ++;
 271
 272                         } break;
 273                         case 5: { // InverseOf
 274
 275                                 int r1 = is.readInt();
 276                                 int r2 = is.readInt();
 277
 278                                 Resource rr1 = getResource(wg, internals, resourceIds, r1);
 279                                 Resource rr2 = getResource(wg, internals, resourceIds, r2);
 280                                 wg.claim(rr1, l0.InverseOf, l0.InverseOf, rr2);
 281                                 statementCounter += 2;
 282
 283                                 inverses.put(r1, rr2);
 284                                 inverses.put(r2, rr1);
 285
 286                         } break;
 287                         }
 288
 289 //                      if((counter % 200000) == 0) {
 290 //                              System.out.println("Written " + counter + " statements.");
 291 //                      }
 292
 293                         double percentageDone = 100.0 * (double) fc.position() / (double) fileLength;
 294                         int newPercentageDone = (int) Math.round(percentageDone);
 295                         if(newPercentageDone > lastPercentageDone) {
 296                                 monitor.setTaskName("Writing database (" + newPercentageDone + "%)");
 297                                 monitor.worked(newPercentageDone - lastPercentageDone);
 298                                 lastPercentageDone = newPercentageDone;
 299                         }
 300                 }
 301         }
 302
 303         private Resource newResource(WriteOnlyGraph wg, Resource[] internals, int s) throws DatabaseException {
 304                 int clusterHint = clusterHints.get(s);
 305
 306                 Resource r;
 307                 if(clusterHint == 0)
 308                         r = wg.newResource();
 309                 else {
 310                         long clusterId = clusterIds.get(clusterHint);
 311                         if(clusterId == 0) {
 312                                 wg.flushCluster();
 313                                 r = wg.newResource();
 314                                 clusterIds.put(clusterHint, clustering.getCluster(r));
 315                         }
 316                         else
 317                                 r = wg.newResource(clusterId);
 318                 }
 319                 internals[s-1] = r;
 320                 resourceIds[s-1] = r.getResourceId();
 321                 return r;
 322         }
 323
 324     @Override
 325     public GraphWriter create() {
 326
 327         current = ++internalCount;
 328         timestamps.add(0);
 329
 330         try {
 331             s.writeByte(4);
 332             s.writeInt(current);
 333         } catch (IOException e) {
 334             throw new RuntimeException("Writing statement failed.", e);
 335         }
 336
 337         return this;
 338
 339     }
 340
 341     @Override
 342     public GraphWriter create(int clusterHint) {
 343
 344         create();
 345         clusterHints.put(current, clusterHint);
 346         return this;
 347
 348     }
 349
 350         @Override
 351         public GraphWriter create(Resource type) {
 352                 assert(type != null);
 353
 354             current = ++internalCount;
 355         timestamps.add(0);
 356
 357         try {
 358             s.writeByte(3);
 359             s.writeInt(current);
 360             s.writeInt(getId(type));
 361         } catch (IOException e) {
 362             throw new RuntimeException("Writing statement failed.", e);
 363         }
 364
 365                 return this;
 366
 367         }
 368
 369         @Override
 370     public GraphWriter create(int clusterHint, Resource type) {
 371                 create(type);
 372         clusterHints.put(current, clusterHint);
 373         return this;
 374
 375     }
 376
 377         @Override
 378         public Resource get() {
 379                 if(current > 0)
 380                         return new InternalResource(current);
 381                 else if(current < 0)
 382                         return externals.get(-1-current);
 383                 else
 384                         return null;
 385         }
 386
 387         @Override
 388         public GraphWriter handle(Resource s) {
 389                 assert(s != null);
 390
 391                 current = getId(s);
 392                 return this;
 393         }
 394
 395         public GraphWriter flush() {
 396         try {
 397             s.writeByte(2);
 398         } catch(IOException e) {
 399             throw new RuntimeException("Writing flush failed.", e);
 400         }
 401         return this;
 402         }
 403
 404 }