bundles/org.simantics.db.impl/src/org/simantics/db/impl/graph/DelayedWriteGraph.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.db.impl.graph;
  13
  14
  15 import java.io.Closeable;
  16 import java.io.File;
  17 import java.io.FileInputStream;
  18 import java.io.FileOutputStream;
  19 import java.io.IOException;
  20 import java.io.InputStream;
  21 import java.io.OutputStream;
  22 import java.nio.ByteBuffer;
  23 import java.nio.channels.FileChannel;
  24 import java.nio.file.Files;
  25 import java.util.ArrayList;
  26 import java.util.HashSet;
  27 import java.util.Set;
  28 import java.util.TreeMap;
  29 import java.util.UUID;
  30
  31 import org.eclipse.core.runtime.Platform;
  32 import org.simantics.databoard.Bindings;
  33 import org.simantics.databoard.accessor.Accessor;
  34 import org.simantics.databoard.binding.Binding;
  35 import org.simantics.databoard.binding.error.BindingConstructionException;
  36 import org.simantics.databoard.serialization.Serializer;
  37 import org.simantics.databoard.type.Datatype;
  38 import org.simantics.databoard.util.binary.RandomAccessBinary;
  39 import org.simantics.db.Metadata;
  40 import org.simantics.db.ReadGraph;
  41 import org.simantics.db.Resource;
  42 import org.simantics.db.Session;
  43 import org.simantics.db.Statement;
  44 import org.simantics.db.VirtualGraph;
  45 import org.simantics.db.WriteGraph;
  46 import org.simantics.db.WriteOnlyGraph;
  47 import org.simantics.db.common.MetadataUtils;
  48 import org.simantics.db.common.request.WriteOnlyRequest;
  49 import org.simantics.db.common.utils.Logger;
  50 import org.simantics.db.exception.BindingException;
  51 import org.simantics.db.exception.ClusterSetExistException;
  52 import org.simantics.db.exception.DatabaseException;
  53 import org.simantics.db.exception.ManyObjectsForFunctionalRelationException;
  54 import org.simantics.db.exception.NoSingleResultException;
  55 import org.simantics.db.exception.ResourceNotFoundException;
  56 import org.simantics.db.exception.ServiceException;
  57 import org.simantics.db.impl.DebugPolicy;
  58 import org.simantics.db.impl.ResourceImpl;
  59 import org.simantics.db.request.DelayedWrite;
  60 import org.simantics.db.request.WriteOnly;
  61 import org.simantics.db.request.WriteResult;
  62 import org.simantics.db.request.WriteTraits;
  63 import org.simantics.db.service.ByteReader;
  64 import org.simantics.db.service.ClusteringSupport;
  65 import org.simantics.db.service.TransferableGraphSupport;
  66 import org.simantics.db.service.XSupport;
  67 import org.simantics.layer0.Layer0;
  68 import org.simantics.utils.datastructures.MapList;
  69
  70 import gnu.trove.map.hash.TIntIntHashMap;
  71 import gnu.trove.map.hash.TObjectIntHashMap;
  72
  73 /**
  74  * Write graph implementation that does not modify the database
  75  * immediately but with an explicit commit method. All read operations
  76  * return results based on the old graph.
  77  */
  78 public class DelayedWriteGraph extends ReadGraphImpl implements WriteGraph, ByteReader, Closeable {
  79
  80         private static final boolean DEBUG = false;
  81         private static final int BUFFER_SIZE = 512*1024;
  82
  83         private static final PassthroughSerializerBinding PASSTHROUGH = new PassthroughSerializerBinding();
  84
  85         private final int TERM = 0;
  86         private final int CLAIM = 1;
  87     private final int CLAIM_NOINVERSE = 2;
  88         private final int CLAIM_VALUE_B = 4;
  89     private final int DENY = 5;
  90     private final int DENY_VALUE = 6;
  91     private final int COMMIT_AND_CONTINUE = 7;
  92
  93     private static class State {
  94         public File tempFile;
  95         public FileOutputStream out;
  96         public FileInputStream in;
  97         public ArrayList<Resource> idToResource = new ArrayList<Resource>();
  98         public TIntIntHashMap externalToId = new TIntIntHashMap();
  99         public ArrayList<Binding> idToBinding = new ArrayList<Binding>();
 100         public TObjectIntHashMap<Binding> bindingToId = new TObjectIntHashMap<Binding>();
 101         public Set<Resource> clusterSets = new HashSet<Resource>();
 102         public Set<Resource> clusterSetsForExistingResources = new HashSet<Resource>();
 103         public int clusterCount = 0;
 104         public long defaultCluster;
 105         public Resource defaultClusterSet;
 106         public int statementCount = 0;
 107         public int valueCount = 0;
 108         public int fileCount = 0;
 109     }
 110
 111     private State writeState;
 112     private TreeMap<String,byte[]> metadata = new TreeMap<String,byte[]>();
 113
 114     private FileChannel channel;
 115     private byte[] bytes = new byte[BUFFER_SIZE];
 116     private ByteBuffer bb = ByteBuffer.wrap(bytes);
 117     private int byteIndex = 0;
 118
 119     private Layer0 b;
 120     private Session session;
 121
 122     public static Resource convertDelayedResource(Resource r) {
 123         if (r instanceof InternalResource)
 124             return ((InternalResource) r).resource;
 125         return r;
 126     }
 127
 128     private static class InternalResource implements Resource {
 129
 130         int id;
 131         long clusterId = 0;
 132         Resource resource = null;
 133         Resource clusterSet = null;
 134
 135         public InternalResource(int id, long clusterId) {
 136             this.id = id;
 137             this.clusterId = clusterId;
 138         }
 139
 140         public InternalResource(int id, Resource clusterSet) {
 141             this.id = id;
 142             this.clusterSet = clusterSet;
 143         }
 144
 145         @Override
 146         public long getResourceId() {
 147             throw new UnsupportedOperationException();
 148         }
 149
 150         @Override
 151         public Resource get() {
 152             return this;
 153         }
 154
 155         @Override
 156         public boolean isPersistent() {
 157             return false;
 158         }
 159
 160         @Override
 161         public int compareTo(Resource o) {
 162             if(o instanceof InternalResource)
 163                 return Integer.compare(id, ((InternalResource)o).id);
 164             return -1;
 165         }
 166
 167         @Override
 168         public int hashCode() {
 169             final int prime = 31;
 170             int result = 1;
 171             result = prime * result + id;
 172             return result;
 173         }
 174
 175         @Override
 176         public int getThreadHash() {
 177             return hashCode();
 178         }
 179
 180         @Override
 181         public boolean equals(Object obj) {
 182             if (this == obj)
 183                 return true;
 184             if (obj == null)
 185                 return false;
 186             if (!(obj instanceof InternalResource))
 187                 return false;
 188             InternalResource other = (InternalResource) obj;
 189             if (id != other.id)
 190                 return false;
 191             return true;
 192         }
 193
 194         @Override
 195         public boolean equalsResource(Resource other) {
 196             return equals(other);
 197         }
 198
 199         @Override
 200         public String toString() {
 201             StringBuilder sb = new StringBuilder(32);
 202             if(DebugPolicy.VERBOSE) {
 203                 sb.append("[delayed id=");
 204                 sb.append(id);
 205                 sb.append("]");
 206             } else {
 207                 sb.append("[did=");
 208                 sb.append(id);
 209                 sb.append("]");
 210             }
 211             return sb.toString();
 212         }
 213     }
 214
 215         private int getId(Resource resource) {
 216                 if(resource instanceof InternalResource)
 217                         return ((InternalResource)resource).id;
 218                 else {
 219                         ResourceImpl r = (ResourceImpl)resource;
 220                         int id = writeState.externalToId.get(r.id);
 221                         if(id != 0) {
 222                                 return id;
 223                         } else {
 224                                 id = writeState.idToResource.size();
 225                                 writeState.idToResource.add(resource);
 226                                 writeState.externalToId.put(r.id, id);
 227                                 return id;
 228                         }
 229                 }
 230         }
 231
 232         private Resource getResource(int id) {
 233                 return writeState.idToResource.get(id);
 234         }
 235
 236         private int getBindingId(Binding binding) {
 237                 if(writeState.bindingToId.contains(binding))
 238                         return writeState.bindingToId.get(binding);
 239                 else {
 240                         int id = writeState.idToBinding.size();
 241                         writeState.idToBinding.add(binding);
 242                         writeState.bindingToId.put(binding, id);
 243                         return id;
 244                 }
 245         }
 246
 247         public DelayedWriteGraph(ReadGraph g) throws IOException {
 248                 super((ReadGraphImpl)g);
 249                 writeState = new State();
 250                 session = g.getSession();
 251                 b = Layer0.getInstance(g);
 252                 writeState.defaultCluster = newCluster();
 253         }
 254
 255     public DelayedWriteGraph(ReadGraph g, State state) {
 256         super((ReadGraphImpl)g);
 257         session = g.getSession();
 258                 b = Layer0.getInstance(g);
 259         this.writeState = state;
 260     }
 261
 262         public DelayedWriteGraph newSync() {
 263         return new DelayedWriteGraph(this, writeState);
 264     }
 265
 266         @Override
 267         public void claim(Resource subject, Resource predicate, Resource object)
 268                         throws ServiceException {
 269                 assert(subject != null);
 270         assert(predicate != null);
 271         assert(object != null);
 272
 273         Resource inverse = getPossibleInverse(predicate);
 274         claim(subject, predicate, inverse, object);
 275         }
 276
 277     @Override
 278     public void addLiteral(Resource resource, Resource predicate, Resource inverse, Resource type, Object value,
 279             Binding binding) throws BindingException,
 280             ManyObjectsForFunctionalRelationException, ServiceException {
 281         Resource valueResource = newResource();
 282         claimValue(valueResource, value, binding);
 283         claim(valueResource, b.InstanceOf, null, type);
 284         claim(resource, predicate, inverse, valueResource);
 285     }
 286
 287     @Override
 288     public void addLiteral(Resource resource, Resource predicate,
 289                 Resource inverse, Object value, Binding binding)
 290                 throws BindingException, ManyObjectsForFunctionalRelationException,
 291                 ServiceException {
 292
 293         Resource type = getType(value);
 294         if(type == null) {
 295             Resource literal = newResource();
 296             type = b.Literal;
 297             Resource dataType = newResource();
 298             claim(dataType, b.InstanceOf, null, b.DataType);
 299             claimValue(dataType, binding.type(), DATA_TYPE_BINDING_INTERNAL);
 300             claim(literal, b.HasDataType, null, dataType);
 301             claim(literal, b.InstanceOf, null, type);
 302             claimValue(literal, value, binding);
 303             claim(resource, predicate, inverse, literal);
 304         } else {
 305                 addLiteral(resource, predicate, inverse, type, value, binding);
 306         }
 307
 308     }
 309
 310     @Override
 311     public <T extends Accessor> T newLiteral(Resource resource, Resource predicate, Datatype datatype, Object initialValue)
 312     throws DatabaseException {
 313
 314         throw new UnsupportedOperationException();
 315
 316     }
 317
 318     @Override
 319     public RandomAccessBinary createRandomAccessBinary (Resource resource, Resource predicate, Datatype datatype, Object initialValue) throws DatabaseException {
 320         throw new UnsupportedOperationException();
 321     }
 322
 323     @Override
 324     public RandomAccessBinary createRandomAccessBinary(Resource resource, Datatype datatype, Object initialValue) throws DatabaseException {
 325         throw new UnsupportedOperationException();
 326     }
 327
 328     @Override
 329     public void claimLiteral(Resource resource, Resource predicate, Object value) throws ManyObjectsForFunctionalRelationException, ServiceException {
 330
 331                 try {
 332                         Binding b = Bindings.getBinding(value.getClass());
 333                 claimLiteral(resource, predicate, value, b);
 334                 } catch (BindingConstructionException e) {
 335                         throw new IllegalArgumentException(e);
 336                 } catch (BindingException e) {
 337                         throw new IllegalArgumentException(e);
 338                 }
 339
 340     }
 341
 342     @Override
 343     public void claimLiteral(Resource resource, Resource predicate, Object value, Binding binding) throws BindingException,
 344                 ManyObjectsForFunctionalRelationException, ServiceException {
 345
 346         Statement valueStatement = null;
 347         if(!(resource instanceof InternalResource)) valueStatement = getPossibleStatement(resource, predicate);
 348
 349         if(valueStatement != null && resource.equals(valueStatement.getSubject())) {
 350
 351             claimValue(valueStatement.getObject(), value, binding);
 352
 353         } else {
 354
 355             Resource type = getType(value);
 356             Resource literal = newResource();
 357             if (type == null) {
 358                 type = b.Literal;
 359                 Resource dataType = newResource();
 360                 claim(dataType, b.InstanceOf, null, b.DataType);
 361                 claimValue(dataType, binding.type(), DATA_TYPE_BINDING_INTERNAL);
 362                 claim(literal, b.HasDataType, dataType);
 363             }
 364             claim(literal, b.InstanceOf, null, type);
 365             claimValue(literal, value, binding);
 366             claim(resource, predicate, literal);
 367
 368         }
 369
 370     }
 371
 372     @Override
 373     public void claimLiteral(Resource resource, Resource predicate,
 374                 Resource inverse, Resource type, Object value)
 375                 throws BindingException, ManyObjectsForFunctionalRelationException,
 376                 ServiceException {
 377
 378                 try {
 379                         Binding b = Bindings.getBinding(value.getClass());
 380                 claimLiteral(resource, predicate, inverse, type, value, b);
 381                 } catch (BindingConstructionException e) {
 382                         throw new IllegalArgumentException(e);
 383                 } catch (BindingException e) {
 384                         throw new IllegalArgumentException(e);
 385                 }
 386
 387     }
 388
 389     @Override
 390     public void claimLiteral(Resource resource, Resource predicate,
 391                 Resource inverse, Resource type, Object value, Binding binding)
 392                 throws BindingException, ManyObjectsForFunctionalRelationException,
 393                 ServiceException {
 394
 395         Statement valueStatement = (resource instanceof InternalResource) ? null : getPossibleStatement(resource, predicate);
 396
 397         if(valueStatement != null && resource.equals(valueStatement.getSubject())) {
 398
 399             claimValue(valueStatement.getObject(), value, binding);
 400
 401         } else {
 402
 403             Resource valueResource = newResource();
 404             claim(valueResource, b.InstanceOf, null, type);
 405             claim(resource, predicate, inverse, valueResource);
 406             claimValue(valueResource, value, binding);
 407
 408         }
 409
 410     }
 411
 412     @Override
 413     public void claimLiteral(Resource resource, Resource predicate,
 414                 Resource type, Object value) throws BindingException,
 415                 ManyObjectsForFunctionalRelationException, ServiceException {
 416
 417                 try {
 418                         Binding b = Bindings.getBinding(value.getClass());
 419                 claimLiteral(resource, predicate, type, value, b);
 420                 } catch (BindingConstructionException e) {
 421                         throw new IllegalArgumentException(e);
 422                 } catch (BindingException e) {
 423                         throw new IllegalArgumentException(e);
 424                 }
 425
 426     }
 427
 428     @Override
 429     public void claimLiteral(Resource resource, Resource predicate,
 430                 Resource type, Object value, Binding binding)
 431                 throws BindingException, ManyObjectsForFunctionalRelationException,
 432                 ServiceException {
 433
 434                 try {
 435                 Resource inverse = getSingleObject(predicate, b.InverseOf);
 436                 claimLiteral(resource, predicate, inverse, type, value, binding);
 437                 } catch (NoSingleResultException e) {
 438                         throw new ServiceException(e);
 439                 }
 440
 441     }
 442
 443         @Override
 444         public void deny(Resource subject) throws ServiceException {
 445                 assert(subject != null);
 446                 if(!(subject instanceof InternalResource)) {
 447                         try {
 448                                 for (Statement statement : getStatements(subject, b.IsWeaklyRelatedTo)) {
 449                                         deny(statement);
 450                                 }
 451                         } catch (ManyObjectsForFunctionalRelationException e) {
 452                                 throw new ServiceException(e);
 453                         }
 454                 }
 455         }
 456
 457         @Override
 458         public void deny(Resource subject, Resource predicate)
 459                         throws ServiceException {
 460                 assert(subject != null);
 461                 if(!(subject instanceof InternalResource)) {
 462                         for (Resource object : getObjects(subject, predicate)) {
 463                                 deny(subject, predicate, object);
 464                         }
 465                 }
 466         }
 467
 468         @Override
 469         public void deny(Resource subject, Resource predicate, Resource object) throws ServiceException {
 470                 denyStatement(subject, predicate, object);
 471         }
 472
 473         @Override
 474         public void denyStatement(Resource subject, Resource predicate, Resource object) throws ServiceException {
 475                 deny(subject, predicate, getPossibleInverse(predicate), object);
 476         }
 477
 478         @Override
 479         public void deny(Statement statement) throws ServiceException {
 480                 Resource predicate = statement.getPredicate();
 481                 deny(statement.getSubject(), predicate, getPossibleInverse(predicate), statement.getObject());
 482         }
 483
 484         @Override
 485         public void denyValue(Resource resource, Resource predicate)
 486                         throws ManyObjectsForFunctionalRelationException, ServiceException {
 487                 assert(resource != null);
 488                 assert(predicate != null);
 489
 490                 if(!(resource instanceof InternalResource)) {
 491                 Statement valueStatement = getPossibleStatement(resource, predicate);
 492
 493                 if (valueStatement != null && !valueStatement.isAsserted(resource)) {
 494                     Resource value = valueStatement.getObject();
 495                     deny(valueStatement);
 496                                 denyValue(value);
 497                 }
 498                 }
 499         }
 500
 501         @Override
 502         public Resource newResource() throws ServiceException {
 503                 if(writeState.defaultClusterSet != null) return newResource(writeState.defaultClusterSet);
 504                 else return newResource(writeState.defaultCluster);
 505         }
 506
 507         @Override
 508         public Resource newResource(long clusterId) throws ServiceException {
 509                 int id = writeState.idToResource.size();
 510                 InternalResource ret = new InternalResource(id, clusterId);
 511                 writeState.idToResource.add(ret);
 512                 return ret;
 513         }
 514
 515     @Override
 516     public Resource newResource(Resource clusterSet) throws ServiceException {
 517
 518         if ((clusterSet instanceof InternalResource)) {
 519                 if(!writeState.clusterSets.contains(clusterSet))
 520                         throw new ClusterSetExistException("Cluster set does not exist. Resource=" + clusterSet);
 521         } else {
 522                 WriteSupport ws = session.getService(WriteSupport.class);
 523                 if (!ws.hasClusterSet(null, clusterSet))
 524                         if(!writeState.clusterSetsForExistingResources.contains(clusterSet))
 525                         throw new ClusterSetExistException("Cluster set does not exist. Resource=" + clusterSet);
 526         }
 527
 528         int id = writeState.idToResource.size();
 529         InternalResource ret = new InternalResource(id, clusterSet);
 530         writeState.idToResource.add(ret);
 531         return ret;
 532     }
 533
 534     @Override
 535     public void newClusterSet(Resource clusterSet) throws ServiceException {
 536         if (DEBUG)
 537             System.out.println("new cluster set=" + clusterSet);
 538         boolean existingResource = !(clusterSet instanceof InternalResource);
 539         if (existingResource) {
 540                 WriteSupport ws = session.getService(WriteSupport.class);
 541                 if (ws.hasClusterSet(null, clusterSet))
 542                         throw new ClusterSetExistException("Cluster set exist already. Resource=" + clusterSet);
 543                 writeState.clusterSetsForExistingResources.add(clusterSet);
 544         } else {
 545                 if(!writeState.clusterSets.add(clusterSet))
 546                         throw new ClusterSetExistException("Cluster set exist already. Resource=" + clusterSet);
 547         }
 548     }
 549     @Override
 550     public Resource setClusterSet4NewResource(Resource clusterSet)
 551     throws ServiceException {
 552         Resource existing = writeState.defaultClusterSet;
 553         writeState.defaultClusterSet = clusterSet;
 554         return existing;
 555     }
 556
 557         @Override
 558         public void claim(Resource subject, Resource predicate, Resource inverse,
 559                         Resource object) throws ServiceException {
 560                 assert(subject != null);
 561                 assert(predicate != null);
 562                 assert(object != null);
 563                 try {
 564                     if(inverse != null && !(subject.equals(object) && inverse.equals(predicate))) {
 565                         writeByte(CLAIM);
 566                         writeInt(getId(subject));
 567                         writeInt(getId(predicate));
 568                         writeInt(getId(inverse));
 569                         writeInt(getId(object));
 570                     } else {
 571                         writeByte(CLAIM_NOINVERSE);
 572                         writeInt(getId(subject));
 573                         writeInt(getId(predicate));
 574                         writeInt(getId(object));
 575                     }
 576                 } catch(Exception e) {
 577                         throw new ServiceException(e);
 578                 }
 579         }
 580
 581     @Override
 582     public void deny(Resource subject, Resource predicate, Resource inverse,
 583             Resource object) throws ServiceException {
 584         assert(subject != null);
 585         assert(predicate != null);
 586         assert(object != null);
 587         try {
 588                 writeByte(DENY);
 589                 writeInt(getId(subject));
 590                 writeInt(getId(predicate));
 591             if(inverse != null) writeInt(getId(inverse));
 592             else writeInt(0);
 593                 writeInt(getId(object));
 594         } catch(Exception e) {
 595             throw new ServiceException(e);
 596         }
 597     }
 598
 599     @Override
 600     public void deny(Resource subject, Resource predicate, Resource inverse,
 601             Resource object, VirtualGraph graph) throws ServiceException {
 602         throw new UnsupportedOperationException();
 603     }
 604
 605         @Override
 606         public void claimValue(Resource resource, Object value)
 607                         throws ServiceException {
 608                 try {
 609                     Binding binding = Bindings.getBinding(value.getClass());
 610                     claimValue(resource, value, binding);
 611                 } catch (BindingConstructionException e) {
 612                         throw new ServiceException(e);
 613                 }
 614         }
 615
 616         private OutputStream valueWriter = new OutputStream() {
 617                 @Override
 618                 public void write(int b) throws IOException {
 619                         writeByte(b);
 620                 }
 621         };
 622
 623         @Override
 624         public void claimValue(Resource resource, Object value, Binding binding)
 625                         throws ServiceException {
 626                 try {
 627
 628                 writeByte(CLAIM_VALUE_B);
 629                 writeInt(getId(resource));
 630                 Serializer serializer = binding.serializer();
 631                 int size = serializer.getSize(value);
 632             writeInt(size);
 633             serializer.serialize(valueWriter, value);
 634
 635                 } catch(IOException e) {
 636                 Logger.defaultLogError(e);
 637                         throw new ServiceException(e);
 638                 }
 639         }
 640
 641         @Override
 642         public void denyValue(Resource resource) throws ServiceException {
 643                 writeByte(DENY_VALUE);
 644                 writeInt(getId(resource));
 645         }
 646
 647         @Override
 648         public void denyValue(Resource resource, VirtualGraph graph) throws ServiceException {
 649         throw new UnsupportedOperationException();
 650         }
 651
 652         @Override
 653         public void flushCluster() throws ServiceException {
 654         writeState.defaultCluster = newCluster();
 655         }
 656
 657     @Override
 658     public void flushCluster(Resource r) throws ServiceException {
 659         throw new ServiceException("Operation flushCluster(" + r + " not implemented.");
 660     }
 661
 662     private void writeReset(int size) {
 663         byteIndex = 0;
 664         bb.position(0);
 665         bb.limit(size);
 666         try {
 667             if (writeState.tempFile == null) {
 668                 File workspace = Platform.getLocation().toFile();
 669                 File temp = new File(workspace, "tempFiles");
 670                 File base = new File(temp, "delayed");
 671                 Files.createDirectories(base.toPath());
 672                 writeState.tempFile = new File(base, UUID.randomUUID().toString());
 673                 writeState.out = new FileOutputStream(writeState.tempFile);
 674                 channel = writeState.out.getChannel();
 675             }
 676
 677             for (int got=0;got < size;) {
 678                 int n = channel.write(bb);
 679                 if (n <= 0) {
 680                     Logger.defaultLogError(new Exception("FileChannel.write returned " + n));
 681                     return;
 682                 }
 683                 got += n;
 684             }
 685         } catch (IOException e) {
 686             Logger.defaultLogError("Failed to write buffer of " + size + " bytes to temporary file " + writeState.tempFile, e);
 687         }
 688     }
 689
 690     private void reset() {
 691         byteIndex = 0;
 692         bb.clear();
 693         if (channel != null) {
 694             try {
 695                 for(int got=0; got < BUFFER_SIZE;) {
 696                     int n = channel.read(bb);
 697                     if (n <= 0)
 698                         return;
 699                     got += n;
 700                 }
 701             } catch (IOException e) {
 702                 Logger.defaultLogError("FileChannel.read failed", e);
 703             }
 704         }
 705     }
 706
 707     private void writeInt(int i) {
 708         if(byteIndex < (BUFFER_SIZE-4)) {
 709                 bytes[byteIndex++] = (byte)(i&0xff);
 710                 bytes[byteIndex++] = (byte)((i>>>8)&0xff);
 711                 bytes[byteIndex++] = (byte)((i>>>16)&0xff);
 712                 bytes[byteIndex++] = (byte)((i>>>24)&0xff);
 713                 if (byteIndex == BUFFER_SIZE)
 714                     writeReset(BUFFER_SIZE);
 715         } else {
 716                 int has = BUFFER_SIZE-byteIndex;
 717                 if(has == 0) writeReset(BUFFER_SIZE);
 718                 bytes[byteIndex++] = (byte)(i&0xff);
 719                 if(has == 1) writeReset(BUFFER_SIZE);
 720                 bytes[byteIndex++] = (byte)((i>>>8)&0xff);
 721                 if(has == 2) writeReset(BUFFER_SIZE);
 722                 bytes[byteIndex++] = (byte)((i>>>16)&0xff);
 723                 if(has == 3) writeReset(BUFFER_SIZE);
 724                 bytes[byteIndex++] = (byte)((i>>>24)&0xff);
 725                 if(has == 4) writeReset(BUFFER_SIZE);
 726         }
 727     }
 728
 729     private int readInt() {
 730         if(byteIndex < (BUFFER_SIZE-4)) {
 731                 int result = (int)
 732                         ((bytes[byteIndex++] & 0xff) |
 733                         ((bytes[byteIndex++] & 0xff)<<8) |
 734                         ((bytes[byteIndex++] & 0xff)<<16) |
 735                         ((bytes[byteIndex++] & 0xff)<<24));
 736                 return result;
 737         } else {
 738                 int has = BUFFER_SIZE-byteIndex;
 739                 int result = 0;
 740                 if(has == 0) reset();
 741                         result = (int)(bytes[byteIndex++] & 0xff);
 742                 if(has == 1) reset();
 743                 result |= (int)((bytes[byteIndex++] & 0xff) <<8);
 744                 if(has == 2) reset();
 745                 result |= (int)((bytes[byteIndex++] & 0xff) <<16);
 746                 if(has == 3) reset();
 747                 result |= (int)((bytes[byteIndex++] & 0xff) <<24);
 748                 if(has == 4) reset();
 749                 return result;
 750         }
 751     }
 752
 753     private byte readByte() {
 754         byte result = bytes[byteIndex++];
 755         if(byteIndex == BUFFER_SIZE) reset();
 756         return result;
 757     }
 758
 759     private void writeByte(int b) {
 760         bytes[byteIndex++] = (byte)b;
 761         if(byteIndex == BUFFER_SIZE) writeReset(BUFFER_SIZE);
 762     }
 763
 764     private void writeBytes(byte[] data) {
 765                 int has = BUFFER_SIZE-byteIndex;
 766                 int amount = data.length;
 767                 if(has > amount) {
 768                         System.arraycopy(data, 0, bytes, byteIndex, amount);
 769                         byteIndex += amount;
 770                 } else {
 771                         System.arraycopy(data, 0, bytes, byteIndex, has);
 772                         writeReset(BUFFER_SIZE);
 773                         ByteBuffer bb2 = ByteBuffer.wrap(data);
 774                         bb2.position(has);
 775                         try {
 776                                 channel.write(bb2);
 777                         } catch (IOException e) {
 778                                 Logger.defaultLogError("FileChannel.write failed", e);
 779                         }
 780                 }
 781     }
 782
 783     public byte[] readBytes(byte[] result, int amount) {
 784         if(result == null) result = new byte[amount];
 785                 int has = BUFFER_SIZE-byteIndex;
 786                 if(has > amount) {
 787                         System.arraycopy(bytes, byteIndex, result, 0, amount);
 788                         byteIndex += amount;
 789                 } else {
 790                         System.arraycopy(bytes, byteIndex, result, 0, has);
 791                         ByteBuffer bb2 = ByteBuffer.wrap(result);
 792                         bb2.position(has);
 793                         for(int got=has;got<amount;)
 794                                 try {
 795                                         got += channel.read(bb2);
 796                                         if(got == -1) {
 797                                                 // End-of-stream, why log this?
 798                                                 return result;
 799                                         }
 800                                 } catch (IOException e) {
 801                                         Logger.defaultLogError("FileChannel.read failed", e);
 802                                 }
 803                         reset();
 804                 }
 805                 return result;
 806     }
 807
 808     public void commit(final WriteOnlyGraph w, final WriteTraits traits) throws ServiceException {
 809             writeState.bindingToId = null;
 810             writeState.externalToId = null;
 811                 writeByte(TERM);
 812
 813                 if (writeState.out != null) {
 814                         // Flush current buffer to file only if backing file has already
 815                         // been taken into use.
 816                         if (byteIndex > 0)
 817                                 writeReset(byteIndex);
 818
 819                         try (OutputStream out = writeState.out) {
 820                                 channel.force(false);
 821                         } catch (IOException e) {
 822                                 throw new ServiceException(e);
 823                         } finally {
 824                                 writeState.out = null;
 825                         }
 826
 827                         try {
 828                                 writeState.in = new FileInputStream(writeState.tempFile);
 829                                 channel = writeState.in.getChannel();
 830                         } catch (IOException e) {
 831                                 throw new ServiceException(e);
 832                         }
 833                 }
 834
 835                 w.getMetadata().putAll(metadata);
 836
 837                 TransferableGraphSupport tgs = w.getService(TransferableGraphSupport.class);
 838
 839                 // First create all resources defined by clusterId
 840                 MapList<Long,InternalResource> clusterAssignment = new MapList<Long,InternalResource>();
 841                 for(Resource r : writeState.idToResource) {
 842                         if(r instanceof InternalResource) {
 843                                 InternalResource ir = (InternalResource)r;
 844                                 if(ir.clusterId < 0) {
 845                                 if (DEBUG)
 846                                         System.out.println("ASSIGN CLUSTER " + ir + " => " + ir.clusterId);
 847                                         clusterAssignment.add(ir.clusterId, ir);
 848                                 } else if(ir.clusterId > 0) {
 849                                 if (DEBUG)
 850                                         System.out.println("-CREATED RESOURCE WITH EXISTING CLUSTER ID: " + ir);
 851                                         ir.resource = w.newResource(ir.clusterId);
 852                                         writeState.idToResource.set(ir.id, ir.resource);
 853                                         if (writeState.clusterSets.contains(ir)) {
 854                                                 w.newClusterSet(ir.resource);
 855                                         if (DEBUG)
 856                                                 System.out.println("--CREATED NEW INTERNAL RESOURCE CLUSTER SET: " + ir.resource);
 857                                         }
 858
 859                                 }
 860                         }
 861                 }
 862
 863                 for(Long clusterKey : clusterAssignment.getKeys()) {
 864                 if (DEBUG)
 865                         System.out.println("CREATE LOGICAL CLUSTER: " + clusterKey);
 866                         w.flushCluster();
 867                         for(InternalResource ir : clusterAssignment.getValuesUnsafe(clusterKey)) {
 868                         if (DEBUG)
 869                                 System.out.println("-CREATED RESOURCE: " + ir);
 870                                 ir.resource = w.newResource();
 871                                 writeState.idToResource.set(ir.id, ir.resource);
 872                                 if (writeState.clusterSets.contains(ir)) {
 873                                         w.newClusterSet(ir.resource);
 874                                 if (DEBUG)
 875                                         System.out.println("--CREATED NEW INTERNAL RESOURCE CLUSTER SET: " + ir.resource);
 876                                 }
 877                         }
 878                 }
 879
 880                 // Create cluster sets for all existing resources (not InternalResource)
 881                 // before proceeding to create resources.
 882                 for(Resource existingResource : writeState.clusterSetsForExistingResources) {
 883                         w.newClusterSet(existingResource);
 884                 if (DEBUG)
 885                         System.out.println("CREATED NEW CLUSTER SET: " + existingResource);
 886                 }
 887
 888                 // Then create all resources defined by cluster set
 889                 for(Resource r : writeState.idToResource) {
 890                         if(r instanceof InternalResource) {
 891                                 InternalResource ir = (InternalResource)r;
 892                                 Resource clusterSet = ir.clusterSet;
 893
 894                                 if (clusterSet != null) {
 895                                 if (DEBUG)
 896                                         System.out.println("NEW RESOURCE " + ir + " for cluster set " + clusterSet);
 897                                         if(clusterSet instanceof InternalResource) {
 898                                                 ir.resource = w.newResource(((InternalResource)clusterSet).resource);
 899                                         } else {
 900                                                 ir.resource = w.newResource(clusterSet);
 901                                         }
 902                                 if (DEBUG)
 903                                         System.out.println(" => " + ir.resource);
 904                                         writeState.idToResource.set(ir.id, ir.resource);
 905                                         if(writeState.clusterSets.contains(ir)) {
 906                                         if (DEBUG)
 907                                                 System.out.println(" ==> NEW CLUSTER SET");
 908                                                 w.newClusterSet(ir.resource);
 909                                         }
 910                                 }
 911
 912                         }
 913                 }
 914
 915                 reset();
 916                 bb.limit(BUFFER_SIZE);
 917
 918                 try {
 919                         while(true) {
 920                                 byte method = readByte();
 921                                 switch(method) {
 922                                 case TERM: {
 923                                         if (DEBUG) {
 924                                                 System.out.println("Resources:  " + writeState.idToResource.size());
 925                                                 System.out.println("Statements: " + writeState.statementCount);
 926                                                 System.out.println("Values:     " + writeState.valueCount);
 927                                                 System.out.println("Files:      " + writeState.fileCount);
 928                                                 System.out.println("Clusters:   " + writeState.clusterCount);
 929                                         }
 930                                         return;
 931                                 }
 932                                 case CLAIM: {
 933                                         writeState.statementCount += 2;
 934                                         Resource subject = getResource(readInt());
 935                                         Resource predicate = getResource(readInt());
 936                     Resource inverse = getResource(readInt());
 937                                         Resource object = getResource(readInt());
 938                                         w.claim(subject, predicate, inverse, object);
 939                                 } break;
 940                 case CLAIM_NOINVERSE: {
 941                     ++writeState.statementCount;
 942                     Resource subject = getResource(readInt());
 943                     Resource predicate = getResource(readInt());
 944                     Resource object = getResource(readInt());
 945                     w.claim(subject, predicate, null, object);
 946                 } break;
 947                 case DENY: {
 948                     Resource subject = getResource(readInt());
 949                     Resource predicate = getResource(readInt());
 950                     int inv = readInt();
 951                     Resource inverse = null;
 952                     if(inv > 0) inverse = getResource(inv);
 953                     Resource object = getResource(readInt());
 954                     if(!subject.isPersistent() || !object.isPersistent()) {
 955                         VirtualGraph statementProvider1 = processor.getProvider(subject, predicate, object);
 956                         if(inv > 0) {
 957                                 VirtualGraph statementProvider2 = processor.getProvider(object, inverse, subject);
 958                                 if(statementProvider2 != null)
 959                                         w.deny(object, inverse, null, subject, statementProvider2);
 960                         }
 961                         if(statementProvider1 != null)
 962                                 w.deny(subject, predicate, null, object, statementProvider1);
 963                     } else {
 964                         w.deny(subject, predicate, inverse, object, null);
 965                     }
 966                 } break;
 967                 case DENY_VALUE: {
 968                     Resource subject = getResource(readInt());
 969                     if(!subject.isPersistent()) {
 970                         VirtualGraph provider = processor.getValueProvider(subject);
 971                         if(provider != null)
 972                                 w.denyValue(subject, provider);
 973                     } else {
 974                         w.denyValue(subject);
 975                     }
 976                 } break;
 977                                 case CLAIM_VALUE_B: {
 978                                         ++writeState.valueCount;
 979                                         Resource resource = getResource(readInt());
 980                                         int len = readInt();
 981                                         tgs.setValue(w, resource, null, this, len);
 982                                 } break;
 983                                 case COMMIT_AND_CONTINUE: {
 984                                         XSupport xs = w.getService(XSupport.class);
 985                                         xs.commitAndContinue(w, traits);
 986                                 } break;
 987                                 }
 988                         }
 989                 } catch(Exception e) {
 990                         if(e instanceof ServiceException)
 991                                 throw (ServiceException)e;
 992                         else
 993                                 throw new ServiceException(e);
 994                 } finally {
 995                         channel = null;
 996                         if (writeState.in != null) {
 997                                 try (InputStream in = writeState.in) {
 998                                 } catch (IOException e) {
 999                                         throw new ServiceException(e);
1000                                 } finally {
1001                                         writeState.in = null;
1002                                         writeState.tempFile.delete();
1003                                 }
1004                         }
1005                 }
1006         }
1007
1008         private Resource getType(Object value) {
1009                 Class<?> clazz = value.getClass();
1010                 Resource dataType =
1011                           clazz == Float.class ? b.Float
1012                         : clazz == Double.class ? b.Double
1013                         : clazz == Integer.class ? b.Integer
1014                         : clazz == String.class ? b.String
1015                         : clazz == Boolean.class ? b.Boolean
1016                         : clazz == Byte.class ? b.Byte
1017                         : clazz == Long.class ? b.Long
1018                         : clazz == float[].class ? b.FloatArray
1019                         : clazz == double[].class ? b.DoubleArray
1020                         : clazz == int[].class ? b.IntegerArray
1021                         : clazz == String[].class ? b.StringArray
1022                         : clazz == boolean[].class ? b.BooleanArray
1023                         : clazz == byte[].class ? b.ByteArray
1024                         : clazz == long[].class ? b.LongArray
1025                         : null
1026                         ;
1027                 return dataType;
1028         }
1029
1030         public long newCluster() {
1031                 return -1 - (++writeState.clusterCount);
1032         }
1033
1034         public long getDefaultCluster() {
1035                 return writeState.defaultCluster;
1036         }
1037
1038         public void setDefaultCluster(long cluster) {
1039             writeState.defaultCluster = cluster;
1040         }
1041
1042     @Override
1043     public void syncRequest(final DelayedWrite request) throws DatabaseException {
1044
1045         try {
1046
1047                 final DelayedWriteGraph dwg = new DelayedWriteGraph(this);
1048                 request.perform(dwg);
1049
1050                 syncRequest(new WriteOnlyRequest() {
1051
1052                         @Override
1053                         public void perform(WriteOnlyGraph graph) throws DatabaseException {
1054                                 dwg.commit(graph, request);
1055                         }
1056
1057                 });
1058
1059         } catch (DatabaseException e) {
1060
1061                 throw e;
1062
1063         } catch (Throwable e) {
1064
1065                 throw new DatabaseException(e);
1066
1067         } finally {
1068         }
1069
1070     }
1071
1072     @Override
1073     public void syncRequest(WriteOnly request) throws DatabaseException {
1074
1075         Resource defaultClusterSet = setClusterSet4NewResource(null);
1076
1077         try {
1078             WriteSupport ws = session.getService(WriteSupport.class);
1079             ws.performWriteRequest(this, request);
1080         } catch (DatabaseException e) {
1081             throw e;
1082         } catch (Throwable t) {
1083             throw new DatabaseException(t);
1084         }  finally {
1085             setClusterSet4NewResource(defaultClusterSet);
1086         }
1087
1088     }
1089
1090     @SuppressWarnings("unchecked")
1091     @Override
1092     public <T> T getService(Class<T> api) {
1093
1094         if(ClusteringSupport.class == api) {
1095
1096             final ClusteringSupport support = (ClusteringSupport)super.getService(api);
1097
1098             return (T)new ClusteringSupport() {
1099
1100                 @Override
1101                 public Resource getResourceByIndexAndCluster(int resourceIndex, long clusterId)
1102                         throws DatabaseException, ResourceNotFoundException {
1103                     return support.getResourceByIndexAndCluster(resourceIndex, clusterId);
1104                 }
1105
1106                 @Override
1107                 public Resource getResourceByKey(int resourceKey) throws ResourceNotFoundException {
1108                     return support.getResourceByKey(resourceKey);
1109                 }
1110
1111                 @Override
1112                 public int getNumberOfResources(long clusterId)
1113                                 throws DatabaseException {
1114                         return support.getNumberOfResources(clusterId);
1115                 }
1116
1117                 @Override
1118                 public long getCluster(Resource r) {
1119                     return support.getCluster(r);
1120                 }
1121
1122                 @Override
1123                 public long createCluster() {
1124                     return newCluster();
1125                 }
1126
1127                 @Override
1128                 public boolean isClusterSet(Resource r) throws DatabaseException {
1129                         return support.isClusterSet(r);
1130                 }
1131
1132                 @Override
1133                 public Resource getClusterSetOfCluster(Resource r) throws DatabaseException {
1134                         return support.getClusterSetOfCluster(r);
1135                 }
1136
1137                 @Override
1138                 public Resource getClusterSetOfCluster(long cluster) throws DatabaseException {
1139                         return support.getClusterSetOfCluster(cluster);
1140                 }
1141
1142             };
1143
1144         } else if (TransferableGraphSupport.class == api) {
1145
1146                 final TransferableGraphSupport parentSupport = session.getService(TransferableGraphSupport.class);
1147
1148                 return (T)new TransferableGraphSupport() {
1149
1150                         @Override
1151                         public void setValue(WriteOnlyGraph graph, Resource resource, VirtualGraph provider, byte[] raw) {
1152                                 writeByte(CLAIM_VALUE_B);
1153                                 writeInt(getId(resource));
1154                                 writeInt(raw.length);
1155                                 writeBytes(raw);
1156                                 writeInt(getBindingId(PASSTHROUGH));
1157                         }
1158
1159                         @Override
1160                         public void setValue(WriteOnlyGraph graph, Resource resource, VirtualGraph provider, ByteReader reader, int amount)
1161                                 throws DatabaseException {
1162                     writeByte(CLAIM_VALUE_B);
1163                     writeInt(getId(resource));
1164                     writeInt(amount);
1165                     writeBytes(reader.readBytes(null, amount));
1166                     writeInt(getBindingId(PASSTHROUGH));
1167                         }
1168
1169                         @Override
1170                         public byte[] getValue(ReadGraph graph, Resource resource) {
1171                                 return parentSupport.getValue(graph, resource);
1172                         }
1173
1174                         @Override
1175                         public InputStream getValueStream(ReadGraph graph, Resource resource) {
1176                                 return parentSupport.getValueStream(graph, resource);
1177                         }
1178
1179                 };
1180
1181         }
1182
1183         return super.getService(api);
1184
1185     }
1186
1187     @Override
1188     public <T> void addMetadata(Metadata data) throws ServiceException {
1189         MetadataUtils.addMetadata(session, metadata, data);
1190     }
1191
1192     public void addCommitAndContinue() {
1193                 writeByte(COMMIT_AND_CONTINUE);
1194     }
1195
1196     @Override
1197     public <T extends Metadata> T getMetadata(Class<T> clazz) throws ServiceException {
1198         return MetadataUtils.getMetadata(session, metadata, clazz);
1199     }
1200
1201     @Override
1202     public TreeMap<String, byte[]> getMetadata() {
1203         return metadata;
1204     }
1205     @Override
1206     public <T> T syncRequest(WriteResult<T> request) throws DatabaseException {
1207
1208         return request.perform(this);
1209
1210     }
1211
1212     @Override
1213     public VirtualGraph getProvider() {
1214         return null;
1215     }
1216
1217     @Override
1218     public void clearUndoList(WriteTraits writeTraits) {
1219         WriteSupport ws = session.getService(WriteSupport.class);
1220         if (null != ws)
1221             ws.clearUndoList(writeTraits);
1222     }
1223
1224         @Override
1225         public void markUndoPoint() {
1226                 // TODO Auto-generated method stub
1227
1228         }
1229
1230         @Override
1231         public <T> T getPossibleRelatedValue(final Resource subject, final Resource relation, final Binding binding)
1232                         throws ManyObjectsForFunctionalRelationException, BindingException, ServiceException {
1233         if(!(subject instanceof InternalResource)) {
1234                 return super.getPossibleRelatedValue(subject, relation, binding);
1235         } else {
1236                 return null;
1237         }
1238         }
1239
1240         @Override
1241         final public Resource getPossibleObject(final Resource subject, final Resource relation)
1242                         throws ManyObjectsForFunctionalRelationException, ServiceException {
1243         if(!(subject instanceof InternalResource)) {
1244                 return super.getPossibleObject(subject, relation);
1245         } else {
1246                 return null;
1247         }
1248         }
1249
1250     public void close() {
1251         if (writeState.out != null) {
1252             try (OutputStream out = writeState.out) {
1253             } catch (IOException e) {
1254                 Logger.defaultLogError("Failed to close delayed write graph temporary commit output stream", e);
1255             } finally {
1256                 writeState.out = null;
1257             }
1258         }
1259         if (writeState.in != null) {
1260             try (InputStream in = writeState.in) {
1261             } catch (IOException e) {
1262                 Logger.defaultLogError("Failed to close delayed write graph temporary commit input stream", e);
1263             } finally {
1264                 writeState.in = null;
1265             }
1266         }
1267         if (writeState.tempFile != null) {
1268             writeState.tempFile.delete();
1269             writeState.tempFile = null;
1270         }
1271     }
1272
1273 }