--- /dev/null
+/*******************************************************************************\r
+ * Copyright (c) 2007, 2010 Association for Decentralized Information Management\r
+ * in Industry THTH ry.\r
+ * All rights reserved. This program and the accompanying materials\r
+ * are made available under the terms of the Eclipse Public License v1.0\r
+ * which accompanies this distribution, and is available at\r
+ * http://www.eclipse.org/legal/epl-v10.html\r
+ *\r
+ * Contributors:\r
+ * VTT Technical Research Centre of Finland - initial API and implementation\r
+ *******************************************************************************/\r
+package org.simantics.layer0.utils.writer;\r
+\r
+\r
+import gnu.trove.map.hash.TIntIntHashMap;\r
+import gnu.trove.map.hash.TIntLongHashMap;\r
+\r
+import java.io.BufferedInputStream;\r
+import java.io.BufferedOutputStream;\r
+import java.io.File;\r
+import java.io.FileInputStream;\r
+import java.io.FileOutputStream;\r
+import java.io.IOException;\r
+import java.io.ObjectInputStream;\r
+import java.io.ObjectOutputStream;\r
+import java.nio.channels.FileChannel;\r
+\r
+import org.eclipse.core.runtime.IProgressMonitor;\r
+import org.simantics.db.ReadGraph;\r
+import org.simantics.db.Resource;\r
+import org.simantics.db.WriteOnlyGraph;\r
+import org.simantics.db.exception.DatabaseException;\r
+\r
+public class DelayedGraphWriter extends AbstractDelayedGraphWriter {\r
+\r
+ protected File file;\r
+ protected ObjectOutputStream s;\r
+ TIntIntHashMap clusterHints = new TIntIntHashMap();\r
+ TIntLongHashMap clusterIds = new TIntLongHashMap();\r
+ \r
+ public DelayedGraphWriter(ReadGraph graph) {\r
+ super(graph);\r
+ try {\r
+ file = File.createTempFile("graph", ".tmp");\r
+ System.out.println("Temp file: " + file.getAbsolutePath());\r
+ s = new ObjectOutputStream(new BufferedOutputStream(new FileOutputStream(file), 1024*1024));\r
+// s = new ObjectOutputStream(backup);\r
+ \r
+ file.deleteOnExit();\r
+ } catch (IOException e) {\r
+ throw new RuntimeException("Opening output stream to temporary file failed", e);\r
+ } \r
+ }\r
+ \r
+ private void writeRef(int ref) throws IOException {\r
+ if(ref > 0) \r
+ timestamps.set(ref-1, time++);\r
+ s.writeInt(ref);\r
+ }\r
+ \r
+ @Override\r
+ public GraphWriter let(Resource p, Resource o) throws DatabaseException {\r
+ assert(p != null);\r
+ assert(o != null);\r
+ try {\r
+ s.writeByte(0);\r
+ writeRef(current);\r
+ writeRef(getPredicateId(p));\r
+ writeRef(getId(o));\r
+ } catch (IOException e) {\r
+ throw new RuntimeException("Writing statement failed.", e);\r
+ }\r
+ return this;\r
+ }\r
+\r
+ @Override\r
+ public GraphWriter let(Resource p, Object value, Resource dataType) throws DatabaseException {\r
+ assert(p != null);\r
+ assert(value != null);\r
+ assert(dataType != null);\r
+ try {\r
+ ++internalCount; \r
+ timestamps.add(0);\r
+ \r
+ s.writeByte(1); \r
+ writeRef(internalCount); \r
+ s.writeUnshared(value);\r
+ writeRef(getId(dataType)); \r
+ \r
+ s.writeByte(0);\r
+ writeRef(current);\r
+ writeRef(getPredicateId(p));\r
+ writeRef(internalCount);\r
+ } catch (IOException e) {\r
+ throw new RuntimeException("Writing statement failed.", e);\r
+ }\r
+ return this;\r
+ }\r
+ \r
+ @Override\r
+ public GraphWriter let(int clusterHint, Resource p, Object value,\r
+ Resource dataType) throws DatabaseException {\r
+ let(p, value, dataType);\r
+ clusterHints.put(internalCount, clusterHint);\r
+ return this;\r
+ }\r
+ \r
+ @Override\r
+ public GraphWriter createLiteral(Object value, Resource dataType) {\r
+ assert(value != null);\r
+ assert(dataType != null);\r
+ try {\r
+ ++internalCount; \r
+ timestamps.add(0);\r
+ \r
+ s.writeByte(1); \r
+ writeRef(internalCount); \r
+ s.writeUnshared(value);\r
+ writeRef(getId(dataType)); \r
+ \r
+ current = internalCount;\r
+ } catch (IOException e) {\r
+ throw new RuntimeException("Writing statement failed.", e);\r
+ }\r
+ return this;\r
+ }\r
+ \r
+ @Override\r
+ public GraphWriter createLiteral(int clusterHint, Object value,\r
+ Resource dataType) {\r
+ createLiteral(value, dataType);\r
+\r
+ clusterHints.put(current, clusterHint);\r
+ return this;\r
+ }\r
+ \r
+ @Override\r
+ public GraphWriter createInverse(int cluster, Resource r) {\r
+ assert(r != null);\r
+ create(cluster);\r
+ try {\r
+ s.writeByte(5);\r
+ writeRef(current);\r
+ writeRef(getId(r));\r
+ } catch (IOException e) {\r
+ throw new RuntimeException("Writing statement failed.", e);\r
+ }\r
+ return this;\r
+ } \r
+ \r
+ @Override\r
+ public GraphWriter createInverse(Resource r) {\r
+ assert(r != null);\r
+ create();\r
+ try {\r
+ s.writeByte(5);\r
+ writeRef(current);\r
+ writeRef(getId(r));\r
+ } catch (IOException e) {\r
+ throw new RuntimeException("Writing statement failed.", e);\r
+ }\r
+ return this;\r
+ } \r
+\r
+ protected Resource getResource(WriteOnlyGraph wg, Resource[] internals, long[] resourceIds, int id) {\r
+ if(id > 0) {\r
+ Resource ret = internals[id-1];\r
+ if(ret == null) {\r
+ throw new Error("Error");\r
+// ret = wg.newResource(types.get(id));\r
+// internals[id-1] = ret;\r
+// resourceIds[id-1] = ret.getResourceId();\r
+ }\r
+ if(timestamps.getQuick(id-1)==time)\r
+ internals[id-1] = null;\r
+ ++time;\r
+ return ret;\r
+ }\r
+ else if(id < 0) {\r
+ return externals.get(-1-id);\r
+ }\r
+ else\r
+ return null;\r
+ }\r
+ \r
+ @Override\r
+ public void commit(IProgressMonitor monitor, WriteOnlyGraph wg) throws DatabaseException {\r
+ \r
+ try {\r
+ if (s != null) {\r
+ s.close();\r
+ s = null;\r
+ }\r
+ externalsInv = null;\r
+\r
+ monitor.beginTask("", 100);\r
+\r
+ int lastPercentageDone = 0;\r
+ long fileLength = file.length();\r
+ \r
+// System.out.println("size of commit file: " + fileLength);\r
+\r
+ Resource[] internals = new Resource[internalCount];\r
+ resourceIds = new long[internalCount];\r
+ \r
+ \r
+ FileInputStream fis = new FileInputStream(file);\r
+ FileChannel fc = fis.getChannel();\r
+ ObjectInputStream is = new ObjectInputStream(new BufferedInputStream(fis, 1024*1024));\r
+ int resourceCounter = 0;\r
+ int statementCounter = 0;\r
+ int valueCounter = 0;\r
+ time = 0;\r
+ loop: while(true) {\r
+ switch(is.read()) {\r
+ case -1:\r
+ break loop;\r
+ case 0: {\r
+ int s = is.readInt();\r
+ int p = is.readInt();\r
+ int o = is.readInt();\r
+ Resource rs = getResource(wg, internals, resourceIds, s);\r
+ Resource rp = getResource(wg, internals, resourceIds, p);\r
+ Resource ro = getResource(wg, internals, resourceIds, o);\r
+ Resource rpInv = inverses.get(p);\r
+ wg.claim(rs, rp, rpInv, ro);\r
+ statementCounter += 2;\r
+ } break; \r
+ case 1: {\r
+ int id = is.readInt();\r
+ Object value = is.readUnshared();\r
+ int type = is.readInt();\r
+ \r
+ Resource r = newResource(wg, internals, id);\r
+ wg.claim(r, l0.InstanceOf, null, getResource(wg, internals, resourceIds, type));\r
+ wg.claimValue(r, value);\r
+ statementCounter ++;\r
+ resourceCounter ++;\r
+ valueCounter ++;\r
+ \r
+ } break;\r
+ case 2: {\r
+ wg.flushCluster();\r
+ } break; \r
+ case 3: {\r
+\r
+ int s = is.readInt();\r
+ int t = is.readInt();\r
+ \r
+ Resource type = getResource(wg, internals, resourceIds, t);\r
+ wg.claim(newResource(wg, internals, s), l0.InstanceOf, null, type);\r
+ statementCounter ++;\r
+ resourceCounter ++;\r
+ \r
+ } break;\r
+ case 4: {\r
+\r
+ int s = is.readInt();\r
+ newResource(wg, internals, s);\r
+ resourceCounter ++;\r
+ \r
+ } break;\r
+ case 5: { // InverseOf\r
+\r
+ int r1 = is.readInt();\r
+ int r2 = is.readInt();\r
+ \r
+ Resource rr1 = getResource(wg, internals, resourceIds, r1);\r
+ Resource rr2 = getResource(wg, internals, resourceIds, r2);\r
+ wg.claim(rr1, l0.InverseOf, l0.InverseOf, rr2);\r
+ statementCounter += 2;\r
+ \r
+ inverses.put(r1, rr2);\r
+ inverses.put(r2, rr1);\r
+ \r
+ } break;\r
+ }\r
+\r
+// if((counter % 200000) == 0) {\r
+// System.out.println("Written " + counter + " statements.");\r
+// }\r
+\r
+ double percentageDone = 100.0 * (double) fc.position() / (double) fileLength;\r
+ int newPercentageDone = (int) Math.round(percentageDone);\r
+ if(newPercentageDone > lastPercentageDone) {\r
+ monitor.setTaskName("Writing database (" + newPercentageDone + "%)");\r
+ monitor.worked(newPercentageDone - lastPercentageDone);\r
+ lastPercentageDone = newPercentageDone;\r
+ }\r
+ }\r
+ \r
+ System.out.println("clusterIds.size() = " + clusterIds.size());\r
+ \r
+ System.out.println("Wrote " + resourceCounter + " resources, " + statementCounter + " statements and " + valueCounter + " values.");\r
+ \r
+ } catch (IOException e) {\r
+ throw new RuntimeException("Commiting delayed graph writings failed.", e);\r
+ } catch (ClassNotFoundException e) {\r
+ throw new RuntimeException("Commiting delayed graph writings failed.", e);\r
+ } finally { \r
+ file.delete();\r
+ monitor.done();\r
+ }\r
+ }\r
+ \r
+ private Resource newResource(WriteOnlyGraph wg, Resource[] internals, int s) throws DatabaseException {\r
+ int clusterHint = clusterHints.get(s);\r
+ \r
+ Resource r;\r
+ if(clusterHint == 0)\r
+ r = wg.newResource();\r
+ else {\r
+ long clusterId = clusterIds.get(clusterHint);\r
+ if(clusterId == 0) {\r
+ wg.flushCluster();\r
+ r = wg.newResource();\r
+ clusterIds.put(clusterHint, clustering.getCluster(r));\r
+ }\r
+ else\r
+ r = wg.newResource(clusterId);\r
+ }\r
+ internals[s-1] = r;\r
+ resourceIds[s-1] = r.getResourceId();\r
+ return r;\r
+ }\r
+ \r
+// public ValueDescriptor createDescriptor(Object obj) {\r
+// \r
+// if(obj instanceof String[])\r
+// return new ValueDescriptor(ValueTrait.StaticValue, (String[])obj);\r
+// else if(obj instanceof int[])\r
+// return new ValueDescriptor(ValueTrait.StaticValue, (int[])obj);\r
+// else if(obj instanceof double[])\r
+// return new ValueDescriptor(ValueTrait.StaticValue, (double[])obj);\r
+// else if(obj instanceof float[])\r
+// return new ValueDescriptor(ValueTrait.StaticValue, (float[])obj);\r
+// else if(obj instanceof boolean[])\r
+// return new ValueDescriptor(ValueTrait.StaticValue, (boolean[])obj);\r
+// else if(obj instanceof long[])\r
+// return new ValueDescriptor(ValueTrait.StaticValue, (long[])obj);\r
+// else if(obj instanceof byte[])\r
+// return new ValueDescriptor(ValueTrait.StaticValue, (byte[])obj);\r
+// \r
+// else if(obj instanceof String)\r
+// return new ValueDescriptor(ValueTrait.StaticValue, new String[] {(String)obj});\r
+// else if(obj instanceof Double)\r
+// return new ValueDescriptor(ValueTrait.StaticValue, new double[] {(Double)obj});\r
+// else if(obj instanceof Float)\r
+// return new ValueDescriptor(ValueTrait.StaticValue, new float[] {(Float)obj});\r
+// else if(obj instanceof Integer)\r
+// return new ValueDescriptor(ValueTrait.StaticValue, new int[] {(Integer)obj});\r
+// else if(obj instanceof Boolean)\r
+// return new ValueDescriptor(ValueTrait.StaticValue, new boolean[] {(Boolean)obj});\r
+// else if(obj instanceof Byte)\r
+// return new ValueDescriptor(ValueTrait.StaticValue, new byte[] {(Byte)obj});\r
+// else if(obj instanceof Long)\r
+// return new ValueDescriptor(ValueTrait.StaticValue, new long[] {(Long)obj});\r
+// \r
+// throw new Error("Wrong type!");\r
+// \r
+// }\r
+\r
+ @Override\r
+ public GraphWriter create() {\r
+ \r
+ current = ++internalCount;\r
+ timestamps.add(0);\r
+\r
+ try {\r
+ s.writeByte(4);\r
+ s.writeInt(current);\r
+ } catch (IOException e) {\r
+ throw new RuntimeException("Writing statement failed.", e);\r
+ }\r
+\r
+ return this;\r
+ \r
+ }\r
+\r
+ @Override\r
+ public GraphWriter create(int clusterHint) {\r
+ \r
+ create(); \r
+ clusterHints.put(current, clusterHint);\r
+ return this;\r
+ \r
+ }\r
+ \r
+ @Override\r
+ public GraphWriter create(Resource type) {\r
+ assert(type != null);\r
+ \r
+ current = ++internalCount;\r
+ timestamps.add(0);\r
+\r
+ try {\r
+ s.writeByte(3);\r
+ s.writeInt(current);\r
+ s.writeInt(getId(type));\r
+ } catch (IOException e) {\r
+ throw new RuntimeException("Writing statement failed.", e);\r
+ }\r
+\r
+ return this;\r
+ \r
+ }\r
+ \r
+ @Override\r
+ public GraphWriter create(int clusterHint, Resource type) {\r
+ \r
+ create(type); \r
+ clusterHints.put(current, clusterHint);\r
+ return this;\r
+ \r
+ }\r
+\r
+ @Override\r
+ public Resource get() {\r
+ if(current > 0)\r
+ return new InternalResource(current);\r
+ else if(current < 0)\r
+ return externals.get(-1-current);\r
+ else\r
+ return null;\r
+ }\r
+ \r
+ @Override\r
+ public GraphWriter handle(Resource s) {\r
+ assert(s != null);\r
+ \r
+ current = getId(s);\r
+ return this;\r
+ }\r
+\r
+ public GraphWriter flush() { \r
+ try {\r
+ s.writeByte(2);\r
+ } catch(IOException e) {\r
+ throw new RuntimeException("Writing flush failed.", e);\r
+ }\r
+ return this;\r
+ }\r
+\r
+}\r
Code Review / simantics / platform.git / blobdiff