bundles/org.simantics.db.procore/src/fi/vtt/simantics/procore/internal/GraphSession.java

   1 /*******************************************************************************\r
   2  * Copyright (c) 2007, 2010 Association for Decentralized Information Management\r
   3  * in Industry THTH ry.\r
   4  * All rights reserved. This program and the accompanying materials\r
   5  * are made available under the terms of the Eclipse Public License v1.0\r
   6  * which accompanies this distribution, and is available at\r
   7  * http://www.eclipse.org/legal/epl-v10.html\r
   8  *\r
   9  * Contributors:\r
  10  *     VTT Technical Research Centre of Finland - initial API and implementation\r
  11  *******************************************************************************/\r
  12 package fi.vtt.simantics.procore.internal;\r
  13 \r
  14 import java.io.ByteArrayInputStream;\r
  15 import java.io.IOException;\r
  16 import java.io.InputStream;\r
  17 import java.lang.ref.SoftReference;\r
  18 import java.util.ArrayList;\r
  19 import java.util.Collection;\r
  20 import java.util.HashSet;\r
  21 import java.util.Iterator;\r
  22 import java.util.Map;\r
  23 import java.util.TreeMap;\r
  24 import java.util.Vector;\r
  25 import java.util.concurrent.CopyOnWriteArrayList;\r
  26 \r
  27 import org.simantics.databoard.Bindings;\r
  28 import org.simantics.databoard.binding.impl.TreeMapBinding;\r
  29 import org.simantics.databoard.serialization.Serializer;\r
  30 import org.simantics.db.ChangeSetIdentifier;\r
  31 import org.simantics.db.Database;\r
  32 import org.simantics.db.Database.Session.ChangeSetData;\r
  33 import org.simantics.db.Database.Session.ChangeSetIds;\r
  34 import org.simantics.db.Database.Session.ClusterChanges;\r
  35 import org.simantics.db.Database.Session.ClusterIds;\r
  36 import org.simantics.db.Database.Session.Refresh;\r
  37 import org.simantics.db.Operation;\r
  38 import org.simantics.db.SessionReference;\r
  39 import org.simantics.db.common.UndoContextEx;\r
  40 import org.simantics.db.common.utils.Logger;\r
  41 import org.simantics.db.exception.DatabaseException;\r
  42 import org.simantics.db.exception.InternalException;\r
  43 import org.simantics.db.impl.ClusterI;\r
  44 import org.simantics.db.impl.ResourceImpl;\r
  45 import org.simantics.db.service.ClusterUID;\r
  46 import org.simantics.db.service.ExternalOperation;\r
  47 import org.simantics.db.service.ManagementSupport.ChangeSetListener;\r
  48 import org.simantics.scl.runtime.function.Function1;\r
  49 import org.simantics.scl.runtime.function.FunctionImpl1;\r
  50 import org.simantics.utils.threads.ThreadUtils;\r
  51 \r
  52 import gnu.trove.iterator.TLongObjectIterator;\r
  53 import gnu.trove.map.hash.THashMap;\r
  54 import gnu.trove.map.hash.TLongObjectHashMap;\r
  55 \r
  56 final class ChangeSetIdentifierImpl implements ChangeSetIdentifier {\r
  57     private final long id;\r
  58     private final GraphSession graphSession;\r
  59     private Map<String, byte[]> metadata;\r
  60     ChangeSetIdentifierImpl(long id, GraphSession graphSession) {\r
  61         this.id = id;\r
  62         this.graphSession = graphSession;\r
  63         this.metadata = null;\r
  64     }\r
  65     public final long getId() {\r
  66         return id;\r
  67     }\r
  68     public final void setMetadata(Map<String, byte[]> metadata) {\r
  69         this.metadata = metadata;\r
  70     }\r
  71     public final byte[] getContext()\r
  72     throws DatabaseException {\r
  73         if (null == graphSession)\r
  74             return new byte[0];\r
  75         return graphSession.getChangeSetContext(id);\r
  76     }\r
  77 \r
  78     static Serializer METADATA_SERIALIZER =\r
  79         Bindings.getSerializerUnchecked( new TreeMapBinding(Bindings.STRING,\r
  80             Bindings.BYTE_ARRAY));\r
  81 \r
  82     @SuppressWarnings("unchecked")\r
  83     public Map<String, byte[]> getMetadata() {\r
  84         try {\r
  85             if (metadata == null) {\r
  86                 byte[] data = getContext();\r
  87                 if (null != data && data.length > 0) {\r
  88                     metadata = (TreeMap<String, byte[]>) METADATA_SERIALIZER.deserialize(data);\r
  89                 }\r
  90             }\r
  91         } catch (DatabaseException e) {\r
  92             Logger.defaultLogError(e);\r
  93             e.printStackTrace();\r
  94         } catch (IOException e) {\r
  95             Logger.defaultLogError(e);\r
  96             e.printStackTrace();\r
  97         }\r
  98         return metadata;\r
  99     }\r
 100 \r
 101     static long[] changeSetIds2ints(Collection<ChangeSetIdentifier> changeSetIds) {\r
 102         if (null == changeSetIds)\r
 103             return new long[0];\r
 104         long[] t = new long[changeSetIds.size()];\r
 105         int i = -1;\r
 106         for (ChangeSetIdentifier id : changeSetIds)\r
 107             t[++i] = ((ChangeSetIdentifierImpl) id).getId();\r
 108         return t;\r
 109     }\r
 110     static long[] operations2ints(Collection<Operation> ops) {\r
 111         if (null == ops)\r
 112             return new long[0];\r
 113         long[] t = new long[ops.size()];\r
 114         int i = -1;\r
 115         for (Operation o : ops)\r
 116             t[++i] = o.getCSId();\r
 117         return t;\r
 118     }\r
 119     static Collection<ChangeSetIdentifier> longs2changeSetIds(GraphSession gs, long firstChangeSetIds, int count) {\r
 120         ArrayList<ChangeSetIdentifier> t = new ArrayList<ChangeSetIdentifier>();\r
 121         t.ensureCapacity(count);\r
 122         for (int i=0; i<count; ++i)\r
 123             t.add(new ChangeSetIdentifierImpl(firstChangeSetIds + i, gs));\r
 124         return t;\r
 125     }\r
 126     static Collection<ChangeSetIdentifier> longs2changeSetIds(GraphSession gs, long min, long max) {\r
 127         ArrayList<ChangeSetIdentifier> t = new ArrayList<ChangeSetIdentifier>();\r
 128         if (max<min)\r
 129             return t;\r
 130         long length = max - min + 1;\r
 131         t.ensureCapacity((int)length);\r
 132         for (int i=0; i<length; ++i) {\r
 133             t.add(new ChangeSetIdentifierImpl(min+i, gs));\r
 134         }\r
 135         return t;\r
 136     }\r
 137 }\r
 138 \r
 139 public abstract class GraphSession {\r
 140     final protected boolean DEBUG = false;\r
 141     final protected boolean VERBOSE = false;\r
 142     protected Listener listener;\r
 143     protected SessionImplSocket session;\r
 144     protected final Database.Session dbSession;\r
 145     private SessionReference sessionReference;\r
 146     private TLongObjectHashMap<ClusterI> clusterMap = new TLongObjectHashMap<ClusterI>();\r
 147     protected THashMap<String, BuiltinData> builtinMap = null;\r
 148     private long firstChangeSetId = 0;\r
 149     protected SynchronizeContextI synchronizeContext;\r
 150     \r
 151     final UndoContextEx undoContext = new UndoContextEx("GraphSession");\r
 152     final CopyOnWriteArrayList<ChangeSetListener> changeSetListeners = new CopyOnWriteArrayList<ChangeSetListener>();\r
 153     private long lastChangeSetId = 0;\r
 154     protected MetadataCache metadataCache = new MetadataCache();\r
 155     \r
 156     public GraphSession(SessionImplSocket sessionImpl, SessionReference sessionReference, Database.Session dbSession) {\r
 157         this.dbSession = dbSession;\r
 158         if (null == dbSession)\r
 159             throw new RuntimeException("Failed to initialize GraphSession. Database.Session can not be null.");\r
 160 //        if (!(sessionReference instanceof ProCoreSessionReference))\r
 161 //            throw new RuntimeException("Failed to initialize GraphSession. SessionReference must be instance of ProCoreSessionReference.");\r
 162         this.sessionReference = sessionReference;\r
 163         this.session = sessionImpl;\r
 164     }\r
 165     void addChangeSetListener(ChangeSetListener csl) {\r
 166         changeSetListeners.add(csl);\r
 167     }\r
 168     void removeChangeSetListener(ChangeSetListener csl) {\r
 169         changeSetListeners.remove(csl);\r
 170     }\r
 171     SessionReference getSessionReference() {\r
 172         return sessionReference;\r
 173     }\r
 174 \r
 175     public void open() throws InternalException {\r
 176         dbSession.open();\r
 177     }\r
 178 \r
 179     public void close() throws InternalException {\r
 180         dbSession.close();\r
 181     }\r
 182 \r
 183     public long getSessionId() {\r
 184         return sessionReference.getSessionId();\r
 185     }\r
 186 //\r
 187 //    public String getHost() {\r
 188 //        return sessionReference.serverReference.socketAddress.getHostName();\r
 189 //    }\r
 190 //\r
 191 //    public int getPort() {\r
 192 //        return sessionReference.serverReference.socketAddress.getPort();\r
 193 //    }\r
 194 \r
 195     long getFirstChangeSetId() {\r
 196         return firstChangeSetId;\r
 197     }\r
 198 \r
 199     String getClusterFileName(long clusterId) {\r
 200         String fileName = "cluster" + clusterId + ".dat";\r
 201         return fileName;\r
 202     }\r
 203 \r
 204     int computeClusterMemoryUse() {\r
 205 \r
 206         int size = 0;\r
 207         TLongObjectIterator<ClusterI> i = clusterMap.iterator();\r
 208         while (i.hasNext()) {\r
 209             i.advance();\r
 210             ClusterI c = i.value();\r
 211             try {\r
 212                 size += c.getUsedSpace();\r
 213             } catch (DatabaseException e) {\r
 214                 Logger.defaultLogInfo("GetUsedSpace faile for cluster " + c);\r
 215             }\r
 216         }\r
 217 \r
 218         return size;\r
 219 \r
 220     }\r
 221 \r
 222     void printClusterSize() {\r
 223 \r
 224         long size = 0;\r
 225 \r
 226         long amount = 0;\r
 227         TLongObjectIterator<ClusterI> i = clusterMap.iterator();\r
 228         while (i.hasNext()) {\r
 229             i.advance();\r
 230             ClusterI c = i.value();\r
 231             if (c.isLoaded()) {\r
 232                 try {\r
 233                     size += c.getUsedSpace();\r
 234                 } catch (DatabaseException e) {\r
 235                     Logger.defaultLogInfo("GetUsedSpace faile for cluster " + c);\r
 236                 }\r
 237                 amount++;\r
 238             }\r
 239         }\r
 240 \r
 241         if (amount > 50)\r
 242             loadedClusters.clear();\r
 243         System.out.println("" + size / (1024 * 1024) + "M in " + amount + " clusters.");\r
 244 \r
 245     }\r
 246 \r
 247     private HashSet<Long> loadedClusters = new HashSet<Long>();\r
 248 \r
 249     ClusterI freeClusterFromCacheAndFreeClusterMemory(long clusterId) {\r
 250         ClusterI c = clusterMap.remove(clusterId);\r
 251         if (null != c)\r
 252             c.releaseMemory();\r
 253         return c;\r
 254     }\r
 255 \r
 256     private void debugPrint(UpdateClusterFunction f) {\r
 257         for (int i = 0; i < f.operation.length; ++i) {\r
 258             System.out.println("op=" + f.operation[i]);\r
 259         }\r
 260     }\r
 261 \r
 262     long total = 0;\r
 263 \r
 264     public boolean updateCluster(UpdateClusterFunction f) {\r
 265         session.clusterStream.markDirty();\r
 266         assert (null != f);\r
 267         try {\r
 268             if (DEBUG) {\r
 269                 System.out.println("update cluster");\r
 270                 if (VERBOSE)\r
 271                     debugPrint(f);\r
 272             }\r
 273 // long start = System.nanoTime();\r
 274             dbSession.updateCluster(f.operation);\r
 275 // long duration = System.nanoTime() - start;\r
 276 // total += duration;\r
 277 // System.err.println("graphClient.call = " + 1e-9*duration + " " + 1e-9*total);\r
 278 // System.out.println("GraphSession.updateCluster done.");\r
 279             return true; // ok\r
 280         } catch (Throwable e) {\r
 281             e.printStackTrace();\r
 282             return false;\r
 283         }\r
 284     }\r
 285     public byte[] getChangeSetContext(long id)\r
 286     throws DatabaseException {\r
 287         byte[] data = metadataCache.get(id);\r
 288         if (null != data)\r
 289             return data;\r
 290         if (id < getServerInformation().firstChangeSetId) {\r
 291             Logger.defaultLogInfo("Asking purged change set metadata. uid=" + id);\r
 292             return new byte[0];\r
 293         }\r
 294         return dbSession.getChangeSetMetadata(id);\r
 295     }\r
 296 \r
 297     public StatementImpl getStatement(ResourceImpl s, ResourceImpl p, ResourceImpl o) {\r
 298         return new StatementImplOld(s, p, o);\r
 299     }\r
 300     public long newClusterId() throws DatabaseException {\r
 301         long id = dbSession.reserveIds(1);\r
 302         if (id <= ClusterUID.Builtin.second)\r
 303             return newClusterId();\r
 304         return id;\r
 305     }\r
 306 \r
 307     long load    = 0;\r
 308 \r
 309     long inflate = 0;\r
 310 \r
 311 //    public ClusterI getClusterImpl(ClusterUID clusterUID, int clusterKey)\r
 312 //    throws DatabaseException {\r
 313 //        assert(ClusterUID.isLegal(clusterUID));\r
 314 //\r
 315 //              if(Development.DEVELOPMENT) {\r
 316 //            if(Development.isTrue(DevelopmentKeys.READGRAPH_COUNT)) {\r
 317 //              ReadGraphImpl.counters.adjustOrPutValue("load " + clusterUID + " " + clusterKey, 1, 1);\r
 318 //            }\r
 319 //              }\r
 320 //\r
 321 //// ITask task = ThreadLogger.getInstance().begin("Load cluster " + clusterId);\r
 322 //        if (DEBUG)\r
 323 //            System.out.println("DEBUG: loading cluster=" + clusterUID);\r
 324 //// System.out.println("getClusterImpl " + clusterId);\r
 325 //        long start = System.nanoTime();\r
 326 //        Cluster t = dbSession.getCluster(clusterUID.asBytes());\r
 327 //        long duration = System.nanoTime() - start;\r
 328 //        load += duration;\r
 329 //        //System.out.println("loaded in " + 1e-9*duration + " total " + 1e-9*load);\r
 330 //// if (Constants.NullClusterId == t.clusterId) {\r
 331 //// throw new DatabaseException("Cluster " + t.clusterId + " not in server.");\r
 332 //// }\r
 333 //\r
 334 //        ByteBuffer deflated = t.getDeflated();\r
 335 //        start = System.nanoTime();\r
 336 //        Object[] arrays = clusterDecompressor.inflateCluster(t.getInflateSize(), deflated);\r
 337 //        duration = System.nanoTime() - start;\r
 338 //        inflate += duration;\r
 339 //        //System.out.println("inflated in " + 1e-9*duration + " total " + 1e-9*inflate);\r
 340 //        long[] longs = (long[]) arrays[0];\r
 341 //        int[] ints = (int[]) arrays[1];\r
 342 //        byte[] bytes = (byte[]) arrays[2];\r
 343 //\r
 344 //// System.out.println("Got cluster " + clusterId + " from core.");\r
 345 //        ClusterImpl cluster = ClusterImpl.make(longs, ints, bytes, session.clusterTranslator, clusterKey);\r
 346 ////        System.out.println("cluster[ " + clusterId + "].size = " + cluster.getUsedSpace(-1));\r
 347 //// task.finish();\r
 348 //        return cluster;\r
 349 //    }\r
 350 \r
 351     ClusterI getCluster(long clusterId) {\r
 352         return clusterMap.get(clusterId);\r
 353     }\r
 354 \r
 355     ClusterI getClusterOrThrow(long clusterId) throws IllegalArgumentException {\r
 356         ClusterI c = this.getCluster(clusterId);\r
 357         if (c == null)\r
 358             throw new IllegalArgumentException("Cluster " + clusterId + " does not exist.");\r
 359         return c;\r
 360     }\r
 361 \r
 362     public String execute(String command) throws DatabaseException {\r
 363         return dbSession.execute(command);\r
 364     }\r
 365 \r
 366     public Collection<ChangeSetIdentifier> getChangeSets()\r
 367     throws DatabaseException {\r
 368         ChangeSetIds t = dbSession.getChangeSetIds();\r
 369         return ChangeSetIdentifierImpl.longs2changeSetIds(this, t.getFirstChangeSetId(), t.getCount());\r
 370     }\r
 371     public long getLastChangeSetId() {\r
 372         return lastChangeSetId;\r
 373     }\r
 374     public Collection<ChangeSetIdentifier> getChangeSets(long from, long to, long upperLimit)\r
 375     throws DatabaseException {\r
 376         if (from > to)\r
 377             throw new DatabaseException("Illegal argument: from=" + from + " > to=" + to + ".");\r
 378         long min = Math.max(from, 1);\r
 379         long max = Math.min(to, upperLimit);\r
 380         return ChangeSetIdentifierImpl.longs2changeSetIds(this, min, max);\r
 381     }\r
 382 \r
 383     public boolean getChangeSets(long minCS, long maxCS, SynchronizeContextI context)\r
 384     throws DatabaseException {\r
 385         try {\r
 386             synchronizeContext = context;\r
 387             ChangeSetData t = dbSession.getChangeSetData(minCS, maxCS, context);\r
 388             return !t.isOk();\r
 389         } catch (Throwable e) {\r
 390             if (DEBUG)\r
 391                 e.printStackTrace();\r
 392             if (e instanceof DatabaseException)\r
 393                 throw (DatabaseException) e;\r
 394             else\r
 395                 throw new DatabaseException("GetChangeSetData call to server failed.");\r
 396         } finally {\r
 397             synchronizeContext = null;\r
 398         }\r
 399     }\r
 400 \r
 401     public static void forExternals(Collection<Operation> operations, Function1<ExternalOperation, Boolean> fn) {\r
 402         for(Operation o : operations) {\r
 403                 for(ExternalOperation e : o.getExternalOperations()) {\r
 404                         if(!e.isDisposed()) fn.apply(e);\r
 405                 }\r
 406         }\r
 407     }\r
 408 \r
 409     public boolean undo(Collection<Operation> changeSetIds, SynchronizeContextI context)\r
 410     throws DatabaseException {\r
 411         try {\r
 412 \r
 413                 forExternals(changeSetIds, new FunctionImpl1<ExternalOperation, Boolean>() {\r
 414                         @Override\r
 415                         public Boolean apply(ExternalOperation op) {\r
 416                                 op.undo();\r
 417                                 return true;\r
 418                         }\r
 419                 });\r
 420 \r
 421             long[] cids = ChangeSetIdentifierImpl.operations2ints(changeSetIds);\r
 422             synchronizeContext = context;\r
 423             if (DEBUG)\r
 424                 for (long id : cids)\r
 425                     System.out.println("DEBUG: Undo cs=" + id);\r
 426             return dbSession.undo(cids, context);\r
 427         } catch (Throwable e) {\r
 428             if (DEBUG)\r
 429                 e.printStackTrace();\r
 430             if (e instanceof DatabaseException)\r
 431                 throw (DatabaseException) e;\r
 432             else\r
 433                 throw new DatabaseException("Undo call to server failed.", e);\r
 434         } finally {\r
 435             synchronizeContext = null;\r
 436         }\r
 437     }\r
 438 \r
 439     public ClusterUID[] getRefresh2(long csid) throws DatabaseException {\r
 440         Refresh t = dbSession.getRefresh(csid);\r
 441         if (DEBUG)\r
 442           System.err.println("getRefresh2 old cs=" + csid + " head cs=" + t.getHeadChangeSetId() + " chanaged clusters=" + t.getFirst().length);\r
 443         ClusterUID[] clusters = new ClusterUID[t.getFirst().length];\r
 444         for (int i=0; i<t.getFirst().length; ++i)\r
 445             clusters[i] = new ClusterUID(t.getFirst()[i], t.getSecond()[i]);\r
 446         return clusters;\r
 447     }\r
 448 \r
 449     public ClusterChanges getClusterChanges(ClusterUID cluster, long csid) throws DatabaseException {\r
 450         ClusterChanges t = dbSession.getClusterChanges(csid, cluster.asBytes());\r
 451         if (DEBUG)\r
 452             System.err.println("getClusterChanges old cs=" + csid + " head cs=" + t.getHeadChangeSetId() + " cluster=" + cluster + " statements=" + t.getPredicateIndex().length + " values=" + t.getValueIndex().length);\r
 453         return t;\r
 454     }\r
 455     public long getCluster(int id) {\r
 456         return session.getCluster(id);\r
 457     }\r
 458 \r
 459     private static class ResourceSegment {\r
 460         public long   valueSize;\r
 461 \r
 462         public byte[] bytes;\r
 463 \r
 464         ResourceSegment(long valueSize, byte[] bytes) {\r
 465             this.valueSize = valueSize;\r
 466             this.bytes = bytes;\r
 467         }\r
 468     }\r
 469     public ResourceSegment getResourceSegment(int resourceIndex, ClusterUID clusterUID, long offset, short size)\r
 470     throws DatabaseException {\r
 471         if (DEBUG)\r
 472             System.out.println("DEBUG: getResourceSegment ri=" + resourceIndex + " cid=" + clusterUID + " offset=" + offset + " size=" + size);\r
 473         org.simantics.db.Database.Session.ResourceSegment t = dbSession.getResourceSegment(clusterUID.asBytes(), resourceIndex, offset, size);\r
 474         return new ResourceSegment(t.getValueSize(), t.getSegment());\r
 475     }\r
 476 \r
 477     public byte[] getResourceValue(int resourceIndex, ClusterUID clusterUID)\r
 478     throws DatabaseException {\r
 479         return getResourceValue(resourceIndex, clusterUID, 0, 0);\r
 480     }\r
 481 \r
 482     public InputStream getResourceValueStream(final int resourceIndex, final ClusterUID clusterUID, long offset, int length) throws DatabaseException {\r
 483 \r
 484         if (DEBUG)\r
 485             System.out.println("DEBUG: getResourceFile ri=" + resourceIndex + " cid=" + clusterUID + " off=" + offset + " len=" + length);\r
 486         final int IMAX = 0xFFFF;\r
 487         short slen = (short)Math.min(length != 0 ? length : IMAX, IMAX);\r
 488         final ResourceSegment s = getResourceSegment(resourceIndex, clusterUID, offset, slen);\r
 489         if (s.valueSize < 0)\r
 490             throw new DatabaseException("Failed to get value for resource index=" + resourceIndex +\r
 491             " cluster=" + clusterUID + " off=" + offset + " len=" + length + " (1).");\r
 492         int ilen = (int)slen & 0xFFFF;\r
 493         assert(s.bytes.length <= ilen);\r
 494         if (0 == length) {\r
 495             if (s.valueSize > Integer.MAX_VALUE)\r
 496                 throw new DatabaseException("Failed to get value for resource index=" + resourceIndex +\r
 497                 " cluster=" + clusterUID + " off=" + offset + " len=" + length +\r
 498                 ". Value size=" + s.valueSize + " (2).");\r
 499             length = (int)s.valueSize;\r
 500         }\r
 501         long rSize = s.valueSize - offset;\r
 502         if (rSize < length)\r
 503             throw new DatabaseException("Failed to get value for resource index=" + resourceIndex +\r
 504             " cluster=" + clusterUID + " off=" + offset + " len=" + length +\r
 505             ". Value size=" + s.valueSize + " (3).");\r
 506         else if (length <= IMAX)\r
 507             return new ByteArrayInputStream(s.bytes);\r
 508 \r
 509         final int finalLength = length;\r
 510 \r
 511         return new InputStream() {\r
 512 \r
 513             int left = finalLength;\r
 514             long valueOffset = 0;\r
 515             int offset = 0;\r
 516             ResourceSegment _s = s;\r
 517 \r
 518             @Override\r
 519             public int read() throws IOException {\r
 520 \r
 521                 if(left <= 0) return -1;\r
 522 \r
 523                 if(offset == _s.bytes.length) {\r
 524                     short slen = (short)Math.min(left, IMAX);\r
 525                     valueOffset += _s.bytes.length;\r
 526                     try {\r
 527                         _s = getResourceSegment(resourceIndex, clusterUID, valueOffset, slen);\r
 528                     } catch (DatabaseException e) {\r
 529                         throw new IOException(e);\r
 530                     }\r
 531                     offset = 0;\r
 532                 }\r
 533 \r
 534                 left--;\r
 535                 int result = _s.bytes[offset++];\r
 536                 if(result < 0) result += 256;\r
 537                 return result;\r
 538 \r
 539             }\r
 540 \r
 541         };\r
 542     }\r
 543 \r
 544 \r
 545     public byte[] getResourceValue(int resourceIndex, ClusterUID clusterUID, long offset, int length)\r
 546     throws DatabaseException {\r
 547         if (DEBUG)\r
 548             System.out.println("DEBUG: getResourceFile ri=" + resourceIndex + " cid=" + clusterUID + " off=" + offset + " len=" + length);\r
 549         final int IMAX = 0xFFFF;\r
 550         short slen = (short)Math.min(length != 0 ? length : IMAX, IMAX);\r
 551         ResourceSegment s = getResourceSegment(resourceIndex, clusterUID, offset, slen);\r
 552         final long VALUE_SIZE = s.valueSize;\r
 553         if (s.valueSize < 0)\r
 554             throw new DatabaseException("Failed to get value for resource index=" + resourceIndex +\r
 555             " cluster=" + clusterUID + " off=" + offset + " len=" + length + " (1).");\r
 556         int ilen = (int)slen & 0xFFFF;\r
 557         assert(s.bytes.length <= ilen);\r
 558         if (0 == length) {\r
 559             if (s.valueSize > Integer.MAX_VALUE)\r
 560                 throw new DatabaseException("Failed to get value for resource index=" + resourceIndex +\r
 561                 " cluster=" + clusterUID + " off=" + offset + " len=" + length +\r
 562                 ". Value size=" + s.valueSize + " (2).");\r
 563             length = (int)s.valueSize;\r
 564         }\r
 565         long rSize = s.valueSize - offset;\r
 566         if (rSize < length)\r
 567             throw new DatabaseException("Failed to get value for resource index=" + resourceIndex +\r
 568             " cluster=" + clusterUID + " off=" + offset + " len=" + length +\r
 569             ". Value size=" + s.valueSize + " (3).");\r
 570         else if (length <= IMAX)\r
 571             return s.bytes;\r
 572         byte[] bytes = new byte[length];\r
 573         int left = (int)length - s.bytes.length;\r
 574         int cur = s.bytes.length;\r
 575         offset += s.bytes.length;\r
 576         System.arraycopy(s.bytes, 0, bytes, 0, cur);\r
 577         while (left > 0) {\r
 578             slen = (short)Math.min(left, IMAX);\r
 579             s = getResourceSegment(resourceIndex, clusterUID, offset, slen);\r
 580             ilen = (int)slen & 0xFFFF;\r
 581             if (s.valueSize != VALUE_SIZE || s.bytes.length != ilen)\r
 582                 throw new DatabaseException("Failed to get value for resource index=" + resourceIndex +\r
 583                 " cluster=" + clusterUID + " off=" + offset + " len=" + length +\r
 584                 ". Value size=" + s.valueSize + " blen=" + s.bytes.length + " (4).");\r
 585             System.arraycopy(s.bytes, 0, bytes, cur, s.bytes.length);\r
 586             left -= s.bytes.length;\r
 587             cur += s.bytes.length;\r
 588             offset += s.bytes.length;\r
 589         }\r
 590         return bytes;\r
 591     }\r
 592 \r
 593     public long getResourceValueSize(int resourceIndex, ClusterUID clusterUID)\r
 594     throws DatabaseException {\r
 595         if (DEBUG)\r
 596             System.out.println("DEBUG: getResourceFile ri=" + resourceIndex + " cid=" + clusterUID);\r
 597         final short SMAX = (short)0xFFFF;\r
 598         ResourceSegment s = getResourceSegment(resourceIndex, clusterUID, 0, SMAX);\r
 599         if (s.valueSize < 0)\r
 600             throw new DatabaseException("Failed to get value for resource index=" + resourceIndex +\r
 601             " cluster=" + clusterUID + " size=" + s.valueSize);\r
 602         return s.valueSize;\r
 603     }\r
 604     int wait4RequestsLess(int limit)\r
 605     throws DatabaseException {\r
 606         dbSession.execute("");\r
 607         return 0;\r
 608     }\r
 609     interface Listener {\r
 610         void onChangeSetId(int thread, long csid, boolean refresh);\r
 611     }\r
 612 \r
 613     void setListener(Listener listener) {\r
 614         this.listener = listener;\r
 615     }\r
 616 \r
 617     protected void updateLastChangeSetId(int thread, final long csid, boolean refresh) {\r
 618         if (null != listener)\r
 619             listener.onChangeSetId(thread, csid, refresh);\r
 620         if (csid > lastChangeSetId) {\r
 621             lastChangeSetId = csid;\r
 622             final Iterator<ChangeSetListener> it = changeSetListeners.iterator();\r
 623             ThreadUtils.getBlockingWorkExecutor().execute(new Runnable() {\r
 624                 @Override\r
 625                 public void run() {\r
 626                     while (it.hasNext()) {\r
 627                         ChangeSetListener l = it.next();\r
 628                         try {\r
 629                             l.onChanged(csid);\r
 630                         } catch (Throwable t) {\r
 631                             Logger.defaultLogError(t);\r
 632                         }\r
 633                     }\r
 634                 }\r
 635             });\r
 636         }\r
 637     }\r
 638     protected abstract ServerInformationImpl getServerInformation()\r
 639     throws DatabaseException;\r
 640     public abstract void acceptCommit(long transactionId, long csid, byte[] metadata)\r
 641     throws DatabaseException;\r
 642     public abstract void cancelCommit(long transactionId, long csid, byte[] metadata, SynchronizeContextI context)\r
 643     throws DatabaseException;\r
 644     public abstract void endTransaction(long transactionId, boolean write)\r
 645     throws DatabaseException;\r
 646     public abstract long askWriteTransaction(int thread, long transactionId)\r
 647     throws DatabaseException;\r
 648     public abstract long askReadTransaction(int thread)\r
 649     throws DatabaseException;\r
 650     public abstract void stop()\r
 651     throws DatabaseException;\r
 652 \r
 653     public abstract long reserveIds(int count)\r
 654     throws DatabaseException;\r
 655     static class MetadataCache {\r
 656         private final boolean DEBUG = false;\r
 657         private final int SIZE = 10;\r
 658         private int lastInd; // index of last inserted element\r
 659         private long lastId; // id of last inserted element\r
 660         private int count; // number of inserted elements\r
 661         Vector<byte[]> buffer;\r
 662         MetadataCache() {\r
 663             init();\r
 664         }\r
 665         public int clear() {\r
 666             int ret = count;\r
 667             init();\r
 668             return ret;\r
 669         }\r
 670         private void init() {\r
 671             if (DEBUG)\r
 672                 System.out.println("DEBUG: MetadataCache init");\r
 673             lastInd = -1;\r
 674             lastId = 0;\r
 675             count = 0;\r
 676             buffer = new Vector<byte[]>();\r
 677             buffer.setSize(SIZE);\r
 678         }\r
 679         private boolean incLastInd() {\r
 680             ++lastInd;\r
 681             if (lastInd >= SIZE) {\r
 682                 lastInd = 0;\r
 683                 return true;\r
 684             } else\r
 685                 return false;\r
 686         }\r
 687         synchronized void addNext(long id, byte[] data)\r
 688         throws DatabaseException {\r
 689             if (DEBUG)\r
 690                 System.out.println("DEBUG: MetadataCache add id=" + id);\r
 691             if (lastId != 0 && lastId != id - 1)\r
 692                 init(); // Only continuous sequence supported.\r
 693             incLastInd();\r
 694             buffer.set(lastInd, data);\r
 695             lastId = id;\r
 696             if (count < SIZE)\r
 697                 ++count;\r
 698         }\r
 699         synchronized byte[] get(long id) {\r
 700             if (DEBUG)\r
 701                 System.out.println("DEBUG: MetadataCache get id=" + id);\r
 702             if (id > lastId || id <= lastId - count)\r
 703                 return null;\r
 704             int ind = lastInd - (int)(lastId - id);\r
 705             if (ind < 0)\r
 706                 ind = SIZE + ind;\r
 707             byte[] found = buffer.get(ind);\r
 708             if (DEBUG)\r
 709                 if (null != found)\r
 710                     System.out.println("DEBUG: MetadataCache found id=" + id);\r
 711             return found;\r
 712         }\r
 713     }\r
 714     public ClusterUID[] listClusters() throws InternalException {\r
 715         ClusterIds t = dbSession.getClusterIds();\r
 716         long[] first = t.getFirst();\r
 717         long[] second = t.getSecond();\r
 718         int N1 = (null == first) ? 0 : first.length;\r
 719         N1 = Math.min(N1, t.getStatus());\r
 720         int N2 = (null == second) ? 0 : second.length;\r
 721         N2 = Math.min(N1, N2);\r
 722         ClusterUID[] clusters = new ClusterUID[N2];\r
 723         for (int i=0; i<N2; ++i)\r
 724             clusters[i] = new ClusterUID(first[i], second[i]);\r
 725         return clusters;\r
 726     }\r
 727     public boolean rolledback() {\r
 728         return dbSession.rolledback();\r
 729     }\r
 730 }\r
 731 class BuiltinData {\r
 732     final long                  id;\r
 733     final long                  cluster;\r
 734     SoftReference<ResourceImpl> weakResource = null;\r
 735     BuiltinData(long id, long cluster) {\r
 736         this.id = id;\r
 737         this.cluster = cluster;\r
 738     }\r
 739 }\r
 740 class UpdateClusterFunction {\r
 741     public byte[] operation = null;\r
 742     public UpdateClusterFunction(byte[] operation) {\r
 743         this.operation = operation;\r
 744     }\r
 745 }\r