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.task("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.task("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         requestManager.scheduleRead(new SessionRead(throwable, notify) {
1500
1501             @Override
1502             public void run0(int thread) {
1503
1504                 fireSessionVariableChange(SessionVariables.QUEUED_READS);
1505
1506                 ITask task = ThreadLogger.task(request);
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                             T t = request.perform(newGraph);
1553
1554                             try {
1555
1556                                 if(result != null) result.set(t);
1557                                 procedure.execute(newGraph, t);
1558
1559                             } catch (Throwable th) {
1560
1561                                 if(throwable != null) {
1562                                     throwable.set(th);
1563                                 } else {
1564                                     Logger.defaultLogError("Unhandled exception", th);
1565                                 }
1566
1567                             }
1568
1569                         } catch (Throwable t) {
1570
1571                             if (DEBUG)
1572                                 t.printStackTrace();
1573
1574                             if(throwable != null) {
1575                                 throwable.set(t);
1576                             } else {
1577                                 Logger.defaultLogError("Unhandled exception", t);
1578                             }
1579
1580                             try {
1581
1582                                 procedure.exception(newGraph, t);
1583
1584                             } catch (Throwable t2) {
1585
1586                                 if(throwable != null) {
1587                                     throwable.set(t2);
1588                                 } else {
1589                                     Logger.defaultLogError("Unhandled exception", t2);
1590                                 }
1591
1592                             }
1593
1594                         }
1595
1596                         task.finish();
1597
1598                     }
1599
1600                 } finally {
1601
1602                     fireSessionVariableChange(SessionVariables.QUEUED_READS);
1603
1604                 }
1605
1606             }
1607
1608         });
1609
1610     }
1611
1612     public <T> void scheduleRequest(final AsyncRead<T> request, final AsyncProcedure<T> procedure, final ListenerBase listener, final Semaphore notify) {
1613
1614         assert (request != null);
1615         assert (procedure != null);
1616
1617         int thread = request.hashCode() & queryProvider2.THREAD_MASK;
1618
1619         requestManager.scheduleRead(new SessionRead(null, notify) {
1620
1621             @Override
1622             public void run0(int thread) {
1623
1624                 fireSessionVariableChange(SessionVariables.QUEUED_READS);
1625
1626                 ITask task = ThreadLogger.task(request);
1627
1628                 final ReadGraphImpl newGraph = ReadGraphImpl.create(getQueryProvider2());
1629
1630                 try {
1631
1632                     if (listener != null) {
1633
1634                         try {
1635                             QueryCacheBase.resultAsyncReadEntry(newGraph, request, null, listener, procedure);
1636                         } catch (DatabaseException e) {
1637                             LOGGER.error("Unhandled query exception", e);
1638                         }
1639
1640                     } else {
1641
1642                         BlockingAsyncProcedure<T> wrap = new BlockingAsyncProcedure<T>(newGraph, null, procedure, request, true) {
1643
1644                             public void execute(AsyncReadGraph graph_, T result) {
1645                                 task.finish();
1646                                 super.execute(graph_, result);
1647                             }
1648
1649                             public void exception(AsyncReadGraph graph_, Throwable t) {
1650                                 task.finish();
1651                                 super.exception(graph_, t);
1652                             }
1653
1654                         };
1655
1656                         try {
1657                             wrap.performSync(request);
1658                         } catch (DatabaseException e) {
1659                             LOGGER.error("Unhandled query exception", e);
1660                         }
1661
1662                     }
1663
1664                 } finally {
1665
1666                     fireSessionVariableChange(SessionVariables.QUEUED_READS);
1667
1668                 }
1669
1670             }
1671
1672         });
1673
1674     }
1675
1676     public <T> void scheduleRequest(final MultiRead<T> request, final SyncMultiProcedure<T> procedure, final Semaphore notify) {
1677
1678         assert (request != null);
1679         assert (procedure != null);
1680
1681         int thread = request.hashCode() & queryProvider2.THREAD_MASK;
1682
1683         int sync = notify != null ? thread : -1;
1684
1685         requestManager.scheduleRead(new SessionRead(null, notify) {
1686
1687             @Override
1688             public void run0(int thread) {
1689
1690                 fireSessionVariableChange(SessionVariables.QUEUED_READS);
1691
1692                 ListenerBase listener = getListenerBase(procedure);
1693
1694                 final ReadGraphImpl newGraph = ReadGraphImpl.create(getQueryProvider2());
1695
1696                 try {
1697
1698                     if (listener != null) {
1699
1700                         newGraph.processor.query(newGraph, request, null, procedure, listener);
1701
1702 //                        newGraph.waitAsync(request);
1703
1704                     } else {
1705
1706                         final ResultCallWrappedSyncQueryProcedure<T> wrapper = new ResultCallWrappedSyncQueryProcedure<T>(procedure);
1707
1708                         try {
1709
1710                             request.perform(newGraph, wrapper);
1711
1712                         } catch (Throwable t) {
1713
1714                             t.printStackTrace();
1715
1716                         }
1717
1718                     }
1719
1720                 } finally {
1721
1722                     fireSessionVariableChange(SessionVariables.QUEUED_READS);
1723
1724                 }
1725
1726             }
1727
1728         });
1729
1730     }
1731
1732     public <T> void scheduleRequest(final AsyncMultiRead<T> request, final AsyncMultiProcedure<T> procedure, final Semaphore notify) {
1733
1734         assert (request != null);
1735         assert (procedure != null);
1736
1737         int thread = request.hashCode() & queryProvider2.THREAD_MASK;
1738
1739         int sync = notify != null ? thread : -1;
1740
1741         requestManager.scheduleRead(new SessionRead(null, notify) {
1742
1743             @Override
1744             public void run0(int thread) {
1745
1746                 fireSessionVariableChange(SessionVariables.QUEUED_READS);
1747
1748                 ListenerBase listener = getListenerBase(procedure);
1749
1750                 final ReadGraphImpl newGraph = ReadGraphImpl.create(getQueryProvider2());
1751
1752                 try {
1753
1754                     if (listener != null) {
1755
1756                         newGraph.processor.query(newGraph, request, null, procedure, listener);
1757
1758 //                        newGraph.waitAsync(request);
1759
1760                     } else {
1761
1762                         final ResultCallWrappedQueryProcedure4<T> wrapper = new ResultCallWrappedQueryProcedure4<T>(procedure);
1763
1764                         try {
1765
1766                             request.perform(newGraph, wrapper);
1767
1768                         } catch (Throwable t) {
1769
1770                             t.printStackTrace();
1771
1772                         }
1773
1774                     }
1775
1776                 } finally {
1777
1778                     fireSessionVariableChange(SessionVariables.QUEUED_READS);
1779
1780                 }
1781
1782             }
1783
1784         });
1785
1786     }
1787
1788     public <T> void scheduleRequest(final ExternalRead<T> request, final Procedure<T> procedure, final Semaphore notify, final DataContainer<Throwable> throwable, final DataContainer<T> result) {
1789
1790         assert (request != null);
1791         assert (procedure != null);
1792
1793         int thread = request.hashCode() & queryProvider2.THREAD_MASK;
1794
1795         requestManager.scheduleRead(new SessionRead(throwable, notify) {
1796
1797             @Override
1798             public void run0(int thread) {
1799
1800                 fireSessionVariableChange(SessionVariables.QUEUED_READS);
1801
1802                 ListenerBase listener = getListenerBase(procedure);
1803
1804                 final ReadGraphImpl newGraph = ReadGraphImpl.create(getQueryProvider2());
1805
1806                 try {
1807
1808                     if (listener != null) {
1809
1810                         try {
1811                             QueryCacheBase.resultExternalReadEntry(newGraph, request, null, listener, procedure);
1812                         } catch (DatabaseException e) {
1813                             Logger.defaultLogError(e);
1814                         }
1815
1816                     } else {
1817
1818 //                        newGraph.state.barrier.inc();
1819
1820                         request.register(newGraph, new Listener<T>() {
1821
1822                             @Override
1823                             public void exception(Throwable t) {
1824                                 if(throwable != null) throwable.set(t);
1825                                 procedure.exception(t);
1826 //                                newGraph.state.barrier.dec();
1827                             }
1828
1829                             @Override
1830                             public void execute(T t) {
1831                                 if(result != null) result.set(t);
1832                                 procedure.execute(t);
1833 //                                newGraph.state.barrier.dec();
1834                             }
1835
1836                             @Override
1837                             public boolean isDisposed() {
1838                                 return true;
1839                             }
1840
1841                         });
1842
1843 //                        newGraph.waitAsync(request);
1844
1845                     }
1846
1847                 } finally {
1848
1849                     fireSessionVariableChange(SessionVariables.QUEUED_READS);
1850
1851                 }
1852
1853             }
1854
1855         });
1856
1857     }
1858
1859
1860     @Override
1861     public <T> void asyncRequest(final Read<T> request, final AsyncProcedure<T> procedure) {
1862
1863         scheduleRequest(request, procedure, null, null, null);
1864
1865     }
1866
1867     @Override
1868     public <T> T syncRequest(Read<T> request, AsyncProcedure<T> procedure) throws DatabaseException {
1869         assertNotSession();
1870         Semaphore notify = new Semaphore(0);
1871         DataContainer<Throwable> container = new DataContainer<Throwable>();
1872         DataContainer<T> result = new DataContainer<T>();
1873         scheduleRequest(request, procedure, notify, container, result);
1874         acquire(notify, request);
1875         Throwable throwable = container.get();
1876         if(throwable != null) {
1877             if(throwable instanceof DatabaseException) throw (DatabaseException)throwable;
1878             else throw new DatabaseException("Unexpected exception", throwable);
1879         }
1880         return result.get();
1881     }
1882
1883     @Override
1884     public <T> T syncRequest(AsyncRead<T> request, final AsyncProcedure<T> procedure) throws DatabaseException {
1885         assertNotSession();
1886         Semaphore notify = new Semaphore(0);
1887         final DataContainer<Throwable> exceptionContainer = new DataContainer<Throwable>();
1888         final DataContainer<T> resultContainer = new DataContainer<T>();
1889         scheduleRequest(request, new AsyncProcedure<T>() {
1890             @Override
1891             public void exception(AsyncReadGraph graph, Throwable throwable) {
1892                 exceptionContainer.set(throwable);
1893                 procedure.exception(graph, throwable);
1894             }
1895             @Override
1896             public void execute(AsyncReadGraph graph, T result) {
1897                 resultContainer.set(result);
1898                 procedure.execute(graph, result);
1899             }
1900         }, getListenerBase(procedure), notify);
1901         acquire(notify, request, procedure);
1902         Throwable throwable = exceptionContainer.get();
1903         if (throwable != null) {
1904             if(throwable instanceof DatabaseException) throw (DatabaseException)throwable;
1905             else throw new DatabaseException("Unexpected exception", throwable);
1906         }
1907         return resultContainer.get();
1908     }
1909
1910
1911     @Override
1912     public <T> Collection<T> syncRequest(AsyncMultiRead<T> request, final AsyncMultiProcedure<T> procedure) throws DatabaseException {
1913         assertNotSession();
1914         Semaphore notify = new Semaphore(0);
1915         final DataContainer<Throwable> exceptionContainer = new DataContainer<Throwable>();
1916         scheduleRequest(request, new AsyncMultiProcedure<T>() {
1917             @Override
1918             public void exception(AsyncReadGraph graph, Throwable throwable) {
1919                 exceptionContainer.set(throwable);
1920                 procedure.exception(graph, throwable);
1921             }
1922             @Override
1923             public void execute(AsyncReadGraph graph, T result) {
1924                 procedure.execute(graph, result);
1925             }
1926             @Override
1927             public void finished(AsyncReadGraph graph) {
1928                 procedure.finished(graph);
1929             }
1930         }, notify);
1931         acquire(notify, request, procedure);
1932         Throwable throwable = exceptionContainer.get();
1933         if (throwable != null) {
1934             if(throwable instanceof DatabaseException) throw (DatabaseException)throwable;
1935             else throw new DatabaseException("Unexpected exception", throwable);
1936         }
1937         // TODO: implement return value
1938         System.err.println("TODO: implement return value for syncRequest(AsyncMultiRead, AsyncMultiProcedure)");
1939         return null;
1940     }
1941
1942     @Override
1943     public <T> T syncRequest(final ExternalRead<T> request) throws DatabaseException {
1944         return syncRequest(request, new ProcedureAdapter<T>());
1945
1946
1947 //        assert(request != null);
1948 //
1949 //        final DataContainer<T> result = new DataContainer<T>();
1950 //        final DataContainer<Throwable> exception = new DataContainer<Throwable>();
1951 //
1952 //        syncRequest(request, new Procedure<T>() {
1953 //
1954 //            @Override
1955 //            public void execute(T t) {
1956 //                result.set(t);
1957 //            }
1958 //
1959 //            @Override
1960 //            public void exception(Throwable t) {
1961 //                exception.set(t);
1962 //            }
1963 //
1964 //        });
1965 //
1966 //        Throwable t = exception.get();
1967 //        if(t != null) {
1968 //            if(t instanceof DatabaseException) throw (DatabaseException)t;
1969 //            else throw new DatabaseException("Unexpected exception in ReadGraph.syncRequest(PrimitiveRead)", t);
1970 //        }
1971 //
1972 //        return result.get();
1973
1974     }
1975
1976     @Override
1977     public <T> T syncRequest(ExternalRead<T> request, final Listener<T> procedure) throws DatabaseException {
1978         return syncRequest(request, (Procedure<T>)procedure);
1979     }
1980
1981
1982 //    @Override
1983 //    public <T> T syncRequest(Read<T> request, AsyncProcedure<T> procedure) throws DatabaseException {
1984 //        assertNotSession();
1985 //        Semaphore notify = new Semaphore(0);
1986 //        DataContainer<Throwable> container = new DataContainer<Throwable>();
1987 //        DataContainer<T> result = new DataContainer<T>();
1988 //        scheduleRequest(request, procedure, notify, container, result);
1989 //        acquire(notify, request);
1990 //        Throwable throwable = container.get();
1991 //        if(throwable != null) {
1992 //            if(throwable instanceof DatabaseException) throw (DatabaseException)throwable;
1993 //            else throw new DatabaseException("Unexpected exception", throwable);
1994 //        }
1995 //        return result.get();
1996 //    }
1997
1998
1999     @Override
2000     public <T> T syncRequest(ExternalRead<T> request, final Procedure<T> procedure) throws DatabaseException {
2001         assertNotSession();
2002         Semaphore notify = new Semaphore(0);
2003         final DataContainer<Throwable> container = new DataContainer<Throwable>();
2004         final DataContainer<T> result = new DataContainer<T>();
2005         scheduleRequest(request, procedure, notify, container, result);
2006         acquire(notify, request);
2007         Throwable throwable = container.get();
2008         if (throwable != null) {
2009             if(throwable instanceof DatabaseException) throw (DatabaseException)throwable;
2010             else throw new DatabaseException("Unexpected exception", throwable);
2011         }
2012         return result.get();
2013     }
2014
2015     @Override
2016     public void syncRequest(Write request) throws DatabaseException  {
2017         assertNotSession();
2018         assertAlive();
2019         Semaphore notify = new Semaphore(0);
2020         final DataContainer<DatabaseException> exception = new DataContainer<DatabaseException>();
2021         scheduleRequest(request, e -> exception.set(e), notify);
2022         acquire(notify, request);
2023         if(exception.get() != null) throw exception.get();
2024     }
2025
2026     @Override
2027     public <T> T syncRequest(WriteResult<T> request) throws DatabaseException  {
2028         assertNotSession();
2029         Semaphore notify = new Semaphore(0);
2030         final DataContainer<Throwable> exception = new DataContainer<Throwable>();
2031         final DataContainer<T> result = new DataContainer<T>();
2032         scheduleRequest(request, new Procedure<T>() {
2033
2034             @Override
2035             public void exception(Throwable t) {
2036               exception.set(t);
2037             }
2038
2039             @Override
2040             public void execute(T t) {
2041               result.set(t);
2042             }
2043
2044         }, notify);
2045         acquire(notify, request);
2046         if(exception.get() != null) {
2047             Throwable t = exception.get();
2048             if(t instanceof DatabaseException) throw (DatabaseException)t;
2049             else throw new DatabaseException(t);
2050         }
2051         return result.get();
2052     }
2053
2054     @Override
2055     public <T> T syncRequest(WriteOnlyResult<T> request) throws DatabaseException  {
2056         assertNotSession();
2057         Semaphore notify = new Semaphore(0);
2058         final DataContainer<Throwable> exception = new DataContainer<Throwable>();
2059         final DataContainer<T> result = new DataContainer<T>();
2060         scheduleRequest(request, new Procedure<T>() {
2061
2062             @Override
2063             public void exception(Throwable t) {
2064               exception.set(t);
2065             }
2066
2067             @Override
2068             public void execute(T t) {
2069               result.set(t);
2070             }
2071
2072         }, notify);
2073         acquire(notify, request);
2074         if(exception.get() != null) {
2075             Throwable t = exception.get();
2076             if(t instanceof DatabaseException) throw (DatabaseException)t;
2077             else throw new DatabaseException(t);
2078         }
2079         return result.get();
2080     }
2081
2082     @Override
2083     public <T> T syncRequest(DelayedWriteResult<T> request) throws DatabaseException  {
2084         assertNotSession();
2085         Semaphore notify = new Semaphore(0);
2086         final DataContainer<Throwable> exception = new DataContainer<Throwable>();
2087         final DataContainer<T> result = new DataContainer<T>();
2088         scheduleRequest(request, new Procedure<T>() {
2089
2090             @Override
2091             public void exception(Throwable t) {
2092               exception.set(t);
2093             }
2094
2095             @Override
2096             public void execute(T t) {
2097               result.set(t);
2098             }
2099
2100         }, notify);
2101         acquire(notify, request);
2102         if(exception.get() != null) {
2103             Throwable t = exception.get();
2104             if(t instanceof DatabaseException) throw (DatabaseException)t;
2105             else throw new DatabaseException(t);
2106         }
2107         return result.get();
2108     }
2109
2110     @Override
2111     public void syncRequest(DelayedWrite request) throws DatabaseException  {
2112         assertNotSession();
2113         Semaphore notify = new Semaphore(0);
2114         final DataContainer<DatabaseException> exception = new DataContainer<DatabaseException>();
2115         scheduleRequest(request, e -> exception.set(e), notify);
2116         acquire(notify, request);
2117         if(exception.get() != null) throw exception.get();
2118     }
2119
2120     @Override
2121     public void syncRequest(WriteOnly request) throws DatabaseException  {
2122         assertNotSession();
2123         assertAlive();
2124         Semaphore notify = new Semaphore(0);
2125         final DataContainer<DatabaseException> exception = new DataContainer<DatabaseException>();
2126         scheduleRequest(request, e -> exception.set(e), notify);
2127         acquire(notify, request);
2128         if(exception.get() != null) throw exception.get();
2129     }
2130
2131     /*
2132      *
2133      * GraphRequestProcessor interface
2134      */
2135
2136     @Override
2137     public <T> void asyncRequest(final AsyncRead<T> request, final AsyncProcedure<T> procedure) {
2138
2139         scheduleRequest(request, procedure, null, null);
2140
2141     }
2142
2143     @Override
2144     public <T> void asyncRequest(final AsyncMultiRead<T> request, final AsyncMultiProcedure<T> procedure) {
2145
2146         scheduleRequest(request, procedure, null);
2147
2148     }
2149
2150     @Override
2151     public <T> void asyncRequest(final ExternalRead<T> request, final Procedure<T> procedure) {
2152
2153         scheduleRequest(request, procedure, null, null, null);
2154
2155     }
2156
2157     @Override
2158     public void asyncRequest(final Write request, final Consumer<DatabaseException> callback) {
2159
2160         scheduleRequest(request, callback, null);
2161
2162     }
2163
2164     @Override
2165     public <T> void asyncRequest(final WriteResult<T> request, final Procedure<T> procedure) {
2166
2167         scheduleRequest(request, procedure, null);
2168
2169     }
2170
2171     @Override
2172     public <T> void asyncRequest(final WriteOnlyResult<T> request, final Procedure<T> procedure) {
2173
2174         scheduleRequest(request, procedure, null);
2175
2176     }
2177
2178     @Override
2179     public <T> void asyncRequest(final DelayedWriteResult<T> request, final Procedure<T> procedure) {
2180
2181         scheduleRequest(request, procedure, null);
2182
2183     }
2184
2185     @Override
2186     public void asyncRequest(final DelayedWrite request, final Consumer<DatabaseException> callback) {
2187
2188         scheduleRequest(request, callback, null);
2189
2190     }
2191
2192     @Override
2193     public void asyncRequest(final Write r) {
2194         asyncRequest(r, null);
2195     }
2196
2197     @Override
2198     public void asyncRequest(final DelayedWrite r) {
2199         asyncRequest(r, null);
2200     }
2201
2202     @Override
2203     public void asyncRequest(final WriteOnly request, final Consumer<DatabaseException> callback) {
2204
2205         scheduleRequest(request, callback, null);
2206
2207     }
2208
2209     @Override
2210     public void asyncRequest(final WriteOnly request) {
2211
2212         asyncRequest(request, null);
2213
2214     }
2215
2216     @Override
2217     public <T> void async(ReadInterface<T> r, AsyncProcedure<T> procedure) {
2218         r.request(this, procedure);
2219     }
2220
2221     @Override
2222     public <T> void async(ReadInterface<T> r, Procedure<T> procedure) {
2223         r.request(this, procedure);
2224     }
2225
2226     @Override
2227     public <T> void async(ReadInterface<T> r, SyncProcedure<T> procedure) {
2228         r.request(this, procedure);
2229     }
2230
2231     @Override
2232     public <T> void async(ReadInterface<T> r, AsyncListener<T> procedure) {
2233         r.request(this, procedure);
2234     }
2235
2236     @Override
2237     public <T> void async(ReadInterface<T> r, Listener<T> procedure) {
2238         r.request(this, procedure);
2239     }
2240
2241     @Override
2242     public <T> void async(ReadInterface<T> r, SyncListener<T> procedure) {
2243         r.request(this, procedure);
2244     }
2245
2246     @Override
2247     public <T> T sync(ReadInterface<T> r) throws DatabaseException {
2248         return r.request(this);
2249     }
2250
2251     @Override
2252     public <T> T sync(WriteInterface<T> r) throws DatabaseException {
2253         return r.request(this);
2254     }
2255
2256     @Override
2257     public <T> void async(WriteInterface<T> r, Procedure<T> procedure) {
2258         r.request(this, procedure);
2259     }
2260
2261     @Override
2262     public <T> void async(WriteInterface<T> r) {
2263         r.request(this, new ProcedureAdapter<T>());
2264     }
2265
2266     //@Override
2267     public void incAsync() {
2268         state.incAsync();
2269     }
2270
2271     //@Override
2272     public void decAsync() {
2273         state.decAsync();
2274     }
2275
2276     public long getCluster(ResourceImpl resource) {
2277         ClusterI cluster = clusterTable.getClusterByResourceKey(resource.id);
2278         return cluster.getClusterId();
2279     }
2280
2281     public long getCluster(int id) {
2282         if (clusterTable == null)
2283             System.out.println("SessionImplSocket.getCluster() clusterTable == null !!!!! how come");
2284         return clusterTable.getClusterIdByResourceKeyNoThrow(id);
2285     }
2286
2287     public ResourceImpl getResource(int id) {
2288         return new ResourceImpl(resourceSupport, id);
2289     }
2290
2291     public ResourceImpl getResource(int resourceIndex, long clusterId) {
2292         assert (!ClusterTraitsBase.isIllegalResourceIndex(resourceIndex));
2293         ClusterI proxy = clusterTable.getClusterByClusterId(clusterId);
2294         int key = proxy.getClusterKey();
2295         int resourceKey = ClusterTraitsBase.createResourceKeyNoThrow(key, resourceIndex);
2296         return new ResourceImpl(resourceSupport, resourceKey);
2297     }
2298
2299     public ResourceImpl getResource2(int id) {
2300         assert (id != 0);
2301         return new ResourceImpl(resourceSupport, id);
2302     }
2303
2304     final public int getId(ResourceImpl impl) {
2305         return impl.id;
2306     }
2307
2308     public static final Charset UTF8 = Charset.forName("utf-8");
2309
2310
2311
2312
2313     static boolean areVirtualStatementsLoaded(final VirtualGraphServerSupportImpl support, int subject) {
2314         for(TransientGraph g : support.providers) {
2315             if(g.isPending(subject)) return false;
2316         }
2317         return true;
2318     }
2319
2320     static boolean areVirtualStatementsLoaded(final VirtualGraphServerSupportImpl support, int subject, int predicate) {
2321         for(TransientGraph g : support.providers) {
2322             if(g.isPending(subject, predicate)) return false;
2323         }
2324         return true;
2325     }
2326
2327     static void loadVirtualStatements(final VirtualGraphServerSupportImpl support, ReadGraphImpl graph, int subject, final Consumer<ReadGraphImpl> runnable) {
2328
2329         Consumer<ReadGraphImpl> composite = new Consumer<ReadGraphImpl>() {
2330
2331             AtomicInteger ready = new AtomicInteger(support.providers.size() + 1);
2332
2333             @Override
2334             public void accept(ReadGraphImpl graph) {
2335                 if(ready.decrementAndGet() == 0) {
2336                     runnable.accept(graph);
2337                 }
2338             }
2339
2340         };
2341
2342         for(TransientGraph g : support.providers) {
2343             if(g.isPending(subject)) {
2344                 try {
2345                     g.load(graph, subject, composite);
2346                 } catch (DatabaseException e) {
2347                     e.printStackTrace();
2348                 }
2349             } else {
2350                 composite.accept(graph);
2351             }
2352         }
2353
2354         composite.accept(graph);
2355
2356     }
2357
2358     static  void loadVirtualStatements(final VirtualGraphServerSupportImpl support, ReadGraphImpl graph, int subject, int predicate, final Consumer<ReadGraphImpl> runnable) {
2359
2360         Consumer<ReadGraphImpl> composite = new Consumer<ReadGraphImpl>() {
2361
2362             AtomicInteger ready = new AtomicInteger(support.providers.size() + 1);
2363
2364             @Override
2365             public void accept(ReadGraphImpl graph) {
2366                 if(ready.decrementAndGet() == 0) {
2367                     runnable.accept(graph);
2368                 }
2369             }
2370
2371         };
2372
2373         for(TransientGraph g : support.providers) {
2374             if(g.isPending(subject, predicate)) {
2375                 try {
2376                     g.load(graph, subject, predicate, composite);
2377                 } catch (DatabaseException e) {
2378                     e.printStackTrace();
2379                 }
2380             } else {
2381                 composite.accept(graph);
2382             }
2383         }
2384
2385         composite.accept(graph);
2386
2387     }
2388
2389 //    void dumpHeap() {
2390 //
2391 //        try {
2392 //            ManagementFactory.newPlatformMXBeanProxy(ManagementFactory.getPlatformMBeanServer(),
2393 //                    "com.sun.management:type=HotSpotDiagnostic", HotSpotDiagnosticMXBean.class).dumpHeap(
2394 //                    "d:/heap" + flushCounter + ".txt", true);
2395 //        } catch (IOException e) {
2396 //            e.printStackTrace();
2397 //        }
2398 //
2399 //    }
2400
2401     void fireReactionsToSynchronize(ChangeSet cs) {
2402
2403         // Do not fire empty events
2404         if (cs.isEmpty())
2405             return;
2406
2407         ReadGraphImpl g = ReadGraphImpl.create(getQueryProvider2());
2408 //        g.state.barrier.inc();
2409
2410         try {
2411
2412             if (!cs.isEmpty()) {
2413                 ChangeEvent e2 = new ChangeEvent(g.getSession(), g, null, cs);
2414                 for (ChangeListener l : changeListeners2) {
2415                     try {
2416                         l.graphChanged(e2);
2417                     } catch (Exception ex) {
2418                         ex.printStackTrace();
2419                     }
2420                 }
2421             }
2422
2423         } finally {
2424
2425 //            g.state.barrier.dec();
2426 //            g.waitAsync(null);
2427
2428         }
2429
2430     }
2431
2432     void fireReactionsToCommit(ReadGraphImpl graph, ChangeSet cs2) {
2433         try {
2434
2435             // Do not fire empty events
2436             if (cs2.isEmpty())
2437                 return;
2438
2439 //            graph.restart();
2440 //            graph.state.barrier.inc();
2441
2442             try {
2443
2444                 if (!cs2.isEmpty()) {
2445
2446                     ChangeEvent e2 = new ChangeEvent(graph.getSession(), graph, null, cs2);
2447                     for (ChangeListener l : changeListeners2) {
2448                         try {
2449                             l.graphChanged(e2);
2450 //                            System.out.println("changelistener " + l);
2451                         } catch (Exception ex) {
2452                             ex.printStackTrace();
2453                         }
2454                     }
2455
2456                 }
2457
2458             } finally {
2459
2460 //                graph.state.barrier.dec();
2461 //                graph.waitAsync(null);
2462
2463             }
2464         } catch (Throwable t) {
2465             t.printStackTrace();
2466
2467         }
2468     }
2469     void fireMetadataListeners(WriteGraphImpl graph, ChangeSet cs2) {
2470
2471         try {
2472
2473             // Do not fire empty events
2474             if (cs2.isEmpty())
2475                 return;
2476
2477             // Do not fire on virtual requests
2478             if(graph.getProvider() != null)
2479                 return;
2480
2481             WriteGraphImpl reactionGraph = WriteGraphImpl.create(graph.processor, writeSupport, null);
2482
2483             try {
2484
2485                 ChangeEvent e2 = new ChangeEvent(graph.getSession(), reactionGraph, graph, cs2);
2486                 for (ChangeListener l : metadataListeners) {
2487                     try {
2488                         l.graphChanged(e2);
2489                     } catch (Throwable ex) {
2490                         LOGGER.error("Could not invoke listener {} with event {}", l, e2, ex);
2491                     }
2492                 }
2493
2494             } finally {
2495
2496             }
2497
2498         } catch (Throwable t) {
2499             LOGGER.error("Could not fire metadata listeners {} {}", graph, cs2, t);
2500         }
2501
2502     }
2503
2504
2505     /*
2506      * (non-Javadoc)
2507      *
2508      * @see org.simantics.db.ServiceLocator#getService(java.lang.Class)
2509      */
2510     @Override
2511     public <T> T getService(Class<T> api) {
2512         T t = peekService(api);
2513         if (t == null) {
2514             if (state.isClosed())
2515                 throw new ServiceNotFoundException(this, api, "Session has been shut down");
2516             throw new ServiceNotFoundException(this, api);
2517         }
2518         return t;
2519     }
2520
2521     /**
2522      * @return
2523      */
2524     protected abstract ServerInformation getCachedServerInformation();
2525
2526         private Class<?> serviceKey1 = null;
2527         private Class<?> serviceKey2 = null;
2528         private Object service1 = null;
2529         private Object service2 = null;
2530
2531     /*
2532      * (non-Javadoc)
2533      *
2534      * @see
2535      * fi.vtt.simantics.db.connection.ServiceLocator#peekService(java.lang.Class)
2536      */
2537     @SuppressWarnings("unchecked")
2538     @Override
2539     public synchronized <T> T peekService(Class<T> api) {
2540
2541                 if(serviceKey1 == api) {
2542                         return (T)service1;
2543                 } else if (serviceKey2 == api) {
2544                         // Promote this key
2545                         Object result = service2;
2546                         service2 = service1;
2547                         serviceKey2 = serviceKey1;
2548                         service1 = result;
2549                         serviceKey1 = api;
2550                         return (T)result;
2551                 }
2552
2553         if (Layer0.class == api)
2554                 return (T) L0;
2555         if (ServerInformation.class == api)
2556             return (T) getCachedServerInformation();
2557         else if (WriteGraphImpl.class == api)
2558             return (T) writeState.getGraph();
2559         else if (ClusterBuilder.class == api)
2560             return (T)new ClusterBuilderImpl(this, (WriteOnlySupport)writeState.getGraph().writeSupport);
2561         else if (ClusterBuilderFactory.class == api)
2562             return (T)new ClusterBuilderFactoryImpl(this);
2563
2564                 service2 = service1;
2565                 serviceKey2 = serviceKey1;
2566
2567         service1 = serviceLocator.peekService(api);
2568         serviceKey1 = api;
2569
2570         return (T)service1;
2571
2572     }
2573
2574     /*
2575      * (non-Javadoc)
2576      *
2577      * @see
2578      * fi.vtt.simantics.db.connection.ServiceLocator#hasService(java.lang.Class)
2579      */
2580     @Override
2581     public boolean hasService(Class<?> api) {
2582         return serviceLocator.hasService(api);
2583     }
2584
2585     /**
2586      * @param api the api that must be implemented by the specified service
2587      * @param service the service implementation
2588      */
2589     @Override
2590     public <T> void registerService(Class<T> api, T service) {
2591         if(Layer0.class == api) {
2592                 L0 = (Layer0)service;
2593                 return;
2594         }
2595         serviceLocator.registerService(api, service);
2596         if (TransactionPolicySupport.class == api) {
2597             transactionPolicy = (TransactionPolicySupport)service;
2598             state.resetTransactionPolicy();
2599         }
2600         if (api == serviceKey1)
2601                 service1 = service;
2602         else if (api == serviceKey2)
2603                 service2 = service;
2604     }
2605
2606     // ----------------
2607     // SessionMonitor
2608     // ----------------
2609
2610     void fireSessionVariableChange(String variable) {
2611         for (MonitorHandler h : monitorHandlers) {
2612             MonitorContext ctx = monitorContexts.getLeft(h);
2613             assert ctx != null;
2614             // SafeRunner functionality repeated here to avoid dependency.
2615             try {
2616                 h.valuesChanged(ctx);
2617             } catch (Exception e) {
2618                 Logger.defaultLogError("monitor handler notification produced the following exception", e);
2619             } catch (LinkageError e) {
2620                 Logger.defaultLogError("monitor handler notification produced a linkage error", e);
2621             }
2622         }
2623     }
2624
2625
2626     class ResourceSerializerImpl implements ResourceSerializer {
2627
2628         public long createRandomAccessId(int id) throws DatabaseException {
2629             if(id < 0)
2630                 return id;
2631             int index = ClusterTraitsBase.getResourceIndexFromResourceKey(id);
2632             long cluster = getCluster(id);
2633             if (0 == cluster)
2634                 return 0; // Better to return 0 then invalid id.
2635             long result = ClusterTraitsBase.createResourceId(cluster, index);
2636             return result;
2637         }
2638
2639         @Override
2640         public long getRandomAccessId(Resource resource) throws DatabaseException {
2641
2642             ResourceImpl resourceImpl = (ResourceImpl) resource;
2643             return createRandomAccessId(resourceImpl.id);
2644
2645         }
2646
2647         @Override
2648         public int getTransientId(Resource resource) throws DatabaseException {
2649
2650             ResourceImpl resourceImpl = (ResourceImpl) resource;
2651             return resourceImpl.id;
2652
2653         }
2654
2655         @Override
2656         public String createRandomAccessId(Resource resource)
2657         throws InvalidResourceReferenceException {
2658
2659             if(resource == null) throw new IllegalArgumentException();
2660
2661             ResourceImpl resourceImpl = (ResourceImpl) resource;
2662
2663             if (resourceImpl.id < 0) return String.valueOf(resourceImpl.id) + "_0";
2664
2665             int r;
2666             try {
2667                 r = ClusterTraits.getResourceIndexFromResourceKey(resourceImpl.id);
2668             } catch (DatabaseException e1) {
2669                 throw new InvalidResourceReferenceException(e1);
2670             }
2671             try {
2672                 // Serialize as '<resource index>_<cluster id>'
2673                 return "" + r + "_" + getCluster(resourceImpl);
2674             } catch (Throwable e) {
2675                 e.printStackTrace();
2676                 throw new InvalidResourceReferenceException(e);
2677             } finally {
2678             }
2679         }
2680
2681         public int getTransientId(long serialized) throws DatabaseException {
2682             if (serialized <= 0)
2683                 return (int)serialized;
2684             int index = ClusterTraitsBase.getResourceIndexFromResourceId(serialized);
2685             long c = ClusterTraitsBase.getClusterIdFromResourceId(serialized);
2686             ClusterI cluster = clusterTranslator.getClusterByClusterId(c);
2687             if (cluster == null)
2688                 throw new DatabaseException("Couldn't load cluster for id " + c + " for resource id " + serialized);
2689             int key = ClusterTraits.createResourceKey(cluster.getClusterKey(), index);
2690             return key;
2691         }
2692
2693         @Override
2694         public Resource getResource(long randomAccessId) throws DatabaseException {
2695             return getResourceByKey(getTransientId(randomAccessId));
2696         }
2697
2698         @Override
2699         public Resource getResource(int transientId) throws DatabaseException {
2700             return getResourceByKey(transientId);
2701         }
2702
2703         @Override
2704         public Resource getResource(String randomAccessId)
2705         throws InvalidResourceReferenceException {
2706             try {
2707                 int i = randomAccessId.indexOf('_');
2708                 if (i == -1)
2709                     throw new InvalidResourceReferenceException("Could not parse resource id + cluster id from '"
2710                             + randomAccessId + "'");
2711                 int r = Integer.parseInt(randomAccessId.substring(0, i));
2712                 if(r < 0) return getResourceByKey(r);
2713                 long c = Long.parseLong(randomAccessId.substring(i + 1));
2714                 ClusterI cluster = clusterTranslator.getClusterByClusterId(c);
2715                 int key = ClusterTraits.createResourceKey(cluster.getClusterKey(), r);
2716                 if (cluster.hasResource(key, clusterTranslator))
2717                     return getResourceByKey(key);
2718             } catch (InvalidResourceReferenceException e) {
2719                 throw e;
2720             } catch (NumberFormatException e) {
2721                 throw new InvalidResourceReferenceException(e);
2722             } catch (Throwable e) {
2723                 e.printStackTrace();
2724                 throw new InvalidResourceReferenceException(e);
2725             } finally {
2726             }
2727             throw new InvalidResourceReferenceException("Resource does not exist " + randomAccessId);
2728         }
2729
2730         @Override
2731         public boolean disposeRandomAccessId(String randomAccessId) throws InvalidResourceReferenceException {
2732             try {
2733                 return true;
2734             } catch (Throwable e) {
2735                 e.printStackTrace();
2736                 throw new InvalidResourceReferenceException(e);
2737             } finally {
2738             }
2739         }
2740     }
2741
2742     // Copied from old SessionImpl
2743
2744     void check() {
2745         if (state.isClosed())
2746             throw new Error("Session closed.");
2747     }
2748
2749
2750
2751     public ResourceImpl getNewResource(long clusterId) {
2752         ClusterI cluster = clusterTable.getClusterByClusterIdOrThrow(clusterId);
2753         int newId;
2754         try {
2755             newId = cluster.createResource(clusterTranslator);
2756         } catch (DatabaseException e) {
2757             Logger.defaultLogError(e);
2758             return null;
2759         }
2760         return new ResourceImpl(resourceSupport, newId);
2761     }
2762
2763     public ResourceImpl getNewResource(Resource clusterSet)
2764     throws DatabaseException {
2765         long resourceId = clusterSet.getResourceId();
2766         Long clusterId = clusterSetsSupport.get(resourceId);
2767         if (null == clusterId)
2768             throw new ClusterSetExistException("Cluster set does not exist. Resource=" + clusterSet);
2769         if (Constants.NewClusterId == clusterId) {
2770             clusterId = getService(ClusteringSupport.class).createCluster();
2771             if ((serviceMode & SERVICE_MODE_CREATE) > 0)
2772                 createdClusters.add(clusterId);
2773             clusterSetsSupport.put(resourceId, clusterId);
2774             return getNewResource(clusterId);
2775         } else {
2776             ClusterBase cb = clusterTranslator.getClusterByClusterId(clusterId);
2777             ResourceImpl result;
2778             if (cb.getNumberOfResources(clusterTranslator) >= ClusterTable.CLUSTER_FILL_SIZE) {
2779                 clusterId = getService(ClusteringSupport.class).createCluster();
2780                 if ((serviceMode & SERVICE_MODE_CREATE) > 0)
2781                     createdClusters.add(clusterId);
2782                 clusterSetsSupport.put(resourceId, clusterId);
2783                 return getNewResource(clusterId);
2784             } else {
2785                 result = getNewResource(clusterId);
2786                 int resultKey = querySupport.getId(result);
2787                 long resultCluster = querySupport.getClusterId(resultKey);
2788                 if (clusterId != resultCluster)
2789                     clusterSetsSupport.put(resourceId, resultCluster);
2790                 return result;
2791             }
2792         }
2793     }
2794
2795     public void getNewClusterSet(Resource clusterSet)
2796     throws DatabaseException {
2797         if (DEBUG)
2798             System.out.println("new cluster set=" + clusterSet);
2799         long resourceId = clusterSet.getResourceId();
2800         if (clusterSetsSupport.containsKey(resourceId))
2801             throw new ClusterSetExistException("Cluster set exist already. Resource=" + clusterSet);
2802         clusterSetsSupport.put(resourceId, Constants.NewClusterId);
2803     }
2804     public boolean containsClusterSet(Resource clusterSet)
2805     throws ServiceException {
2806         long resourceId = clusterSet.getResourceId();
2807         return clusterSetsSupport.containsKey(resourceId);
2808     }
2809     public Resource setDefaultClusterSet4NewResource(Resource clusterSet) {
2810         Resource r = defaultClusterSet;
2811         defaultClusterSet = clusterSet;
2812         return r;
2813     }
2814     void printDiagnostics() {
2815
2816         if (DIAGNOSTICS) {
2817
2818             final DataContainer<Long> totalMem = new DataContainer<Long>(0L);
2819             final DataContainer<Long> residentClusters = new DataContainer<Long>(0L);
2820             for(ClusterI cluster : clusterTable.getClusters()) {
2821                 try {
2822                     if (cluster.isLoaded() && !cluster.isEmpty()) {
2823                         residentClusters.set(residentClusters.get() + 1);
2824                         totalMem.set(totalMem.get() + cluster.getUsedSpace());
2825                     }
2826                 } catch (DatabaseException e) {
2827                     Logger.defaultLogError(e);
2828                 }
2829             }
2830
2831             System.out.println("--------------------------------");
2832             System.out.println("Cluster information:");
2833             System.out.println("-amount of resident clusters=" + residentClusters.get());
2834             System.out.println("-total memory usage=" + totalMem.get() / 1024 + "kB");
2835
2836             for(ClusterI cluster : clusterTable.getClusters()) {
2837                 System.out.print("Cluster " + cluster.getClusterId() + " [");
2838                 try {
2839                     if (!cluster.isLoaded())
2840                         System.out.println("not loaded]");
2841                     else if (cluster.isEmpty())
2842                         System.out.println("is empty]");
2843                     else {
2844                         System.out.print("resources=" + cluster.getNumberOfResources(clusterTranslator));
2845                         System.out.print(",references=" + cluster.getReferenceCount());
2846                         System.out.println(",memory=" + cluster.getUsedSpace() / 1024 + "kB]");
2847                     }
2848                 } catch (DatabaseException e) {
2849                     Logger.defaultLogError(e);
2850                     System.out.println("is corrupted]");
2851                 }
2852             }
2853
2854             queryProvider2.printDiagnostics();
2855             System.out.println("--------------------------------");
2856         }
2857
2858     }
2859
2860     public void fireStartReadTransaction() {
2861
2862         if (DIAGNOSTICS)
2863             System.out.println("StartReadTransaction");
2864
2865         for (SessionEventListener listener : eventListeners) {
2866             try {
2867                 listener.readTransactionStarted();
2868             } catch (Throwable t) {
2869                 t.printStackTrace();
2870             }
2871         }
2872
2873     }
2874
2875     public void fireFinishReadTransaction() {
2876
2877         if (DIAGNOSTICS)
2878             System.out.println("FinishReadTransaction");
2879
2880         for (SessionEventListener listener : eventListeners) {
2881             try {
2882                 listener.readTransactionFinished();
2883             } catch (Throwable t) {
2884                 t.printStackTrace();
2885             }
2886         }
2887
2888     }
2889
2890     public void fireStartWriteTransaction() {
2891
2892         if (DIAGNOSTICS)
2893             System.out.println("StartWriteTransaction");
2894
2895         for (SessionEventListener listener : eventListeners) {
2896             try {
2897                 listener.writeTransactionStarted();
2898             } catch (Throwable t) {
2899                 t.printStackTrace();
2900             }
2901         }
2902
2903     }
2904
2905     public void fireFinishWriteTransaction() {
2906
2907         if (DIAGNOSTICS)
2908             System.out.println("FinishWriteTransaction");
2909
2910         for (SessionEventListener listener : eventListeners) {
2911             try {
2912                 listener.writeTransactionFinished();
2913             } catch (Throwable t) {
2914                 t.printStackTrace();
2915             }
2916         }
2917
2918         Indexing.resetDependenciesIndexingDisabled();
2919
2920     }
2921
2922     Resource deserialize(final long resourceId, final long clusterId) throws IOException {
2923         throw new Error("Not supported at the moment.");
2924     }
2925
2926     State getState() {
2927         return state;
2928     }
2929
2930     ClusterTable getClusterTable() {
2931         return clusterTable;
2932     }
2933
2934     public GraphSession getGraphSession() {
2935         return graphSession;
2936     }
2937
2938     public QueryProcessor getQueryProvider2() {
2939         return queryProvider2;
2940     }
2941
2942     ClientChangesImpl getClientChanges() {
2943         return clientChanges;
2944     }
2945
2946     boolean getWriteOnly() {
2947         return writeOnly;
2948     }
2949
2950     static int counter = 0;
2951
2952     public void onClusterLoaded(long clusterId) {
2953
2954         clusterTable.updateSize();
2955
2956     }
2957
2958
2959
2960
2961     /*
2962      * Implementation of the interface RequestProcessor
2963      */
2964
2965     @Override
2966     public <T> T syncRequest(final Read<T> request) throws DatabaseException {
2967
2968         assertNotSession();
2969         assertAlive();
2970
2971         assert(request != null);
2972
2973         final DataContainer<T> result = new DataContainer<T>();
2974         final DataContainer<Throwable> exception = new DataContainer<Throwable>();
2975
2976         syncRequest(request, new AsyncProcedure<T>() {
2977
2978             @Override
2979             public void execute(AsyncReadGraph graph, T t) {
2980                 result.set(t);
2981             }
2982
2983             @Override
2984             public void exception(AsyncReadGraph graph, Throwable t) {
2985                 exception.set(t);
2986             }
2987
2988         });
2989
2990         Throwable t = exception.get();
2991         if(t != null) {
2992             if(t instanceof DatabaseException) throw (DatabaseException)t;
2993             else throw new DatabaseException("Unexpected exception in ReadGraph.syncRequest(Read)", t);
2994         }
2995
2996         return result.get();
2997
2998     }
2999
3000 //    @Override
3001 //    public <T> T syncRequest(Read<T> request, AsyncListener<T> procedure) throws DatabaseException {
3002 //        assertNotSession();
3003 //        return syncRequest(request, (AsyncProcedure<T>)procedure);
3004 //    }
3005
3006     @Override
3007     public <T> T syncRequest(Read<T> request, SyncListener<T> procedure) throws DatabaseException {
3008         assertNotSession();
3009         return syncRequest(request, new SyncToAsyncListener<T>(procedure));
3010     }
3011
3012     @Override
3013     public <T> T syncRequest(Read<T> request, final Listener<T> procedure) throws DatabaseException {
3014         assertNotSession();
3015         return syncRequest(request, new NoneToAsyncListener<T>(procedure));
3016     }
3017
3018     @Override
3019     public <T> T syncRequest(final Read<T> request, final SyncProcedure<T> procedure) throws DatabaseException {
3020         assertNotSession();
3021         return syncRequest(request, new SyncToAsyncProcedure<T>(procedure));
3022     }
3023
3024     @Override
3025     public <T> T syncRequest(Read<T> request, Procedure<T> procedure) throws DatabaseException {
3026         assertNotSession();
3027         return syncRequest(request, new NoneToAsyncProcedure<T>(procedure));
3028     }
3029
3030     @Override
3031     public <T> T syncRequest(final AsyncRead<T> request) throws DatabaseException {
3032
3033         assertNotSession();
3034
3035         assert(request != null);
3036
3037         final DataContainer<T> result = new DataContainer<T>();
3038         final DataContainer<Throwable> exception = new DataContainer<Throwable>();
3039
3040         syncRequest(request, new AsyncProcedure<T>() {
3041
3042             @Override
3043             public void execute(AsyncReadGraph graph, T t) {
3044                 result.set(t);
3045             }
3046
3047             @Override
3048             public void exception(AsyncReadGraph graph, Throwable t) {
3049                 exception.set(t);
3050             }
3051
3052         });
3053
3054         Throwable t = exception.get();
3055         if(t != null) {
3056             if(t instanceof DatabaseException) throw (DatabaseException)t;
3057             else throw new DatabaseException("Unexpected exception in ReadGraph.syncRequest(Read)", t);
3058         }
3059
3060         return result.get();
3061
3062     }
3063
3064     @Override
3065     public <T> T syncRequest(AsyncRead<T> request, AsyncListener<T> procedure) throws DatabaseException {
3066         assertNotSession();
3067         return syncRequest(request, (AsyncProcedure<T>)procedure);
3068     }
3069
3070     @Override
3071     public <T> T syncRequest(AsyncRead<T> request, SyncListener<T> procedure) throws DatabaseException {
3072         assertNotSession();
3073         return syncRequest(request, new SyncToAsyncListener<T>(procedure));
3074     }
3075
3076     @Override
3077     public <T> T syncRequest(AsyncRead<T> request, Listener<T> procedure) throws DatabaseException {
3078         assertNotSession();
3079         return syncRequest(request, new NoneToAsyncListener<T>(procedure));
3080     }
3081
3082     @Override
3083     public <T> T syncRequest(AsyncRead<T> request, final SyncProcedure<T> procedure) throws DatabaseException {
3084         assertNotSession();
3085         return syncRequest(request, new SyncToAsyncProcedure<T>(procedure));
3086     }
3087
3088     @Override
3089     final public <T> T syncRequest(final AsyncRead<T> request, final Procedure<T> procedure) throws DatabaseException {
3090         assertNotSession();
3091         return syncRequest(request, new NoneToAsyncProcedure<T>(procedure));
3092     }
3093
3094     @Override
3095     public <T> Collection<T> syncRequest(MultiRead<T> request) throws DatabaseException {
3096
3097         assertNotSession();
3098
3099         assert(request != null);
3100
3101         final ArrayList<T> result = new ArrayList<T>();
3102         final DataContainer<Throwable> exception = new DataContainer<Throwable>();
3103
3104         syncRequest(request, new SyncMultiProcedure<T>() {
3105
3106             @Override
3107             public void execute(ReadGraph graph, T t) {
3108                 synchronized(result) {
3109                     result.add(t);
3110                 }
3111             }
3112
3113             @Override
3114             public void finished(ReadGraph graph) {
3115             }
3116
3117             @Override
3118             public void exception(ReadGraph graph, Throwable t) {
3119                 exception.set(t);
3120             }
3121
3122
3123         });
3124
3125         Throwable t = exception.get();
3126         if(t != null) {
3127             if(t instanceof DatabaseException) throw (DatabaseException)t;
3128             else throw new DatabaseException("Unexpected exception in ReadGraph.syncRequest(Read)", t);
3129         }
3130
3131         return result;
3132
3133     }
3134
3135     @Override
3136     public <T> Collection<T> syncRequest(MultiRead<T> arg0, SyncMultiProcedure<T> arg1) throws DatabaseException {
3137         assertNotSession();
3138         throw new Error("Not implemented!");
3139     }
3140
3141     @Override
3142     public <T> Collection<T> syncRequest(MultiRead<T> request, SyncMultiListener<T> procedure) throws DatabaseException {
3143         assertNotSession();
3144         return syncRequest(request, (SyncMultiProcedure<T>)procedure);
3145     }
3146
3147     @Override
3148     public <T> Collection<T> syncRequest(MultiRead<T> request, MultiListener<T> procedure) throws DatabaseException {
3149         assertNotSession();
3150         return syncRequest(request, new NoneToSyncMultiListener<T>(procedure));
3151     }
3152
3153     @Override
3154     public <T> Collection<T> syncRequest(MultiRead<T> request, MultiProcedure<T> procedure) throws DatabaseException {
3155         assertNotSession();
3156         return syncRequest(request, new NoneToSyncMultiProcedure<T>(procedure));
3157     }
3158
3159     @Override
3160     final public <T> Collection<T> syncRequest(final AsyncMultiRead<T> request) throws DatabaseException {
3161
3162         assertNotSession();
3163
3164         assert(request != null);
3165
3166         final ArrayList<T> result = new ArrayList<T>();
3167         final DataContainer<Throwable> exception = new DataContainer<Throwable>();
3168
3169         syncRequest(request, new AsyncMultiProcedure<T>() {
3170
3171             @Override
3172             public void execute(AsyncReadGraph graph, T t) {
3173                 synchronized(result) {
3174                     result.add(t);
3175                 }
3176             }
3177
3178             @Override
3179             public void finished(AsyncReadGraph graph) {
3180             }
3181
3182             @Override
3183             public void exception(AsyncReadGraph graph, Throwable t) {
3184                 exception.set(t);
3185             }
3186
3187
3188         });
3189
3190         Throwable t = exception.get();
3191         if (t != null) {
3192             if (t instanceof DatabaseException)
3193                 throw (DatabaseException) t;
3194             else
3195                 throw new DatabaseException("Unexpected exception in ReadGraph.syncRequest(Read)", t);
3196         }
3197
3198         return result;
3199
3200     }
3201
3202     @Override
3203     public <T> Collection<T> syncRequest(AsyncMultiRead<T> request, AsyncMultiListener<T> procedure) throws DatabaseException {
3204         assertNotSession();
3205         return syncRequest(request, (AsyncMultiProcedure<T>)procedure);
3206     }
3207
3208     @Override
3209     public <T> Collection<T> syncRequest(AsyncMultiRead<T> request, SyncMultiListener<T> procedure) throws DatabaseException {
3210         assertNotSession();
3211         return syncRequest(request, new SyncToAsyncMultiListener<T>(procedure));
3212     }
3213
3214     @Override
3215     public <T> Collection<T> syncRequest(AsyncMultiRead<T> request, MultiListener<T> procedure) throws DatabaseException {
3216         assertNotSession();
3217        return syncRequest(request, new NoneToAsyncMultiListener<T>(procedure));
3218     }
3219
3220     @Override
3221     public <T> Collection<T> syncRequest(AsyncMultiRead<T> request, final SyncMultiProcedure<T> procedure) throws DatabaseException {
3222         assertNotSession();
3223         return syncRequest(request, new SyncToAsyncMultiProcedure<T>(procedure));
3224     }
3225
3226     @Override
3227     final public <T> Collection<T> syncRequest(final AsyncMultiRead<T> request, final MultiProcedure<T> procedure) throws DatabaseException {
3228         assertNotSession();
3229         return syncRequest(request, new NoneToAsyncMultiProcedure<T>(procedure));
3230     }
3231
3232
3233     @Override
3234     public <T> void asyncRequest(Read<T> request) {
3235
3236         asyncRequest(request, new ProcedureAdapter<T>() {
3237             @Override
3238             public void exception(Throwable t) {
3239                 t.printStackTrace();
3240             }
3241         });
3242
3243     }
3244
3245     @Override
3246     public <T> void asyncRequest(Read<T> request, AsyncListener<T> procedure) {
3247         asyncRequest(request, (AsyncProcedure<T>)procedure);
3248     }
3249
3250     @Override
3251     public <T> void asyncRequest(Read<T> request, final SyncListener<T> procedure) {
3252         asyncRequest(request, new SyncToAsyncListener<T>(procedure));
3253     }
3254
3255     @Override
3256     public <T> void asyncRequest(Read<T> request, final Listener<T> procedure) {
3257         asyncRequest(request, new NoneToAsyncListener<T>(procedure));
3258     }
3259
3260     @Override
3261     public <T> void asyncRequest(Read<T> request, SyncProcedure<T> procedure) {
3262         asyncRequest(request, new SyncToAsyncProcedure<T>(procedure));
3263     }
3264
3265     @Override
3266     public <T> void asyncRequest(Read<T> request, final Procedure<T> procedure) {
3267         asyncRequest(request, new NoneToAsyncProcedure<T>(procedure));
3268     }
3269
3270     @Override
3271     final public <T> void asyncRequest(final AsyncRead<T> request) {
3272
3273         assert(request != null);
3274
3275         asyncRequest(request, new ProcedureAdapter<T>() {
3276             @Override
3277             public void exception(Throwable t) {
3278                 t.printStackTrace();
3279             }
3280         });
3281
3282     }
3283
3284     @Override
3285     public <T> void asyncRequest(AsyncRead<T> request, AsyncListener<T> procedure) {
3286         scheduleRequest(request, procedure, procedure, null);
3287     }
3288
3289     @Override
3290     public <T> void asyncRequest(AsyncRead<T> request, final SyncListener<T> procedure) {
3291         asyncRequest(request, new SyncToAsyncListener<T>(procedure));
3292     }
3293
3294     @Override
3295     public <T> void asyncRequest(AsyncRead<T> request, final Listener<T> procedure) {
3296         asyncRequest(request, new NoneToAsyncListener<T>(procedure));
3297     }
3298
3299     @Override
3300     public <T> void asyncRequest(AsyncRead<T> request, SyncProcedure<T> procedure) {
3301         asyncRequest(request, new SyncToAsyncProcedure<T>(procedure));
3302     }
3303
3304     @Override
3305     final public <T> void asyncRequest(final AsyncRead<T> request, final Procedure<T> procedure) {
3306         asyncRequest(request, new NoneToAsyncProcedure<T>(procedure));
3307     }
3308
3309     @Override
3310     public <T> void asyncRequest(MultiRead<T> request) {
3311
3312         assert(request != null);
3313
3314         asyncRequest(request, new SyncMultiProcedureAdapter<T>() {
3315             @Override
3316             public void exception(ReadGraph graph, Throwable t) {
3317                 t.printStackTrace();
3318             }
3319         });
3320
3321     }
3322
3323     @Override
3324     public <T> void asyncRequest(MultiRead<T> request, SyncMultiListener<T> procedure) {
3325         asyncRequest(request, (SyncMultiProcedure<T>)procedure);
3326     }
3327
3328     @Override
3329     public <T> void asyncRequest(MultiRead<T> request, MultiListener<T> procedure) {
3330         asyncRequest(request, new NoneToSyncMultiListener<T>(procedure));
3331     }
3332
3333     @Override
3334     public <T> void asyncRequest(MultiRead<T> request, SyncMultiProcedure<T> procedure) {
3335         scheduleRequest(request, procedure, null);
3336     }
3337
3338     @Override
3339     public <T> void asyncRequest(MultiRead<T> request, MultiProcedure<T> procedure) {
3340         asyncRequest(request, new NoneToSyncMultiProcedure<T>(procedure));
3341     }
3342
3343     @Override
3344     final public <T> void asyncRequest(AsyncMultiRead<T> request) {
3345
3346         assert(request != null);
3347
3348         asyncRequest(request, new AsyncMultiProcedureAdapter<T>() {
3349             @Override
3350             public void exception(AsyncReadGraph graph, Throwable t) {
3351                 t.printStackTrace();
3352             }
3353         });
3354
3355     }
3356
3357     @Override
3358     public <T> void asyncRequest(AsyncMultiRead<T> request, AsyncMultiListener<T> procedure) {
3359         asyncRequest(request, (AsyncMultiProcedure<T>)procedure);
3360     }
3361
3362     @Override
3363     public <T> void asyncRequest(AsyncMultiRead<T> request, SyncMultiListener<T> procedure) {
3364         asyncRequest(request, new SyncToAsyncMultiListener<T>(procedure));
3365     }
3366
3367     @Override
3368     public <T> void asyncRequest(AsyncMultiRead<T> request, MultiListener<T> procedure) {
3369         asyncRequest(request, new NoneToAsyncMultiListener<T>(procedure));
3370     }
3371
3372     @Override
3373     public <T> void asyncRequest(AsyncMultiRead<T> request, SyncMultiProcedure<T> procedure) {
3374         asyncRequest(request, new SyncToAsyncMultiProcedure<T>(procedure));
3375     }
3376
3377     @Override
3378     final public <T> void asyncRequest(AsyncMultiRead<T> request, final MultiProcedure<T> procedure) {
3379         asyncRequest(request, new NoneToAsyncMultiProcedure<T>(procedure));
3380     }
3381
3382     @Override
3383     final public <T> void asyncRequest(final ExternalRead<T> request) {
3384
3385         assert(request != null);
3386
3387         asyncRequest(request, new ProcedureAdapter<T>() {
3388             @Override
3389             public void exception(Throwable t) {
3390                 t.printStackTrace();
3391             }
3392         });
3393
3394     }
3395
3396     @Override
3397     public <T> void asyncRequest(ExternalRead<T> request, final Listener<T> procedure) {
3398         asyncRequest(request, (Procedure<T>)procedure);
3399     }
3400
3401     boolean sameProvider(Write request) {
3402         if(writeState.getGraph().provider != null) {
3403             return writeState.getGraph().provider.equals(request.getProvider());
3404         } else {
3405             return request.getProvider() == null;
3406         }
3407     }
3408
3409     boolean plainWrite(WriteGraphImpl graph) {
3410         if(graph == null) return false;
3411         if(graph.writeSupport.writeOnly()) return false;
3412         return true;
3413     }
3414
3415
3416     public static final ThreadGroup SessionThreadGroup = new ThreadGroup("Session Thread Group");
3417
3418     private void assertNotSession() throws DatabaseException {
3419         Thread current = Thread.currentThread();
3420         if (sessionThreads.contains(current))
3421             throw new ServiceException("Caller is already inside a transaction.");
3422     }
3423
3424     void assertAlive() {
3425         if (!state.isAlive())
3426             throw new RuntimeDatabaseException("Session has been shut down.");
3427     }
3428
3429     public InputStream getValueStream(ReadGraphImpl graph, Resource resource) {
3430         return querySupport.getValueStream(graph, querySupport.getId(resource));
3431     }
3432
3433     public byte[] getValue(ReadGraphImpl graph, Resource resource) {
3434         return querySupport.getValue(graph, querySupport.getId(resource));
3435     }
3436
3437     <T> void acquire(Semaphore semaphore, T request) throws DatabaseException {
3438         acquire(semaphore, request, null);
3439     }
3440
3441     private <T,P> void acquire(Semaphore semaphore, T request, P procedure) throws DatabaseException {
3442         assertAlive();
3443         try {
3444             long tryCount = 0;
3445             // Loop until the semaphore is acquired reporting requests that take a long time.
3446             while (true) {
3447
3448                 if (semaphore.tryAcquire(DebugPolicy.LONG_EXECUTION_REPORT_PERIOD, DebugPolicy.LONG_EXECUTION_REPORT_PERIOD_UNIT)) {
3449                     // Acquired OK.
3450                     return;
3451                 } else {
3452                     if (DebugPolicy.REPORT_TIME_CONSUMING_REQUESTS) {
3453                         ++tryCount;
3454                         long waitTime = DebugPolicy.LONG_EXECUTION_REPORT_PERIOD_UNIT
3455                                 .toMillis(DebugPolicy.LONG_EXECUTION_REPORT_PERIOD)
3456                                 * tryCount;
3457                         System.err.println("DB client request '" + request + "' is taking long to execute, so far "
3458                                 + (waitTime) + " ms. (procedure=" + procedure + ")");
3459                     }
3460                 }
3461
3462                 assertAlive();
3463
3464             }
3465         } catch (InterruptedException e) {
3466             e.printStackTrace();
3467             // FIXME: Should perhaps do something else in this case ??
3468         }
3469     }
3470
3471
3472
3473 //    @Override
3474 //    public boolean holdOnToTransactionAfterCancel() {
3475 //        return transactionPolicy.holdOnToTransactionAfterCancel();
3476 //    }
3477 //
3478 //    @Override
3479 //    public boolean holdOnToTransactionAfterCommit() {
3480 //        return transactionPolicy.holdOnToTransactionAfterCommit();
3481 //    }
3482 //
3483 //    @Override
3484 //    public boolean holdOnToTransactionAfterRead() {
3485 //        return transactionPolicy.holdOnToTransactionAfterRead();
3486 //    }
3487 //
3488 //    @Override
3489 //    public void onRelinquish() {
3490 //        transactionPolicy.onRelinquish();
3491 //    }
3492 //
3493 //    @Override
3494 //    public void onRelinquishDone() {
3495 //        transactionPolicy.onRelinquishDone();
3496 //    }
3497 //
3498 //    @Override
3499 //    public void onRelinquishError() {
3500 //        transactionPolicy.onRelinquishError();
3501 //    }
3502
3503
3504     @Override
3505     public Session getSession() {
3506         return this;
3507     }
3508
3509
3510     protected abstract VirtualGraph getProvider(VirtualGraph virtualGraph);
3511     protected abstract ResourceImpl getNewResource() throws DatabaseException;
3512
3513     public void ceased(int thread) {
3514
3515         requestManager.ceased(thread);
3516
3517     }
3518
3519     public int getAmountOfQueryThreads() {
3520         // This must be a power of two
3521         return 1;
3522 //        return Integer.highestOneBit(Runtime.getRuntime().availableProcessors());
3523     }
3524
3525     public Resource getResourceByKey(int key) throws ResourceNotFoundException {
3526
3527         return new ResourceImpl(resourceSupport, key);
3528
3529     }
3530
3531     public void acquireWriteOnly() {
3532         writeOnly = true;
3533     }
3534
3535     public void releaseWriteOnly(ReadGraphImpl graph) {
3536         writeOnly = false;
3537         queryProvider2.releaseWrite(graph);
3538     }
3539
3540     public void handleUpdatesAndMetadata(WriteGraphImpl writer) {
3541
3542         long start = System.nanoTime();
3543
3544         while(dirtyPrimitives) {
3545             dirtyPrimitives = false;
3546             getQueryProvider2().propagateChangesInQueryCache(writer);
3547             getQueryProvider2().listening.fireListeners(writer);
3548         }
3549
3550         fireMetadataListeners(writer, clientChanges);
3551
3552         if(DebugPolicy.PERFORMANCE_DATA) {
3553             long end = System.nanoTime();
3554             System.err.println("handleUpdatesAndMetadata " + 1e-9*(end-start));
3555         }
3556
3557     }
3558
3559     void removeTemporaryData() {
3560         File platform = Platform.getLocation().toFile();
3561         File tempFiles = new File(platform, "tempFiles");
3562         File temp = new File(tempFiles, "db");
3563         if(!temp.exists())
3564             return;
3565         try {
3566             Files.walkFileTree(temp.toPath(), new SimpleFileVisitor<Path>() {
3567                 @Override
3568                 public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) {
3569                     try {
3570                         Files.delete(file);
3571                     } catch (IOException e) {
3572                         Logger.defaultLogError(e);
3573                     }
3574                     return FileVisitResult.CONTINUE;
3575                 }
3576             });
3577         } catch (IOException e) {
3578             Logger.defaultLogError(e);
3579         }
3580     }
3581
3582     public void handleCreatedClusters() {
3583         if ((serviceMode & SERVICE_MODE_CREATE) > 0) {
3584             createdClusters.forEach(new TLongProcedure() {
3585
3586                 @Override
3587                 public boolean execute(long value) {
3588                     ClusterImpl cluster = clusterTable.getClusterByClusterId(value);
3589                     cluster.setImmutable(true, clusterTranslator);
3590                     return true;
3591                 }
3592             });
3593         }
3594         createdClusters.clear();
3595     }
3596
3597     @Override
3598     public Object getModificationCounter() {
3599         return queryProvider2.modificationCounter;
3600     }
3601
3602     @Override
3603     public void markUndoPoint() {
3604         state.setCombine(false);
3605     }
3606
3607 }