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

   1 /*******************************************************************************
   2  * Copyright (c) 2007, 2018 Association for Decentralized Information Management
   3  * in Industry THTH ry.
   4  * All rights reserved. This program and the accompanying materials
   5  * are made available under the terms of the Eclipse Public License v1.0
   6  * which accompanies this distribution, and is available at
   7  * http://www.eclipse.org/legal/epl-v10.html
   8  *
   9  * Contributors:
  10  *     VTT Technical Research Centre of Finland - initial API and implementation
  11  *******************************************************************************/
  12 package fi.vtt.simantics.procore.internal;
  13
  14 import java.io.File;
  15 import java.io.IOException;
  16 import java.io.InputStream;
  17 import java.nio.charset.Charset;
  18 import java.nio.file.FileVisitResult;
  19 import java.nio.file.Files;
  20 import java.nio.file.Path;
  21 import java.nio.file.SimpleFileVisitor;
  22 import java.nio.file.attribute.BasicFileAttributes;
  23 import java.util.ArrayList;
  24 import java.util.Collection;
  25 import java.util.HashSet;
  26 import java.util.TreeMap;
  27 import java.util.concurrent.CopyOnWriteArrayList;
  28 import java.util.concurrent.Semaphore;
  29 import java.util.concurrent.atomic.AtomicInteger;
  30 import java.util.function.Consumer;
  31
  32 import org.eclipse.core.runtime.Platform;
  33 import org.simantics.databoard.Bindings;
  34 import org.simantics.db.AsyncReadGraph;
  35 import org.simantics.db.ChangeSet;
  36 import org.simantics.db.DevelopmentKeys;
  37 import org.simantics.db.Disposable;
  38 import org.simantics.db.ExternalValueSupport;
  39 import org.simantics.db.Metadata;
  40 import org.simantics.db.MonitorContext;
  41 import org.simantics.db.MonitorHandler;
  42 import org.simantics.db.ReadGraph;
  43 import org.simantics.db.Resource;
  44 import org.simantics.db.ResourceSerializer;
  45 import org.simantics.db.Session;
  46 import org.simantics.db.SessionManager;
  47 import org.simantics.db.SessionVariables;
  48 import org.simantics.db.VirtualGraph;
  49 import org.simantics.db.WriteGraph;
  50 import org.simantics.db.authentication.UserAuthenticationAgent;
  51 import org.simantics.db.authentication.UserAuthenticator;
  52 import org.simantics.db.common.Indexing;
  53 import org.simantics.db.common.TransactionPolicyRelease;
  54 import org.simantics.db.common.procedure.adapter.AsyncMultiProcedureAdapter;
  55 import org.simantics.db.common.procedure.adapter.ProcedureAdapter;
  56 import org.simantics.db.common.procedure.adapter.SyncMultiProcedureAdapter;
  57 import org.simantics.db.common.procedure.wrapper.NoneToAsyncListener;
  58 import org.simantics.db.common.procedure.wrapper.NoneToAsyncMultiListener;
  59 import org.simantics.db.common.procedure.wrapper.NoneToAsyncMultiProcedure;
  60 import org.simantics.db.common.procedure.wrapper.NoneToAsyncProcedure;
  61 import org.simantics.db.common.procedure.wrapper.NoneToSyncMultiListener;
  62 import org.simantics.db.common.procedure.wrapper.NoneToSyncMultiProcedure;
  63 import org.simantics.db.common.procedure.wrapper.SyncToAsyncListener;
  64 import org.simantics.db.common.procedure.wrapper.SyncToAsyncMultiListener;
  65 import org.simantics.db.common.procedure.wrapper.SyncToAsyncMultiProcedure;
  66 import org.simantics.db.common.procedure.wrapper.SyncToAsyncProcedure;
  67 import org.simantics.db.common.service.ServiceActivityMonitorImpl;
  68 import org.simantics.db.common.service.ServiceActivityUpdaterForWriteTransactions;
  69 import org.simantics.db.common.utils.Logger;
  70 import org.simantics.db.event.ChangeEvent;
  71 import org.simantics.db.event.ChangeListener;
  72 import org.simantics.db.event.SessionEventListener;
  73 import org.simantics.db.exception.CancelTransactionException;
  74 import org.simantics.db.exception.ClusterSetExistException;
  75 import org.simantics.db.exception.DatabaseException;
  76 import org.simantics.db.exception.ImmutableException;
  77 import org.simantics.db.exception.InvalidResourceReferenceException;
  78 import org.simantics.db.exception.ResourceNotFoundException;
  79 import org.simantics.db.exception.RuntimeDatabaseException;
  80 import org.simantics.db.exception.ServiceException;
  81 import org.simantics.db.exception.ServiceNotFoundException;
  82 import org.simantics.db.impl.BlockingAsyncProcedure;
  83 import org.simantics.db.impl.ClusterBase;
  84 import org.simantics.db.impl.ClusterI;
  85 import org.simantics.db.impl.ClusterTraitsBase;
  86 import org.simantics.db.impl.ClusterTranslator;
  87 import org.simantics.db.impl.ResourceImpl;
  88 import org.simantics.db.impl.TransientGraph;
  89 import org.simantics.db.impl.VirtualGraphImpl;
  90 import org.simantics.db.impl.graph.DelayedWriteGraph;
  91 import org.simantics.db.impl.graph.ReadGraphImpl;
  92 import org.simantics.db.impl.graph.WriteGraphImpl;
  93 import org.simantics.db.impl.graph.WriteSupport;
  94 import org.simantics.db.impl.internal.RandomAccessValueSupport;
  95 import org.simantics.db.impl.procedure.ResultCallWrappedQueryProcedure4;
  96 import org.simantics.db.impl.procedure.ResultCallWrappedSyncQueryProcedure;
  97 import org.simantics.db.impl.query.QueryCache;
  98 import org.simantics.db.impl.query.QueryCacheBase;
  99 import org.simantics.db.impl.query.QueryProcessor;
 100 import org.simantics.db.impl.query.QueryProcessor.SessionRead;
 101 import org.simantics.db.impl.query.QueryProcessor.SessionTask;
 102 import org.simantics.db.impl.service.QueryDebug;
 103 import org.simantics.db.impl.support.VirtualGraphServerSupport;
 104 import org.simantics.db.impl.support.WriteRequestScheduleSupport;
 105 import org.simantics.db.procedure.AsyncListener;
 106 import org.simantics.db.procedure.AsyncMultiListener;
 107 import org.simantics.db.procedure.AsyncMultiProcedure;
 108 import org.simantics.db.procedure.AsyncProcedure;
 109 import org.simantics.db.procedure.Listener;
 110 import org.simantics.db.procedure.ListenerBase;
 111 import org.simantics.db.procedure.MultiListener;
 112 import org.simantics.db.procedure.MultiProcedure;
 113 import org.simantics.db.procedure.Procedure;
 114 import org.simantics.db.procedure.SyncListener;
 115 import org.simantics.db.procedure.SyncMultiListener;
 116 import org.simantics.db.procedure.SyncMultiProcedure;
 117 import org.simantics.db.procedure.SyncProcedure;
 118 import org.simantics.db.procore.cluster.ClusterImpl;
 119 import org.simantics.db.procore.cluster.ClusterTraits;
 120 import org.simantics.db.procore.protocol.Constants;
 121 import org.simantics.db.procore.protocol.DebugPolicy;
 122 import org.simantics.db.request.AsyncMultiRead;
 123 import org.simantics.db.request.AsyncRead;
 124 import org.simantics.db.request.DelayedWrite;
 125 import org.simantics.db.request.DelayedWriteResult;
 126 import org.simantics.db.request.ExternalRead;
 127 import org.simantics.db.request.MultiRead;
 128 import org.simantics.db.request.Read;
 129 import org.simantics.db.request.ReadInterface;
 130 import org.simantics.db.request.Write;
 131 import org.simantics.db.request.WriteInterface;
 132 import org.simantics.db.request.WriteOnly;
 133 import org.simantics.db.request.WriteOnlyResult;
 134 import org.simantics.db.request.WriteResult;
 135 import org.simantics.db.request.WriteTraits;
 136 import org.simantics.db.service.ByteReader;
 137 import org.simantics.db.service.ClusterBuilder;
 138 import org.simantics.db.service.ClusterBuilderFactory;
 139 import org.simantics.db.service.ClusterControl;
 140 import org.simantics.db.service.ClusterSetsSupport;
 141 import org.simantics.db.service.ClusterUID;
 142 import org.simantics.db.service.ClusteringSupport;
 143 import org.simantics.db.service.CollectionSupport;
 144 import org.simantics.db.service.DebugSupport;
 145 import org.simantics.db.service.DirectQuerySupport;
 146 import org.simantics.db.service.EventSupport;
 147 import org.simantics.db.service.GraphChangeListenerSupport;
 148 import org.simantics.db.service.InitSupport;
 149 import org.simantics.db.service.LifecycleSupport;
 150 import org.simantics.db.service.ManagementSupport;
 151 import org.simantics.db.service.QueryControl;
 152 import org.simantics.db.service.SerialisationSupport;
 153 import org.simantics.db.service.ServerInformation;
 154 import org.simantics.db.service.ServiceActivityMonitor;
 155 import org.simantics.db.service.SessionEventSupport;
 156 import org.simantics.db.service.SessionMonitorSupport;
 157 import org.simantics.db.service.SessionUserSupport;
 158 import org.simantics.db.service.StatementSupport;
 159 import org.simantics.db.service.TransactionPolicySupport;
 160 import org.simantics.db.service.TransactionSupport;
 161 import org.simantics.db.service.TransferableGraphSupport;
 162 import org.simantics.db.service.UndoRedoSupport;
 163 import org.simantics.db.service.VirtualGraphSupport;
 164 import org.simantics.db.service.XSupport;
 165 import org.simantics.layer0.Layer0;
 166 import org.simantics.utils.DataContainer;
 167 import org.simantics.utils.Development;
 168 import org.simantics.utils.threads.logger.ITask;
 169 import org.simantics.utils.threads.logger.ThreadLogger;
 170 import org.slf4j.LoggerFactory;
 171
 172 import gnu.trove.procedure.TLongProcedure;
 173 import gnu.trove.set.hash.TLongHashSet;
 174
 175
 176 public abstract class SessionImplSocket implements Session, WriteRequestScheduleSupport {
 177
 178     private static org.slf4j.Logger LOGGER = LoggerFactory.getLogger(SessionImplSocket.class);
 179
 180     protected static final boolean DEBUG        = false;
 181
 182     private static final boolean DIAGNOSTICS       = false;
 183
 184     private TransactionPolicySupport transactionPolicy;
 185
 186     final private ClusterControl clusterControl;
 187
 188     final protected BuiltinSupportImpl builtinSupport;
 189     final protected VirtualGraphServerSupportImpl virtualGraphServerSupport;
 190     final protected ClusterSetsSupport clusterSetsSupport;
 191     final protected LifecycleSupportImpl lifecycleSupport;
 192
 193     protected QuerySupportImpl querySupport;
 194     protected ResourceSupportImpl resourceSupport;
 195     protected WriteSupport writeSupport;
 196     public ClusterTranslator clusterTranslator;
 197
 198     boolean dirtyPrimitives = false;
 199
 200     public static final int SERVICE_MODE_CREATE = 2;
 201     public static final int SERVICE_MODE_ALLOW = 1;
 202     public int serviceMode = 0;
 203     public boolean createdImmutableClusters = false;
 204     public TLongHashSet createdClusters = new TLongHashSet();
 205
 206     private Layer0 L0;
 207
 208     /**
 209      * The service locator maintained by the workbench. These services are
 210      * initialized during workbench during the <code>init</code> method.
 211      */
 212     final protected ServiceLocatorImpl                           serviceLocator     = new ServiceLocatorImpl();
 213
 214     final public ResourceSerializerImpl                       resourceSerializer = new ResourceSerializerImpl();
 215
 216     final CopyOnWriteArrayList<ChangeListener>         changeListeners2   = new CopyOnWriteArrayList<ChangeListener>();
 217
 218     final CopyOnWriteArrayList<ChangeListener>         metadataListeners   = new CopyOnWriteArrayList<ChangeListener>();
 219
 220     final CopyOnWriteArrayList<SessionEventListener>   eventListeners     = new CopyOnWriteArrayList<SessionEventListener>();
 221
 222     final HashSet<Thread>                              sessionThreads     = new HashSet<Thread>();
 223
 224     final BijectionMap<MonitorContext, MonitorHandler> monitorContexts    = new BijectionMap<MonitorContext, MonitorHandler>();
 225
 226     final protected State                                        state              = new State();
 227
 228     protected GraphSession                                       graphSession       = null;
 229
 230     protected SessionManager                                     sessionManagerImpl = null;
 231
 232     protected UserAuthenticationAgent                            authAgent          = null;
 233
 234     protected UserAuthenticator                                  authenticator      = null;
 235
 236     protected Resource                                           user               = null;
 237
 238     protected ClusterStream                                      clusterStream      = null;
 239
 240     protected SessionRequestManager                         requestManager = null;
 241
 242     public ClusterTable                                       clusterTable       = null;
 243
 244     public QueryProcessor                                     queryProvider2 = null;
 245
 246     ClientChangesImpl                                  clientChanges      = null;
 247
 248     MonitorHandler[]                                   monitorHandlers    = new MonitorHandler[0];
 249
 250 //    protected long                                               newClusterId       = Constants.NullClusterId;
 251
 252     protected int                                                flushCounter       = 0;
 253
 254     protected boolean                                            writeOnly          = false;
 255
 256     WriteState<?>                                                writeState = null;
 257     WriteStateBase<?>                                            delayedWriteState = null;
 258     protected Resource defaultClusterSet = null; // If not null then used for newResource().
 259
 260     public SessionImplSocket(SessionManager sessionManagerImpl, UserAuthenticationAgent authAgent) {
 261
 262 //        if (authAgent == null)
 263 //            throw new IllegalArgumentException("null authentication agent");
 264
 265         File t = StaticSessionProperties.virtualGraphStoragePath;
 266         if (null == t)
 267             t = new File(".");
 268         this.clusterTable = new ClusterTable(this, t);
 269         this.builtinSupport = new BuiltinSupportImpl(this);
 270         this.sessionManagerImpl = sessionManagerImpl;
 271         this.user = null;
 272 //        this.authAgent = authAgent;
 273
 274         serviceLocator.registerService(Session.class, this);
 275         serviceLocator.registerService(InitSupport.class, new InitSupportImpl(this));
 276         serviceLocator.registerService(ManagementSupport.class, new ManagementSupportImpl(this));
 277         serviceLocator.registerService(QueryControl.class, new QueryControlImpl(this));
 278         serviceLocator.registerService(SessionUserSupport.class, new SessionUserSupportImpl(this));
 279         serviceLocator.registerService(GraphChangeListenerSupport.class, new GraphChangeListenerSupportImpl(this));
 280         serviceLocator.registerService(SessionEventSupport.class, new SessionEventSupportImpl(this));
 281         serviceLocator.registerService(SerialisationSupport.class, new SerialisationSupportImpl(this));
 282         serviceLocator.registerService(UndoRedoSupport.class, new UndoRedoSupportImpl(this));
 283         serviceLocator.registerService(ClusteringSupport.class, new ClusteringSupportImpl(this));
 284         serviceLocator.registerService(TransactionSupport.class, new TransactionSupportImpl(this));
 285         serviceLocator.registerService(SessionMonitorSupport.class, new SessionMonitorSupportImpl(this));
 286         serviceLocator.registerService(TransferableGraphSupport.class, new TransferableGraphSupportImpl(this));
 287         serviceLocator.registerService(QueryDebug.class, new QueryDebugImpl(this));
 288         serviceLocator.registerService(CollectionSupport.class, new CollectionSupportImpl(this));
 289         serviceLocator.registerService(StatementSupport.class, new StatementSupportImpl(this));
 290         serviceLocator.registerService(DirectQuerySupport.class, new DirectQuerySupportImpl(this));
 291         serviceLocator.registerService(XSupport.class, new XSupportImpl(this));
 292         serviceLocator.registerService(DebugSupport.class, new DebugSupportImpl());
 293         serviceLocator.registerService(ExternalValueSupport.class, new ExternalValueSupportImpl(this));
 294         serviceLocator.registerService(RandomAccessValueSupport.class, new RandomAccessValueSupportImpl());
 295         serviceLocator.registerService(ServiceActivityMonitor.class, new ServiceActivityMonitorImpl());
 296         serviceLocator.registerService(EventSupport.class, new EventSupportImpl());
 297         ServiceActivityUpdaterForWriteTransactions.register(this);
 298
 299         this.virtualGraphServerSupport = new VirtualGraphServerSupportImpl(this, t);
 300         serviceLocator.registerService(VirtualGraphSupport.class, virtualGraphServerSupport);
 301         serviceLocator.registerService(VirtualGraphServerSupport.class, virtualGraphServerSupport);
 302         this.lifecycleSupport = new LifecycleSupportImpl(this);
 303         serviceLocator.registerService(LifecycleSupport.class, lifecycleSupport);
 304         this.transactionPolicy = new TransactionPolicyRelease();
 305         serviceLocator.registerService(TransactionPolicySupport.class, transactionPolicy);
 306         this.clusterControl = new ClusterControlImpl(this);
 307         serviceLocator.registerService(ClusterControl.class, clusterControl);
 308         this.clusterSetsSupport = new ClusterSetsSupportImpl2(); // Using same path as virtual graphs.
 309         this.clusterSetsSupport.setReadDirectory(t.toPath());
 310         this.clusterSetsSupport.updateWriteDirectory(t.toPath());
 311         serviceLocator.registerService(ClusterSetsSupport.class, clusterSetsSupport);
 312     }
 313
 314     /*
 315      *
 316      * Session interface
 317      */
 318
 319
 320     @Override
 321     public Resource getRootLibrary() {
 322         return queryProvider2.getRootLibraryResource();
 323     }
 324
 325     void refresh(int thread, ClusterUID[] clusterUID, long csid) throws DatabaseException {
 326         if (!graphSession.dbSession.refreshEnabled())
 327             return;
 328         try {
 329             getClusterTable().refresh(csid, this, clusterUID);
 330         } catch (Throwable t) {
 331             Logger.defaultLogError("Refesh failed.", t);
 332         }
 333     }
 334
 335     static final class TaskHelper {
 336         private final String name;
 337         private Object result;
 338         final Semaphore sema = new Semaphore(0);
 339         private Throwable throwable = null;
 340         final Consumer<DatabaseException> callback = e -> {
 341             synchronized (TaskHelper.this) {
 342                 throwable = e;
 343             }
 344         };
 345         final Procedure<Object> proc = new Procedure<Object>() {
 346             @Override
 347             public void execute(Object result) {
 348                 callback.accept(null);
 349             }
 350             @Override
 351             public void exception(Throwable t) {
 352                 if (t instanceof DatabaseException)
 353                     callback.accept((DatabaseException)t);
 354                 else
 355                     callback.accept(new DatabaseException("" + name + "operation failed.", t));
 356             }
 357         };
 358         final WriteTraits writeTraits = new WriteTraits() {};
 359         TaskHelper(String name) {
 360             this.name = name;
 361         }
 362         @SuppressWarnings("unchecked")
 363         <T> T getResult() {
 364             return (T)result;
 365         }
 366         void setResult(Object result) {
 367             this.result = result;
 368         }
 369 //        Throwable throwableGet() {
 370 //            return throwable;
 371 //        }
 372         synchronized void throwableSet(Throwable t) {
 373             throwable = t;
 374         }
 375 //        void throwableSet(String t) {
 376 //            throwable = new InternalException("" + name + " operation failed. " + t);
 377 //        }
 378         synchronized void throwableCheck()
 379         throws DatabaseException {
 380             if (null != throwable)
 381                 if (throwable instanceof DatabaseException)
 382                     throw (DatabaseException)throwable;
 383                 else
 384                     throw new DatabaseException("Undo operation failed.", throwable);
 385         }
 386         void throw_(String message)
 387         throws DatabaseException {
 388             throw new DatabaseException("" + name + " operation failed. " + message);
 389         }
 390     }
 391
 392
 393
 394     private ListenerBase getListenerBase(Object procedure) {
 395         if (procedure instanceof ListenerBase)
 396             return (ListenerBase) procedure;
 397         else
 398             return null;
 399     }
 400
 401     public <T> void scheduleRequest(final Write request, final Consumer<DatabaseException> callback, final Semaphore notify) {
 402         scheduleRequest(request, callback, notify, null);
 403     }
 404
 405     /* (non-Javadoc)
 406      * @see fi.vtt.simantics.procore.internal.WriteRequestScheduler#scheduleRequest(org.simantics.db.request.Write, org.simantics.utils.datastructures.Callback, java.util.concurrent.Semaphore, java.lang.Boolean)
 407      */
 408     @Override
 409     public <T> void scheduleRequest(final Write request, final Consumer<DatabaseException> callback, final Semaphore notify, Boolean combine) {
 410
 411         assert (request != null);
 412
 413         if(Development.DEVELOPMENT) {
 414             try {
 415             if(Development.<Boolean>getProperty(DevelopmentKeys.SESSION_LOG_WRITES, Bindings.BOOLEAN))
 416                 System.err.println("schedule write '" + request + "'");
 417             } catch (Throwable t) {
 418                 Logger.defaultLogError(t);
 419             }
 420         }
 421
 422         requestManager.scheduleWrite(new SessionTask(null) {
 423
 424             @Override
 425             public void run0(int thread) {
 426
 427                 if(Development.DEVELOPMENT) {
 428                     try {
 429                     if(Development.<Boolean>getProperty(DevelopmentKeys.SESSION_LOG_WRITES, Bindings.BOOLEAN))
 430                         System.err.println("perform write '" + request + "'");
 431                     } catch (Throwable t) {
 432                         Logger.defaultLogError(t);
 433                     }
 434                 }
 435
 436                 ITask task = ThreadLogger.getInstance().begin("WriteRequest " + request);
 437
 438                 fireSessionVariableChange(SessionVariables.QUEUED_WRITES);
 439
 440                 try {
 441
 442                     flushCounter = 0;
 443                     Disposable.safeDispose(clientChanges);
 444                     clientChanges = new ClientChangesImpl(SessionImplSocket.this);
 445
 446                     VirtualGraph vg = getProvider(request.getProvider());
 447
 448                     WriteGraphImpl writer = WriteGraphImpl.create(getQueryProvider2(), writeSupport, vg);
 449                     writeState = new WriteState<Object>(writer, request, notify, new Procedure<Object>() {
 450
 451                         @Override
 452                         public void execute(Object result) {
 453                             if(callback != null) callback.accept(null);
 454                         }
 455
 456                         @Override
 457                         public void exception(Throwable t) {
 458                             if(callback != null) callback.accept((DatabaseException)t);
 459                         }
 460
 461                     });
 462
 463                     assert (null != writer);
 464 //                    writer.state.barrier.inc();
 465
 466                     try {
 467                         request.perform(writer);
 468                         assert (null != writer);
 469                     } catch (Throwable t) {
 470                         if (!(t instanceof CancelTransactionException))
 471                             Logger.defaultLogError("Write transaction caused an unexpected error, see exception.", t);
 472                         writeState.except(t);
 473                     } finally {
 474 //                        writer.state.barrier.dec();
 475 //                        writer.waitAsync(request);
 476                     }
 477
 478
 479                     assert(!queryProvider2.cache.dirty);
 480
 481                 } catch (Throwable e) {
 482
 483                     // Log it first, just to be safe that the error is always logged.
 484                     if (!(e instanceof CancelTransactionException))
 485                     Logger.defaultLogError("Write transaction caused an unexpected error, see exception.", e);
 486
 487 //                    writeState.getGraph().state.barrier.dec();
 488 //                    writeState.getGraph().waitAsync(request);
 489
 490                     writeState.except(e);
 491
 492 //                    try {
 493 //                        // Callback is client code, we have to be prepared for it to throw unexpected exceptions.
 494 //                        // All we can do here is to log those, can't really pass them anywhere.
 495 //                        if (callback != null) {
 496 //                            if(e instanceof DatabaseException) callback.run((DatabaseException)e);
 497 //                            else callback.run(new DatabaseException(e));
 498 //                        }
 499 //                    } catch (Throwable e2) {
 500 //                        Logger.defaultLogError("Write request callback caused an unexpected error, see exception.", e2);
 501 //                    }
 502 //                    boolean empty = clusterStream.reallyFlush();
 503 //                    int callerThread = -1;
 504 //                    ClientChangesImpl cs = new ClientChangesImpl(SessionImplSocket.this);
 505 //                    SynchronizeContext context = new SynchronizeContext(callerThread, SessionImplSocket.this, cs, 1);
 506 //                    try {
 507 //                        // Send all local changes to server so it can calculate correct reverse change set.
 508 //                        // This will call clusterStream.accept().
 509 //                        state.cancelCommit(context, clusterStream);
 510 //                        if (!empty) {
 511 //                            if (!context.isOk()) // this is a blocking operation
 512 //                                throw new InternalException("Cancel failed. This should never happen.");
 513 //                            getQueryProvider2().performDirtyUpdates(writeState.getGraph());
 514 //                        }
 515 //                        state.cancelCommit2(context, clusterStream);
 516 //                    } catch (DatabaseException e3) {
 517 //                        Logger.defaultLogError("Write request cancel caused an unexpected error.", e3);
 518 //                    } finally {
 519 //                        clusterStream.setOff(false);
 520 //                        clientChanges = new ClientChangesImpl(SessionImplSocket.this);
 521 //                    }
 522
 523                 } finally {
 524
 525                     fireSessionVariableChange(SessionVariables.QUEUED_WRITES);
 526
 527                 }
 528
 529                 task.finish();
 530
 531                 if(Development.DEVELOPMENT) {
 532                     try {
 533                     if(Development.<Boolean>getProperty(DevelopmentKeys.SESSION_LOG_WRITES, Bindings.BOOLEAN))
 534                         System.err.println("finish write '" + request + "'");
 535                     } catch (Throwable t) {
 536                         Logger.defaultLogError(t);
 537                     }
 538                 }
 539
 540             }
 541
 542         }, combine);
 543
 544     }
 545
 546     public <T> void scheduleRequest(final WriteResult<T> request, final Procedure<T> procedure, final Semaphore notify) {
 547         scheduleRequest(request, procedure, notify, null);
 548     }
 549
 550     /* (non-Javadoc)
 551      * @see fi.vtt.simantics.procore.internal.WriteRequestScheduler#scheduleRequest(org.simantics.db.request.WriteResult, org.simantics.db.procedure.Procedure, java.util.concurrent.Semaphore, java.lang.Boolean)
 552      */
 553     @Override
 554     public <T> void scheduleRequest(final WriteResult<T> request, final Procedure<T> procedure, final Semaphore notify, Boolean combine) {
 555
 556         assert (request != null);
 557
 558         requestManager.scheduleWrite(new SessionTask(null) {
 559
 560             @Override
 561             public void run0(int thread) {
 562
 563                 ITask task = ThreadLogger.getInstance().begin("WriteRequest " + request);
 564
 565                 fireSessionVariableChange(SessionVariables.QUEUED_WRITES);
 566
 567                 flushCounter = 0;
 568                 Disposable.safeDispose(clientChanges);
 569                 clientChanges = new ClientChangesImpl(SessionImplSocket.this);
 570
 571                 VirtualGraph vg = getProvider(request.getProvider());
 572                 WriteGraphImpl writer = WriteGraphImpl.create(getQueryProvider2(), writeSupport, vg);
 573
 574                 try {
 575                     WriteState<T> writeStateT = new WriteState<T>(writer, request, notify, procedure);
 576                     writeState = writeStateT;
 577
 578                     assert (null != writer);
 579 //                    writer.state.barrier.inc();
 580                     writeStateT.setResult(request.perform(writer));
 581                     assert (null != writer);
 582
 583 //                    writer.state.barrier.dec();
 584 //                    writer.waitAsync(null);
 585
 586                 } catch (Throwable e) {
 587
 588 //                    writer.state.barrier.dec();
 589 //                    writer.waitAsync(null);
 590
 591                     writeState.except(e);
 592
 593 //                  state.stopWriteTransaction(clusterStream);
 594 //
 595 //              } catch (Throwable e) {
 596 //                  // Log it first, just to be safe that the error is always logged.
 597 //                  Logger.defaultLogError("Write transaction caused an unexpected error, see exception.", e);
 598 //
 599 //                  try {
 600 //                      // Callback is client code, we have to be prepared for it to throw unexpected exceptions.
 601 //                      // All we can do here is to log those, can't really pass them anywhere.
 602 //                      if (procedure != null) {
 603 //                          if(e instanceof DatabaseException) procedure.exception((DatabaseException)e);
 604 //                          else procedure.exception(new DatabaseException(e));
 605 //                      }
 606 //                  } catch (Throwable e2) {
 607 //                      Logger.defaultLogError("Write request callback caused an unexpected error, see exception.", e2);
 608 //                  }
 609 //
 610 //                  clientChanges = new ClientChangesImpl(SessionImplSocket.this);
 611 //
 612 //                  state.stopWriteTransaction(clusterStream);
 613
 614                 } finally {
 615                     fireSessionVariableChange(SessionVariables.QUEUED_WRITES);
 616                 }
 617
 618 //              if(notify != null) notify.release();
 619
 620                 task.finish();
 621
 622             }
 623
 624         }, combine);
 625
 626     }
 627
 628     public <T> void scheduleRequest(final DelayedWrite request, final Consumer<DatabaseException> callback, final Semaphore notify) {
 629         scheduleRequest(request, callback, notify, null);
 630     }
 631
 632     /* (non-Javadoc)
 633      * @see fi.vtt.simantics.procore.internal.WriteRequestScheduler#scheduleRequest(org.simantics.db.request.DelayedWrite, org.simantics.utils.datastructures.Callback, java.util.concurrent.Semaphore, java.lang.Boolean)
 634      */
 635     @Override
 636     public <T> void scheduleRequest(final DelayedWrite request, final Consumer<DatabaseException> callback, final Semaphore notify, Boolean combine) {
 637
 638         final ITask total = ThreadLogger.getInstance().begin("ScheduleDelayedWrite");
 639
 640         assert (request != null);
 641
 642         requestManager.scheduleWrite(new SessionTask(null) {
 643
 644             @Override
 645             public void run0(int thread) {
 646                 fireSessionVariableChange(SessionVariables.QUEUED_READS);
 647
 648                 Procedure<Object> stateProcedure = new Procedure<Object>() {
 649                     @Override
 650                     public void execute(Object result) {
 651                         if (callback != null)
 652                             callback.accept(null);
 653                     }
 654                     @Override
 655                     public void exception(Throwable t) {
 656                         if (callback != null) {
 657                             if (t instanceof DatabaseException) callback.accept((DatabaseException) t);
 658                             else callback.accept(new DatabaseException(t));
 659                         } else
 660                             Logger.defaultLogError("Unhandled exception", t);
 661                     }
 662                 };
 663
 664                 final ReadGraphImpl newGraph = ReadGraphImpl.create(getQueryProvider2());
 665                 delayedWriteState = new WriteStateBase<Object>(request, notify, stateProcedure);
 666                 DelayedWriteGraph dwg = null;
 667 //                newGraph.state.barrier.inc();
 668
 669                 try {
 670                     dwg = new DelayedWriteGraph(newGraph);
 671                     request.perform(dwg);
 672                 } catch (Throwable e) {
 673                     delayedWriteState.except(e);
 674                     total.finish();
 675                     dwg.close();
 676                     return;
 677                 } finally {
 678 //                    newGraph.state.barrier.dec();
 679 //                    newGraph.waitAsync(request);
 680                     fireSessionVariableChange(SessionVariables.QUEUED_READS);
 681                 }
 682
 683                 delayedWriteState = null;
 684
 685                 ITask task2 = ThreadLogger.getInstance().begin("DelayedWriteCommit");
 686                 fireSessionVariableChange(SessionVariables.QUEUED_WRITES);
 687
 688                 flushCounter = 0;
 689                 Disposable.safeDispose(clientChanges);
 690                 clientChanges = new ClientChangesImpl(SessionImplSocket.this);
 691
 692                 acquireWriteOnly();
 693
 694                 VirtualGraph vg = getProvider(request.getProvider());
 695                 WriteSupport writeSupport = vg != null ? new VirtualWriteOnlySupport() : new WriteOnlySupport();
 696
 697                 WriteGraphImpl writer = WriteGraphImpl.create(getQueryProvider2(), writeSupport, vg);
 698
 699                 writeState = new WriteState<Object>(writer, request, notify, stateProcedure);
 700
 701                 assert (null != writer);
 702 //                writer.state.barrier.inc();
 703
 704                 try {
 705                     // Cannot cancel
 706                     dwg.commit(writer, request);
 707
 708                         if(defaultClusterSet != null) {
 709                                 XSupport xs = getService(XSupport.class);
 710                                 defaultClusterSet = xs.convertDelayedResourceToResource(defaultClusterSet);
 711                                 ClusteringSupport cs = getService(ClusteringSupport.class);
 712                                 clusterSetsSupport.put(defaultClusterSet.getResourceId(), cs.getCluster(defaultClusterSet));
 713                         }
 714
 715                     // This makes clusters available from server
 716                     clusterStream.reallyFlush();
 717
 718                     releaseWriteOnly(writer);
 719                     handleUpdatesAndMetadata(writer);
 720
 721                 } catch (ServiceException e) {
 722 //                    writer.state.barrier.dec();
 723 //                    writer.waitAsync(null);
 724
 725                     // These shall be requested from server
 726                     clusterTable.removeWriteOnlyClusters();
 727                     // This makes clusters available from server
 728                     clusterStream.reallyFlush();
 729
 730                     releaseWriteOnly(writer);
 731                     writeState.except(e);
 732                 } finally {
 733                     // Debugging & Profiling
 734                     fireSessionVariableChange(SessionVariables.QUEUED_WRITES);
 735                     task2.finish();
 736                     total.finish();
 737                 }
 738             }
 739
 740         }, combine);
 741
 742     }
 743
 744     public <T> void scheduleRequest(final DelayedWriteResult<T> request, final Procedure<T> procedure, final Semaphore notify) {
 745         scheduleRequest(request, procedure, notify, null);
 746     }
 747
 748     /* (non-Javadoc)
 749      * @see fi.vtt.simantics.procore.internal.WriteRequestScheduler#scheduleRequest(org.simantics.db.request.DelayedWriteResult, org.simantics.db.procedure.Procedure, java.util.concurrent.Semaphore, java.lang.Boolean)
 750      */
 751     @Override
 752     public <T> void scheduleRequest(final DelayedWriteResult<T> request, final Procedure<T> procedure, final Semaphore notify, Boolean combine) {
 753         throw new Error("Not implemented");
 754     }
 755
 756     protected ClusterImpl getNewResourceCluster() throws DatabaseException {
 757         ClusterImpl cluster = clusterTable.getNewResourceCluster(clusterTranslator, graphSession, writeOnly);
 758         if((serviceMode & SERVICE_MODE_CREATE) > 0) {
 759             createdClusters.add(cluster.clusterId);
 760         }
 761         return cluster;
 762     }
 763
 764     class WriteOnlySupport implements WriteSupport {
 765
 766         ClusterStream stream;
 767         ClusterImpl currentCluster;
 768
 769         public WriteOnlySupport() {
 770             this.stream = clusterStream;
 771         }
 772
 773         @Override
 774         public void claim(VirtualGraph provider, Resource subject, Resource predicate, Resource object) throws ServiceException {
 775             claim(provider, ((ResourceImpl)subject).id, ((ResourceImpl)predicate).id, ((ResourceImpl)object).id );
 776         }
 777
 778         @Override
 779         public void claim(VirtualGraph provider, int s, int p, int o) throws ServiceException {
 780
 781             ClusterImpl cluster = clusterTable.getClusterByResourceKey(s);
 782
 783             if (cluster.getImmutable() && (serviceMode & SessionImplSocket.SERVICE_MODE_ALLOW) == 0)
 784                 if(s != queryProvider2.getRootLibrary())
 785                     throw new ImmutableException("Trying to modify immutable resource key=" + s);
 786
 787             try {
 788                 maintainCluster(cluster, cluster.addRelation(s, p, o, clusterTranslator));
 789             } catch (DatabaseException e) {
 790                 Logger.defaultLogError(e);
 791                 //throw e;
 792             }
 793
 794             clientChanges.invalidate(s);
 795
 796             if (cluster.isWriteOnly())
 797                 return;
 798             queryProvider2.updateStatements(s, p);
 799
 800         }
 801
 802         @Override
 803         public void claimValue(VirtualGraph provider, Resource resource, byte[] value) throws DatabaseException {
 804             claimValue(provider, ((ResourceImpl)resource).id, value, value.length);
 805         }
 806
 807         @Override
 808         public void claimValue(VirtualGraph provider, int rid, byte[] value, int length) {
 809
 810             ClusterImpl cluster = clusterTable.getClusterByResourceKey(rid);
 811             try {
 812                 maintainCluster(cluster, cluster.setValue(rid, value, length, clusterTranslator));
 813             } catch (DatabaseException e) {
 814                 Logger.defaultLogError(e);
 815             }
 816
 817             clientChanges.invalidate(rid);
 818
 819             if (cluster.isWriteOnly())
 820                 return;
 821             queryProvider2.updateValue(rid);
 822
 823         }
 824
 825         @Override
 826         public void claimValue(VirtualGraph provider, Resource resource, ByteReader reader, int amount) throws DatabaseException {
 827
 828             if(amount < 65536) {
 829                 claimValue(provider,resource, reader.readBytes(null, amount));
 830                 return;
 831             }
 832
 833             byte[] bytes = new byte[65536];
 834
 835             int rid = ((ResourceImpl)resource).id;
 836             ClusterImpl cluster = clusterTable.getClusterByResourceKey(rid);
 837             try {
 838                 int left = amount;
 839                 while(left > 0) {
 840                     int block = Math.min(left, 65536);
 841                     reader.readBytes(bytes, block);
 842                     maintainCluster(cluster, cluster.modiValueEx(rid, amount-left, block, bytes, 0, clusterTranslator));
 843                     left -= block;
 844                 }
 845             } catch (DatabaseException e) {
 846                 Logger.defaultLogError(e);
 847             }
 848
 849             clientChanges.invalidate(rid);
 850
 851             if (cluster.isWriteOnly())
 852                 return;
 853             queryProvider2.updateValue(rid);
 854
 855         }
 856
 857         private void maintainCluster(ClusterImpl before, ClusterI after_) {
 858             if(after_ != null && after_ != before) {
 859                 ClusterImpl after = (ClusterImpl)after_;
 860                 if(currentCluster == before) {
 861                     currentCluster = after;
 862                 }
 863                 clusterTable.replaceCluster(after);
 864             }
 865         }
 866
 867         public int createResourceKey(int foreignCounter) throws DatabaseException {
 868             if(currentCluster == null) {
 869                 currentCluster = getNewResourceCluster();
 870             }
 871             if(currentCluster.getNumberOfResources(clusterTranslator) == ClusterTable.CLUSTER_FILL_SIZE) {
 872                 ClusterWriteOnly newCluster = (ClusterWriteOnly)getNewResourceCluster();
 873                 newCluster.foreignLookup = new byte[foreignCounter];
 874                 currentCluster = newCluster;
 875                 if (DEBUG)
 876                     System.err.println("foreignLookup " + currentCluster + " " + foreignCounter);
 877             }
 878             return currentCluster.createResource(clusterTranslator);
 879         }
 880
 881         @Override
 882         public Resource createResource(VirtualGraph provider) throws DatabaseException {
 883             if(currentCluster == null) {
 884                 if (null != defaultClusterSet) {
 885                         ResourceImpl result = getNewResource(defaultClusterSet);
 886                     currentCluster = clusterTable.getClusterByResourceKey(result.id);
 887                     return result;
 888                 } else {
 889                     currentCluster = getNewResourceCluster();
 890                 }
 891             }
 892             if(currentCluster.getNumberOfResources(clusterTranslator) >= ClusterTable.CLUSTER_FILL_SIZE) {
 893                 if (null != defaultClusterSet) {
 894                         ResourceImpl result = getNewResource(defaultClusterSet);
 895                     currentCluster = clusterTable.getClusterByResourceKey(result.id);
 896                     return result;
 897                 } else {
 898                     currentCluster = getNewResourceCluster();
 899                 }
 900             }
 901             return new ResourceImpl(resourceSupport, currentCluster.createResource(clusterTranslator));
 902         }
 903
 904         @Override
 905         public Resource createResource(VirtualGraph provider, long clusterId)
 906         throws DatabaseException {
 907             return getNewResource(clusterId);
 908         }
 909
 910         @Override
 911         public Resource createResource(VirtualGraph provider, Resource clusterSet)
 912         throws DatabaseException {
 913             return getNewResource(clusterSet);
 914         }
 915
 916         @Override
 917         public void createClusterSet(VirtualGraph provider, Resource clusterSet)
 918                 throws DatabaseException {
 919             getNewClusterSet(clusterSet);
 920         }
 921
 922         @Override
 923         public boolean hasClusterSet(VirtualGraph dummy, Resource clusterSet)
 924         throws ServiceException {
 925             return containsClusterSet(clusterSet);
 926         }
 927
 928         public void selectCluster(long cluster) {
 929                 currentCluster = clusterTable.getClusterByClusterId(cluster);
 930                 long setResourceId = clusterSetsSupport.getSet(cluster);
 931                 clusterSetsSupport.put(setResourceId, cluster);
 932         }
 933
 934         @Override
 935         public Resource setDefaultClusterSet(Resource clusterSet)
 936         throws ServiceException {
 937                 Resource result = setDefaultClusterSet4NewResource(clusterSet);
 938                 if(clusterSet != null) {
 939                         long id = clusterSetsSupport.get(clusterSet.getResourceId());
 940                         currentCluster = clusterTable.getClusterByClusterId(id);
 941                         return result;
 942                 } else {
 943                         currentCluster = null;
 944                         return null;
 945                 }
 946         }
 947
 948         @Override
 949         public void denyValue(VirtualGraph provider, Resource resource) throws ServiceException {
 950
 951                  provider = getProvider(provider);
 952              if (null == provider) {
 953                  int key = ((ResourceImpl)resource).id;
 954
 955
 956
 957                  ClusterI cluster = clusterTable.getClusterProxyByResourceKey(key);
 958 //                 if (cluster.getImmutable() && (serviceMode & SessionImplSocket.SERVICE_MODE_ALLOW) == 0)
 959 //                      if(key != queryProvider2.getRootLibrary())
 960 //                              throw new ImmutableException("Trying to modify immutable resource key=" + key);
 961
 962 //                 try {
 963 //                     cluster.removeValue(key, clusterTranslator);
 964 //                 } catch (DatabaseException e) {
 965 //                     Logger.defaultLogError(e);
 966 //                     return;
 967 //                 }
 968
 969                  clusterTable.writeOnlyInvalidate(cluster);
 970
 971                  try {
 972                          int sResourceKey = ClusterTraits.getResourceIndexFromResourceKey(key);
 973                          clusterTranslator.addStatementIndex(cluster, sResourceKey, cluster.getClusterUID(), ClusterChange.DELETE_OPERATION);
 974                          clusterTranslator.removeValue(cluster);
 975                  } catch (DatabaseException e) {
 976                          Logger.defaultLogError(e);
 977                  }
 978
 979                  queryProvider2.invalidateResource(key);
 980                  clientChanges.invalidate(key);
 981
 982              } else {
 983                  ((VirtualGraphImpl)provider).denyValue(((ResourceImpl) resource).id);
 984                  queryProvider2.updateValue(querySupport.getId(resource));
 985                  clientChanges.claimValue(resource);
 986              }
 987
 988
 989         }
 990
 991         @Override
 992         public void flush(boolean intermediate) {
 993             throw new UnsupportedOperationException();
 994         }
 995
 996         @Override
 997         public void flushCluster() {
 998             clusterTable.flushCluster(graphSession);
 999             if(defaultClusterSet != null) {
1000                 clusterSetsSupport.put(defaultClusterSet.getResourceId(), Constants.NewClusterId);
1001             }
1002             currentCluster = null;
1003         }
1004
1005         @Override
1006         public void flushCluster(Resource r) {
1007             throw new UnsupportedOperationException("flushCluster resource " + r);
1008         }
1009
1010         @Override
1011         public void gc() {
1012         }
1013
1014         @Override
1015         public boolean removeStatement(VirtualGraph provider, Resource subject, Resource predicate,
1016                 Resource object) {
1017
1018                 int s = ((ResourceImpl)subject).id;
1019                 int p = ((ResourceImpl)predicate).id;
1020                 int o = ((ResourceImpl)object).id;
1021
1022                 provider = getProvider(provider);
1023                 if (null == provider) {
1024
1025                         ClusterI cluster = clusterTable.getClusterProxyByResourceKey(s);
1026                         clusterTable.writeOnlyInvalidate(cluster);
1027
1028                         try {
1029
1030                                 int sResourceKey = ClusterTraits.getResourceIndexFromResourceKey(s);
1031
1032                                 ClusterI pc = clusterTable.getClusterProxyByResourceKey(p);
1033                                 ClusterI oc = clusterTable.getClusterProxyByResourceKey(o);
1034
1035                                 clusterTranslator.addStatementIndex(cluster, sResourceKey, cluster.getClusterUID(), ClusterChange.REMOVE_OPERATION);
1036                                 clusterTranslator.addStatementIndex(cluster, p, pc.getClusterUID(), ClusterStream.NULL_OPERATION);
1037                                 clusterTranslator.addStatementIndex(cluster, o, oc.getClusterUID(), ClusterStream.NULL_OPERATION);
1038                                 clusterTranslator.removeStatement(cluster);
1039
1040                                 queryProvider2.invalidateResource(s);
1041                                 clientChanges.invalidate(s);
1042
1043                         } catch (DatabaseException e) {
1044
1045                                 Logger.defaultLogError(e);
1046
1047                         }
1048
1049                         return true;
1050
1051                 } else {
1052
1053                         ((VirtualGraphImpl)provider).deny(s, p, o);
1054                         queryProvider2.invalidateResource(s);
1055                         clientChanges.invalidate(s);
1056
1057                         return true;
1058
1059                 }
1060
1061         }
1062
1063         @Override
1064         public void setValue(VirtualGraph provider, Resource resource, byte[] value) {
1065             throw new UnsupportedOperationException();
1066         }
1067
1068         @Override
1069         public boolean writeOnly() {
1070             return true;
1071         }
1072
1073         @Override
1074         public void performWriteRequest(WriteGraph graph, Write request) throws DatabaseException {
1075             writeSupport.performWriteRequest(graph, request);
1076         }
1077
1078         @Override
1079         public <T> T performWriteRequest(WriteGraph graph, WriteResult<T> request) throws DatabaseException {
1080             throw new UnsupportedOperationException();
1081         }
1082
1083         @Override
1084         public void performWriteRequest(WriteGraph graph, WriteOnly request) throws DatabaseException {
1085             throw new UnsupportedOperationException();
1086         }
1087
1088         @Override
1089         public <T> void addMetadata(Metadata data) throws ServiceException {
1090             writeSupport.addMetadata(data);
1091         }
1092
1093         @Override
1094         public <T extends Metadata> T getMetadata(Class<T> clazz) throws ServiceException {
1095             return writeSupport.getMetadata(clazz);
1096         }
1097
1098         @Override
1099         public TreeMap<String, byte[]> getMetadata() {
1100             return writeSupport.getMetadata();
1101         }
1102
1103         @Override
1104         public void commitDone(WriteTraits writeTraits, long csid) {
1105             writeSupport.commitDone(writeTraits, csid);
1106         }
1107
1108         @Override
1109         public void clearUndoList(WriteTraits writeTraits) {
1110             writeSupport.clearUndoList(writeTraits);
1111         }
1112         @Override
1113         public int clearMetadata() {
1114             return writeSupport.clearMetadata();
1115         }
1116
1117                 @Override
1118                 public void startUndo() {
1119                         writeSupport.startUndo();
1120                 }
1121     }
1122
1123     class VirtualWriteOnlySupport implements WriteSupport {
1124
1125 //        @Override
1126 //        public void addStatement(VirtualGraph provider, Resource subject, Resource predicate,
1127 //                Resource object) {
1128 //            throw new UnsupportedOperationException();
1129 //        }
1130
1131         @Override
1132         public void claim(VirtualGraph provider, Resource subject, Resource predicate, Resource object) {
1133
1134             TransientGraph impl = (TransientGraph)provider;
1135             impl.claim(querySupport.getId(subject), querySupport.getId(predicate), querySupport.getId(object));
1136             getQueryProvider2().updateStatements(querySupport.getId(subject), querySupport.getId(predicate));
1137             clientChanges.claim(subject, predicate, object);
1138
1139         }
1140
1141         @Override
1142         public void claim(VirtualGraph provider, int subject, int predicate, int object) {
1143
1144             TransientGraph impl = (TransientGraph)provider;
1145             impl.claim(subject, predicate, object);
1146             getQueryProvider2().updateStatements(subject, predicate);
1147             clientChanges.claim(subject, predicate, object);
1148
1149         }
1150
1151         @Override
1152         public void claimValue(VirtualGraph provider, Resource resource, byte[] value) throws DatabaseException {
1153             claimValue(provider, ((ResourceImpl)resource).id, value, value.length);
1154         }
1155
1156         @Override
1157         public void claimValue(VirtualGraph provider, int resource, byte[] value, int length) {
1158             ((VirtualGraphImpl)provider).claimValue(resource, value, length);
1159             getQueryProvider2().updateValue(resource);
1160             clientChanges.claimValue(resource);
1161         }
1162
1163         @Override
1164         public void claimValue(VirtualGraph provider, Resource resource, ByteReader reader, int amount) throws DatabaseException {
1165             byte[] value = reader.readBytes(null, amount);
1166             claimValue(provider, resource, value);
1167         }
1168
1169         @Override
1170         public Resource createResource(VirtualGraph provider) {
1171             TransientGraph impl = (TransientGraph)provider;
1172             return impl.getResource(impl.newResource(false));
1173         }
1174
1175         @Override
1176         public Resource createResource(VirtualGraph provider, long clusterId) {
1177             throw new UnsupportedOperationException();
1178         }
1179
1180         @Override
1181         public Resource createResource(VirtualGraph provider, Resource clusterSet)
1182         throws DatabaseException {
1183             throw new UnsupportedOperationException();
1184         }
1185
1186         @Override
1187         public void createClusterSet(VirtualGraph provider, Resource clusterSet)
1188         throws DatabaseException {
1189             throw new UnsupportedOperationException();
1190         }
1191
1192         @Override
1193         public boolean hasClusterSet(VirtualGraph provider, Resource clusterSet)
1194         throws ServiceException {
1195             throw new UnsupportedOperationException();
1196         }
1197
1198         @Override
1199         public Resource setDefaultClusterSet(Resource clusterSet)
1200         throws ServiceException {
1201                 return null;
1202         }
1203
1204         @Override
1205         public void denyValue(VirtualGraph provider, Resource resource) {
1206             ((VirtualGraphImpl)provider).denyValue(((ResourceImpl) resource).id);
1207             getQueryProvider2().updateValue(querySupport.getId(resource));
1208             // NOTE: this only keeps track of value changes by-resource.
1209             clientChanges.claimValue(resource);
1210         }
1211
1212         @Override
1213         public void flush(boolean intermediate) {
1214             throw new UnsupportedOperationException();
1215         }
1216
1217         @Override
1218         public void flushCluster() {
1219             throw new UnsupportedOperationException();
1220         }
1221
1222         @Override
1223         public void flushCluster(Resource r) {
1224             throw new UnsupportedOperationException("Resource " + r);
1225         }
1226
1227         @Override
1228         public void gc() {
1229         }
1230
1231         @Override
1232         public boolean removeStatement(VirtualGraph provider, Resource subject, Resource predicate, Resource object) {
1233             TransientGraph impl = (TransientGraph) provider;
1234             impl.deny(querySupport.getId(subject), querySupport.getId(predicate), querySupport.getId(object));
1235             getQueryProvider2().updateStatements(querySupport.getId(subject), querySupport.getId(predicate));
1236             clientChanges.deny(subject, predicate, object);
1237             return true;
1238         }
1239
1240         @Override
1241         public void setValue(VirtualGraph provider, Resource resource, byte[] value) {
1242             throw new UnsupportedOperationException();
1243         }
1244
1245         @Override
1246         public boolean writeOnly() {
1247             return true;
1248         }
1249
1250         @Override
1251         public void performWriteRequest(WriteGraph graph, Write request) throws DatabaseException {
1252             throw new UnsupportedOperationException();
1253         }
1254
1255         @Override
1256         public <T> T performWriteRequest(WriteGraph graph, WriteResult<T> request) throws DatabaseException {
1257             throw new UnsupportedOperationException();
1258         }
1259
1260         @Override
1261         public void performWriteRequest(WriteGraph graph, WriteOnly request) throws DatabaseException {
1262             throw new UnsupportedOperationException();
1263         }
1264
1265         @Override
1266         public <T> void addMetadata(Metadata data) throws ServiceException {
1267             throw new UnsupportedOperationException();
1268         }
1269
1270         @Override
1271         public <T extends Metadata> T getMetadata(Class<T> clazz) throws ServiceException {
1272             throw new UnsupportedOperationException();
1273         }
1274
1275         @Override
1276         public TreeMap<String, byte[]> getMetadata() {
1277             throw new UnsupportedOperationException();
1278         }
1279
1280         @Override
1281         public void commitDone(WriteTraits writeTraits, long csid) {
1282         }
1283
1284         @Override
1285         public void clearUndoList(WriteTraits writeTraits) {
1286         }
1287
1288         @Override
1289         public int clearMetadata() {
1290             return 0;
1291         }
1292
1293                 @Override
1294                 public void startUndo() {
1295                 }
1296     }
1297
1298     private <T> void performWriteOnly(WriteOnlyResult<T> request, Semaphore notify, Procedure<T> callback) {
1299
1300         try {
1301
1302             //int thread = request.hashCode() & queryProvider2.THREAD_MASK;
1303
1304             fireSessionVariableChange(SessionVariables.QUEUED_WRITES);
1305
1306             flushCounter = 0;
1307             Disposable.safeDispose(clientChanges);
1308             clientChanges = new ClientChangesImpl(SessionImplSocket.this);
1309
1310             acquireWriteOnly();
1311
1312             WriteSupport writeSupport = request.getProvider() != null ? new VirtualWriteOnlySupport() : new WriteOnlySupport();
1313
1314             WriteGraphImpl writer = WriteGraphImpl.create(getQueryProvider2(), writeSupport, request.getProvider());
1315
1316             WriteState<T> writeStateT = new WriteState<T>(writer, request, notify, callback);
1317             writeState = writeStateT;
1318
1319             assert (null != writer);
1320 //            writer.state.barrier.inc();
1321             long start = System.nanoTime();
1322             T result = request.perform(writer);
1323             long duration = System.nanoTime() - start;
1324             if (DEBUG) {
1325                 System.err.println("################");
1326                 System.err.println("WriteOnly duration " + 1e-9*duration);
1327             }
1328             writeStateT.setResult(result);
1329
1330             // This makes clusters available from server
1331             clusterStream.reallyFlush();
1332
1333             // This will trigger query updates
1334             releaseWriteOnly(writer);
1335             assert (null != writer);
1336
1337             handleUpdatesAndMetadata(writer);
1338
1339         } catch (CancelTransactionException e) {
1340
1341             releaseWriteOnly(writeState.getGraph());
1342
1343             clusterTable.removeWriteOnlyClusters();
1344             state.stopWriteTransaction(clusterStream);
1345
1346         } catch (Throwable e) {
1347
1348             e.printStackTrace();
1349
1350             releaseWriteOnly(writeState.getGraph());
1351
1352             clusterTable.removeWriteOnlyClusters();
1353
1354             if (callback != null)
1355                 callback.exception(new DatabaseException(e));
1356
1357             state.stopWriteTransaction(clusterStream);
1358             Logger.defaultLogError("Write transaction caused an unexpected error, see exception.", e);
1359
1360         } finally  {
1361
1362             fireSessionVariableChange(SessionVariables.QUEUED_WRITES);
1363
1364         }
1365
1366     }
1367
1368     public <T> void scheduleRequest(final WriteOnly request, final Consumer<DatabaseException> callback, final Semaphore notify) {
1369         scheduleRequest(request, callback, notify, null);
1370     }
1371
1372     @Override
1373     public <T> void scheduleRequest(final WriteOnly request, final Consumer<DatabaseException> callback, final Semaphore notify, Boolean combine) {
1374
1375         assertAlive();
1376
1377         assert (request != null);
1378
1379         requestManager.scheduleWrite(new SessionTask(null) {
1380
1381             @Override
1382             public void run0(int thread) {
1383
1384                 ITask task = ThreadLogger.getInstance().begin("WriteRequest " + request);
1385
1386                     try {
1387
1388                         fireSessionVariableChange(SessionVariables.QUEUED_WRITES);
1389
1390                         flushCounter = 0;
1391                         Disposable.safeDispose(clientChanges);
1392                         clientChanges = new ClientChangesImpl(SessionImplSocket.this);
1393
1394                         acquireWriteOnly();
1395
1396                         VirtualGraph vg = getProvider(request.getProvider());
1397                         WriteSupport writeSupport = vg != null ? new VirtualWriteOnlySupport() : new WriteOnlySupport();
1398
1399                         WriteGraphImpl writer = WriteGraphImpl.create(getQueryProvider2(), writeSupport, vg);
1400
1401                         writeState = new WriteState<Object>(writer, request, notify, new Procedure<Object>() {
1402
1403                             @Override
1404                             public void execute(Object result) {
1405                                 if(callback != null) callback.accept(null);
1406                             }
1407
1408                             @Override
1409                             public void exception(Throwable t) {
1410                                 if(callback != null) callback.accept((DatabaseException)t);
1411                             }
1412
1413                         });
1414
1415                         assert (null != writer);
1416 //                        writer.state.barrier.inc();
1417
1418                         try {
1419
1420                             request.perform(writer);
1421
1422                         } catch (Throwable e) {
1423
1424 //                            writer.state.barrier.dec();
1425 //                            writer.waitAsync(null);
1426
1427                             releaseWriteOnly(writer);
1428
1429                             clusterTable.removeWriteOnlyClusters();
1430
1431                             if(!(e instanceof CancelTransactionException)) {
1432                                 if (callback != null)
1433                                     callback.accept(new DatabaseException(e));
1434                             }
1435
1436                             writeState.except(e);
1437
1438                             return;
1439
1440                         }
1441
1442                         // This makes clusters available from server
1443                         boolean empty = clusterStream.reallyFlush();
1444                         // This was needed to make WO requests call metadata listeners.
1445                         // NOTE: the calling event does not contain clientChanges information.
1446                         if (!empty && clientChanges.isEmpty())
1447                             clientChanges.setNotEmpty(true);
1448                         releaseWriteOnly(writer);
1449                         assert (null != writer);
1450                     } finally  {
1451
1452                         fireSessionVariableChange(SessionVariables.QUEUED_WRITES);
1453
1454                     }
1455
1456
1457                 task.finish();
1458
1459             }
1460
1461         }, combine);
1462
1463     }
1464
1465     public <T> void scheduleRequest(final WriteOnlyResult<T> request, final Procedure<T> callback, final Semaphore notify) {
1466         scheduleRequest(request, callback, notify, null);
1467     }
1468
1469     /* (non-Javadoc)
1470      * @see org.simantics.db.impl.support.WriteRequestScheduleSupport#scheduleRequest(org.simantics.db.request.WriteOnlyResult, org.simantics.db.procedure.Procedure, java.util.concurrent.Semaphore, java.lang.Boolean)
1471      */
1472     @Override
1473     public <T> void scheduleRequest(final WriteOnlyResult<T> request, final Procedure<T> callback, final Semaphore notify, Boolean combine) {
1474
1475         assert (request != null);
1476
1477         requestManager.scheduleWrite(new SessionTask(null) {
1478
1479             @Override
1480             public void run0(int thread) {
1481
1482                 ITask task = ThreadLogger.getInstance().begin("WriteRequest " + request);
1483
1484                 performWriteOnly(request, notify, callback);
1485
1486                 task.finish();
1487
1488             }
1489
1490         }, combine);
1491
1492     }
1493
1494     public <T> void scheduleRequest(final Read<T> request, final AsyncProcedure<T> procedure, final Semaphore notify, final DataContainer<Throwable> throwable, final DataContainer<T> result) {
1495
1496         assert (request != null);
1497         assert (procedure != null);
1498
1499         //int thread = request.hashCode() & queryProvider2.THREAD_MASK;
1500
1501         requestManager.scheduleRead(new SessionRead(throwable, notify) {
1502
1503             @Override
1504             public void run0(int thread) {
1505
1506                 fireSessionVariableChange(SessionVariables.QUEUED_READS);
1507
1508                 ListenerBase listener = getListenerBase(procedure);
1509
1510                 final ReadGraphImpl newGraph = ReadGraphImpl.create(getQueryProvider2());
1511
1512                 // This is never synced but increase to prevent it from visiting 0
1513                 newGraph.asyncBarrier.inc();
1514
1515                 try {
1516
1517                     if (listener != null) {
1518
1519                         try {
1520
1521                                 AsyncProcedure ap = new AsyncProcedure<T>() {
1522
1523                                         @Override
1524                                         public void exception(AsyncReadGraph graph, Throwable t) {
1525                                                 procedure.exception(graph, t);
1526                                                 if(throwable != null) {
1527                                                         throwable.set(t);
1528                                                 } else {
1529                                                         //                                    ErrorLogger.defaultLogError("Unhandled exception", t);
1530                                                 }
1531                                         }
1532
1533                                         @Override
1534                                         public void execute(AsyncReadGraph graph, T t) {
1535                                                 if(result != null) result.set(t);
1536                                                 procedure.execute(graph, t);
1537                                         }
1538
1539                                 };
1540
1541                                 QueryCache.runnerReadEntry(newGraph, request, null, listener, ap, true);
1542
1543                         } catch (Throwable t) {
1544                             // This is handled by the AsyncProcedure
1545                                 //Logger.defaultLogError("Internal error", t);
1546                         }
1547
1548                     } else {
1549
1550                         try {
1551
1552 //                            newGraph.state.barrier.inc();
1553
1554                             T t = request.perform(newGraph);
1555
1556                             try {
1557
1558                                 if(result != null) result.set(t);
1559                                 procedure.execute(newGraph, t);
1560
1561                             } catch (Throwable th) {
1562
1563                                 if(throwable != null) {
1564                                     throwable.set(th);
1565                                 } else {
1566                                     Logger.defaultLogError("Unhandled exception", th);
1567                                 }
1568
1569                             }
1570
1571                         } catch (Throwable t) {
1572
1573                             if (DEBUG)
1574                                 t.printStackTrace();
1575
1576                             if(throwable != null) {
1577                                 throwable.set(t);
1578                             } else {
1579                                 Logger.defaultLogError("Unhandled exception", t);
1580                             }
1581
1582                             try {
1583
1584                                 procedure.exception(newGraph, t);
1585
1586                             } catch (Throwable t2) {
1587
1588                                 if(throwable != null) {
1589                                     throwable.set(t2);
1590                                 } else {
1591                                     Logger.defaultLogError("Unhandled exception", t2);
1592                                 }
1593
1594                             }
1595
1596                         }
1597
1598 //                        newGraph.state.barrier.dec();
1599 //                        newGraph.waitAsync(request);
1600
1601                     }
1602
1603                 } finally {
1604
1605                     fireSessionVariableChange(SessionVariables.QUEUED_READS);
1606
1607                 }
1608
1609             }
1610
1611         });
1612
1613     }
1614
1615     public <T> void scheduleRequest(final AsyncRead<T> request, final AsyncProcedure<T> procedure, final ListenerBase listener, final Semaphore notify) {
1616
1617         assert (request != null);
1618         assert (procedure != null);
1619
1620         int thread = request.hashCode() & queryProvider2.THREAD_MASK;
1621
1622         requestManager.scheduleRead(new SessionRead(null, notify) {
1623
1624             @Override
1625             public void run0(int thread) {
1626
1627                 fireSessionVariableChange(SessionVariables.QUEUED_READS);
1628
1629                 final ReadGraphImpl newGraph = ReadGraphImpl.create(getQueryProvider2());
1630
1631                 try {
1632
1633                     if (listener != null) {
1634
1635                         try {
1636                                 QueryCacheBase.resultAsyncReadEntry(newGraph, request, null, listener, procedure);
1637                                 //QueryCache.runnerAsyncReadEntry(newGraph, request, null, listener, procedure, true);
1638                                                         //newGraph.processor.query(newGraph, request, null, procedure, listener);
1639                                                 } catch (DatabaseException e) {
1640                                                         Logger.defaultLogError(e);
1641                                                 }
1642
1643                     } else {
1644
1645 //                        final ResultCallWrappedSingleQueryProcedure4<T> wrapper = new ResultCallWrappedSingleQueryProcedure4<T>(
1646 //                                procedure, "request");
1647
1648                         BlockingAsyncProcedure<T> wrap = new BlockingAsyncProcedure<T>(newGraph.asyncBarrier, newGraph, procedure, request);
1649
1650                         try {
1651
1652                             request.perform(newGraph, wrap);
1653                             wrap.get();
1654
1655                         } catch (DatabaseException e) {
1656
1657                                                         Logger.defaultLogError(e);
1658
1659                         }
1660
1661                     }
1662
1663                 } finally {
1664
1665                     fireSessionVariableChange(SessionVariables.QUEUED_READS);
1666
1667                 }
1668
1669             }
1670
1671         });
1672
1673     }
1674
1675     public <T> void scheduleRequest(final MultiRead<T> request, final SyncMultiProcedure<T> procedure, final Semaphore notify) {
1676
1677         assert (request != null);
1678         assert (procedure != null);
1679
1680         int thread = request.hashCode() & queryProvider2.THREAD_MASK;
1681
1682         int sync = notify != null ? thread : -1;
1683
1684         requestManager.scheduleRead(new SessionRead(null, notify) {
1685
1686             @Override
1687             public void run0(int thread) {
1688
1689                 fireSessionVariableChange(SessionVariables.QUEUED_READS);
1690
1691                 ListenerBase listener = getListenerBase(procedure);
1692
1693                 final ReadGraphImpl newGraph = ReadGraphImpl.create(getQueryProvider2());
1694
1695                 try {
1696
1697                     if (listener != null) {
1698
1699                         newGraph.processor.query(newGraph, request, null, procedure, listener);
1700
1701 //                        newGraph.waitAsync(request);
1702
1703                     } else {
1704
1705                         final ResultCallWrappedSyncQueryProcedure<T> wrapper = new ResultCallWrappedSyncQueryProcedure<T>(procedure);
1706
1707                         try {
1708
1709                             request.perform(newGraph, wrapper);
1710
1711                         } catch (Throwable t) {
1712
1713                             t.printStackTrace();
1714
1715                         }
1716
1717                     }
1718
1719                 } finally {
1720
1721                     fireSessionVariableChange(SessionVariables.QUEUED_READS);
1722
1723                 }
1724
1725             }
1726
1727         });
1728
1729     }
1730
1731     public <T> void scheduleRequest(final AsyncMultiRead<T> request, final AsyncMultiProcedure<T> procedure, final Semaphore notify) {
1732
1733         assert (request != null);
1734         assert (procedure != null);
1735
1736         int thread = request.hashCode() & queryProvider2.THREAD_MASK;
1737
1738         int sync = notify != null ? thread : -1;
1739
1740         requestManager.scheduleRead(new SessionRead(null, notify) {
1741
1742             @Override
1743             public void run0(int thread) {
1744
1745                 fireSessionVariableChange(SessionVariables.QUEUED_READS);
1746
1747                 ListenerBase listener = getListenerBase(procedure);
1748
1749                 final ReadGraphImpl newGraph = ReadGraphImpl.create(getQueryProvider2());
1750
1751                 try {
1752
1753                     if (listener != null) {
1754
1755                         newGraph.processor.query(newGraph, request, null, procedure, listener);
1756
1757 //                        newGraph.waitAsync(request);
1758
1759                     } else {
1760
1761                         final ResultCallWrappedQueryProcedure4<T> wrapper = new ResultCallWrappedQueryProcedure4<T>(procedure);
1762
1763                         try {
1764
1765                             request.perform(newGraph, wrapper);
1766
1767                         } catch (Throwable t) {
1768
1769                             t.printStackTrace();
1770
1771                         }
1772
1773                     }
1774
1775                 } finally {
1776
1777                     fireSessionVariableChange(SessionVariables.QUEUED_READS);
1778
1779                 }
1780
1781             }
1782
1783         });
1784
1785     }
1786
1787     public <T> void scheduleRequest(final ExternalRead<T> request, final Procedure<T> procedure, final Semaphore notify, final DataContainer<Throwable> throwable, final DataContainer<T> result) {
1788
1789         assert (request != null);
1790         assert (procedure != null);
1791
1792         int thread = request.hashCode() & queryProvider2.THREAD_MASK;
1793
1794         requestManager.scheduleRead(new SessionRead(throwable, notify) {
1795
1796             @Override
1797             public void run0(int thread) {
1798
1799                 fireSessionVariableChange(SessionVariables.QUEUED_READS);
1800
1801                 ListenerBase listener = getListenerBase(procedure);
1802
1803                 final ReadGraphImpl newGraph = ReadGraphImpl.create(getQueryProvider2());
1804
1805                 try {
1806
1807                     if (listener != null) {
1808
1809                         try {
1810                             QueryCacheBase.resultExternalReadEntry(newGraph, request, null, listener, procedure);
1811                         } catch (DatabaseException e) {
1812                             Logger.defaultLogError(e);
1813                         }
1814
1815                     } else {
1816
1817 //                        newGraph.state.barrier.inc();
1818
1819                         request.register(newGraph, new Listener<T>() {
1820
1821                             @Override
1822                             public void exception(Throwable t) {
1823                                 if(throwable != null) throwable.set(t);
1824                                 procedure.exception(t);
1825 //                                newGraph.state.barrier.dec();
1826                             }
1827
1828                             @Override
1829                             public void execute(T t) {
1830                                 if(result != null) result.set(t);
1831                                 procedure.execute(t);
1832 //                                newGraph.state.barrier.dec();
1833                             }
1834
1835                             @Override
1836                             public boolean isDisposed() {
1837                                 return true;
1838                             }
1839
1840                         });
1841
1842 //                        newGraph.waitAsync(request);
1843
1844                     }
1845
1846                 } finally {
1847
1848                     fireSessionVariableChange(SessionVariables.QUEUED_READS);
1849
1850                 }
1851
1852             }
1853
1854         });
1855
1856     }
1857
1858
1859     @Override
1860     public <T> void asyncRequest(final Read<T> request, final AsyncProcedure<T> procedure) {
1861
1862         scheduleRequest(request, procedure, null, null, null);
1863
1864     }
1865
1866     @Override
1867     public <T> T syncRequest(Read<T> request, AsyncProcedure<T> procedure) throws DatabaseException {
1868         assertNotSession();
1869         Semaphore notify = new Semaphore(0);
1870         DataContainer<Throwable> container = new DataContainer<Throwable>();
1871         DataContainer<T> result = new DataContainer<T>();
1872         scheduleRequest(request, procedure, notify, container, result);
1873         acquire(notify, request);
1874         Throwable throwable = container.get();
1875         if(throwable != null) {
1876             if(throwable instanceof DatabaseException) throw (DatabaseException)throwable;
1877             else throw new DatabaseException("Unexpected exception", throwable);
1878         }
1879         return result.get();
1880     }
1881
1882     @Override
1883     public <T> T syncRequest(AsyncRead<T> request, final AsyncProcedure<T> procedure) throws DatabaseException {
1884         assertNotSession();
1885         Semaphore notify = new Semaphore(0);
1886         final DataContainer<Throwable> exceptionContainer = new DataContainer<Throwable>();
1887         final DataContainer<T> resultContainer = new DataContainer<T>();
1888         scheduleRequest(request, new AsyncProcedure<T>() {
1889             @Override
1890             public void exception(AsyncReadGraph graph, Throwable throwable) {
1891                 exceptionContainer.set(throwable);
1892                 procedure.exception(graph, throwable);
1893             }
1894             @Override
1895             public void execute(AsyncReadGraph graph, T result) {
1896                 resultContainer.set(result);
1897                 procedure.execute(graph, result);
1898             }
1899         }, getListenerBase(procedure), notify);
1900         acquire(notify, request, procedure);
1901         Throwable throwable = exceptionContainer.get();
1902         if (throwable != null) {
1903             if(throwable instanceof DatabaseException) throw (DatabaseException)throwable;
1904             else throw new DatabaseException("Unexpected exception", throwable);
1905         }
1906         return resultContainer.get();
1907     }
1908
1909
1910     @Override
1911     public <T> Collection<T> syncRequest(AsyncMultiRead<T> request, final AsyncMultiProcedure<T> procedure) throws DatabaseException {
1912         assertNotSession();
1913         Semaphore notify = new Semaphore(0);
1914         final DataContainer<Throwable> exceptionContainer = new DataContainer<Throwable>();
1915         scheduleRequest(request, new AsyncMultiProcedure<T>() {
1916             @Override
1917             public void exception(AsyncReadGraph graph, Throwable throwable) {
1918                 exceptionContainer.set(throwable);
1919                 procedure.exception(graph, throwable);
1920             }
1921             @Override
1922             public void execute(AsyncReadGraph graph, T result) {
1923                 procedure.execute(graph, result);
1924             }
1925             @Override
1926             public void finished(AsyncReadGraph graph) {
1927                 procedure.finished(graph);
1928             }
1929         }, notify);
1930         acquire(notify, request, procedure);
1931         Throwable throwable = exceptionContainer.get();
1932         if (throwable != null) {
1933             if(throwable instanceof DatabaseException) throw (DatabaseException)throwable;
1934             else throw new DatabaseException("Unexpected exception", throwable);
1935         }
1936         // TODO: implement return value
1937         System.err.println("TODO: implement return value for syncRequest(AsyncMultiRead, AsyncMultiProcedure)");
1938         return null;
1939     }
1940
1941     @Override
1942     public <T> T syncRequest(final ExternalRead<T> request) throws DatabaseException {
1943         return syncRequest(request, new ProcedureAdapter<T>());
1944
1945
1946 //        assert(request != null);
1947 //
1948 //        final DataContainer<T> result = new DataContainer<T>();
1949 //        final DataContainer<Throwable> exception = new DataContainer<Throwable>();
1950 //
1951 //        syncRequest(request, new Procedure<T>() {
1952 //
1953 //            @Override
1954 //            public void execute(T t) {
1955 //                result.set(t);
1956 //            }
1957 //
1958 //            @Override
1959 //            public void exception(Throwable t) {
1960 //                exception.set(t);
1961 //            }
1962 //
1963 //        });
1964 //
1965 //        Throwable t = exception.get();
1966 //        if(t != null) {
1967 //            if(t instanceof DatabaseException) throw (DatabaseException)t;
1968 //            else throw new DatabaseException("Unexpected exception in ReadGraph.syncRequest(PrimitiveRead)", t);
1969 //        }
1970 //
1971 //        return result.get();
1972
1973     }
1974
1975     @Override
1976     public <T> T syncRequest(ExternalRead<T> request, final Listener<T> procedure) throws DatabaseException {
1977         return syncRequest(request, (Procedure<T>)procedure);
1978     }
1979
1980
1981 //    @Override
1982 //    public <T> T syncRequest(Read<T> request, AsyncProcedure<T> procedure) throws DatabaseException {
1983 //        assertNotSession();
1984 //        Semaphore notify = new Semaphore(0);
1985 //        DataContainer<Throwable> container = new DataContainer<Throwable>();
1986 //        DataContainer<T> result = new DataContainer<T>();
1987 //        scheduleRequest(request, procedure, notify, container, result);
1988 //        acquire(notify, request);
1989 //        Throwable throwable = container.get();
1990 //        if(throwable != null) {
1991 //            if(throwable instanceof DatabaseException) throw (DatabaseException)throwable;
1992 //            else throw new DatabaseException("Unexpected exception", throwable);
1993 //        }
1994 //        return result.get();
1995 //    }
1996
1997
1998     @Override
1999     public <T> T syncRequest(ExternalRead<T> request, final Procedure<T> procedure) throws DatabaseException {
2000         assertNotSession();
2001         Semaphore notify = new Semaphore(0);
2002         final DataContainer<Throwable> container = new DataContainer<Throwable>();
2003         final DataContainer<T> result = new DataContainer<T>();
2004         scheduleRequest(request, procedure, notify, container, result);
2005         acquire(notify, request);
2006         Throwable throwable = container.get();
2007         if (throwable != null) {
2008             if(throwable instanceof DatabaseException) throw (DatabaseException)throwable;
2009             else throw new DatabaseException("Unexpected exception", throwable);
2010         }
2011         return result.get();
2012     }
2013
2014     @Override
2015     public void syncRequest(Write request) throws DatabaseException  {
2016         assertNotSession();
2017         assertAlive();
2018         Semaphore notify = new Semaphore(0);
2019         final DataContainer<DatabaseException> exception = new DataContainer<DatabaseException>();
2020         scheduleRequest(request, e -> exception.set(e), notify);
2021         acquire(notify, request);
2022         if(exception.get() != null) throw exception.get();
2023     }
2024
2025     @Override
2026     public <T> T syncRequest(WriteResult<T> request) throws DatabaseException  {
2027         assertNotSession();
2028         Semaphore notify = new Semaphore(0);
2029         final DataContainer<Throwable> exception = new DataContainer<Throwable>();
2030         final DataContainer<T> result = new DataContainer<T>();
2031         scheduleRequest(request, new Procedure<T>() {
2032
2033             @Override
2034             public void exception(Throwable t) {
2035               exception.set(t);
2036             }
2037
2038             @Override
2039             public void execute(T t) {
2040               result.set(t);
2041             }
2042
2043         }, notify);
2044         acquire(notify, request);
2045         if(exception.get() != null) {
2046             Throwable t = exception.get();
2047             if(t instanceof DatabaseException) throw (DatabaseException)t;
2048             else throw new DatabaseException(t);
2049         }
2050         return result.get();
2051     }
2052
2053     @Override
2054     public <T> T syncRequest(WriteOnlyResult<T> request) throws DatabaseException  {
2055         assertNotSession();
2056         Semaphore notify = new Semaphore(0);
2057         final DataContainer<Throwable> exception = new DataContainer<Throwable>();
2058         final DataContainer<T> result = new DataContainer<T>();
2059         scheduleRequest(request, new Procedure<T>() {
2060
2061             @Override
2062             public void exception(Throwable t) {
2063               exception.set(t);
2064             }
2065
2066             @Override
2067             public void execute(T t) {
2068               result.set(t);
2069             }
2070
2071         }, notify);
2072         acquire(notify, request);
2073         if(exception.get() != null) {
2074             Throwable t = exception.get();
2075             if(t instanceof DatabaseException) throw (DatabaseException)t;
2076             else throw new DatabaseException(t);
2077         }
2078         return result.get();
2079     }
2080
2081     @Override
2082     public <T> T syncRequest(DelayedWriteResult<T> request) throws DatabaseException  {
2083         assertNotSession();
2084         Semaphore notify = new Semaphore(0);
2085         final DataContainer<Throwable> exception = new DataContainer<Throwable>();
2086         final DataContainer<T> result = new DataContainer<T>();
2087         scheduleRequest(request, new Procedure<T>() {
2088
2089             @Override
2090             public void exception(Throwable t) {
2091               exception.set(t);
2092             }
2093
2094             @Override
2095             public void execute(T t) {
2096               result.set(t);
2097             }
2098
2099         }, notify);
2100         acquire(notify, request);
2101         if(exception.get() != null) {
2102             Throwable t = exception.get();
2103             if(t instanceof DatabaseException) throw (DatabaseException)t;
2104             else throw new DatabaseException(t);
2105         }
2106         return result.get();
2107     }
2108
2109     @Override
2110     public void syncRequest(DelayedWrite request) throws DatabaseException  {
2111         assertNotSession();
2112         Semaphore notify = new Semaphore(0);
2113         final DataContainer<DatabaseException> exception = new DataContainer<DatabaseException>();
2114         scheduleRequest(request, e -> exception.set(e), notify);
2115         acquire(notify, request);
2116         if(exception.get() != null) throw exception.get();
2117     }
2118
2119     @Override
2120     public void syncRequest(WriteOnly request) throws DatabaseException  {
2121         assertNotSession();
2122         assertAlive();
2123         Semaphore notify = new Semaphore(0);
2124         final DataContainer<DatabaseException> exception = new DataContainer<DatabaseException>();
2125         scheduleRequest(request, e -> exception.set(e), notify);
2126         acquire(notify, request);
2127         if(exception.get() != null) throw exception.get();
2128     }
2129
2130     /*
2131      *
2132      * GraphRequestProcessor interface
2133      */
2134
2135     @Override
2136     public <T> void asyncRequest(final AsyncRead<T> request, final AsyncProcedure<T> procedure) {
2137
2138         scheduleRequest(request, procedure, null, null);
2139
2140     }
2141
2142     @Override
2143     public <T> void asyncRequest(final AsyncMultiRead<T> request, final AsyncMultiProcedure<T> procedure) {
2144
2145         scheduleRequest(request, procedure, null);
2146
2147     }
2148
2149     @Override
2150     public <T> void asyncRequest(final ExternalRead<T> request, final Procedure<T> procedure) {
2151
2152         scheduleRequest(request, procedure, null, null, null);
2153
2154     }
2155
2156     @Override
2157     public void asyncRequest(final Write request, final Consumer<DatabaseException> callback) {
2158
2159         scheduleRequest(request, callback, null);
2160
2161     }
2162
2163     @Override
2164     public <T> void asyncRequest(final WriteResult<T> request, final Procedure<T> procedure) {
2165
2166         scheduleRequest(request, procedure, null);
2167
2168     }
2169
2170     @Override
2171     public <T> void asyncRequest(final WriteOnlyResult<T> request, final Procedure<T> procedure) {
2172
2173         scheduleRequest(request, procedure, null);
2174
2175     }
2176
2177     @Override
2178     public <T> void asyncRequest(final DelayedWriteResult<T> request, final Procedure<T> procedure) {
2179
2180         scheduleRequest(request, procedure, null);
2181
2182     }
2183
2184     @Override
2185     public void asyncRequest(final DelayedWrite request, final Consumer<DatabaseException> callback) {
2186
2187         scheduleRequest(request, callback, null);
2188
2189     }
2190
2191     @Override
2192     public void asyncRequest(final Write r) {
2193         asyncRequest(r, null);
2194     }
2195
2196     @Override
2197     public void asyncRequest(final DelayedWrite r) {
2198         asyncRequest(r, null);
2199     }
2200
2201     @Override
2202     public void asyncRequest(final WriteOnly request, final Consumer<DatabaseException> callback) {
2203
2204         scheduleRequest(request, callback, null);
2205
2206     }
2207
2208     @Override
2209     public void asyncRequest(final WriteOnly request) {
2210
2211         asyncRequest(request, null);
2212
2213     }
2214
2215     @Override
2216     public <T> void async(ReadInterface<T> r, AsyncProcedure<T> procedure) {
2217         r.request(this, procedure);
2218     }
2219
2220     @Override
2221     public <T> void async(ReadInterface<T> r, Procedure<T> procedure) {
2222         r.request(this, procedure);
2223     }
2224
2225     @Override
2226     public <T> void async(ReadInterface<T> r, SyncProcedure<T> procedure) {
2227         r.request(this, procedure);
2228     }
2229
2230     @Override
2231     public <T> void async(ReadInterface<T> r, AsyncListener<T> procedure) {
2232         r.request(this, procedure);
2233     }
2234
2235     @Override
2236     public <T> void async(ReadInterface<T> r, Listener<T> procedure) {
2237         r.request(this, procedure);
2238     }
2239
2240     @Override
2241     public <T> void async(ReadInterface<T> r, SyncListener<T> procedure) {
2242         r.request(this, procedure);
2243     }
2244
2245     @Override
2246     public <T> T sync(ReadInterface<T> r) throws DatabaseException {
2247         return r.request(this);
2248     }
2249
2250     @Override
2251     public <T> T sync(WriteInterface<T> r) throws DatabaseException {
2252         return r.request(this);
2253     }
2254
2255     @Override
2256     public <T> void async(WriteInterface<T> r, Procedure<T> procedure) {
2257         r.request(this, procedure);
2258     }
2259
2260     @Override
2261     public <T> void async(WriteInterface<T> r) {
2262         r.request(this, new ProcedureAdapter<T>());
2263     }
2264
2265     //@Override
2266     public void incAsync() {
2267         state.incAsync();
2268     }
2269
2270     //@Override
2271     public void decAsync() {
2272         state.decAsync();
2273     }
2274
2275     public long getCluster(ResourceImpl resource) {
2276         ClusterI cluster = clusterTable.getClusterByResourceKey(resource.id);
2277         return cluster.getClusterId();
2278     }
2279
2280     public long getCluster(int id) {
2281         if (clusterTable == null)
2282             System.out.println("SessionImplSocket.getCluster() clusterTable == null !!!!! how come");
2283         return clusterTable.getClusterIdByResourceKeyNoThrow(id);
2284     }
2285
2286     public ResourceImpl getResource(int id) {
2287         return new ResourceImpl(resourceSupport, id);
2288     }
2289
2290     public ResourceImpl getResource(int resourceIndex, long clusterId) {
2291         assert (!ClusterTraitsBase.isIllegalResourceIndex(resourceIndex));
2292         ClusterI proxy = clusterTable.getClusterByClusterId(clusterId);
2293         int key = proxy.getClusterKey();
2294         int resourceKey = ClusterTraitsBase.createResourceKeyNoThrow(key, resourceIndex);
2295         return new ResourceImpl(resourceSupport, resourceKey);
2296     }
2297
2298     public ResourceImpl getResource2(int id) {
2299         assert (id != 0);
2300         return new ResourceImpl(resourceSupport, id);
2301     }
2302
2303     final public int getId(ResourceImpl impl) {
2304         return impl.id;
2305     }
2306
2307     public static final Charset UTF8 = Charset.forName("utf-8");
2308
2309
2310
2311
2312     static boolean areVirtualStatementsLoaded(final VirtualGraphServerSupportImpl support, int subject) {
2313         for(TransientGraph g : support.providers) {
2314             if(g.isPending(subject)) return false;
2315         }
2316         return true;
2317     }
2318
2319     static boolean areVirtualStatementsLoaded(final VirtualGraphServerSupportImpl support, int subject, int predicate) {
2320         for(TransientGraph g : support.providers) {
2321             if(g.isPending(subject, predicate)) return false;
2322         }
2323         return true;
2324     }
2325
2326     static void loadVirtualStatements(final VirtualGraphServerSupportImpl support, ReadGraphImpl graph, int subject, final Consumer<ReadGraphImpl> runnable) {
2327
2328         Consumer<ReadGraphImpl> composite = new Consumer<ReadGraphImpl>() {
2329
2330             AtomicInteger ready = new AtomicInteger(support.providers.size() + 1);
2331
2332             @Override
2333             public void accept(ReadGraphImpl graph) {
2334                 if(ready.decrementAndGet() == 0) {
2335                     runnable.accept(graph);
2336                 }
2337             }
2338
2339         };
2340
2341         for(TransientGraph g : support.providers) {
2342             if(g.isPending(subject)) {
2343                 try {
2344                     g.load(graph, subject, composite);
2345                 } catch (DatabaseException e) {
2346                     e.printStackTrace();
2347                 }
2348             } else {
2349                 composite.accept(graph);
2350             }
2351         }
2352
2353         composite.accept(graph);
2354
2355     }
2356
2357     static  void loadVirtualStatements(final VirtualGraphServerSupportImpl support, ReadGraphImpl graph, int subject, int predicate, final Consumer<ReadGraphImpl> runnable) {
2358
2359         Consumer<ReadGraphImpl> composite = new Consumer<ReadGraphImpl>() {
2360
2361             AtomicInteger ready = new AtomicInteger(support.providers.size() + 1);
2362
2363             @Override
2364             public void accept(ReadGraphImpl graph) {
2365                 if(ready.decrementAndGet() == 0) {
2366                     runnable.accept(graph);
2367                 }
2368             }
2369
2370         };
2371
2372         for(TransientGraph g : support.providers) {
2373             if(g.isPending(subject, predicate)) {
2374                 try {
2375                     g.load(graph, subject, predicate, composite);
2376                 } catch (DatabaseException e) {
2377                     e.printStackTrace();
2378                 }
2379             } else {
2380                 composite.accept(graph);
2381             }
2382         }
2383
2384         composite.accept(graph);
2385
2386     }
2387
2388 //    void dumpHeap() {
2389 //
2390 //        try {
2391 //            ManagementFactory.newPlatformMXBeanProxy(ManagementFactory.getPlatformMBeanServer(),
2392 //                    "com.sun.management:type=HotSpotDiagnostic", HotSpotDiagnosticMXBean.class).dumpHeap(
2393 //                    "d:/heap" + flushCounter + ".txt", true);
2394 //        } catch (IOException e) {
2395 //            e.printStackTrace();
2396 //        }
2397 //
2398 //    }
2399
2400     void fireReactionsToSynchronize(ChangeSet cs) {
2401
2402         // Do not fire empty events
2403         if (cs.isEmpty())
2404             return;
2405
2406         ReadGraphImpl g = ReadGraphImpl.create(getQueryProvider2());
2407 //        g.state.barrier.inc();
2408
2409         try {
2410
2411             if (!cs.isEmpty()) {
2412                 ChangeEvent e2 = new ChangeEvent(g.getSession(), g, null, cs);
2413                 for (ChangeListener l : changeListeners2) {
2414                     try {
2415                         l.graphChanged(e2);
2416                     } catch (Exception ex) {
2417                         ex.printStackTrace();
2418                     }
2419                 }
2420             }
2421
2422         } finally {
2423
2424 //            g.state.barrier.dec();
2425 //            g.waitAsync(null);
2426
2427         }
2428
2429     }
2430
2431     void fireReactionsToCommit(ReadGraphImpl graph, ChangeSet cs2) {
2432         try {
2433
2434             // Do not fire empty events
2435             if (cs2.isEmpty())
2436                 return;
2437
2438 //            graph.restart();
2439 //            graph.state.barrier.inc();
2440
2441             try {
2442
2443                 if (!cs2.isEmpty()) {
2444
2445                     ChangeEvent e2 = new ChangeEvent(graph.getSession(), graph, null, cs2);
2446                     for (ChangeListener l : changeListeners2) {
2447                         try {
2448                             l.graphChanged(e2);
2449 //                            System.out.println("changelistener " + l);
2450                         } catch (Exception ex) {
2451                             ex.printStackTrace();
2452                         }
2453                     }
2454
2455                 }
2456
2457             } finally {
2458
2459 //                graph.state.barrier.dec();
2460 //                graph.waitAsync(null);
2461
2462             }
2463         } catch (Throwable t) {
2464             t.printStackTrace();
2465
2466         }
2467     }
2468     void fireMetadataListeners(WriteGraphImpl graph, ChangeSet cs2) {
2469
2470         try {
2471
2472             // Do not fire empty events
2473             if (cs2.isEmpty())
2474                 return;
2475
2476             // Do not fire on virtual requests
2477             if(graph.getProvider() != null)
2478                 return;
2479
2480             WriteGraphImpl reactionGraph = WriteGraphImpl.create(graph.processor, writeSupport, null);
2481
2482             try {
2483
2484                 ChangeEvent e2 = new ChangeEvent(graph.getSession(), reactionGraph, graph, cs2);
2485                 for (ChangeListener l : metadataListeners) {
2486                     try {
2487                         l.graphChanged(e2);
2488                     } catch (Throwable ex) {
2489                         LOGGER.error("Could not invoke listener {} with event {}", l, e2, ex);
2490                     }
2491                 }
2492
2493             } finally {
2494
2495             }
2496
2497         } catch (Throwable t) {
2498             LOGGER.error("Could not fire metadata listeners {} {}", graph, cs2, t);
2499         }
2500
2501     }
2502
2503
2504     /*
2505      * (non-Javadoc)
2506      *
2507      * @see org.simantics.db.ServiceLocator#getService(java.lang.Class)
2508      */
2509     @Override
2510     public <T> T getService(Class<T> api) {
2511         T t = peekService(api);
2512         if (t == null) {
2513             if (state.isClosed())
2514                 throw new ServiceNotFoundException(this, api, "Session has been shut down");
2515             throw new ServiceNotFoundException(this, api);
2516         }
2517         return t;
2518     }
2519
2520     /**
2521      * @return
2522      */
2523     protected abstract ServerInformation getCachedServerInformation();
2524
2525         private Class<?> serviceKey1 = null;
2526         private Class<?> serviceKey2 = null;
2527         private Object service1 = null;
2528         private Object service2 = null;
2529
2530     /*
2531      * (non-Javadoc)
2532      *
2533      * @see
2534      * fi.vtt.simantics.db.connection.ServiceLocator#peekService(java.lang.Class)
2535      */
2536     @SuppressWarnings("unchecked")
2537     @Override
2538     public synchronized <T> T peekService(Class<T> api) {
2539
2540                 if(serviceKey1 == api) {
2541                         return (T)service1;
2542                 } else if (serviceKey2 == api) {
2543                         // Promote this key
2544                         Object result = service2;
2545                         service2 = service1;
2546                         serviceKey2 = serviceKey1;
2547                         service1 = result;
2548                         serviceKey1 = api;
2549                         return (T)result;
2550                 }
2551
2552         if (Layer0.class == api)
2553                 return (T) L0;
2554         if (ServerInformation.class == api)
2555             return (T) getCachedServerInformation();
2556         else if (WriteGraphImpl.class == api)
2557             return (T) writeState.getGraph();
2558         else if (ClusterBuilder.class == api)
2559             return (T)new ClusterBuilderImpl(this, (WriteOnlySupport)writeState.getGraph().writeSupport);
2560         else if (ClusterBuilderFactory.class == api)
2561             return (T)new ClusterBuilderFactoryImpl(this);
2562
2563                 service2 = service1;
2564                 serviceKey2 = serviceKey1;
2565
2566         service1 = serviceLocator.peekService(api);
2567         serviceKey1 = api;
2568
2569         return (T)service1;
2570
2571     }
2572
2573     /*
2574      * (non-Javadoc)
2575      *
2576      * @see
2577      * fi.vtt.simantics.db.connection.ServiceLocator#hasService(java.lang.Class)
2578      */
2579     @Override
2580     public boolean hasService(Class<?> api) {
2581         return serviceLocator.hasService(api);
2582     }
2583
2584     /**
2585      * @param api the api that must be implemented by the specified service
2586      * @param service the service implementation
2587      */
2588     @Override
2589     public <T> void registerService(Class<T> api, T service) {
2590         if(Layer0.class == api) {
2591                 L0 = (Layer0)service;
2592                 return;
2593         }
2594         serviceLocator.registerService(api, service);
2595         if (TransactionPolicySupport.class == api) {
2596             transactionPolicy = (TransactionPolicySupport)service;
2597             state.resetTransactionPolicy();
2598         }
2599         if (api == serviceKey1)
2600                 service1 = service;
2601         else if (api == serviceKey2)
2602                 service2 = service;
2603     }
2604
2605     // ----------------
2606     // SessionMonitor
2607     // ----------------
2608
2609     void fireSessionVariableChange(String variable) {
2610         for (MonitorHandler h : monitorHandlers) {
2611             MonitorContext ctx = monitorContexts.getLeft(h);
2612             assert ctx != null;
2613             // SafeRunner functionality repeated here to avoid dependency.
2614             try {
2615                 h.valuesChanged(ctx);
2616             } catch (Exception e) {
2617                 Logger.defaultLogError("monitor handler notification produced the following exception", e);
2618             } catch (LinkageError e) {
2619                 Logger.defaultLogError("monitor handler notification produced a linkage error", e);
2620             }
2621         }
2622     }
2623
2624
2625     class ResourceSerializerImpl implements ResourceSerializer {
2626
2627         public long createRandomAccessId(int id) throws DatabaseException {
2628             if(id < 0)
2629                 return id;
2630             int index = ClusterTraitsBase.getResourceIndexFromResourceKey(id);
2631             long cluster = getCluster(id);
2632             if (0 == cluster)
2633                 return 0; // Better to return 0 then invalid id.
2634             long result = ClusterTraitsBase.createResourceId(cluster, index);
2635             return result;
2636         }
2637
2638         @Override
2639         public long getRandomAccessId(Resource resource) throws DatabaseException {
2640
2641             ResourceImpl resourceImpl = (ResourceImpl) resource;
2642             return createRandomAccessId(resourceImpl.id);
2643
2644         }
2645
2646         @Override
2647         public int getTransientId(Resource resource) throws DatabaseException {
2648
2649             ResourceImpl resourceImpl = (ResourceImpl) resource;
2650             return resourceImpl.id;
2651
2652         }
2653
2654         @Override
2655         public String createRandomAccessId(Resource resource)
2656         throws InvalidResourceReferenceException {
2657
2658             if(resource == null) throw new IllegalArgumentException();
2659
2660             ResourceImpl resourceImpl = (ResourceImpl) resource;
2661
2662             if (resourceImpl.id < 0) return String.valueOf(resourceImpl.id) + "_0";
2663
2664             int r;
2665             try {
2666                 r = ClusterTraits.getResourceIndexFromResourceKey(resourceImpl.id);
2667             } catch (DatabaseException e1) {
2668                 throw new InvalidResourceReferenceException(e1);
2669             }
2670             try {
2671                 // Serialize as '<resource index>_<cluster id>'
2672                 return "" + r + "_" + getCluster(resourceImpl);
2673             } catch (Throwable e) {
2674                 e.printStackTrace();
2675                 throw new InvalidResourceReferenceException(e);
2676             } finally {
2677             }
2678         }
2679
2680         public int getTransientId(long serialized) throws DatabaseException {
2681             if (serialized <= 0)
2682                 return (int)serialized;
2683             int index = ClusterTraitsBase.getResourceIndexFromResourceId(serialized);
2684             long c = ClusterTraitsBase.getClusterIdFromResourceId(serialized);
2685             ClusterI cluster = clusterTranslator.getClusterByClusterId(c);
2686             if (cluster == null)
2687                 throw new DatabaseException("Couldn't load cluster for id " + c + " for resource id " + serialized);
2688             int key = ClusterTraits.createResourceKey(cluster.getClusterKey(), index);
2689             return key;
2690         }
2691
2692         @Override
2693         public Resource getResource(long randomAccessId) throws DatabaseException {
2694             return getResourceByKey(getTransientId(randomAccessId));
2695         }
2696
2697         @Override
2698         public Resource getResource(int transientId) throws DatabaseException {
2699             return getResourceByKey(transientId);
2700         }
2701
2702         @Override
2703         public Resource getResource(String randomAccessId)
2704         throws InvalidResourceReferenceException {
2705             try {
2706                 int i = randomAccessId.indexOf('_');
2707                 if (i == -1)
2708                     throw new InvalidResourceReferenceException("Could not parse resource id + cluster id from '"
2709                             + randomAccessId + "'");
2710                 int r = Integer.parseInt(randomAccessId.substring(0, i));
2711                 if(r < 0) return getResourceByKey(r);
2712                 long c = Long.parseLong(randomAccessId.substring(i + 1));
2713                 ClusterI cluster = clusterTranslator.getClusterByClusterId(c);
2714                 int key = ClusterTraits.createResourceKey(cluster.getClusterKey(), r);
2715                 if (cluster.hasResource(key, clusterTranslator))
2716                     return getResourceByKey(key);
2717             } catch (InvalidResourceReferenceException e) {
2718                 throw e;
2719             } catch (NumberFormatException e) {
2720                 throw new InvalidResourceReferenceException(e);
2721             } catch (Throwable e) {
2722                 e.printStackTrace();
2723                 throw new InvalidResourceReferenceException(e);
2724             } finally {
2725             }
2726             throw new InvalidResourceReferenceException("Resource does not exist " + randomAccessId);
2727         }
2728
2729         @Override
2730         public boolean disposeRandomAccessId(String randomAccessId) throws InvalidResourceReferenceException {
2731             try {
2732                 return true;
2733             } catch (Throwable e) {
2734                 e.printStackTrace();
2735                 throw new InvalidResourceReferenceException(e);
2736             } finally {
2737             }
2738         }
2739     }
2740
2741     // Copied from old SessionImpl
2742
2743     void check() {
2744         if (state.isClosed())
2745             throw new Error("Session closed.");
2746     }
2747
2748
2749
2750     public ResourceImpl getNewResource(long clusterId) {
2751         ClusterI cluster = clusterTable.getClusterByClusterIdOrThrow(clusterId);
2752         int newId;
2753         try {
2754             newId = cluster.createResource(clusterTranslator);
2755         } catch (DatabaseException e) {
2756             Logger.defaultLogError(e);
2757             return null;
2758         }
2759         return new ResourceImpl(resourceSupport, newId);
2760     }
2761
2762     public ResourceImpl getNewResource(Resource clusterSet)
2763     throws DatabaseException {
2764         long resourceId = clusterSet.getResourceId();
2765         Long clusterId = clusterSetsSupport.get(resourceId);
2766         if (null == clusterId)
2767             throw new ClusterSetExistException("Cluster set does not exist. Resource=" + clusterSet);
2768         if (Constants.NewClusterId == clusterId) {
2769             clusterId = getService(ClusteringSupport.class).createCluster();
2770             if ((serviceMode & SERVICE_MODE_CREATE) > 0)
2771                 createdClusters.add(clusterId);
2772             clusterSetsSupport.put(resourceId, clusterId);
2773             return getNewResource(clusterId);
2774         } else {
2775             ClusterBase cb = clusterTranslator.getClusterByClusterId(clusterId);
2776             ResourceImpl result;
2777             if (cb.getNumberOfResources(clusterTranslator) >= ClusterTable.CLUSTER_FILL_SIZE) {
2778                 clusterId = getService(ClusteringSupport.class).createCluster();
2779                 if ((serviceMode & SERVICE_MODE_CREATE) > 0)
2780                     createdClusters.add(clusterId);
2781                 clusterSetsSupport.put(resourceId, clusterId);
2782                 return getNewResource(clusterId);
2783             } else {
2784                 result = getNewResource(clusterId);
2785                 int resultKey = querySupport.getId(result);
2786                 long resultCluster = querySupport.getClusterId(resultKey);
2787                 if (clusterId != resultCluster)
2788                     clusterSetsSupport.put(resourceId, resultCluster);
2789                 return result;
2790             }
2791         }
2792     }
2793
2794     public void getNewClusterSet(Resource clusterSet)
2795     throws DatabaseException {
2796         if (DEBUG)
2797             System.out.println("new cluster set=" + clusterSet);
2798         long resourceId = clusterSet.getResourceId();
2799         if (clusterSetsSupport.containsKey(resourceId))
2800             throw new ClusterSetExistException("Cluster set exist already. Resource=" + clusterSet);
2801         clusterSetsSupport.put(resourceId, Constants.NewClusterId);
2802     }
2803     public boolean containsClusterSet(Resource clusterSet)
2804     throws ServiceException {
2805         long resourceId = clusterSet.getResourceId();
2806         return clusterSetsSupport.containsKey(resourceId);
2807     }
2808     public Resource setDefaultClusterSet4NewResource(Resource clusterSet) {
2809         Resource r = defaultClusterSet;
2810         defaultClusterSet = clusterSet;
2811         return r;
2812     }
2813     void printDiagnostics() {
2814
2815         if (DIAGNOSTICS) {
2816
2817             final DataContainer<Long> totalMem = new DataContainer<Long>(0L);
2818             final DataContainer<Long> residentClusters = new DataContainer<Long>(0L);
2819             for(ClusterI cluster : clusterTable.getClusters()) {
2820                 try {
2821                     if (cluster.isLoaded() && !cluster.isEmpty()) {
2822                         residentClusters.set(residentClusters.get() + 1);
2823                         totalMem.set(totalMem.get() + cluster.getUsedSpace());
2824                     }
2825                 } catch (DatabaseException e) {
2826                     Logger.defaultLogError(e);
2827                 }
2828             }
2829
2830             System.out.println("--------------------------------");
2831             System.out.println("Cluster information:");
2832             System.out.println("-amount of resident clusters=" + residentClusters.get());
2833             System.out.println("-total memory usage=" + totalMem.get() / 1024 + "kB");
2834
2835             for(ClusterI cluster : clusterTable.getClusters()) {
2836                 System.out.print("Cluster " + cluster.getClusterId() + " [");
2837                 try {
2838                     if (!cluster.isLoaded())
2839                         System.out.println("not loaded]");
2840                     else if (cluster.isEmpty())
2841                         System.out.println("is empty]");
2842                     else {
2843                         System.out.print("resources=" + cluster.getNumberOfResources(clusterTranslator));
2844                         System.out.print(",references=" + cluster.getReferenceCount());
2845                         System.out.println(",memory=" + cluster.getUsedSpace() / 1024 + "kB]");
2846                     }
2847                 } catch (DatabaseException e) {
2848                     Logger.defaultLogError(e);
2849                     System.out.println("is corrupted]");
2850                 }
2851             }
2852
2853             queryProvider2.printDiagnostics();
2854             System.out.println("--------------------------------");
2855         }
2856
2857     }
2858
2859     public void fireStartReadTransaction() {
2860
2861         if (DIAGNOSTICS)
2862             System.out.println("StartReadTransaction");
2863
2864         for (SessionEventListener listener : eventListeners) {
2865             try {
2866                 listener.readTransactionStarted();
2867             } catch (Throwable t) {
2868                 t.printStackTrace();
2869             }
2870         }
2871
2872     }
2873
2874     public void fireFinishReadTransaction() {
2875
2876         if (DIAGNOSTICS)
2877             System.out.println("FinishReadTransaction");
2878
2879         for (SessionEventListener listener : eventListeners) {
2880             try {
2881                 listener.readTransactionFinished();
2882             } catch (Throwable t) {
2883                 t.printStackTrace();
2884             }
2885         }
2886
2887     }
2888
2889     public void fireStartWriteTransaction() {
2890
2891         if (DIAGNOSTICS)
2892             System.out.println("StartWriteTransaction");
2893
2894         for (SessionEventListener listener : eventListeners) {
2895             try {
2896                 listener.writeTransactionStarted();
2897             } catch (Throwable t) {
2898                 t.printStackTrace();
2899             }
2900         }
2901
2902     }
2903
2904     public void fireFinishWriteTransaction() {
2905
2906         if (DIAGNOSTICS)
2907             System.out.println("FinishWriteTransaction");
2908
2909         for (SessionEventListener listener : eventListeners) {
2910             try {
2911                 listener.writeTransactionFinished();
2912             } catch (Throwable t) {
2913                 t.printStackTrace();
2914             }
2915         }
2916
2917         Indexing.resetDependenciesIndexingDisabled();
2918
2919     }
2920
2921     Resource deserialize(final long resourceId, final long clusterId) throws IOException {
2922         throw new Error("Not supported at the moment.");
2923     }
2924
2925     State getState() {
2926         return state;
2927     }
2928
2929     ClusterTable getClusterTable() {
2930         return clusterTable;
2931     }
2932
2933     public GraphSession getGraphSession() {
2934         return graphSession;
2935     }
2936
2937     public QueryProcessor getQueryProvider2() {
2938         return queryProvider2;
2939     }
2940
2941     ClientChangesImpl getClientChanges() {
2942         return clientChanges;
2943     }
2944
2945     boolean getWriteOnly() {
2946         return writeOnly;
2947     }
2948
2949     static int counter = 0;
2950
2951     public void onClusterLoaded(long clusterId) {
2952
2953         clusterTable.updateSize();
2954
2955     }
2956
2957
2958
2959
2960     /*
2961      * Implementation of the interface RequestProcessor
2962      */
2963
2964     @Override
2965     public <T> T syncRequest(final Read<T> request) throws DatabaseException {
2966
2967         assertNotSession();
2968         assertAlive();
2969
2970         assert(request != null);
2971
2972         final DataContainer<T> result = new DataContainer<T>();
2973         final DataContainer<Throwable> exception = new DataContainer<Throwable>();
2974
2975         syncRequest(request, new AsyncProcedure<T>() {
2976
2977             @Override
2978             public void execute(AsyncReadGraph graph, T t) {
2979                 result.set(t);
2980             }
2981
2982             @Override
2983             public void exception(AsyncReadGraph graph, Throwable t) {
2984                 exception.set(t);
2985             }
2986
2987         });
2988
2989         Throwable t = exception.get();
2990         if(t != null) {
2991             if(t instanceof DatabaseException) throw (DatabaseException)t;
2992             else throw new DatabaseException("Unexpected exception in ReadGraph.syncRequest(Read)", t);
2993         }
2994
2995         return result.get();
2996
2997     }
2998
2999 //    @Override
3000 //    public <T> T syncRequest(Read<T> request, AsyncListener<T> procedure) throws DatabaseException {
3001 //        assertNotSession();
3002 //        return syncRequest(request, (AsyncProcedure<T>)procedure);
3003 //    }
3004
3005     @Override
3006     public <T> T syncRequest(Read<T> request, SyncListener<T> procedure) throws DatabaseException {
3007         assertNotSession();
3008         return syncRequest(request, new SyncToAsyncListener<T>(procedure));
3009     }
3010
3011     @Override
3012     public <T> T syncRequest(Read<T> request, final Listener<T> procedure) throws DatabaseException {
3013         assertNotSession();
3014         return syncRequest(request, new NoneToAsyncListener<T>(procedure));
3015     }
3016
3017     @Override
3018     public <T> T syncRequest(final Read<T> request, final SyncProcedure<T> procedure) throws DatabaseException {
3019         assertNotSession();
3020         return syncRequest(request, new SyncToAsyncProcedure<T>(procedure));
3021     }
3022
3023     @Override
3024     public <T> T syncRequest(Read<T> request, Procedure<T> procedure) throws DatabaseException {
3025         assertNotSession();
3026         return syncRequest(request, new NoneToAsyncProcedure<T>(procedure));
3027     }
3028
3029     @Override
3030     public <T> T syncRequest(final AsyncRead<T> request) throws DatabaseException {
3031
3032         assertNotSession();
3033
3034         assert(request != null);
3035
3036         final DataContainer<T> result = new DataContainer<T>();
3037         final DataContainer<Throwable> exception = new DataContainer<Throwable>();
3038
3039         syncRequest(request, new AsyncProcedure<T>() {
3040
3041             @Override
3042             public void execute(AsyncReadGraph graph, T t) {
3043                 result.set(t);
3044             }
3045
3046             @Override
3047             public void exception(AsyncReadGraph graph, Throwable t) {
3048                 exception.set(t);
3049             }
3050
3051         });
3052
3053         Throwable t = exception.get();
3054         if(t != null) {
3055             if(t instanceof DatabaseException) throw (DatabaseException)t;
3056             else throw new DatabaseException("Unexpected exception in ReadGraph.syncRequest(Read)", t);
3057         }
3058
3059         return result.get();
3060
3061     }
3062
3063     @Override
3064     public <T> T syncRequest(AsyncRead<T> request, AsyncListener<T> procedure) throws DatabaseException {
3065         assertNotSession();
3066         return syncRequest(request, (AsyncProcedure<T>)procedure);
3067     }
3068
3069     @Override
3070     public <T> T syncRequest(AsyncRead<T> request, SyncListener<T> procedure) throws DatabaseException {
3071         assertNotSession();
3072         return syncRequest(request, new SyncToAsyncListener<T>(procedure));
3073     }
3074
3075     @Override
3076     public <T> T syncRequest(AsyncRead<T> request, Listener<T> procedure) throws DatabaseException {
3077         assertNotSession();
3078         return syncRequest(request, new NoneToAsyncListener<T>(procedure));
3079     }
3080
3081     @Override
3082     public <T> T syncRequest(AsyncRead<T> request, final SyncProcedure<T> procedure) throws DatabaseException {
3083         assertNotSession();
3084         return syncRequest(request, new SyncToAsyncProcedure<T>(procedure));
3085     }
3086
3087     @Override
3088     final public <T> T syncRequest(final AsyncRead<T> request, final Procedure<T> procedure) throws DatabaseException {
3089         assertNotSession();
3090         return syncRequest(request, new NoneToAsyncProcedure<T>(procedure));
3091     }
3092
3093     @Override
3094     public <T> Collection<T> syncRequest(MultiRead<T> request) throws DatabaseException {
3095
3096         assertNotSession();
3097
3098         assert(request != null);
3099
3100         final ArrayList<T> result = new ArrayList<T>();
3101         final DataContainer<Throwable> exception = new DataContainer<Throwable>();
3102
3103         syncRequest(request, new SyncMultiProcedure<T>() {
3104
3105             @Override
3106             public void execute(ReadGraph graph, T t) {
3107                 synchronized(result) {
3108                     result.add(t);
3109                 }
3110             }
3111
3112             @Override
3113             public void finished(ReadGraph graph) {
3114             }
3115
3116             @Override
3117             public void exception(ReadGraph graph, Throwable t) {
3118                 exception.set(t);
3119             }
3120
3121
3122         });
3123
3124         Throwable t = exception.get();
3125         if(t != null) {
3126             if(t instanceof DatabaseException) throw (DatabaseException)t;
3127             else throw new DatabaseException("Unexpected exception in ReadGraph.syncRequest(Read)", t);
3128         }
3129
3130         return result;
3131
3132     }
3133
3134     @Override
3135     public <T> Collection<T> syncRequest(MultiRead<T> arg0, SyncMultiProcedure<T> arg1) throws DatabaseException {
3136         assertNotSession();
3137         throw new Error("Not implemented!");
3138     }
3139
3140     @Override
3141     public <T> Collection<T> syncRequest(MultiRead<T> request, SyncMultiListener<T> procedure) throws DatabaseException {
3142         assertNotSession();
3143         return syncRequest(request, (SyncMultiProcedure<T>)procedure);
3144     }
3145
3146     @Override
3147     public <T> Collection<T> syncRequest(MultiRead<T> request, MultiListener<T> procedure) throws DatabaseException {
3148         assertNotSession();
3149         return syncRequest(request, new NoneToSyncMultiListener<T>(procedure));
3150     }
3151
3152     @Override
3153     public <T> Collection<T> syncRequest(MultiRead<T> request, MultiProcedure<T> procedure) throws DatabaseException {
3154         assertNotSession();
3155         return syncRequest(request, new NoneToSyncMultiProcedure<T>(procedure));
3156     }
3157
3158     @Override
3159     final public <T> Collection<T> syncRequest(final AsyncMultiRead<T> request) throws DatabaseException {
3160
3161         assertNotSession();
3162
3163         assert(request != null);
3164
3165         final ArrayList<T> result = new ArrayList<T>();
3166         final DataContainer<Throwable> exception = new DataContainer<Throwable>();
3167
3168         syncRequest(request, new AsyncMultiProcedure<T>() {
3169
3170             @Override
3171             public void execute(AsyncReadGraph graph, T t) {
3172                 synchronized(result) {
3173                     result.add(t);
3174                 }
3175             }
3176
3177             @Override
3178             public void finished(AsyncReadGraph graph) {
3179             }
3180
3181             @Override
3182             public void exception(AsyncReadGraph graph, Throwable t) {
3183                 exception.set(t);
3184             }
3185
3186
3187         });
3188
3189         Throwable t = exception.get();
3190         if (t != null) {
3191             if (t instanceof DatabaseException)
3192                 throw (DatabaseException) t;
3193             else
3194                 throw new DatabaseException("Unexpected exception in ReadGraph.syncRequest(Read)", t);
3195         }
3196
3197         return result;
3198
3199     }
3200
3201     @Override
3202     public <T> Collection<T> syncRequest(AsyncMultiRead<T> request, AsyncMultiListener<T> procedure) throws DatabaseException {
3203         assertNotSession();
3204         return syncRequest(request, (AsyncMultiProcedure<T>)procedure);
3205     }
3206
3207     @Override
3208     public <T> Collection<T> syncRequest(AsyncMultiRead<T> request, SyncMultiListener<T> procedure) throws DatabaseException {
3209         assertNotSession();
3210         return syncRequest(request, new SyncToAsyncMultiListener<T>(procedure));
3211     }
3212
3213     @Override
3214     public <T> Collection<T> syncRequest(AsyncMultiRead<T> request, MultiListener<T> procedure) throws DatabaseException {
3215         assertNotSession();
3216        return syncRequest(request, new NoneToAsyncMultiListener<T>(procedure));
3217     }
3218
3219     @Override
3220     public <T> Collection<T> syncRequest(AsyncMultiRead<T> request, final SyncMultiProcedure<T> procedure) throws DatabaseException {
3221         assertNotSession();
3222         return syncRequest(request, new SyncToAsyncMultiProcedure<T>(procedure));
3223     }
3224
3225     @Override
3226     final public <T> Collection<T> syncRequest(final AsyncMultiRead<T> request, final MultiProcedure<T> procedure) throws DatabaseException {
3227         assertNotSession();
3228         return syncRequest(request, new NoneToAsyncMultiProcedure<T>(procedure));
3229     }
3230
3231
3232     @Override
3233     public <T> void asyncRequest(Read<T> request) {
3234
3235         asyncRequest(request, new ProcedureAdapter<T>() {
3236             @Override
3237             public void exception(Throwable t) {
3238                 t.printStackTrace();
3239             }
3240         });
3241
3242     }
3243
3244     @Override
3245     public <T> void asyncRequest(Read<T> request, AsyncListener<T> procedure) {
3246         asyncRequest(request, (AsyncProcedure<T>)procedure);
3247     }
3248
3249     @Override
3250     public <T> void asyncRequest(Read<T> request, final SyncListener<T> procedure) {
3251         asyncRequest(request, new SyncToAsyncListener<T>(procedure));
3252     }
3253
3254     @Override
3255     public <T> void asyncRequest(Read<T> request, final Listener<T> procedure) {
3256         asyncRequest(request, new NoneToAsyncListener<T>(procedure));
3257     }
3258
3259     @Override
3260     public <T> void asyncRequest(Read<T> request, SyncProcedure<T> procedure) {
3261         asyncRequest(request, new SyncToAsyncProcedure<T>(procedure));
3262     }
3263
3264     @Override
3265     public <T> void asyncRequest(Read<T> request, final Procedure<T> procedure) {
3266         asyncRequest(request, new NoneToAsyncProcedure<T>(procedure));
3267     }
3268
3269     @Override
3270     final public <T> void asyncRequest(final AsyncRead<T> request) {
3271
3272         assert(request != null);
3273
3274         asyncRequest(request, new ProcedureAdapter<T>() {
3275             @Override
3276             public void exception(Throwable t) {
3277                 t.printStackTrace();
3278             }
3279         });
3280
3281     }
3282
3283     @Override
3284     public <T> void asyncRequest(AsyncRead<T> request, AsyncListener<T> procedure) {
3285         scheduleRequest(request, procedure, procedure, null);
3286     }
3287
3288     @Override
3289     public <T> void asyncRequest(AsyncRead<T> request, final SyncListener<T> procedure) {
3290         asyncRequest(request, new SyncToAsyncListener<T>(procedure));
3291     }
3292
3293     @Override
3294     public <T> void asyncRequest(AsyncRead<T> request, final Listener<T> procedure) {
3295         asyncRequest(request, new NoneToAsyncListener<T>(procedure));
3296     }
3297
3298     @Override
3299     public <T> void asyncRequest(AsyncRead<T> request, SyncProcedure<T> procedure) {
3300         asyncRequest(request, new SyncToAsyncProcedure<T>(procedure));
3301     }
3302
3303     @Override
3304     final public <T> void asyncRequest(final AsyncRead<T> request, final Procedure<T> procedure) {
3305         asyncRequest(request, new NoneToAsyncProcedure<T>(procedure));
3306     }
3307
3308     @Override
3309     public <T> void asyncRequest(MultiRead<T> request) {
3310
3311         assert(request != null);
3312
3313         asyncRequest(request, new SyncMultiProcedureAdapter<T>() {
3314             @Override
3315             public void exception(ReadGraph graph, Throwable t) {
3316                 t.printStackTrace();
3317             }
3318         });
3319
3320     }
3321
3322     @Override
3323     public <T> void asyncRequest(MultiRead<T> request, SyncMultiListener<T> procedure) {
3324         asyncRequest(request, (SyncMultiProcedure<T>)procedure);
3325     }
3326
3327     @Override
3328     public <T> void asyncRequest(MultiRead<T> request, MultiListener<T> procedure) {
3329         asyncRequest(request, new NoneToSyncMultiListener<T>(procedure));
3330     }
3331
3332     @Override
3333     public <T> void asyncRequest(MultiRead<T> request, SyncMultiProcedure<T> procedure) {
3334         scheduleRequest(request, procedure, null);
3335     }
3336
3337     @Override
3338     public <T> void asyncRequest(MultiRead<T> request, MultiProcedure<T> procedure) {
3339         asyncRequest(request, new NoneToSyncMultiProcedure<T>(procedure));
3340     }
3341
3342     @Override
3343     final public <T> void asyncRequest(AsyncMultiRead<T> request) {
3344
3345         assert(request != null);
3346
3347         asyncRequest(request, new AsyncMultiProcedureAdapter<T>() {
3348             @Override
3349             public void exception(AsyncReadGraph graph, Throwable t) {
3350                 t.printStackTrace();
3351             }
3352         });
3353
3354     }
3355
3356     @Override
3357     public <T> void asyncRequest(AsyncMultiRead<T> request, AsyncMultiListener<T> procedure) {
3358         asyncRequest(request, (AsyncMultiProcedure<T>)procedure);
3359     }
3360
3361     @Override
3362     public <T> void asyncRequest(AsyncMultiRead<T> request, SyncMultiListener<T> procedure) {
3363         asyncRequest(request, new SyncToAsyncMultiListener<T>(procedure));
3364     }
3365
3366     @Override
3367     public <T> void asyncRequest(AsyncMultiRead<T> request, MultiListener<T> procedure) {
3368         asyncRequest(request, new NoneToAsyncMultiListener<T>(procedure));
3369     }
3370
3371     @Override
3372     public <T> void asyncRequest(AsyncMultiRead<T> request, SyncMultiProcedure<T> procedure) {
3373         asyncRequest(request, new SyncToAsyncMultiProcedure<T>(procedure));
3374     }
3375
3376     @Override
3377     final public <T> void asyncRequest(AsyncMultiRead<T> request, final MultiProcedure<T> procedure) {
3378         asyncRequest(request, new NoneToAsyncMultiProcedure<T>(procedure));
3379     }
3380
3381     @Override
3382     final public <T> void asyncRequest(final ExternalRead<T> request) {
3383
3384         assert(request != null);
3385
3386         asyncRequest(request, new ProcedureAdapter<T>() {
3387             @Override
3388             public void exception(Throwable t) {
3389                 t.printStackTrace();
3390             }
3391         });
3392
3393     }
3394
3395     @Override
3396     public <T> void asyncRequest(ExternalRead<T> request, final Listener<T> procedure) {
3397         asyncRequest(request, (Procedure<T>)procedure);
3398     }
3399
3400     boolean sameProvider(Write request) {
3401         if(writeState.getGraph().provider != null) {
3402             return writeState.getGraph().provider.equals(request.getProvider());
3403         } else {
3404             return request.getProvider() == null;
3405         }
3406     }
3407
3408     boolean plainWrite(WriteGraphImpl graph) {
3409         if(graph == null) return false;
3410         if(graph.writeSupport.writeOnly()) return false;
3411         return true;
3412     }
3413
3414
3415     public static final ThreadGroup SessionThreadGroup = new ThreadGroup("Session Thread Group");
3416
3417     private void assertNotSession() throws DatabaseException {
3418         Thread current = Thread.currentThread();
3419         if (sessionThreads.contains(current))
3420             throw new ServiceException("Caller is already inside a transaction.");
3421     }
3422
3423     void assertAlive() {
3424         if (!state.isAlive())
3425             throw new RuntimeDatabaseException("Session has been shut down.");
3426     }
3427
3428     public InputStream getValueStream(ReadGraphImpl graph, Resource resource) {
3429         return querySupport.getValueStream(graph, querySupport.getId(resource));
3430     }
3431
3432     public byte[] getValue(ReadGraphImpl graph, Resource resource) {
3433         return querySupport.getValue(graph, querySupport.getId(resource));
3434     }
3435
3436     <T> void acquire(Semaphore semaphore, T request) throws DatabaseException {
3437         acquire(semaphore, request, null);
3438     }
3439
3440     private <T,P> void acquire(Semaphore semaphore, T request, P procedure) throws DatabaseException {
3441         assertAlive();
3442         try {
3443             long tryCount = 0;
3444             // Loop until the semaphore is acquired reporting requests that take a long time.
3445             while (true) {
3446
3447                 if (semaphore.tryAcquire(DebugPolicy.LONG_EXECUTION_REPORT_PERIOD, DebugPolicy.LONG_EXECUTION_REPORT_PERIOD_UNIT)) {
3448                     // Acquired OK.
3449                     return;
3450                 } else {
3451                     if (DebugPolicy.REPORT_TIME_CONSUMING_REQUESTS) {
3452                         ++tryCount;
3453                         long waitTime = DebugPolicy.LONG_EXECUTION_REPORT_PERIOD_UNIT
3454                                 .toMillis(DebugPolicy.LONG_EXECUTION_REPORT_PERIOD)
3455                                 * tryCount;
3456                         System.err.println("DB client request '" + request + "' is taking long to execute, so far "
3457                                 + (waitTime) + " ms. (procedure=" + procedure + ")");
3458                     }
3459                 }
3460
3461                 assertAlive();
3462
3463             }
3464         } catch (InterruptedException e) {
3465             e.printStackTrace();
3466             // FIXME: Should perhaps do something else in this case ??
3467         }
3468     }
3469
3470
3471
3472 //    @Override
3473 //    public boolean holdOnToTransactionAfterCancel() {
3474 //        return transactionPolicy.holdOnToTransactionAfterCancel();
3475 //    }
3476 //
3477 //    @Override
3478 //    public boolean holdOnToTransactionAfterCommit() {
3479 //        return transactionPolicy.holdOnToTransactionAfterCommit();
3480 //    }
3481 //
3482 //    @Override
3483 //    public boolean holdOnToTransactionAfterRead() {
3484 //        return transactionPolicy.holdOnToTransactionAfterRead();
3485 //    }
3486 //
3487 //    @Override
3488 //    public void onRelinquish() {
3489 //        transactionPolicy.onRelinquish();
3490 //    }
3491 //
3492 //    @Override
3493 //    public void onRelinquishDone() {
3494 //        transactionPolicy.onRelinquishDone();
3495 //    }
3496 //
3497 //    @Override
3498 //    public void onRelinquishError() {
3499 //        transactionPolicy.onRelinquishError();
3500 //    }
3501
3502
3503     @Override
3504     public Session getSession() {
3505         return this;
3506     }
3507
3508
3509     protected abstract VirtualGraph getProvider(VirtualGraph virtualGraph);
3510     protected abstract ResourceImpl getNewResource() throws DatabaseException;
3511
3512     public void ceased(int thread) {
3513
3514         requestManager.ceased(thread);
3515
3516     }
3517
3518     public int getAmountOfQueryThreads() {
3519         // This must be a power of two
3520         return 1;
3521 //        return Integer.highestOneBit(Runtime.getRuntime().availableProcessors());
3522     }
3523
3524     public Resource getResourceByKey(int key) throws ResourceNotFoundException {
3525
3526         return new ResourceImpl(resourceSupport, key);
3527
3528     }
3529
3530     public void acquireWriteOnly() {
3531         writeOnly = true;
3532     }
3533
3534     public void releaseWriteOnly(ReadGraphImpl graph) {
3535         writeOnly = false;
3536         queryProvider2.releaseWrite(graph);
3537     }
3538
3539     public void handleUpdatesAndMetadata(WriteGraphImpl writer) {
3540
3541         long start = System.nanoTime();
3542
3543         while(dirtyPrimitives) {
3544             dirtyPrimitives = false;
3545             getQueryProvider2().propagateChangesInQueryCache(writer);
3546             getQueryProvider2().listening.fireListeners(writer);
3547         }
3548
3549         fireMetadataListeners(writer, clientChanges);
3550
3551         if(DebugPolicy.PERFORMANCE_DATA) {
3552             long end = System.nanoTime();
3553             System.err.println("handleUpdatesAndMetadata " + 1e-9*(end-start));
3554         }
3555
3556     }
3557
3558     void removeTemporaryData() {
3559         File platform = Platform.getLocation().toFile();
3560         File tempFiles = new File(platform, "tempFiles");
3561         File temp = new File(tempFiles, "db");
3562         if(!temp.exists())
3563             return;
3564         try {
3565             Files.walkFileTree(temp.toPath(), new SimpleFileVisitor<Path>() {
3566                 @Override
3567                 public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) {
3568                     try {
3569                         Files.delete(file);
3570                     } catch (IOException e) {
3571                         Logger.defaultLogError(e);
3572                     }
3573                     return FileVisitResult.CONTINUE;
3574                 }
3575             });
3576         } catch (IOException e) {
3577             Logger.defaultLogError(e);
3578         }
3579     }
3580
3581     public void handleCreatedClusters() {
3582         if ((serviceMode & SERVICE_MODE_CREATE) > 0) {
3583             createdClusters.forEach(new TLongProcedure() {
3584
3585                 @Override
3586                 public boolean execute(long value) {
3587                     ClusterImpl cluster = clusterTable.getClusterByClusterId(value);
3588                     cluster.setImmutable(true, clusterTranslator);
3589                     return true;
3590                 }
3591             });
3592         }
3593         createdClusters.clear();
3594     }
3595
3596     @Override
3597     public Object getModificationCounter() {
3598         return queryProvider2.modificationCounter;
3599     }
3600
3601     @Override
3602     public void markUndoPoint() {
3603         state.setCombine(false);
3604     }
3605
3606 }