package fi.vtt.simantics.procore.internal; import java.util.TreeMap; import org.simantics.db.Metadata; import org.simantics.db.Operation; import org.simantics.db.Resource; import org.simantics.db.VirtualGraph; import org.simantics.db.WriteGraph; import org.simantics.db.common.MetadataUtils; import org.simantics.db.common.exception.DebugException; import org.simantics.db.common.utils.Logger; import org.simantics.db.exception.DatabaseException; import org.simantics.db.exception.ImmutableException; import org.simantics.db.exception.ServiceException; import org.simantics.db.impl.ClusterI; import org.simantics.db.impl.MemWatch; import org.simantics.db.impl.ResourceImpl; import org.simantics.db.impl.VirtualGraphImpl; import org.simantics.db.impl.graph.ReadGraphImpl; import org.simantics.db.impl.graph.WriteGraphImpl; import org.simantics.db.impl.graph.WriteSupport; import org.simantics.db.impl.query.QueryProcessor; import org.simantics.db.impl.query.QuerySupport; import org.simantics.db.procore.cluster.ClusterImpl; import org.simantics.db.procore.protocol.Constants; import org.simantics.db.request.Write; import org.simantics.db.request.WriteOnly; import org.simantics.db.request.WriteResult; import org.simantics.db.request.WriteTraits; import org.simantics.db.service.ByteReader; public class WriteSupportImpl implements WriteSupport { final private SessionImplSocket session; final private QueryProcessor queryProcessor; final private State state; final private QuerySupport querySupport; final private TreeMap metadata; WriteSupportImpl(SessionImplSocket session) { this.session = session; this.queryProcessor = session.getQueryProvider2(); this.state = session.state; this.querySupport = session.querySupport; this.metadata = new TreeMap(); assert(this.session != null); assert(this.queryProcessor != null); assert(this.state != null); assert(this.querySupport != null); } @Override public void flushCluster() { session.clusterTable.flushCluster(session.graphSession); if(session.defaultClusterSet != null) { long resourceId = session.defaultClusterSet.getResourceId(); session.clusterSetsSupport.put(resourceId, Constants.NewClusterId); } } @Override public void flushCluster(Resource r) { session.clusterStream.reallyFlush(); } @Override public boolean writeOnly() { return session.writeOnly; } @Override public void flush(boolean intermediate) { if (!session.state.isWriteTransaction()) throw new IllegalStateException("Can only flush during transaction."); gc(); } final public void claim(VirtualGraph provider, Resource subject, Resource predicate, Resource object) throws ServiceException { claim(provider, querySupport.getId(subject), querySupport.getId(predicate), querySupport.getId(object)); } @Override final public void claim(VirtualGraph provider, int subject, int predicate, int object) throws ServiceException { provider = session.getProvider(provider); if (writeOnly()) { if (provider != null) { ((VirtualGraphImpl)provider).claim(subject, predicate, object); queryProcessor.updateStatements(subject, predicate); session.clientChanges.claim(subject, predicate, object); } else { claimImpl2(subject, predicate, object); } } else { // queryProcessor.acquireWrite(writeState.getGraph()); if (provider != null) { ((VirtualGraphImpl)provider).claim(subject, predicate, object); queryProcessor.updateStatements(subject, predicate); session.clientChanges.claim(subject, predicate, object); } else { claimImpl(subject, predicate, object); } queryProcessor.releaseWrite(session.writeState.getGraph()); } } @Override public void setValue(VirtualGraph provider, Resource resource, byte[] value) throws ServiceException { provider = session.getProvider(provider); if (writeOnly()) { if (provider != null) { ((VirtualGraphImpl)provider).claimValue(((ResourceImpl) resource).id, value, value.length); queryProcessor.updateValue(querySupport.getId(resource)); session.clientChanges.claimValue(resource); } else { try { addSetValue(((ResourceImpl) resource).id, value, value.length); } catch (DatabaseException e) { Logger.defaultLogError(e); } } } else { if (provider != null) { ((VirtualGraphImpl)provider).claimValue(((ResourceImpl) resource).id, value, value.length); queryProcessor.updateValue(querySupport.getId(resource)); session.clientChanges.claimValue(resource); } else { try { addSetValue(((ResourceImpl) resource).id, value, value.length); } catch (DatabaseException e) { Logger.defaultLogError(e); } } queryProcessor.releaseWrite(session.writeState.getGraph()); } } @Override public Resource createResource(VirtualGraph provider) throws DatabaseException { if (provider != null) { int newId = ((VirtualGraphImpl)provider).newResource(false); return new ResourceImpl(session.resourceSupport, newId); } else { return session.getNewResource(); } } @Override public Resource createResource(VirtualGraph provider, long clusterId) throws DatabaseException { assert (provider == null); return session.getNewResource(clusterId); } @Override public Resource createResource(VirtualGraph provider, Resource clusterSet) throws DatabaseException { assert(provider == null); assert(clusterSet != null); return session.getNewResource(clusterSet); } @Override public void createClusterSet(VirtualGraph provider, Resource clusterSet) throws DatabaseException { assert(provider == null); assert(clusterSet != null); session.getNewClusterSet(clusterSet); } @Override public boolean hasClusterSet(VirtualGraph dummy, Resource clusterSet) throws ServiceException { return session.containsClusterSet(clusterSet); } @Override public Resource setDefaultClusterSet(Resource clusterSet) throws ServiceException { return session.setDefaultClusterSet4NewResource(clusterSet); } @Override public void denyValue(VirtualGraph provider, Resource resource) throws ServiceException { provider = session.getProvider(provider); if (null == provider) { int key = ((ResourceImpl)resource).id; ClusterI cluster = session.clusterTable.getClusterByResourceKey(key); if (cluster.getImmutable() && (session.serviceMode & SessionImplSocket.SERVICE_MODE_ALLOW) == 0) if(key != queryProcessor.getRootLibrary()) throw new ImmutableException("Trying to modify immutable resource key=" + key); try { cluster.removeValue(key, session.clusterTranslator); } catch (DatabaseException e) { Logger.defaultLogError(e); return; } queryProcessor.updateValue(key); session.clientChanges.claimValue(resource); } else { ((VirtualGraphImpl)provider).denyValue(((ResourceImpl) resource).id); queryProcessor.updateValue(querySupport.getId(resource)); session.clientChanges.claimValue(resource); } if (!writeOnly()) queryProcessor.releaseWrite(session.writeState.getGraph()); } @Override public boolean removeStatement(VirtualGraph provider, Resource subject, Resource predicate, Resource object) throws ServiceException { boolean ret = true; if (writeOnly()) { if (provider != null) { ((VirtualGraphImpl)provider).deny(querySupport.getId(subject), querySupport.getId(predicate), querySupport.getId(object)); queryProcessor.updateStatements(querySupport.getId(subject), querySupport.getId(predicate)); } else { ret = removeStatement(querySupport.getId(subject), querySupport.getId(predicate), querySupport.getId(object)); } } else { if (provider != null) { // queryProcessor.acquireWrite(writeState.getGraph()); ((VirtualGraphImpl)provider).deny(querySupport.getId(subject), querySupport.getId(predicate), querySupport.getId(object)); queryProcessor.updateStatements(querySupport.getId(subject), querySupport.getId(predicate)); queryProcessor.releaseWrite(session.writeState.getGraph()); } else { int sid = querySupport.getId(subject); int pid = querySupport.getId(predicate); int oid = querySupport.getId(object); if (sid < 0 || pid < 0 || oid < 0) { // One of the resources is virtual, cannot remove such // statement from persistent storage. return false; } // queryProcessor.acquireWrite(writeState.getGraph()); ret = removeStatement(sid, pid, oid); queryProcessor.releaseWrite(session.writeState.getGraph()); } } session.clientChanges.deny(subject, predicate, object); return ret; } @Override public synchronized void performWriteRequest(WriteGraph graph_, Write request) throws DatabaseException { WriteGraphImpl graph = (WriteGraphImpl)graph_; // graph.state.barrier.inc(); try { request.perform(graph); } catch (Throwable t) { t.printStackTrace(); } // graph.state.barrier.dec(); // graph.waitAsync(request); queryProcessor.performDirtyUpdates(graph); // Do not fire metadata listeners for virtual requests if(graph.getProvider() == null) { //session.fireMetadataListeners(graph, session.clientChanges); state.commitAndContinue(graph, session.clusterStream, request); //session.clientChanges = new ClientChangesImpl(session); } // graph.state.barrier.assertReady(); } @Override public synchronized T performWriteRequest(WriteGraph graph_, WriteResult request) throws DatabaseException { WriteGraphImpl graph = (WriteGraphImpl)graph_; // graph.state.barrier.inc(); T result = null; Throwable t = null; try { result = request.perform(graph); } catch (Throwable t2) { if(DebugException.DEBUG) new DebugException(t2).printStackTrace(); t = t2; } // graph.state.barrier.dec(); // graph.waitAsync(request); queryProcessor.performDirtyUpdates(graph); // Do not fire metadata listeners for virtual requests if(graph.getProvider() == null) { //session.fireMetadataListeners((WriteGraphImpl)graph, session.clientChanges); state.commitAndContinue(graph, session.clusterStream, request); //session.clientChanges = new ClientChangesImpl(session); } if(t != null) { if(t instanceof DatabaseException) throw (DatabaseException)t; else throw new DatabaseException(t); } return result; } @Override public synchronized void performWriteRequest(WriteGraph graph, WriteOnly request) throws DatabaseException { session.acquireWriteOnly(); request.perform(graph); ReadGraphImpl impl = (ReadGraphImpl)graph; queryProcessor.performDirtyUpdates(impl); // Do not fire metadata listeners for virtual requests if(graph.getProvider() == null) { //session.fireMetadataListeners(impl, session.clientChanges); state.commitAndContinue(session.writeState.getGraph(), session.clusterStream, request); //session.clientChanges = new ClientChangesImpl(session); } session.releaseWriteOnly(impl); } @Override public void gc() { if (MemWatch.isLowOnMemory()) { session.clusterTable.gc(); queryProcessor.gc(0, Integer.MAX_VALUE); System.gc(); } } @Override public void claimValue(VirtualGraph provider, Resource resource, byte[] value) throws DatabaseException { claimValue(provider, ((ResourceImpl)resource).id, value, value.length); } @Override public void claimValue(VirtualGraph provider, int resource, byte[] value, int length) throws DatabaseException { provider = session.getProvider(provider); if (writeOnly()) { if (provider != null) { ((VirtualGraphImpl)provider).claimValue(resource, value, length); queryProcessor.updateValue(resource); session.clientChanges.claimValue(resource); } else { addSetValue(resource, value, length); } } else { if (provider != null) { ((VirtualGraphImpl)provider).claimValue(resource, value, length); queryProcessor.updateValue(resource); session.clientChanges.claimValue(resource); queryProcessor.releaseWrite(session.writeState.getGraph()); } else { try { addSetValue(resource, value, length); } catch (DatabaseException e) { throw e; } catch (Throwable t) { throw new DatabaseException(t); } finally { queryProcessor.releaseWrite(session.writeState.getGraph()); } } } } @Override public void claimValue(VirtualGraph provider, Resource resource, ByteReader reader, int amount) throws DatabaseException { claimValue(provider, resource, reader.readBytes(null, amount)); } @Override public void addMetadata(Metadata data) throws ServiceException { MetadataUtils.addMetadata(session, metadata, data); } @SuppressWarnings("unchecked") @Override public T getMetadata(Class clazz) throws ServiceException { return MetadataUtils.getMetadata(session, metadata, clazz); } @Override public TreeMap getMetadata() { return metadata; } @Override public void commitDone(WriteTraits writeTraits, long csid) { metadata.clear(); if (this.writeTraits == writeTraits) { session.graphSession.undoContext.clear(); this.writeTraits = null; } // if (null != operation) // operation = null; } @Override public int clearMetadata() { int ret = metadata.size(); metadata.clear(); return ret; } @Override public void clearUndoList(WriteTraits writeTraits) { this.writeTraits = writeTraits; } @Override public void startUndo() { session.state.setCombine(false); } private WriteTraits writeTraits = null; private void addSetValue(int subject, byte[] value, int length) throws DatabaseException { ClusterI cluster = session.clusterTable.getClusterByResourceKey(subject); if (cluster.getImmutable() && (session.serviceMode & SessionImplSocket.SERVICE_MODE_ALLOW) == 0) if(subject != queryProcessor.getRootLibrary()) throw new ImmutableException("Trying to modify immutable resource key=" + subject); ClusterI cluster2 = cluster.setValue(subject, value, length, session.clusterTranslator); if (cluster2 != cluster) session.clusterTable.replaceCluster(cluster2); session.clientChanges.claimValue(subject); if (cluster2.isWriteOnly()) return; queryProcessor.updateValue(subject); } final private void claimImpl(int subject, int predicate, int object) throws ServiceException { assert (subject != 0); assert (predicate != 0); assert (object != 0); ClusterImpl cluster = session.clusterTable.getClusterByResourceKey(subject); assert (null != cluster); if (cluster.getImmutable() && (session.serviceMode & SessionImplSocket.SERVICE_MODE_ALLOW) == 0) if(subject != queryProcessor.getRootLibrary()) throw new ImmutableException("Trying to modify immutable resource key=" + subject); try { ClusterI c = cluster.addRelation(subject, predicate, object, session.clusterTranslator); if (null != c && c != cluster) session.clusterTable.replaceCluster(c); } catch (DatabaseException e) { Logger.defaultLogError(e); throw new RuntimeException(e); } queryProcessor.updateStatements(subject, predicate); session.clientChanges.claim(subject, predicate, object); } final private void claimImpl2(int subject, int predicate, int object) { assert (subject != 0); assert (predicate != 0); assert (object != 0); ClusterI cluster = session.clusterTable.getClusterByResourceKey(subject); try { ClusterI c = cluster.addRelation(subject, predicate, object, session.clusterTranslator); if (null != c && c != cluster) session.clusterTable.replaceCluster(c); } catch (DatabaseException e) { Logger.defaultLogError(e); } if (cluster.isWriteOnly()) return; queryProcessor.updateStatements(subject, predicate); } private boolean removeStatement(int subject, int predicate, int object) throws ImmutableException { assert (subject != 0); assert (predicate != 0); assert (object != 0); ClusterI cluster = session.clusterTable.getClusterByResourceKey(subject); assert (null != cluster); if (cluster.getImmutable() && (session.serviceMode & SessionImplSocket.SERVICE_MODE_ALLOW) == 0) if(subject != queryProcessor.getRootLibrary()) throw new ImmutableException("Trying to modify immutable resource key=" + subject); try { cluster.denyRelation(subject, predicate, object, session.clusterTranslator); } catch (DatabaseException e) { Logger.defaultLogError(e); return false; } queryProcessor.updateStatements(subject, predicate); return true; } }