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

   1 package fi.vtt.simantics.procore.internal;
   2
   3 import java.io.IOException;
   4 import java.util.Collection;
   5 import java.util.LinkedList;
   6
   7 import org.simantics.db.Disposable;
   8 import org.simantics.db.Resource;
   9 import org.simantics.db.common.utils.Logger;
  10 import org.simantics.db.exception.CancelTransactionException;
  11 import org.simantics.db.exception.DatabaseException;
  12 import org.simantics.db.impl.graph.WriteGraphImpl;
  13 import org.simantics.db.impl.internal.RandomAccessValueSupport;
  14 import org.simantics.db.impl.internal.ResourceData;
  15 import org.simantics.db.impl.query.QueryProcessor.SessionRead;
  16 import org.simantics.db.impl.query.QueryProcessor.SessionTask;
  17 import org.simantics.db.request.WriteTraits;
  18 import org.simantics.utils.datastructures.Pair;
  19
  20 public class SessionRequestManager {
  21
  22         private static boolean DEBUG = false;
  23
  24         enum State {
  25
  26                 INIT, IDLE, READ, WRITE, WRITE_UPDATE, READ_UPDATE, ERROR
  27
  28         }
  29
  30         final SessionImplSocket session;
  31         final fi.vtt.simantics.procore.internal.State transactionState;
  32         final LinkedList<SessionTask> writes = new LinkedList<SessionTask>();
  33         final LinkedList<SessionRead> reads = new LinkedList<SessionRead>();
  34
  35         State state = State.INIT;
  36
  37         private boolean legal(State current, State target) {
  38
  39                 if(State.IDLE == current) {
  40
  41                         return State.INIT != target && State.WRITE_UPDATE != target && State.READ_UPDATE != target;
  42
  43                 } else if (State.WRITE == current) {
  44
  45                         return State.WRITE_UPDATE == target;
  46
  47                 } else if (State.READ == current) {
  48
  49                         return State.READ_UPDATE == target;
  50
  51                 } else if (State.ERROR == current) {
  52
  53                         return State.IDLE == target;
  54
  55                 } else {
  56
  57                         return State.IDLE == target;
  58
  59                 }
  60
  61         }
  62
  63         private synchronized void setState(int thread, State state, SessionTask task) {
  64
  65                 if(DEBUG) System.err.println("Session state: " + this.state + " => " + state + " - " + task);
  66                 if(this.state == state) return;
  67
  68                 assert(legal(this.state, state));
  69
  70                 if(State.READ == state) {
  71
  72                         assert(State.IDLE == this.state);
  73
  74                 }
  75
  76                 State oldState = this.state;
  77                 this.state = state;
  78
  79                 if(State.IDLE == oldState && State.IDLE != state) {
  80                     session.incAsync();
  81                 }
  82
  83                 /*
  84                  * We are now executing in new state.
  85                  * The methods below SHALL NOT start any task
  86                  * immediately. All tasks shall be scheduled.
  87                  *
  88                  */
  89                 if(State.READ == state) startRead(thread, (SessionRead)task);
  90                 else if(State.WRITE == state) startWrite(thread, task);
  91                 else if(State.READ == oldState) startReadUpdate(thread);
  92                 else if(State.WRITE == oldState) startWriteUpdate(thread);
  93
  94                 // Check the need for new transaction
  95                 if(state == State.IDLE) {
  96
  97                         closeRandomAccessValues();
  98
  99                         if(!writes.isEmpty()) {
 100                                 setState(thread, State.WRITE, writes.poll());
 101                         } else if (!reads.isEmpty()) {
 102                                 setState(thread, State.READ, reads.poll());
 103                         }
 104
 105             session.decAsync();
 106
 107                 }
 108
 109         }
 110
 111         public SessionRequestManager(SessionImplSocket session, fi.vtt.simantics.procore.internal.State transactionState) {
 112                 this.session = session;
 113                 this.transactionState = transactionState;
 114         }
 115
 116         public synchronized void startRead(int thread, final SessionRead task) {
 117
 118                 session.queryProvider2.scheduleAlways(thread, new SessionTask(task.object, task.thread, task.syncCaller) {
 119
 120                         @Override
 121             public void run(int thread) {
 122                 try {
 123                     transactionState.startReadTransaction(thread);
 124                     task.run(thread);
 125                 } catch (Throwable t) {
 126                     Logger.defaultLogError(new DatabaseException("Read transaction could not be started", t));
 127                     if(task.throwable != null)
 128                         task.throwable.set(t);
 129                     state = State.ERROR;
 130                 } finally {
 131                     if(task.notify != null)
 132                         task.notify.release();
 133                 }
 134             }
 135
 136                 });
 137
 138         }
 139
 140         public synchronized void startReadUpdate(int thread) {
 141
 142                 session.queryProvider2.scheduleAlways(thread, new SessionTask(null, thread) {
 143
 144                         @Override
 145                         public void run(int thread) {
 146
 147                                 session.fireFinishReadTransaction();
 148
 149                                 try {
 150                                         transactionState.stopReadTransaction();
 151                                 } catch (DatabaseException e) {
 152                                         e.printStackTrace();
 153                                 }
 154
 155                         }
 156
 157                 });
 158
 159         }
 160
 161 //      public synchronized void stopRead(int thread) {
 162 //
 163 //              try {
 164 //                      transactionState.stopReadTransaction();
 165 //              } catch (DatabaseException e) {
 166 //                      e.printStackTrace();
 167 //              }
 168 //
 169 //      }
 170
 171         public synchronized void startWrite(int thread, final SessionTask task) {
 172
 173                 session.queryProvider2.scheduleAlways(thread, new SessionTask((WriteTraits)task.object, task.thread) {
 174
 175                         @Override
 176                         public void run(int thread) {
 177
 178                         try {
 179                                 transactionState.startWriteTransaction(thread);
 180                         } catch (Throwable t) {
 181                             DatabaseException e = new DatabaseException("Write transaction could not be started", t);
 182                             Logger.defaultLogError(e);
 183                             return;
 184                         }
 185                                 task.run(thread);
 186
 187                         }
 188
 189                 });
 190
 191         }
 192
 193         public synchronized void startWriteUpdate(int thread) {
 194
 195                 session.queryProvider2.scheduleAlways(thread, new SessionTask(null, thread) {
 196
 197                         @Override
 198                         public void run(int thread) {
 199
 200                                 // Support for DelayedWriteRequest cancels during the
 201                                 // read-only part of the request.
 202                                 WriteStateBase<?> delayedState = session.delayedWriteState;
 203                                 session.delayedWriteState = null;
 204                                 if (delayedState != null) {
 205                                         if (session.writeState != null)
 206                                                 throw new AssertionError("Both session delayedWriteState and writeState are set, only one can be set.");
 207
 208                         if (delayedState.isExcepted()) {
 209                         // There can never be any changes in the cluster stream.
 210                         // In this case, don't bother rolling back.
 211                             boolean empty = session.clusterStream.reallyFlush();
 212                         assert(empty);
 213                             transactionState.stopWriteTransaction(session.clusterStream);
 214                         } else {
 215                             throw new UnsupportedOperationException("delayedWriteState may only exist when request fails.");
 216                         }
 217                         Disposable.safeDispose(session.clientChanges);
 218                         session.clientChanges = new ClientChangesImpl(session);
 219                                         delayedState.finish();
 220                                         return;
 221                                 }
 222
 223                                 // The session could have been terminated
 224                                 if(!session.state.isAlive()) return;
 225
 226                                 WriteState<?> writeState = session.writeState;
 227                                 WriteGraphImpl graph = writeState.getGraph();
 228
 229                                 if(writeState.isExcepted()) {
 230
 231                                     if(!(writeState.exception instanceof CancelTransactionException))
 232                                         writeState.exception.printStackTrace();
 233
 234                                     transactionState.cancelWriteTransaction(graph);
 235
 236                                 } else {
 237
 238                                         session.handleUpdatesAndMetadata(graph);
 239
 240 //                    long start = System.nanoTime();
 241                                         transactionState.commitWriteTransaction(writeState.getGraph(), session.clusterStream, session.clientChanges, writeState.getRequest(), null);
 242 //                                      long duration = System.nanoTime() - start;
 243 //                    System.out.println("commitWriteTransaction " + 1e-9*duration + "s. ");
 244
 245                                 }
 246
 247                                 Disposable.safeDispose(session.clientChanges);
 248                                 session.clientChanges = new ClientChangesImpl(session);
 249
 250                                 WriteState<?> state = session.writeState;
 251                                 state.finish();
 252                                 session.writeState = null;
 253
 254                         }
 255
 256                 });
 257
 258         }
 259
 260 //      public synchronized void stopWrite(int thread) {
 261 //
 262 //        session.clientChanges = new ClientChangesImpl(session);
 263 //
 264 //        WriteState<?> state = session.writeState;
 265 //
 266 //        System.err.println("D");
 267 //        state.finish();
 268 //        System.err.println("E");
 269 //
 270 //        session.writeState = null;
 271 //
 272 //      }
 273
 274         public synchronized void ceased(int thread) {
 275
 276                 if(State.WRITE == state) {
 277
 278                         setState(thread, State.WRITE_UPDATE, null);
 279
 280                 } else if (State.WRITE_UPDATE == state) {
 281
 282                         setState(thread, State.IDLE, null);
 283
 284                 } else if (State.READ_UPDATE == state) {
 285
 286                         setState(thread, State.IDLE, null);
 287
 288                 } else if (State.READ == state) {
 289
 290                         if (!reads.isEmpty()) {
 291
 292                                 final SessionRead read = reads.poll();
 293                                 session.queryProvider2.scheduleAlways(thread, new SessionTask(read.object, read.thread, read.syncCaller) {
 294
 295                                         @Override
 296                                         public void run(int thread) {
 297                                                 read.run(thread);
 298                                                 if(read.notify != null) read.notify.release();
 299                                         }
 300
 301                                 });
 302
 303                         } else {
 304
 305                                 setState(thread, State.READ_UPDATE, null);
 306
 307                         }
 308
 309                 } else if (State.INIT == state) {
 310
 311                         assert(reads.isEmpty());
 312                         assert(writes.isEmpty());
 313                         setState(thread, State.IDLE, null);
 314
 315                 } else if (State.ERROR == state) {
 316
 317                         setState(thread, State.IDLE, null);
 318
 319                 } else {
 320
 321                     throw new IllegalStateException("State in ceased should be WRITE or READ or INIT (was " + state + ")");
 322
 323                 }
 324
 325         }
 326
 327     public synchronized void scheduleRead(final SessionRead task) {
 328
 329                 assert(State.INIT != state);
 330
 331                 if(State.READ == state) {
 332                         session.queryProvider2.schedule(Integer.MIN_VALUE, new SessionTask(task.object, task.thread, task.syncCaller) {
 333
 334                                 @Override
 335                                 public void run(int thread) {
 336                                         try {
 337                                                 task.run(thread);
 338                                         } finally {
 339                                                 if (task.notify != null) task.notify.release();
 340                                         }
 341                                 }
 342
 343                         });
 344                 } else if (State.IDLE == state) {
 345                         setState(Integer.MIN_VALUE, State.READ, task);
 346                 } else {
 347                         reads.offer(task);
 348                 }
 349
 350         }
 351
 352     public int lastUpdateIndex() {
 353         return writes.size();
 354     }
 355
 356         public synchronized void scheduleWrite(SessionTask task) {
 357
 358                 scheduleWrite(task, null);
 359
 360         }
 361
 362         /**
 363          * @param task the task to schedule
 364          * @param combine <code>true</code> or <code>false</code> to explicitly
 365          *        specify whether to combine the write with the previous operation
 366          *        or <code>null</code> to use the default logic, i.e. combine when
 367          *        this schedule is performed from within update processing (state ==
 368          *        State.WRITE_UPDATE).
 369          */
 370         public synchronized void scheduleWrite(SessionTask task, Boolean combine) {
 371
 372                 boolean inUpdate = state == State.WRITE_UPDATE;
 373
 374                 assert(State.INIT != state);
 375                 //task.combine = combine != null ? combine : inUpdate;
 376                 if(State.IDLE == state) {
 377                         setState(Integer.MIN_VALUE, State.WRITE, task);
 378                 } else {
 379                         if(inUpdate) {
 380                                 int index = lastUpdateIndex();
 381                                 if(index == writes.size()) writes.offer(task);
 382                                 else writes.add(index, task);
 383                         } else {
 384                                 writes.offer(task);
 385                         }
 386                 }
 387
 388         }
 389
 390         /**
 391          * Unregisters and closes all currently registered ResourceData random
 392          * access value file handles. Any I/O exceptions occurring during closing
 393          * are logged into {@link Logger}.
 394          */
 395         private void closeRandomAccessValues() {
 396                 RandomAccessValueSupport ravs = session.peekService(RandomAccessValueSupport.class);
 397                 if (ravs == null)
 398                         return;
 399                 Collection<Pair<Resource, ResourceData>> values = ravs.removeAll();
 400                 for (Pair<Resource, ResourceData> value : values) {
 401                         try {
 402                                 value.second.binaryFile.close();
 403                         } catch (IOException e) {
 404                                 Logger.defaultLogError("I/O exception while closing random access value file " + value.second.binaryFile.file() + " for resource " + value.first , e);
 405                         }
 406                 }
 407         }
 408
 409 }