--- /dev/null
+package org.simantics.acorn.internal;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import org.simantics.acorn.internal.ClusterStream.ClusterEnum;
+import org.simantics.acorn.internal.ClusterStream.Data;
+import org.simantics.acorn.internal.ClusterStream.DebugInfo;
+import org.simantics.acorn.internal.ClusterStream.OpEnum;
+import org.simantics.acorn.internal.ClusterStream.StmEnum;
+import org.simantics.compressions.Compressions;
+import org.simantics.db.exception.RuntimeDatabaseException;
+import org.simantics.db.impl.ClusterTraitsBase;
+import org.simantics.db.procore.cluster.ClusterTraits;
+import org.simantics.db.procore.cluster.ClusterTraitsSmall;
+import org.simantics.db.service.Bytes;
+import org.simantics.db.service.ClusterUID;
+import org.simantics.utils.datastructures.Pair;
+
+import gnu.trove.map.hash.TIntByteHashMap;
+import gnu.trove.map.hash.TLongIntHashMap;
+
+
+public final class ClusterChange {
+
+ public static final int VERSION = 1;
+ public static final byte ADD_OPERATION = 2;
+ public static final byte REMOVE_OPERATION = 3;
+ public static final byte DELETE_OPERATION = 5;
+
+ public static final boolean DEBUG = false;
+ public static final boolean DEBUG_STAT = false;
+ public static final boolean DEBUG_CCS = false;
+
+ private static DebugInfo sum = new DebugInfo();
+
+ public final TIntByteHashMap foreignTable = new TIntByteHashMap();
+ private final DebugInfo info;
+// private final GraphSession graphSession;
+ public final ClusterUID clusterUID;
+ private final int SIZE_OFFSET;
+// private final int HEADER_SIZE;
+ // How much buffer is used before stream is flushed to server. The bigger the better.
+ public static final int MAX_FIXED_BYTES = (1<<15) + (1<<14);
+ private static final int MAX_FIXED_OPERATION_SIZE = 17 + 16;
+ private static final int MAX_FIXED_OPERATION_SIZE_AND_ROOM_FOR_ERROR = MAX_FIXED_OPERATION_SIZE + 36;
+ private int nextSize = MAX_FIXED_BYTES;
+ int byteIndex = 0;
+ private byte[] bytes = null; // Operation data.
+// private final byte[] header;
+ private boolean flushed = false;
+ private ArrayList<Pair<ClusterUID, byte[]>> stream;
+
+// public ClusterImpl clusterImpl;
+
+ public ClusterChange( ArrayList<Pair<ClusterUID, byte[]>> stream, ClusterUID clusterUID) {
+ this.clusterUID = clusterUID;
+ long[] longs = new long[ClusterUID.getLongLength()];
+ clusterUID.toLong(longs, 0);
+ this.stream = stream;
+// this.graphSession = clusterStream.graphSession;
+ info = new DebugInfo();
+// HEADER_SIZE = 8 + longs.length * 8;
+// header = new byte[HEADER_SIZE];
+ SIZE_OFFSET = 0;
+// Bytes.writeLE(header, SIZE_OFFSET + 0, 0); // Correct byte vector size is set with setHeaderVectorSize() later.
+// Bytes.writeLE(header, SIZE_OFFSET + 4, VERSION);
+// for (int i=0, offset=8; i<longs.length; ++i, offset+=8)
+// Bytes.writeLE(header, offset, longs[i]);
+ //initBuffer();
+// this.clusterStream = clusterStream;
+// this.clusterChange2 = new ClusterChange2(clusterUID, clusterImpl);
+// clusterStream.changes.add(this);
+ }
+
+// private void setHeaderVectorSize(int size) {
+// if (size < 0)
+// throw new RuntimeDatabaseException("Change set size can't be negative.");
+// int len = size + HEADER_SIZE - SIZE_OFFSET - 4;
+// Bytes.writeLE(header, SIZE_OFFSET, len);
+// }
+ @Override
+ public String toString() {
+ return super.toString() + " cluster=" + clusterUID + " off=" + byteIndex;
+ }
+ final public void initBuffer() {
+ flushed = false;
+ if (null == bytes || bytes.length < nextSize) {
+ bytes = new byte[nextSize];
+ nextSize = MAX_FIXED_BYTES;
+ }
+ byteIndex = 0;
+ }
+ private final void clear() {
+// if(clusterImpl != null && clusterImpl.change != null)
+// clusterImpl.change.init();
+ foreignTable.clear();
+ //initBuffer();
+ bytes = null;
+ byteIndex = 0;
+ if (DEBUG_STAT)
+ info.clear();
+ }
+ private final void checkInitialization() {
+// if (0 == byteIndex)
+// clusterStream.changes.addChange(this);
+ }
+ private final void printlnd(String s) {
+ System.out.println("DEBUG: ClusterChange " + clusterUID + ": " + s);
+ }
+ public final void createResource(short index) {
+ checkInitialization();
+ if (DEBUG)
+ printlnd("New ri=" + index + " offset=" + byteIndex);
+ if (index > ClusterTraits.getMaxNumberOfResources())
+ throw new RuntimeDatabaseException("Illegal resource index=" + index + ".");
+ checkBufferSpace(null);
+ bytes[byteIndex++] = (byte)52;
+ bytes[byteIndex++] = (byte)index;
+ bytes[byteIndex++] = (byte)(index>>>8);
+ }
+ void flushCollect(Change c) {
+ throw new UnsupportedOperationException();
+// flushInternal(graphSession, clusterUID);
+// if (DEBUG)
+// printlnd("Cluster change data was flushed.");
+// if (null != c) {
+// if (DEBUG)
+// printlnd("Clearing lookup for " + c.toString());
+// c.lookup1 = null;
+// c.lookup2 = null;
+// }
+// if (null != clusterImpl) {
+// clusterImpl.foreignLookup = null;
+// }
+ }
+
+ private final boolean checkBufferSpace(Change c) {
+// clusterStream.changes.checkFlush();
+ if(bytes == null) initBuffer();
+ if (MAX_FIXED_BYTES - byteIndex > MAX_FIXED_OPERATION_SIZE_AND_ROOM_FOR_ERROR) {
+ return false;
+ }
+ flush();
+// initBuffer();
+ return true;
+ }
+
+ private final void checkBufferSpace(int size) {
+ if(bytes == null) initBuffer();
+ if (bytes.length - byteIndex >= size)
+ return;
+ nextSize = Math.max(MAX_FIXED_BYTES, size);
+ flush();
+ initBuffer();
+ }
+
+ public final void addChange(Change c) {
+ checkInitialization();
+ checkBufferSpace(c);
+ byte operation = c.op0;
+ if(operation == ADD_OPERATION)
+ addStm(c, StmEnum.Add);
+ else if (operation == REMOVE_OPERATION)
+ addStm(c, StmEnum.Remove);
+ else if (operation == DELETE_OPERATION) {
+ if (DEBUG)
+ printlnd("Delete value offset=" + byteIndex + " " + c);
+ addByte(OpEnum.Delete.getOrMask());
+ addShort(ClusterTraitsBase.getResourceIndexFromResourceKeyNoThrow(c.key0));
+ }
+ c.lastArg = 0;
+ }
+
+ private final void addForeignLong(short index, ClusterUID clusterUID) {
+ byteIndex = clusterUID.toByte(bytes, byteIndex);
+ bytes[byteIndex++] = (byte)(index & 0xFF);
+ bytes[byteIndex++] = (byte)(index >>> 8);
+ }
+
+ private final ClusterEnum addIndexAndCluster(int key, ClusterUID clusterUID, byte lookIndex, byte[] lookup) {
+ assert(!clusterUID.equals(ClusterUID.Null));
+ short resourceIndex = ClusterTraitsBase.getResourceIndexFromResourceKeyNoThrow(key);
+ if (clusterUID.equals(this.clusterUID)) {
+ bytes[byteIndex++] = (byte)(resourceIndex & 0xFF);
+ bytes[byteIndex++] = (byte)(resourceIndex >>> 8);
+ return ClusterEnum.Local;
+ }
+
+ byte foreign = 0;
+ if(lookIndex > 0) {
+ if(lookup != null)
+ foreign = lookup[lookIndex];
+ } else {
+ foreign = foreignTable.get(key);
+ }
+ if (0 != foreign) {
+ if (foreign > 256)
+ throw new RuntimeDatabaseException("Internal error, contact application support." +
+ "Too big foreing index=" + foreign + " max=256");
+ --foreign;
+ bytes[byteIndex++] = foreign;
+ return ClusterEnum.ForeignShort;
+ } else {
+ byte position = (byte) (foreignTable.size() + 1);
+ if(lookup != null)
+ lookup[lookIndex] = position;
+ foreignTable.put(key, position);
+ if (DEBUG_STAT)
+ info.sForeign = foreignTable.size();
+ if (clusterUID.equals(ClusterUID.Null))
+ throw new RuntimeDatabaseException("Internal error, contact application support." +
+ "Cluster unique id not defined for foreing cluster.");
+ addForeignLong(resourceIndex, clusterUID);
+ return ClusterEnum.ForeignLong;
+ }
+ }
+
+ private final void addByte(byte b) {
+ bytes[byteIndex++] = b;
+ }
+
+ private final void addShort(short s) {
+ bytes[byteIndex++] = (byte)(s & 0xFF);
+ bytes[byteIndex++] = (byte)(s >>> 8);
+ }
+
+// private final void addShort(int s) {
+// bytes[byteIndex++] = (byte) (s & 0xFF);
+// bytes[byteIndex++] = (byte) ((s >>> 8) & 0xFF);
+// }
+
+ private final void addInt(int i) {
+// System.err.println("addInt " + i + " " + i);
+ bytes[byteIndex++] = (byte) (i & 0xFF);
+ bytes[byteIndex++] = (byte) ((i >>> 8) & 0xFF);
+ bytes[byteIndex++] = (byte) ((i >>> 16) & 0xFF);
+ bytes[byteIndex++] = (byte) ((i >>> 24) & 0xFF);
+ // buffer.asIntBuffer().put(i);
+ // buffer.position(buffer.position()+4);
+ }
+
+// private void addLong6(long l) {
+//// System.err.println("addLong " + l);
+// bytes[byteIndex++] = (byte) (l & 0xFF);
+// bytes[byteIndex++] = (byte) ((l >>> 8) & 0xFF);
+// bytes[byteIndex++] = (byte) ((l >>> 16) & 0xFF);
+// bytes[byteIndex++] = (byte) ((l >>> 24) & 0xFF);
+// bytes[byteIndex++] = (byte) ((l >>> 32) & 0xFF);
+// bytes[byteIndex++] = (byte) ((l >>> 40) & 0xFF);
+// // buffer.asLongBuffer().put(l);
+// // buffer.position(buffer.position() + 6);
+// }
+
+ private void addLong7(long l) {
+ bytes[byteIndex++] = (byte) (l & 0xFF);
+ bytes[byteIndex++] = (byte) ((l >>> 8) & 0xFF);
+ bytes[byteIndex++] = (byte) ((l >>> 16) & 0xFF);
+ bytes[byteIndex++] = (byte) ((l >>> 24) & 0xFF);
+ bytes[byteIndex++] = (byte) ((l >>> 32) & 0xFF);
+ bytes[byteIndex++] = (byte) ((l >>> 40) & 0xFF);
+ bytes[byteIndex++] = (byte) ((l >>> 48) & 0xFF);
+ // buffer.asLongBuffer().put(l);
+ // buffer.position(buffer.position() + 7);
+ }
+
+// private void addLong(long l) {
+// bytes[byteIndex++] = (byte) (l & 0xFF);
+// bytes[byteIndex++] = (byte) ((l >>> 8) & 0xFF);
+// bytes[byteIndex++] = (byte) ((l >>> 16) & 0xFF);
+// bytes[byteIndex++] = (byte) ((l >>> 24) & 0xFF);
+// bytes[byteIndex++] = (byte) ((l >>> 32) & 0xFF);
+// bytes[byteIndex++] = (byte) ((l >>> 40) & 0xFF);
+// bytes[byteIndex++] = (byte) ((l >>> 48) & 0xFF);
+// bytes[byteIndex++] = (byte) ((l >>> 56) & 0xFF);
+// }
+ private final byte bufferPop() {
+ return bytes[--byteIndex];
+ }
+
+ final class DebugStm {
+ StmEnum e;
+ int r;
+ int p;
+ int o;
+ ClusterUID pc;
+ ClusterUID oc;
+
+ DebugStm(StmEnum e, int r, int p, ClusterUID pc, int o, ClusterUID oc) {
+ this.e = e;
+ this.r = r;
+ this.p = p;
+ this.o = o;
+ this.pc = pc;
+ this.oc = oc;
+ }
+
+ @Override
+ public String toString() {
+ short ri = ClusterTraitsBase.getResourceIndexFromResourceKeyNoThrow(r);
+ short pi = ClusterTraitsBase.getResourceIndexFromResourceKeyNoThrow(p);
+ short oi = ClusterTraitsBase.getResourceIndexFromResourceKeyNoThrow(o);
+ return "" + e + " rk=" + r + " ri=" + ri + " rc=" + clusterUID
+ + " pk=" + p + " pi=" + pi + " pc=" + pc
+ + " ok=" + o + " oi=" + oi + " oc=" + oc;
+ }
+
+ public String toString2() {
+ return "" + e + " r=" + r + " rc=" + clusterUID + " p=" + p
+ + " pc=" + pc + " o=" + o + " oc=" + oc;
+ }
+
+ public String toString3() {
+ short ri = ClusterTraitsBase.getResourceIndexFromResourceKeyNoThrow(r);
+ short pi = ClusterTraitsBase.getResourceIndexFromResourceKeyNoThrow(p);
+ short oi = ClusterTraitsBase.getResourceIndexFromResourceKeyNoThrow(o);
+ return "" + e + " ri=" + ri
+ + " pi=" + pi + " pc=" + pc
+ + " oi=" + oi + " oc=" + oc;
+ }
+ }
+
+ private List<DebugStm> debugStms = new ArrayList<DebugStm>();
+
+ @SuppressWarnings("unused")
+ private final void addStm(Change c, StmEnum stmEnum) {
+
+ if (DEBUG_STAT)
+ ++info.nStms;
+ if (DEBUG || DEBUG_CCS) {
+ DebugStm d = new DebugStm(stmEnum, c.key0, c.key1, c.clusterUID1, c.key2, c.clusterUID2);
+ if (DEBUG_CCS)
+ debugStms.add(d);
+ if (DEBUG) {
+ printlnd(d.toString3() + " offset=" + byteIndex);
+ }
+ }
+ // int opPos = buffer.position();
+ int opPos = byteIndex++;
+ // buffer.put((byte)0); // operation code
+ // addByte((byte)0);
+
+ boolean done = true;
+
+ ClusterEnum a = addIndexAndCluster(c.key1, c.clusterUID1, c.lookIndex1, c.lookup1);
+ byte ab = 0;
+
+ // ForeignShort = byte
+ // Local = short
+ // ForeignLong = 8 byte
+ if (a != ClusterEnum.ForeignShort) {
+ ab = bufferPop();
+ done = false;
+ }
+
+ ClusterEnum b = addIndexAndCluster(c.key2, c.clusterUID2, c.lookIndex2, c.lookup2);
+ byte bb = 0;
+ if (b != ClusterEnum.ForeignShort) {
+ bb = bufferPop();
+ done = false;
+ }
+
+ int ri = ClusterTraitsBase.getResourceIndexFromResourceKeyNoThrow(c.key0);
+ if (ClusterTraitsSmall.isIllegalResourceIndex(ri))
+ throw new RuntimeDatabaseException("Assertion error. Illegal resource index=" + ri);
+ bytes[byteIndex++] = (byte)ri; // index low byte
+ if(!done) {
+ Data data = ClusterEnum.getData(stmEnum, a, b);
+ int left = 6 - data.bits;
+ int op = ri >>> (8 + left);
+ ri >>>= 8;
+ ri &= (1 << left) - 1;
+ if (a != ClusterEnum.ForeignShort) {
+ ri |= ab << left;
+ left += 6;
+ }
+ if (b != ClusterEnum.ForeignShort) {
+ ri |= bb << left;
+ left += 6;
+ }
+ switch (data.bytes) {
+ default:
+ throw new RuntimeDatabaseException("Assertion error. Illegal number of bytes=" + data.bytes);
+ case 2:
+ bytes[byteIndex++] = (byte)(ri & 0xFF);
+ bytes[byteIndex++] = (byte)((ri >>> 8) & 0xFF);
+ break;
+ case 1:
+ bytes[byteIndex++] = (byte)(ri & 0xFF);
+ break;
+ case 0:
+ break;
+ }
+ op |= data.mask;
+ this.bytes[opPos] = (byte)op;
+ } else {
+ if (stmEnum == StmEnum.Add)
+ bytes[opPos] = (byte)((ri >>> 8) + 64);
+ else
+ bytes[opPos] = (byte)((ri >>> 8) + 128);
+ }
+ if (DEBUG_STAT) {
+ if (a == ClusterEnum.Local && b == ClusterEnum.Local) {
+ ++info.nLocal;
+ } else if (a == ClusterEnum.Local || b == ClusterEnum.Local) {
+ ++info.nPartly;
+ } else {
+ ++info.nForeign;
+ }
+ }
+ if (foreignTable.size() > 252)
+ flush();
+// throw new UnsupportedOperationException();
+ //flushInternal(graphSession, clusterUID);
+ }
+
+ private final int modiValue(int ri, long value_offset, byte[] bytes, int offset, int size) {
+ if (DEBUG)
+ printlnd("Modify value ri=" + ri + " vo=" + value_offset + " size=" + size + " total=" + bytes.length);
+ if (ClusterTraitsBase.isIllegalResourceIndex(ri))
+ throw new RuntimeDatabaseException("Assertion error. Illegal resource index=" + ri);
+ if (value_offset > (1L << 58 - 1))
+ throw new RuntimeDatabaseException("Illegal value offset="
+ + value_offset);
+ if (size < 0 || size > MAX_FIXED_BYTES - 1)
+ throw new RuntimeDatabaseException("Illegal value size=" + size);
+ if (offset + size > bytes.length)
+ throw new RuntimeDatabaseException("Illegal value size=" + size);
+ checkBufferSpace(12 + size);
+ addByte(OpEnum.Modify.getOrMask());
+ ri |= (value_offset >>> 56) << 14; // add top two bits
+ addShort((short) ri);
+ value_offset &= (1L << 56) - 1;
+ addLong7(value_offset);
+ addShort((short) size);
+ if (DEBUG)
+ System.out.println("Modify value fixed part end offset=" + byteIndex);
+ int copied = Math.min(size, this.bytes.length - byteIndex);
+ System.arraycopy(bytes, offset, this.bytes, byteIndex, copied);
+ byteIndex += size;
+ return copied;
+ }
+
+// private final void modiValueBig(int ri, long voffset, int left, byte[] bytes, int offset) {
+// checkBufferSpace(0);
+// int current = Math.min(this.bytes.length - byteIndex - 12, left);
+// if(current >= 0) {
+// int written = modiValue(ri, voffset, bytes, offset, current);
+// voffset += written;
+// offset += written;
+// left -= written;
+// }
+//// flushInternal(graphSession, clusterUID);
+// while (left > 0) {
+// int length = Math.min(left, (1 << 16) - 1);
+// if (DEBUG)
+// printlnd("Modify big value ri=" + ri + " vo=" + voffset + " len=" + length);
+// int psize = length + 12;
+//// setHeaderVectorSize(psize);
+// byte[] message = new byte[psize/*+HEADER_SIZE*/];
+//// System.arraycopy(header, 0, message, 0, HEADER_SIZE);
+// int to = 0;
+// Bytes.write(message, to++, OpEnum.Modify.getOrMask());
+// short index = (short)(ri | (voffset >>> 56)<<14); // add top two bits
+// Bytes.writeLE(message, to, index); to += 2;
+// Bytes.writeLE7(message, to, voffset & ((1L << 56) - 1)); to += 7;
+// Bytes.writeLE(message, to, (short)length); to += 2;
+// System.arraycopy(bytes, offset, message, to, length);
+//// graphSession.updateCluster(new UpdateClusterFunction(message));
+// voffset += length;
+// offset += length;
+// left -= length;
+// }
+// }
+
+ private final int setValueBig(int ri, byte[] bytes, int length_) {
+ checkBufferSpace(12);
+ int sum = 0;
+ int voffset = 0;
+ int offset = 0;
+ int left = length_;
+ while (left > 0) {
+ int length = Math.min(left, MAX_FIXED_BYTES - 12 - byteIndex);
+ if (DEBUG)
+ printlnd("Set big value ri=" + ri + " vo=" + voffset + " len=" + length);
+ int written = modiValue(ri, voffset, bytes, offset, length);
+ sum += written;
+ voffset += written;
+ offset += written;
+ left -= written;
+ checkBufferSpace(12);
+ }
+ return sum;
+ }
+
+ private final int setValueSmall(int ri, byte[] bytes, int length) {
+ checkBufferSpace(5 + length);
+ int pos = byteIndex;
+ int i = length << 14 | ri;
+ if (length < 32) {
+ byte op = (byte) (OpEnum.SetShort.getOrMask() | length >>> 2);
+ addByte(op);
+ short s = (short) i;
+ addShort(s);
+ } else {
+ addByte(OpEnum.Set.getOrMask());
+ addInt(i);
+ }
+ System.arraycopy(bytes, 0, this.bytes, byteIndex, length);
+ byteIndex += length;
+ int len = byteIndex - pos;
+ return len;
+ }
+
+ final void setValue(short index, byte[] bytes) {
+ setValue(index, bytes, bytes.length);
+ }
+
+ final public void setValue(short index, byte[] bytes, int length) {
+ checkInitialization();
+ if (ClusterTraitsBase.isIllegalResourceIndex(index))
+ throw new RuntimeDatabaseException("Assertion error. Illegal resource index=" + index);
+ if (DEBUG)
+ printlnd("Set value ri=" + index
+ + " len=" + length
+ + " bytes=" + Arrays.toString(Arrays.copyOfRange(bytes, 0, Math.min(10, length))));
+ int len;
+ /*
+ * The limit for the cluster stream is (1<18)-1 but this avoids the
+ * conversion to big cluster.
+ */
+ if (length > ClusterTraitsSmall.VALUE_SIZE_MAX)
+ len = setValueBig(index, bytes, length);
+ else
+ len = setValueSmall(index, bytes, length);
+ if (DEBUG_STAT) {
+ ++info.nValues;
+ info.sValues += len + length;
+ }
+ }
+
+// final void setValue(Change c, byte[] bytes, int length) {
+// short ri = ClusterTraitsBase.getResourceIndexFromResourceKeyNoThrow(c.key0);
+// setValue(ri, bytes, length);
+// c.initValue();
+// }
+
+// final void modiValue(Change c, long voffset, int length, byte[] bytes, int offset) {
+// checkInitialization();
+// int ri = ClusterTraitsBase.getResourceIndexFromResourceKeyNoThrow(c.key0);
+// if (DEBUG)
+// printlnd("Modify value ri=" + ri
+// + " voff=" + voffset
+// + " vlen=" + length
+// + " blen=" + bytes.length
+// + " boff=" + offset
+// + " bytes=" + Arrays.toString(Arrays.copyOfRange(bytes, 0, Math.min(10, bytes.length))));
+// modiValueBig(ri, voffset, length, bytes, offset);
+// c.init();
+// if (DEBUG_STAT) {
+// ++info.nValues;
+// info.sValues += length;
+// }
+// }
+ final void setImmutable(boolean immutable) {
+ checkInitialization();
+// clusterChange2.setImmutable(immutable);
+ }
+ final void undoValueEx(int resourceIndex) {
+ checkInitialization();
+// clusterChange2.undoValueEx(resourceIndex);
+ }
+ final void setDeleted(boolean deleted) {
+ checkInitialization();
+// clusterChange2.setDeleted(deleted);
+ }
+ final void corrupt() {
+ checkInitialization();
+ addByte((byte)0);
+ }
+
+ public byte[] getBytes() {
+ byte[] copy = new byte[byteIndex];
+ System.arraycopy(bytes, 0, copy, 0, byteIndex);
+ return copy;
+ }
+
+ /**
+ * @param graphSession
+ * @param clusterId
+ * @return true if actually flushed something
+ */
+ final boolean flush(/*GraphSession graphSession,*/ ClusterUID clusterUID) {
+ throw new UnsupportedOperationException();
+// if (byteIndex > 0) {
+// if(DebugPolicy.REPORT_CLUSTER_STREAM)
+// System.err.println("Flush cluster change set stream " + this);
+// setHeaderVectorSize(byteIndex);
+// byte[] copy = new byte[byteIndex + HEADER_SIZE];
+// System.arraycopy(header, 0, copy, 0, HEADER_SIZE);
+// System.arraycopy(bytes, 0, copy, HEADER_SIZE, byteIndex);
+// UpdateClusterFunction updateClusterFunction = new UpdateClusterFunction(copy);
+// if (DEBUG_CCS) {
+// for (DebugStm stm : debugStms)
+// printlnd(stm.toString2());
+// debugStms.clear();
+// }
+// if (DEBUG_STAT) {
+// info.tot = updateClusterFunction.operation.length;
+// printlnd("ReallyFlush: " + info.toString());
+// sum.add(info);
+// printlnd("ReallyFlush sum: " + sum.toString());
+// }
+// // long start = System.nanoTime();
+// graphSession.updateCluster(updateClusterFunction);
+// // long duration = System.nanoTime() - start;
+// // duration2 += duration;
+// // System.err.println("updateCluster " + 1e-9*duration);
+// // System.err.println("updateCluster total " + 1e-9*duration2);
+// clear();
+// clusterChange2.flush(graphSession);
+// return true;
+// } else if (clusterChange2.isDirty()) {
+// clusterChange2.flush(graphSession);
+// clear();
+// return true;
+// } else if (flushed) {
+// flushed = false;
+// return true;
+// } else {
+// return true;
+// }
+ }
+
+ final void flushInternal(ClusterUID clusterUID) {
+ throw new UnsupportedOperationException();
+// flush(graphSession, clusterUID);
+// flushed = true;
+ }
+
+ final class ForeignTable {
+ private final TLongIntHashMap table = new TLongIntHashMap();
+
+ private long createKey(short index, long cluster) {
+ assert (cluster <= (1L << 48) - 1);
+ return (cluster << 14) | index;
+ }
+
+ public int get(short index, long cluster) {
+ int value = table.get(createKey(index, cluster));
+ if (DEBUG)
+ printlnd("ForeignTable get c=" + clusterUID + " i="
+ + (value - 1) + " r=" + index + " rc=" + cluster);
+ return value;
+ }
+
+ public int put(short index, long cluster, int value) {
+ if (DEBUG)
+ printlnd("ForeignTable put c=" + clusterUID + " i="
+ + (value - 1) + " r=" + index + " rc=" + cluster);
+ return table.put(createKey(index, cluster), value);
+ }
+
+ public int size() {
+ return table.size();
+ }
+
+ public void clear() {
+ table.clear();
+ }
+ }
+
+ @Override
+ public int hashCode() {
+ return 31*clusterUID.hashCode();
+ }
+
+ @Override
+ public boolean equals(Object object) {
+ if (this == object)
+ return true;
+ else if (object == null)
+ return false;
+ else if (!(object instanceof ClusterChange))
+ return false;
+ ClusterChange r = (ClusterChange)object;
+ return r.clusterUID.equals(clusterUID);
+ }
+
+ public void flush() {
+
+ if(byteIndex > 0) {
+
+ final ClusterUID cuid = clusterUID;
+
+ byte[] block = getBytes();
+ byte[] raw = new byte[block.length + 28];
+ Bytes.writeLE(raw, 0, 1);
+ System.arraycopy(cuid.asBytes(), 0, raw, 4, 16);
+ Bytes.writeLE(raw, 20, block.length);
+ System.arraycopy(block, 0, raw, 24, block.length);
+ Bytes.writeLE(raw, 24+block.length, 0);
+
+ ByteBuffer rawBB = ByteBuffer.wrap(raw);
+ ByteBuffer outputBB = ByteBuffer.allocate(raw.length + raw.length/8);
+ //outputBB.order(ByteOrder.LITTLE_ENDIAN);
+ int compressedSize = Compressions.get(Compressions.LZ4).compressBuffer(rawBB, 0, raw.length, outputBB, 0);
+
+ byte[] data_ = null;
+ if(compressedSize < raw.length) {
+ data_ = new byte[compressedSize];
+ outputBB.get(data_,0,compressedSize);
+ } else {
+ data_ = raw;
+ }
+
+ byte[] data = new byte[data_.length+24];
+ Bytes.writeLE(data, 0, 0);
+ Bytes.writeLE(data, 4, 0);
+ Bytes.writeLE(data, 8, raw.length);
+ Bytes.writeLE(data, 12, raw.length);
+ Bytes.writeLE(data, 16, data_.length);
+ System.arraycopy(data_, 0, data, 20, data_.length);
+ Bytes.writeLE(data, 20+data_.length, 0);
+
+ stream.add(Pair.make(clusterUID, data));
+ clear();
+ initBuffer();
+
+ }
+ }
+
+}
Code Review / simantics / platform.git / blobdiff