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