Dispose ClientChangesImpl ChangeSets to minimize memory footprint
[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.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 }