package org.simantics.db.impl.graph;
-import java.io.DataInput;
-import java.io.DataOutput;
+import java.io.Closeable;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.OutputStream;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
+import java.nio.file.Files;
import java.util.ArrayList;
import java.util.HashSet;
-import java.util.IdentityHashMap;
-import java.util.List;
import java.util.Set;
import java.util.TreeMap;
import java.util.UUID;
import org.eclipse.core.runtime.Platform;
import org.simantics.databoard.Bindings;
import org.simantics.databoard.accessor.Accessor;
-import org.simantics.databoard.accessor.reference.ChildReference;
import org.simantics.databoard.binding.Binding;
import org.simantics.databoard.binding.error.BindingConstructionException;
-import org.simantics.databoard.binding.impl.BindingPrintContext;
-import org.simantics.databoard.serialization.SerializationException;
import org.simantics.databoard.serialization.Serializer;
import org.simantics.databoard.type.Datatype;
-import org.simantics.databoard.util.IdentityPair;
import org.simantics.databoard.util.binary.RandomAccessBinary;
import org.simantics.db.Metadata;
import org.simantics.db.ReadGraph;
import org.simantics.layer0.Layer0;
import org.simantics.utils.datastructures.MapList;
-import gnu.trove.list.array.TIntArrayList;
import gnu.trove.map.hash.TIntIntHashMap;
import gnu.trove.map.hash.TObjectIntHashMap;
* immediately but with an explicit commit method. All read operations
* return results based on the old graph.
*/
-public class DelayedWriteGraph extends ReadGraphImpl implements WriteGraph, ByteReader {
-
- private static final boolean DEBUG = false;
- public static int BUFFER = 65536;
-
- static class Haxx extends Binding {
-
- static final Serializer serializer = new Serializer() {
-
- public byte[] serialize(Object obj) throws SerializationException {
- return (byte[])obj;
- }
-
- @Override
- public void serialize(DataOutput out,
- TObjectIntHashMap<Object> identities, Object obj)
- throws IOException {
- throw new Error("Not supported.");
- }
-
- @Override
- public void serialize(DataOutput out, Object obj)
- throws IOException {
- throw new Error("Not supported.");
- }
-
- @Override
- public Object deserialize(DataInput in, List<Object> identities)
- throws IOException {
- throw new Error("Not supported.");
- }
-
- @Override
- public Object deserialize(DataInput in) throws IOException {
- throw new Error("Not supported.");
- }
-
- @Override
- public void deserializeTo(DataInput in, List<Object> identities,
- Object obj) throws IOException {
- throw new Error("Not supported.");
- }
-
- @Override
- public void deserializeTo(DataInput in, Object obj)
- throws IOException {
- throw new Error("Not supported.");
- }
-
- @Override
- public void skip(DataInput in, List<Object> identities)
- throws IOException {
- throw new Error("Not supported.");
- }
-
- @Override
- public void skip(DataInput in) throws IOException {
- throw new Error("Not supported.");
- }
-
- @Override
- public Integer getConstantSize() {
- throw new Error("Not supported.");
- }
-
- @Override
- public int getSize(Object obj, TObjectIntHashMap<Object> identities)
- throws IOException {
- throw new Error("Not supported.");
- }
-
- @Override
- public int getSize(Object obj) throws IOException {
- throw new Error("Not supported.");
- }
-
- @Override
- public int getMinSize() {
- throw new Error("Not supported.");
- }
-
- };
-
- @Override
- public Serializer serializer() {
- return serializer;
- }
-
- @Override
- public void accept(Visitor1 v, Object obj) {
- throw new Error("Not supported.");
- }
-
- @Override
- public <T> T accept(Visitor<T> v) {
- throw new Error("Not supported.");
- }
-
- @Override
- public boolean isInstance(Object obj) {
- throw new Error("Not supported.");
- }
-
- @Override
- public void assertInstaceIsValid(Object obj, Set<Object> validInstances)
- throws org.simantics.databoard.binding.error.BindingException {
- throw new Error("Not supported.");
- }
-
- @Override
- public int deepHashValue(Object value,
- IdentityHashMap<Object, Object> hashedObjects)
- throws org.simantics.databoard.binding.error.BindingException {
- throw new Error("Not supported.");
- }
-
- @Override
- public int deepCompare(Object o1, Object o2,
- Set<IdentityPair<Object, Object>> compareHistory)
- throws org.simantics.databoard.binding.error.BindingException {
- throw new Error("Not supported.");
- }
-
- @Override
- public void readFrom(Binding srcBinding, Object src, Object dst)
- throws org.simantics.databoard.binding.error.BindingException {
- throw new Error("Not supported.");
- }
-
- @Override
- public Object readFromTry(Binding srcBinding, Object src, Object dst)
- throws org.simantics.databoard.binding.error.BindingException {
- throw new Error("Not supported.");
- }
-
- @Override
- protected void toString(Object value, BindingPrintContext ctx) throws org.simantics.databoard.binding.error.BindingException {
- throw new Error("Not supported.");
- }
-
- @Override
- public int getComponentCount() {
- throw new Error("Not supported.");
- }
+public class DelayedWriteGraph extends ReadGraphImpl implements WriteGraph, ByteReader, Closeable {
- @Override
- public Binding getComponentBinding(int index) {
- throw new Error("Not supported.");
- }
+ private static final boolean DEBUG = false;
+ private static final int BUFFER_SIZE = 512*1024;
- @Override
- public Binding getComponentBinding(ChildReference path) {
- throw new Error("Not supported.");
- }
-
- }
-
- private static final Haxx haxx = new Haxx();
+ private static final PassthroughSerializerBinding PASSTHROUGH = new PassthroughSerializerBinding();
private final int TERM = 0;
private final int CLAIM = 1;
private final int DENY = 5;
private final int DENY_VALUE = 6;
private final int COMMIT_AND_CONTINUE = 7;
-
- static class ClusterSet {
- public Resource resource;
- ClusterSet() {
- ids = new TIntArrayList();
- old = false;
- }
- ClusterSet(boolean old, Resource r) {
- ids = new TIntArrayList();
- this.old = old;
- this.resource = r;
- }
- void add(int id) {
- ids.add(id);
- }
- private TIntArrayList ids;
- private final boolean old; // true if
- boolean isNew() {
- return !old;
- }
- }
- public class State {
+
+ private static class State {
public File tempFile;
public FileOutputStream out;
+ public FileInputStream in;
public ArrayList<Resource> idToResource = new ArrayList<Resource>();
public TIntIntHashMap externalToId = new TIntIntHashMap();
public ArrayList<Binding> idToBinding = new ArrayList<Binding>();
public int valueCount = 0;
public int fileCount = 0;
}
-
- public State writeState;
- public TreeMap<String,byte[]> metadata = new TreeMap<String,byte[]>();
-
- Layer0 b;
- Session session;
+
+ private State writeState;
+ private TreeMap<String,byte[]> metadata = new TreeMap<String,byte[]>();
+
+ private FileChannel channel;
+ private byte[] bytes = new byte[BUFFER_SIZE];
+ private ByteBuffer bb = ByteBuffer.wrap(bytes);
+ private int byteIndex = 0;
+
+ private Layer0 b;
+ private Session session;
+
public static Resource convertDelayedResource(Resource r) {
- if (r instanceof InternalResource) {
- InternalResource ri = (InternalResource)r;
- return ri.resource;
- }
+ if (r instanceof InternalResource)
+ return ((InternalResource) r).resource;
return r;
}
+
private static class InternalResource implements Resource {
int id;
long clusterId = 0;
Resource resource = null;
Resource clusterSet = null;
-
+
public InternalResource(int id, long clusterId) {
this.id = id;
this.clusterId = clusterId;
this.id = id;
this.clusterSet = clusterSet;
}
-
+
@Override
public long getResourceId() {
throw new UnsupportedOperationException();
public Resource get() {
return this;
}
-
+
@Override
public boolean isPersistent() {
return false;
@Override
public int compareTo(Resource o) {
- if(o instanceof InternalResource) {
- return Integer.compare(id, ((InternalResource)o).id);
- } else {
- return -1;
- }
+ if(o instanceof InternalResource)
+ return Integer.compare(id, ((InternalResource)o).id);
+ return -1;
}
@Override
result = prime * result + id;
return result;
}
-
+
@Override
public int getThreadHash() {
return hashCode();
return false;
return true;
}
-
@Override
public boolean equalsResource(Resource other) {
return equals(other);
}
+
@Override
public String toString() {
StringBuilder sb = new StringBuilder(32);
session = g.getSession();
b = Layer0.getInstance(g);
this.writeState = state;
- }
+ }
public DelayedWriteGraph newSync() {
return new DelayedWriteGraph(this, writeState);
if (ws.hasClusterSet(null, clusterSet))
throw new ClusterSetExistException("Cluster set exist already. Resource=" + clusterSet);
writeState.clusterSetsForExistingResources.add(clusterSet);
- } else {
+ } else {
if(!writeState.clusterSets.add(clusterSet))
throw new ClusterSetExistException("Cluster set exist already. Resource=" + clusterSet);
}
}
}
+ private OutputStream valueWriter = new OutputStream() {
+ @Override
+ public void write(int b) throws IOException {
+ writeByte(b);
+ }
+ };
+
@Override
public void claimValue(Resource resource, Object value, Binding binding)
throws ServiceException {
try {
-
+
writeByte(CLAIM_VALUE_B);
writeInt(getId(resource));
Serializer serializer = binding.serializer();
int size = serializer.getSize(value);
writeInt(size);
- serializer.serialize(new OutputStream() {
-
- @Override
- public void write(int b) throws IOException {
- writeByte(b);
- }
-
- }, value);
-
+ serializer.serialize(valueWriter, value);
+
} catch(IOException e) {
Logger.defaultLogError(e);
throw new ServiceException(e);
public void flushCluster(Resource r) throws ServiceException {
throw new ServiceException("Operation flushCluster(" + r + " not implemented.");
}
-
- private FileChannel channel;
- byte[] bytes = new byte[BUFFER];
- byte[] buffer = new byte[BUFFER];
- ByteBuffer bb = ByteBuffer.wrap(bytes);
- int byteIndex = 0;
-
- private void writeReset(int size) {
+ private void writeReset(int size) {
byteIndex = 0;
bb.position(0);
bb.limit(size);
try {
-
- if(writeState.tempFile == null) {
-
+ if (writeState.tempFile == null) {
File workspace = Platform.getLocation().toFile();
File temp = new File(workspace, "tempFiles");
- temp.mkdirs();
-
File base = new File(temp, "delayed");
- base.mkdirs();
+ Files.createDirectories(base.toPath());
writeState.tempFile = new File(base, UUID.randomUUID().toString());
writeState.out = new FileOutputStream(writeState.tempFile);
channel = writeState.out.getChannel();
-
}
for (int got=0;got < size;) {
int n = channel.write(bb);
if (n <= 0) {
- new Exception().printStackTrace();
+ Logger.defaultLogError(new Exception("FileChannel.write returned " + n));
return;
}
got += n;
}
} catch (IOException e) {
- e.printStackTrace();
+ Logger.defaultLogError("Failed to write buffer of " + size + " bytes to temporary file " + writeState.tempFile, e);
}
}
private void reset() {
byteIndex = 0;
- try {
- bb.clear();
- for(int got=0; got < BUFFER;) {
- int n = channel.read(bb);
- if (n <= 0)
- return;
- got += n;
+ bb.clear();
+ if (channel != null) {
+ try {
+ for(int got=0; got < BUFFER_SIZE;) {
+ int n = channel.read(bb);
+ if (n <= 0)
+ return;
+ got += n;
+ }
+ } catch (IOException e) {
+ Logger.defaultLogError("FileChannel.read failed", e);
}
- } catch (IOException e) {
- e.printStackTrace();
}
}
private void writeInt(int i) {
- if(byteIndex < (BUFFER-4)) {
+ if(byteIndex < (BUFFER_SIZE-4)) {
bytes[byteIndex++] = (byte)(i&0xff);
bytes[byteIndex++] = (byte)((i>>>8)&0xff);
bytes[byteIndex++] = (byte)((i>>>16)&0xff);
bytes[byteIndex++] = (byte)((i>>>24)&0xff);
- if (byteIndex == BUFFER)
- writeReset(BUFFER);
+ if (byteIndex == BUFFER_SIZE)
+ writeReset(BUFFER_SIZE);
} else {
- int has = BUFFER-byteIndex;
- if(has == 0) writeReset(BUFFER);
+ int has = BUFFER_SIZE-byteIndex;
+ if(has == 0) writeReset(BUFFER_SIZE);
bytes[byteIndex++] = (byte)(i&0xff);
- if(has == 1) writeReset(BUFFER);
+ if(has == 1) writeReset(BUFFER_SIZE);
bytes[byteIndex++] = (byte)((i>>>8)&0xff);
- if(has == 2) writeReset(BUFFER);
+ if(has == 2) writeReset(BUFFER_SIZE);
bytes[byteIndex++] = (byte)((i>>>16)&0xff);
- if(has == 3) writeReset(BUFFER);
+ if(has == 3) writeReset(BUFFER_SIZE);
bytes[byteIndex++] = (byte)((i>>>24)&0xff);
- if(has == 4) writeReset(BUFFER);
+ if(has == 4) writeReset(BUFFER_SIZE);
}
}
-
+
private int readInt() {
- if(byteIndex < (BUFFER-4)) {
+ if(byteIndex < (BUFFER_SIZE-4)) {
int result = (int)
((bytes[byteIndex++] & 0xff) |
((bytes[byteIndex++] & 0xff)<<8) |
((bytes[byteIndex++] & 0xff)<<24));
return result;
} else {
- int has = BUFFER-byteIndex;
+ int has = BUFFER_SIZE-byteIndex;
int result = 0;
if(has == 0) reset();
result = (int)(bytes[byteIndex++] & 0xff);
return result;
}
}
-
+
private byte readByte() {
byte result = bytes[byteIndex++];
- if(byteIndex == BUFFER) reset();
+ if(byteIndex == BUFFER_SIZE) reset();
return result;
}
private void writeByte(int b) {
bytes[byteIndex++] = (byte)b;
- if(byteIndex == BUFFER) writeReset(BUFFER);
+ if(byteIndex == BUFFER_SIZE) writeReset(BUFFER_SIZE);
}
-
+
private void writeBytes(byte[] data) {
- int has = BUFFER-byteIndex;
+ int has = BUFFER_SIZE-byteIndex;
int amount = data.length;
if(has > amount) {
System.arraycopy(data, 0, bytes, byteIndex, amount);
byteIndex += amount;
} else {
System.arraycopy(data, 0, bytes, byteIndex, has);
- writeReset(BUFFER);
+ writeReset(BUFFER_SIZE);
ByteBuffer bb2 = ByteBuffer.wrap(data);
bb2.position(has);
try {
channel.write(bb2);
} catch (IOException e) {
- e.printStackTrace();
+ Logger.defaultLogError("FileChannel.write failed", e);
}
}
}
-
- public byte[] readBytes(int amount) {
- return readBytes(buffer, amount);
- }
-
+
public byte[] readBytes(byte[] result, int amount) {
if(result == null) result = new byte[amount];
- int has = BUFFER-byteIndex;
+ int has = BUFFER_SIZE-byteIndex;
if(has > amount) {
System.arraycopy(bytes, byteIndex, result, 0, amount);
byteIndex += amount;
try {
got += channel.read(bb2);
if(got == -1) {
- new Exception().printStackTrace();
+ // End-of-stream, why log this?
return result;
}
-
} catch (IOException e) {
- e.printStackTrace();
+ Logger.defaultLogError("FileChannel.read failed", e);
}
reset();
}
return result;
}
-
+
public void commit(final WriteOnlyGraph w, final WriteTraits traits) throws ServiceException {
writeState.bindingToId = null;
writeState.externalToId = null;
- try {
-
- writeByte(TERM);
- if(byteIndex > 0) {
+ writeByte(TERM);
+
+ if (writeState.out != null) {
+ // Flush current buffer to file only if backing file has already
+ // been taken into use.
+ if (byteIndex > 0)
writeReset(byteIndex);
+
+ try (OutputStream out = writeState.out) {
+ channel.force(false);
+ } catch (IOException e) {
+ throw new ServiceException(e);
+ } finally {
+ writeState.out = null;
}
- channel.force(false);
- writeState.out.close();
- FileInputStream fs = new FileInputStream(writeState.tempFile);
- channel = fs.getChannel();
-
- } catch (IOException e) {
- throw new ServiceException(e);
+ try {
+ writeState.in = new FileInputStream(writeState.tempFile);
+ channel = writeState.in.getChannel();
+ } catch (IOException e) {
+ throw new ServiceException(e);
+ }
}
-
+
w.getMetadata().putAll(metadata);
-
+
TransferableGraphSupport tgs = w.getService(TransferableGraphSupport.class);
// First create all resources defined by clusterId
}
reset();
- bb.limit(BUFFER);
+ bb.limit(BUFFER_SIZE);
try {
while(true) {
byte method = readByte();
- switch(method) {
+ switch(method) {
case TERM: {
+ if (DEBUG) {
+ System.out.println("Resources: " + writeState.idToResource.size());
+ System.out.println("Statements: " + writeState.statementCount);
+ System.out.println("Values: " + writeState.valueCount);
+ System.out.println("Files: " + writeState.fileCount);
+ System.out.println("Clusters: " + writeState.clusterCount);
+ }
return;
}
case CLAIM: {
Resource resource = getResource(readInt());
int len = readInt();
tgs.setValue(w, resource, null, this, len);
-// byte[] bytes = readBytes(len);
-// tgs.setValue(resource, null, bytes);
} break;
case COMMIT_AND_CONTINUE: {
XSupport xs = w.getService(XSupport.class);
else
throw new ServiceException(e);
} finally {
- try {
- channel.close();
- channel = null;
- } catch (IOException e) {
- throw new ServiceException(e);
+ channel = null;
+ if (writeState.in != null) {
+ try (InputStream in = writeState.in) {
+ } catch (IOException e) {
+ throw new ServiceException(e);
+ } finally {
+ writeState.in = null;
+ writeState.tempFile.delete();
+ }
}
}
-// System.out.println("Resources: " + state.resourceCount);
-// System.out.println("Statements: " + state.statementCount);
-// System.out.println("Values: " + state.valueCount);
}
-
+
private Resource getType(Object value) {
Class<?> clazz = value.getClass();
Resource dataType =
: clazz == String[].class ? b.StringArray
: clazz == boolean[].class ? b.BooleanArray
: clazz == byte[].class ? b.ByteArray
- : clazz == long[].class ? b.LongArray
+ : clazz == long[].class ? b.LongArray
: null
;
return dataType;
}
-
+
public long newCluster() {
return -1 - (++writeState.clusterCount);
}
public long getDefaultCluster() {
return writeState.defaultCluster;
}
-
+
public void setDefaultCluster(long cluster) {
writeState.defaultCluster = cluster;
}
-
+
@Override
public void syncRequest(final DelayedWrite request) throws DatabaseException {
-
+
try {
final DelayedWriteGraph dwg = new DelayedWriteGraph(this);
});
} catch (DatabaseException e) {
-
+
throw e;
-
+
} catch (Throwable e) {
-
+
throw new DatabaseException(e);
-
+
} finally {
-
}
-
+
}
-
+
@Override
public void syncRequest(WriteOnly request) throws DatabaseException {
-
+
Resource defaultClusterSet = setClusterSet4NewResource(null);
-
+
try {
WriteSupport ws = session.getService(WriteSupport.class);
ws.performWriteRequest(this, request);
} finally {
setClusterSet4NewResource(defaultClusterSet);
}
-
+
}
@SuppressWarnings("unchecked")
@Override
public <T> T getService(Class<T> api) {
-
+
if(ClusteringSupport.class == api) {
final ClusteringSupport support = (ClusteringSupport)super.getService(api);
return (T)new ClusteringSupport() {
-
+
@Override
public Resource getResourceByIndexAndCluster(int resourceIndex, long clusterId)
throws DatabaseException, ResourceNotFoundException {
return support.getResourceByIndexAndCluster(resourceIndex, clusterId);
}
-
+
@Override
public Resource getResourceByKey(int resourceKey) throws ResourceNotFoundException {
return support.getResourceByKey(resourceKey);
}
-
+
@Override
public int getNumberOfResources(long clusterId)
throws DatabaseException {
return support.getNumberOfResources(clusterId);
}
-
+
@Override
public long getCluster(Resource r) {
return support.getCluster(r);
}
-
+
@Override
public long createCluster() {
return newCluster();
}
-
+
@Override
public boolean isClusterSet(Resource r) throws DatabaseException {
return support.isClusterSet(r);
public Resource getClusterSetOfCluster(Resource r) throws DatabaseException {
return support.getClusterSetOfCluster(r);
}
-
+
@Override
public Resource getClusterSetOfCluster(long cluster) throws DatabaseException {
return support.getClusterSetOfCluster(cluster);
}
};
-
+
} else if (TransferableGraphSupport.class == api) {
-
+
final TransferableGraphSupport parentSupport = session.getService(TransferableGraphSupport.class);
-
+
return (T)new TransferableGraphSupport() {
@Override
writeInt(getId(resource));
writeInt(raw.length);
writeBytes(raw);
- writeInt(getBindingId(haxx));
+ writeInt(getBindingId(PASSTHROUGH));
}
-
+
@Override
public void setValue(WriteOnlyGraph graph, Resource resource, VirtualGraph provider, ByteReader reader, int amount)
throws DatabaseException {
writeInt(getId(resource));
writeInt(amount);
writeBytes(reader.readBytes(null, amount));
- writeInt(getBindingId(haxx));
+ writeInt(getBindingId(PASSTHROUGH));
}
-
+
@Override
public byte[] getValue(ReadGraph graph, Resource resource) {
return parentSupport.getValue(graph, resource);
}
-
+
@Override
public InputStream getValueStream(ReadGraph graph, Resource resource) {
return parentSupport.getValueStream(graph, resource);
}
-
+
};
}
-
+
return super.getService(api);
-
+
}
@Override
}
}
+ public void close() {
+ if (writeState.out != null) {
+ try (OutputStream out = writeState.out) {
+ } catch (IOException e) {
+ Logger.defaultLogError("Failed to close delayed write graph temporary commit output stream", e);
+ } finally {
+ writeState.out = null;
+ }
+ }
+ if (writeState.in != null) {
+ try (InputStream in = writeState.in) {
+ } catch (IOException e) {
+ Logger.defaultLogError("Failed to close delayed write graph temporary commit input stream", e);
+ } finally {
+ writeState.in = null;
+ }
+ }
+ if (writeState.tempFile != null) {
+ writeState.tempFile.delete();
+ writeState.tempFile = null;
+ }
+ }
+
}
Code Review / simantics / platform.git / blobdiff