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