Java > Open Source Codes > fr > dyade > aaa > util > NTransaction


1 /*
2  * Copyright (C) 2004 - 2006 ScalAgent Distributed Technologies
3  *
4  * This library is free software; you can redistribute it and/or
5  * modify it under the terms of the GNU Lesser General Public
6  * License as published by the Free Software Foundation; either
7  * version 2.1 of the License, or any later version.
8  *
9  * This library is distributed in the hope that it will be useful,
10  * but WITHOUT ANY WARRANTY; without even the implied warranty of
11  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
12  * Lesser General Public License for more details.
13  *
14  * You should have received a copy of the GNU Lesser General Public
15  * License along with this library; if not, write to the Free Software
16  * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
17  * USA.
18  *
19  * Initial developer(s): ScalAgent Distributed Technologies
20  * Contributor(s):
21  */
22 package fr.dyade.aaa.util;
23
24 import java.io.*;
25 import java.util.*;
26
27 import org.objectweb.util.monolog.api.BasicLevel;
28 import org.objectweb.util.monolog.api.Logger;
29
30 public final class NTransaction implements Transaction, NTransactionMBean {
31   // Logging monitor
32 private static Logger logmon = null;
33
34   /**
35    * Global in memory log initial capacity, by default 4096.
36    * This value can be adjusted for a particular server by setting
37    * <code>NTLogMemoryCapacity</code> specific property.
38    * <p>
39    * These property can be fixed either from <code>java</code> launching
40    * command, or in <code>a3servers.xml</code> configuration file.
41    */
42   static int LogMemoryCapacity = 4096;
43
44   /**
45    * Returns the initial capacity of global in memory log (by default 4096).
46    *
47    * @return The initial capacity of global in memory log.
48    */
49   public int getLogMemoryCapacity() {
50     return LogMemoryCapacity;
51   }
52
53   /**
54    * Maximum size of memory log, by default 2048Kb.
55    * This value can be adjusted (Kb) for a particular server by setting
56    * <code>NTLogMemorySize</code> specific property.
57    * <p>
58    * These property can be fixed either from <code>java</code> launching
59    * command, or in <code>a3servers.xml</code> configuration file.
60    */
61   static int MaxLogMemorySize = 2048 * Kb;
62
63   /**
64    * Returns the maximum size of memory log in Kb, by default 2048Kb.
65    *
66    * @return The maximum size of memory log in Kb.
67    */
68   public int getMaxLogMemorySize() {
69     return MaxLogMemorySize/Mb;
70   }
71
72   /**
73    * Sets the maximum size of memory log in Kb.
74    *
75    * @param size The maximum size of memory log in Kb.
76    */
77   public void setMaxLogMemorySize(int size) {
78     if (size > 0) MaxLogMemorySize = size *Mb;
79   }
80
81   /**
82    * Returns the size of memory log in byte.
83    *
84    * @return The size of memory log in byte.
85    */
86   public int getLogMemorySize() {
87     return logFile.logMemorySize;
88   }
89
90   /**
91    * Size of disk log in Mb, by default 16Mb.
92    * This value can be adjusted (Mb) for a particular server by setting
93    * <code>NTLogFileSize</code> specific property.
94    * <p>
95    * These property can be fixed either from <code>java</code> launching
96    * command, or in <code>a3servers.xml</code> configuration file.
97    */
98   static int LogFileSize = 16 * Mb;
99
100   /**
101    * Returns the size of disk log in Mb, by default 16Mb.
102    *
103    * @return The size of disk log in Mb.
104    */
105   public int getLogFileSize() {
106     return LogFileSize/Mb;
107   }
108
109   /**
110    * Sets the size of disk log in Mb.
111    *
112    * @param size The size of disk log in Mb.
113    */
114   public void setLogFileSize(int size) {
115     if (size > 0) LogFileSize = size *Mb;
116   }
117
118   /**
119    * Number of pooled operation, by default 1000.
120    * This value can be adjusted for a particular server by setting
121    * <code>NTLogThresholdOperation</code> specific property.
122    * <p>
123    * These property can be fixed either from <code>java</code> launching
124    * command, or in <code>a3servers.xml</code> configuration file.
125    */
126   static int LogThresholdOperation = 1000;
127
128   /**
129    * Returns the pool size for <code>operation</code> objects, by default 1000.
130    *
131    * @return The pool size for <code>operation</code> objects.
132    */
133   public int getLogThresholdOperation() {
134     return LogThresholdOperation;
135   }
136
137   /**
138    * Returns the number of commit operation since starting up.
139    *
140    * @return The number of commit operation.
141    */
142   public int getCommitCount() {
143     return logFile.commitCount;
144   }
145
146   /**
147    * Returns the number of garbage operation since starting up.
148    *
149    * @return The number of garbage operation.
150    */
151   public int getGarbageCount() {
152     return logFile.garbageCount;
153   }
154
155   /**
156    * Returns the maximum time between two garbages, 0 if disable.
157    *
158    * @return The maximum time between two garbages.
159    */
160   public final int getGarbageDelay() {
161     return (int) (logFile.garbageTimeOut /1000L);
162   }
163
164   /**
165    * Sets the maximum time between two garbages, 0 to disable the
166    * asynchronous garbage mechanism.
167    *
168    * @param timeout The maximum time between two garbages.
169    */
170   public final void setGarbageDelay(int timeout) {
171     logFile.garbageTimeOut = timeout *1000L;
172   }
173
174   private Timer timer = null;
175   private GarbageTask task = null;
176
177   /**
178    * Sets asynchronous garbage.
179    *
180    * @param async If true activates the asynchronous garbage,
181    * deasctivates otherwise.
182    */
183   public void garbageAsync(boolean async) {
184     if (async) {
185       if (task == null) {
186         task = new GarbageTask();
187       }
188     } else {
189       if (task != null) task.cancel();
190       task = null;
191       if (timer != null) timer.cancel();
192       timer = null;
193     }
194   }
195
196   private class GarbageTask extends TimerTask {
197     private GarbageTask() {
198       if (timer == null) timer = new Timer();
199       if (logFile.garbageTimeOut > 0) {
200         try {
201           timer.schedule(this, logFile.garbageTimeOut);
202         } catch (Exception exc) {
203           logmon.log(BasicLevel.ERROR,
204                      "NTransaction, cannot schedule garbage task ", exc);
205         }
206       }
207     }
208     
209     /** Method called when the timer expires. */
210     public void run() {
211       if (logFile.garbageTimeOut > 0) {
212         if (System.currentTimeMillis() > (logFile.lastGarbageTime + logFile.garbageTimeOut)) {
213           garbage();
214         }
215         try {
216           timer.schedule(this, logFile.garbageTimeOut);
217         } catch (Exception exc) {
218           logmon.log(BasicLevel.ERROR,
219                      "NTransaction, cannot schedule garbage task ", exc);
220         }
221       }
222     }
223   }
224
225   long startTime = 0L;
226
227   /**
228    * Returns the starting time.
229    *
230    * @return The starting time.
231    */
232   public long getStartTime() {
233     return startTime;
234   }
235
236   /**
237    * Returns the cumulated time of garbage operations since starting up.
238    *
239    * @return The cumulated time of garbage operations since starting up.
240    */
241   public long getGarbageTime() {
242     return logFile.garbageTime;
243   }
244
245   /**
246    * Returns the ratio of garbage operations since starting up.
247    *
248    * @return The ratio of garbage operations since starting up.
249    */
250   public int getGarbageRatio() {
251     return (int) ((logFile.garbageTime *100) / (System.currentTimeMillis() - startTime));
252   }
253
254   /**
255    * The Repository classname implementation.
256    * This value can be set for a particular server by setting the
257    * <code>NTRepositoryImpl</code> specific property. By default its value
258    * is "fr.dyade.aaa.util.FileRepository".
259    * <p>
260    * These property can be fixed either from <code>java</code> launching
261    * command, or in <code>a3servers.xml</code> configuration file.
262    */
263   String repositoryImpl = "fr.dyade.aaa.util.FileRepository";
264
265   /**
266    * Returns the Repository classname implementation.
267    *
268    * @return The Repository classname implementation.
269    */
270   public String getRepositoryImpl() {
271     return repositoryImpl;
272   }
273
274   /**
275    * Returns the number of save operation to repository.
276    *
277    * @return The number of save operation to repository.
278    */
279   public int getNbSavedObjects() {
280     return repository.getNbSavedObjects();
281   }
282
283   /**
284    * Returns the number of delete operation on repository.
285    *
286    * @return The number of delete operation on repository.
287    */
288   public int getNbDeletedObjects() {
289     return repository.getNbDeletedObjects();
290   }
291
292   /**
293    * Returns the number of useless delete operation on repository.
294    *
295    * @return The number of useless delete operation on repository.
296    */
297   public int getNbBadDeletedObjects() {
298     return repository.getNbBadDeletedObjects();
299   }
300
301   /**
302    * Returns the number of load operation from repository.
303    *
304    * @return The number of load operation from repository.
305    */
306   public int getNbLoadedObjects() {
307     return repository.getNbLoadedObjects();
308   }
309
310   /** Log context associated with each Thread using NTransaction. */
311   private class Context {
312     Hashtable log = null;
313     ByteArrayOutputStream bos = null;
314     ObjectOutputStream oos = null;
315
316     Context() {
317       log = new Hashtable(15);
318       bos = new ByteArrayOutputStream(256);
319     }
320   }
321
322   File dir = null;
323
324   LogFile logFile = null;
325
326   Repository repository = null;
327
328   /**
329    * ThreadLocal variable used to get the log to associate state with each
330    * thread. The log contains all operations do by the current thread since
331    * the last <code>commit</code>. On commit, its content is added to current
332    * log (memory + disk), then it is freed.
333    */
334   private ThreadLocal perThreadContext = null;
335
336   static final boolean debug = false;
337
338   public NTransaction() {}
339
340   /**
341    * Tests if the Transaction component is persistent.
342    *
343    * @return true.
344    */
345   public boolean isPersistent() {
346     return true;
347   }
348
349   public final void init(String path) throws IOException {
350     phase = INIT;
351
352     logmon = Debug.getLogger("fr.dyade.aaa.util.Transaction");
353     if (logmon.isLoggable(BasicLevel.INFO))
354       logmon.log(BasicLevel.INFO, "NTransaction, init()");
355
356     LogMemoryCapacity = Integer.getInteger("NTLogMemoryCapacity",
357                                            LogMemoryCapacity).intValue();
358     LogFileSize = Integer.getInteger("NTLogFileSize",
359                                      LogFileSize /Mb).intValue() *Mb;
360     MaxLogMemorySize = Integer.getInteger("NTLogMemorySize",
361                                           MaxLogMemorySize /Kb).intValue() *Kb;
362
363     dir = new File(path);
364     if (!dir.exists()) dir.mkdir();
365     if (!dir.isDirectory())
366       throw new FileNotFoundException(path + " is not a directory.");
367
368     // Saves the transaction classname in order to prevent use of a
369 // different one after restart (see AgentServer.init).
370 DataOutputStream ldos = null;
371     try {
372       File tfc = new File(dir, "TFC");
373       if (! tfc.exists()) {
374         ldos = new DataOutputStream(new FileOutputStream(tfc));
375         ldos.writeUTF(getClass().getName());
376         ldos.flush();
377       }
378     } finally {
379       if (ldos != null) ldos.close();
380     }
381
382     try {
383       repositoryImpl = System.getProperty("NTRepositoryImpl", repositoryImpl);
384       repository = (Repository) Class.forName(repositoryImpl).newInstance();
385       repository.init(dir);
386     } catch (ClassNotFoundException exc) {
387       logmon.log(BasicLevel.FATAL,
388                  "NTransaction, cannot initializes the repository ", exc);
389       throw new IOException(exc.getMessage());
390     } catch (InstantiationException exc) {
391       logmon.log(BasicLevel.FATAL,
392                  "NTransaction, cannot initializes the repository ", exc);
393       throw new IOException(exc.getMessage());
394     } catch (IllegalAccessException exc) {
395       logmon.log(BasicLevel.FATAL,
396                  "NTransaction, cannot initializes the repository ", exc);
397       throw new IOException(exc.getMessage());
398     }
399     logFile = new LogFile(dir, repository);
400
401     perThreadContext = new ThreadLocal () {
402         protected synchronized Object initialValue() {
403           return new Context();
404         }
405       };
406     
407     // Be careful, setGarbageDelay and garbageAsync use logFile !!
408 setGarbageDelay(Integer.getInteger("NTGarbageDelay",
409                                        getGarbageDelay()).intValue());
410     garbageAsync(Boolean.getBoolean("NTAsyncGarbage"));
411
412     startTime = System.currentTimeMillis();
413
414     if (logmon.isLoggable(BasicLevel.INFO))
415       logmon.log(BasicLevel.INFO, "NTransaction, initialized " + startTime);
416
417     /* The Transaction subsystem is ready */
418     setPhase(FREE);
419   }
420
421   public final File getDir() {
422     return dir;
423   }
424
425   /**
426    * Returns the path of persistence directory.
427    *
428    * @return The path of persistence directory.
429    */
430   public String getPersistenceDir() {
431     return dir.getPath();
432   }
433
434   // State of the transaction monitor.
435 private int phase = INIT;
436   String phaseInfo = PhaseInfo[phase];
437
438   /**
439    *
440    */
441   public int getPhase() {
442     return phase;
443   }
444
445   public String getPhaseInfo() {
446     return phaseInfo;
447   }
448
449   private final void setPhase(int newPhase) {
450     phase = newPhase;
451     phaseInfo = PhaseInfo[phase];
452   }
453
454   public final synchronized void begin() throws IOException {
455     while (phase != FREE) {
456       try {
457     wait();
458       } catch (InterruptedException exc) {
459       }
460     }
461     // Change the transaction state.
462 setPhase(RUN);
463   }
464
465   /**
466    * Returns an array of strings naming the persistent objects denoted by
467    * a name that satisfy the specified prefix. Each string is an object name.
468    *
469    * @param prefix the prefix
470    * @return An array of strings naming the persistent objects
471    * denoted by a name that satisfy the specified prefix. The
472    * array will be empty if no names match.
473    */
474   public synchronized String [] getList(String prefix) {
475     String [] list1 = null;
476     try {
477       list1 = repository.list(prefix);
478     } catch (IOException exc) {
479       // AF: TODO
480 }
481     if (list1 == null) list1 = new String [0];
482     Object [] list2 = logFile.log.keySet().toArray();
483     int nb = list1.length;
484     for (int i=0; i<list2.length; i++) {
485       if ((list2[i] instanceof String ) &&
486           (((String ) list2[i]).startsWith(prefix))) {
487         int j=0;
488         for (; j<list1.length; j++) {
489           if (list2[i].equals(list1[j])) break;
490         }
491         if (j<list1.length) {
492           // The file is already in the directory list, it must be count
493 // at most once.
494 if (((Operation) logFile.log.get(list2[i])).type == Operation.DELETE) {
495             // The file is deleted in transaction log.
496 list1[j] = null;
497             nb -= 1;
498           }
499           list2[i] = null;
500         } else if (((Operation) logFile.log.get(list2[i])).type == Operation.SAVE) {
501           // The file is added in transaction log
502 nb += 1;
503         } else {
504           list2[i] = null;
505         }
506       } else {
507         list2[i] = null;
508       }
509     }
510     String [] list = new String [nb];
511     for (int i=list1.length-1; i>=0; i--) {
512       if (list1[i] != null) list[--nb] = list1[i];
513     }
514     for (int i=list2.length-1; i>=0; i--) {
515       if (list2[i] != null) list[--nb] = (String ) list2[i];
516     }
517         
518     return list;
519   }
520
521   public final void save(Serializable obj, String name) throws IOException {
522     save(obj, null, name);
523   }
524
525   static private final byte[] OOS_STREAM_HEADER = {
526     (byte)((ObjectStreamConstants.STREAM_MAGIC >>> 8) & 0xFF),
527     (byte)((ObjectStreamConstants.STREAM_MAGIC >>> 0) & 0xFF),
528     (byte)((ObjectStreamConstants.STREAM_VERSION >>> 8) & 0xFF),
529     (byte)((ObjectStreamConstants.STREAM_VERSION >>> 0) & 0xFF)
530   };
531
532   public final void save(Serializable obj,
533                          String dirName, String name) throws IOException {
534     if (logmon.isLoggable(BasicLevel.DEBUG))
535       logmon.log(BasicLevel.DEBUG, "NTransaction, save(" + dirName + ", " + name + ")");
536
537     Context ctx = (Context) perThreadContext.get();
538     if (ctx.oos == null) {
539       ctx.bos.reset();
540       ctx.oos = new ObjectOutputStream(ctx.bos);
541     } else {
542       ctx.oos.reset();
543       ctx.bos.reset();
544       ctx.bos.write(OOS_STREAM_HEADER, 0, 4);
545     }
546     ctx.oos.writeObject(obj);
547     ctx.oos.flush();
548
549     saveInLog(ctx.bos.toByteArray(), dirName, name, ctx.log, false);
550   }
551
552   /**
553    * Save an object state already serialized. The byte array keeped in log is
554    * a copy, so the original one may be modified.
555    */
556   public final void saveByteArray(byte[] buf, String name) throws IOException {
557     saveByteArray(buf, null, name);
558   }
559
560   /**
561    * Save an object state already serialized. The byte array keeped in log is
562    * a copy, so the original one may be modified.
563    */
564   public final void saveByteArray(byte[] buf,
565                                   String dirName, String name) throws IOException {
566     Context ctx = (Context) perThreadContext.get();
567     saveInLog(buf,
568               dirName, name,
569               ((Context) perThreadContext.get()).log, true);
570   }
571
572   private final void saveInLog(byte[] buf,
573                                String dirName, String name,
574                                Hashtable log,
575                                boolean copy) throws IOException {
576     Object key = OperationKey.newKey(dirName, name);
577     Operation op = Operation.alloc(Operation.SAVE, dirName, name, buf);
578     Operation old = (Operation) log.put(key, op);
579     if (copy) {
580       if ((old != null) &&
581           (old.type == Operation.SAVE) &&
582           (old.value.length == buf.length)) {
583         // reuse old buffer
584 op.value = old.value;
585       } else {
586         // alloc a new one
587 op.value = new byte[buf.length];
588       }
589       System.arraycopy(buf, 0, op.value, 0, buf.length);
590     }
591     if (old != null) old.free();
592
593   }
594
595   private final byte[] getFromLog(Hashtable log, Object key) throws IOException {
596     // Searchs in the log a new value for the object.
597 Operation op = (Operation) log.get(key);
598     if (op != null) {
599       if (op.type == Operation.SAVE) {
600     return op.value;
601       } else if (op.type == Operation.DELETE) {
602     // The object was deleted.
603 throw new FileNotFoundException();
604       }
605     }
606     return null;
607   }
608
609   private final byte[] getFromLog(String dirName, String name) throws IOException {
610     // First searchs in the logs a new value for the object.
611 Object key = OperationKey.newKey(dirName, name);
612     byte[] buf = getFromLog(((Context) perThreadContext.get()).log, key);
613     if (buf != null) return buf;
614     
615     if ((buf = getFromLog(logFile.log, key)) != null) {
616       return buf;
617     }
618
619     return null;
620   }
621
622   public final Object load(String name) throws IOException, ClassNotFoundException {
623     return load(null, name);
624   }
625
626   public Object load(String dirName, String name) throws IOException, ClassNotFoundException {
627     if (logmon.isLoggable(BasicLevel.DEBUG))
628       logmon.log(BasicLevel.DEBUG, "NTransaction, load(" + dirName + ", " + name + ")");
629
630     // First searchs in the logs a new value for the object.
631 try {
632       byte[] buf = getFromLog(dirName, name);
633       if (buf != null) {
634         ByteArrayInputStream bis = new ByteArrayInputStream(buf);
635     ObjectInputStream ois = new ObjectInputStream(bis);
636     return ois.readObject();
637       }
638       
639       // Gets it from disk.
640 return repository.loadobj(dirName, name);
641     } catch (FileNotFoundException exc) {
642       if (logmon.isLoggable(BasicLevel.DEBUG))
643         logmon.log(BasicLevel.DEBUG, "NTransaction, load(" + dirName + ", " + name + ") NOT FOUND");
644
645       return null;
646     }
647   }
648
649   public final byte[] loadByteArray(String name) throws IOException {
650     return loadByteArray(null, name);
651   }
652
653
654   public byte[] loadByteArray(String dirName, String name) throws IOException {
655     if (logmon.isLoggable(BasicLevel.DEBUG))
656       logmon.log(BasicLevel.DEBUG, "NTransaction, loadByteArray(" + dirName + ", " + name + ")");
657
658     // First searchs in the logs a new value for the object.
659 try {
660       byte[] buf = getFromLog(dirName, name);
661       if (buf != null) return buf;
662
663       // Gets it from disk.
664 return repository.load(dirName, name);
665     } catch (FileNotFoundException exc) {
666       if (logmon.isLoggable(BasicLevel.DEBUG))
667         logmon.log(BasicLevel.DEBUG, "NTransaction, loadByteArray(" + dirName + ", " + name + ") NOT FOUND");
668
669       return null;
670     }
671   }
672
673   public final void delete(String name) {
674     delete(null, name);
675   }
676   
677   public final void delete(String dirName, String name) {
678     if (logmon.isLoggable(BasicLevel.DEBUG))
679       logmon.log(BasicLevel.DEBUG,
680                  "NTransaction, delete(" + dirName + ", " + name + ")");
681
682     Object key = OperationKey.newKey(dirName, name);
683
684     Hashtable log = ((Context) perThreadContext.get()).log;
685     Operation op = Operation.alloc(Operation.DELETE, dirName, name);
686     op = (Operation) log.put(key, op);
687     if (op != null) op.free();
688   }
689
690   public final synchronized void commit() throws IOException {
691     if (phase != RUN)
692       throw new IllegalStateException ("Can not commit.");
693
694     if (logmon.isLoggable(BasicLevel.DEBUG))
695       logmon.log(BasicLevel.DEBUG, "NTransaction, commit");
696     
697     Hashtable log = ((Context) perThreadContext.get()).log;
698     if (! log.isEmpty()) {
699       logFile.commit(log);
700       log.clear();
701     }
702
703     if (logmon.isLoggable(BasicLevel.DEBUG))
704       logmon.log(BasicLevel.DEBUG, "NTransaction, committed");
705
706     setPhase(COMMIT);
707   }
708
709   public final synchronized void rollback() throws IOException {
710     if (phase != RUN)
711       throw new IllegalStateException ("Can not rollback.");
712
713     if (logmon.isLoggable(BasicLevel.DEBUG))
714       logmon.log(BasicLevel.DEBUG, "NTransaction, rollback");
715
716     setPhase(ROLLBACK);
717     ((Context) perThreadContext.get()).log.clear();
718   }
719
720   public final synchronized void release() throws IOException {
721     if ((phase != RUN) && (phase != COMMIT) && (phase != ROLLBACK))
722       throw new IllegalStateException ("Can not release transaction.");
723
724     // Change the transaction state.
725 setPhase(FREE);
726     // wake-up an eventually user's thread in begin
727 notify();
728   }
729
730   /**
731    * Garbage the log file.
732    * It waits all transactions termination, then the log file is garbaged
733    * and all operations are reported to disk.
734    */
735   public final synchronized void garbage() {
736     if (logmon.isLoggable(BasicLevel.INFO))
737       logmon.log(BasicLevel.INFO, "NTransaction, stops");
738
739     while (phase != FREE) {
740       // Wait for the transaction subsystem to be free
741 try {
742         wait();
743       } catch (InterruptedException exc) {
744       }
745     }
746
747     setPhase(GARBAGE);
748     try {
749       logFile.garbage();
750     } catch (IOException exc) {
751       logmon.log(BasicLevel.WARN, "NTransaction, can't garbage logfile", exc);
752     }
753     setPhase(FREE);
754
755     if (logmon.isLoggable(BasicLevel.INFO)) {
756       logmon.log(BasicLevel.INFO,
757                  "NTransaction, stopped: " +
758                  "garbage=" + logFile.garbageCount + ", " +
759                  "commit=" + logFile.commitCount + ", " +
760                  "ratio=" + getGarbageRatio());
761     }
762   }
763
764   /**
765    * Stops the transaction module.
766    * It waits all transactions termination, then the module is kept
767    * in a FREE 'ready to use' state.
768    * The log file is garbaged, all operations are reported to disk.
769    */
770   public synchronized void stop() {
771     if (logmon.isLoggable(BasicLevel.INFO))
772       logmon.log(BasicLevel.INFO, "NTransaction, stops");
773
774     while (phase != FREE) {
775       // Wait for the transaction subsystem to be free
776 try {
777         wait();
778       } catch (InterruptedException exc) {
779       }
780     }
781
782     setPhase(FINALIZE);
783     try {
784       logFile.garbage();
785     } catch (IOException exc) {
786       logmon.log(BasicLevel.WARN, "NTransaction, can't garbage logfile", exc);
787     }
788     setPhase(FREE);
789
790     if (logmon.isLoggable(BasicLevel.INFO)) {
791       logmon.log(BasicLevel.INFO,
792                  "NTransaction, stopped: " +
793                  "garbage=" + logFile.garbageCount + ", " +
794                  "commit=" + logFile.commitCount + ", " +
795                  "ratio=" + getGarbageRatio());
796     }
797   }
798
799   /**
800    * Close the transaction module.
801    * It waits all transactions termination, the module will be initialized
802    * anew before reusing it.
803    * The log file is garbaged then closed.
804    */
805   public synchronized void close() {
806     if (logmon.isLoggable(BasicLevel.INFO))
807       logmon.log(BasicLevel.INFO, "NTransaction, stops");
808
809     if (phase == INIT) return;
810
811     while (phase != FREE) {
812       // Wait for the transaction subsystem to be free
813 try {
814         wait();
815       } catch (InterruptedException exc) {
816       }
817     }
818
819     setPhase(FINALIZE);
820     logFile.stop();
821     setPhase(INIT);
822
823     if (logmon.isLoggable(BasicLevel.INFO)) {
824       logmon.log(BasicLevel.INFO,
825                  "NTransaction, closed: " +
826                  "garbage=" + logFile.garbageCount + ", " +
827                  "commit=" + logFile.commitCount + ", " +
828                  "ratio=" + getGarbageRatio());
829     }
830   }
831
832   /**
833    *
834    */
835   static final class LogFile extends ByteArrayOutputStream {
836     /**
837      * Log of all operations already commited but not reported on disk.
838      */
839     Hashtable log = null;
840     /** log file */
841     RandomAccessFile logFile = null;
842
843     int current = -1;
844
845     /**
846      * Number of commit operation since starting up.
847      */
848     int commitCount = 0;
849
850     /**
851      * Number of garbage operation since starting up.
852      */
853     int garbageCount = 0;
854
855     /**
856      * Cumulated time of garbage operations since starting up.
857      */
858     long garbageTime = 0l;
859
860     /**
861      * Date of last garbage.
862      */
863     long lastGarbageTime = 0L;
864
865     /**
866      * Maximum delay between 2 garbages.
867      */
868     long garbageTimeOut = 0L;
869
870     /** Root directory of transaction storage */
871     private File dir = null;
872     /** Coherency lock filename */
873     static private final String LockPathname = "lock";
874     /** Coherency lock file */
875     private File lockFile = null;
876
877     private Repository repository = null;
878
879     LogFile(File dir, Repository repository) throws IOException {
880       super(4 * Kb);
881       this.dir = dir;
882       this.repository = repository;
883
884       boolean nolock = Boolean.getBoolean("NTNoLockFile");
885       if (! nolock) {
886         lockFile = new File(dir, LockPathname);
887         if (! lockFile.createNewFile()) {
888           logmon.log(BasicLevel.FATAL,
889                      "NTransaction.init(): " +
890                      "Either the server is already running, " +
891                      "either you have to remove lock file: " +
892                      lockFile.getAbsolutePath());
893           throw new IOException("Transaction already running.");
894         }
895         lockFile.deleteOnExit();
896       }
897
898       // Search for old log file, then apply all committed operation,
899 // finally cleans it.
900 log = new Hashtable(LogMemoryCapacity);
901
902       
903       File logFilePN = new File(dir, "log");
904       if ((logFilePN.exists()) && (logFilePN.length() > 0)) {
905         logFile = new RandomAccessFile(logFilePN, "r");
906         try {
907           int optype = logFile.read();
908           while (optype == Operation.COMMIT) {
909             String dirName;
910             String name;
911
912             optype = logFile.read();
913  
914             while ((optype == Operation.SAVE) ||
915                    (optype == Operation.DELETE)) {
916               // Gets all operations of one committed transaction then
917 // adds them to specified log.
918 dirName = logFile.readUTF();
919               if (dirName.length() == 0) dirName = null;
920               name = logFile.readUTF();
921
922               Object key = OperationKey.newKey(dirName, name);
923
924               if (Debug.debug && logmon.isLoggable(BasicLevel.DEBUG))
925                 logmon.log(BasicLevel.DEBUG,
926                            "NTransaction.init(), OPERATION=" +
927                            optype + ", " + name);
928
929               Operation op = null;
930               if (optype == Operation.SAVE) {
931                 byte buf[] = new byte[logFile.readInt()];
932                 logFile.readFully(buf);
933                 op = Operation.alloc(optype, dirName, name, buf);
934                 op = (Operation) log.put(key, op);
935               } else {
936                 op = Operation.alloc(optype, dirName, name);
937                 op = (Operation) log.put(key, op);
938               }
939               if (op != null) op.free();
940
941               optype = logFile.read();
942             }
943             if (Debug.debug && logmon.isLoggable(BasicLevel.DEBUG))
944               logmon.log(BasicLevel.DEBUG,
945                          "NTransaction.init(), COMMIT=" + optype);
946           };
947
948           if (Debug.debug && logmon.isLoggable(BasicLevel.DEBUG))
949             logmon.log(BasicLevel.DEBUG,
950                        "NTransaction.init(), END=" + optype + ", " +
951                        logFile.getFilePointer());
952
953           if (optype != Operation.END) System.exit(-1);
954         } catch (IOException exc) {
955           throw exc;
956         } finally {
957           logFile.close();
958         }
959
960         logFile = new RandomAccessFile(logFilePN, "rwd");
961         garbage();
962       } else {
963         logFile = new RandomAccessFile(logFilePN, "rwd");
964         logFile.setLength(LogFileSize);
965
966         current = 1;
967         // Cleans log file
968 logFile.seek(0);
969         logFile.write(Operation.END);
970       }
971     }
972
973     static private final byte[] emptyUTFString = {0, 0};
974
975     void writeUTF(String str) {
976       int strlen = str.length() ;
977
978       int newcount = count + strlen +2;
979       if (newcount > buf.length) {
980         byte newbuf[] = new byte[Math.max(buf.length << 1, newcount)];
981         System.arraycopy(buf, 0, newbuf, 0, count);
982         buf = newbuf;
983       }
984
985       buf[count++] = (byte) ((strlen >>> 8) & 0xFF);
986       buf[count++] = (byte) ((strlen >>> 0) & 0xFF);
987
988       str.getBytes(0, strlen, buf, count);
989       count = newcount;
990     }
991
992     void writeInt(int v) throws IOException {
993       int newcount = count +4;
994       if (newcount > buf.length) {
995         byte newbuf[] = new byte[buf.length << 1];
996         System.arraycopy(buf, 0, newbuf, 0, count);
997         buf = newbuf;
998       }
999
1000      buf[count++] = (byte) ((v >>> 24) & 0xFF);
1001      buf[count++] = (byte) ((v >>> 16) & 0xFF);
1002      buf[count++] = (byte) ((v >>> 8) & 0xFF);
1003      buf[count++] = (byte) ((v >>> 0) & 0xFF);
1004    }
1005
1006    int logMemorySize = 0;
1007
1008    /**
1009     * Reports all buffered operations in logs.
1010     */
1011    void commit(Hashtable ctxlog) throws IOException {
1012      if (logmon.isLoggable(BasicLevel.DEBUG))
1013        logmon.log(BasicLevel.DEBUG, "NTransaction.LogFile.commit()");
1014
1015      commitCount += 1;
1016      
1017      Operation op = null;
1018      for (Enumeration e = ctxlog.elements(); e.hasMoreElements(); ) {
1019        op = (Operation) e.nextElement();
1020
1021        // Save the log to disk
1022 write(op.type);
1023        if (op.dirName != null) {
1024          writeUTF(op.dirName);
1025        } else {
1026          write(emptyUTFString);
1027        }
1028        writeUTF(op.name);
1029        if (op.type == Operation.SAVE) {
1030          logMemorySize += op.value.length;
1031
1032          writeInt(op.value.length);
1033          write(op.value);
1034        }
1035
1036        // Reports all committed operation in clog
1037 op = (Operation) log.put(OperationKey.newKey(op.dirName, op.name), op);
1038        if (op != null) {
1039          if (op.type == Operation.SAVE)
1040            logMemorySize -= op.value.length;
1041
1042          op.free();
1043        }
1044      }
1045      write(Operation.END);
1046
1047      // AF: Is it needed? Normally the file pointer should already set at
1048 // the right position... To be verified, but not really a big cost.
1049 logFile.seek(current);
1050      logFile.write(buf, 0, count);
1051
1052      // AF: May be we can avoid this second synchronous write, using a
1053 // marker: determination d'un marqueur lié au log courant (date en
1054 // millis par exemple), ecriture du marqueur au debut du log, puis
1055 // ecriture du marqueur apres chaque Operation.COMMIT.
1056 logFile.seek(current -1);
1057      logFile.write(Operation.COMMIT);
1058
1059      current += (count);
1060      reset();
1061
1062      ctxlog.clear();
1063
1064      if ((current > LogFileSize) || (logMemorySize > MaxLogMemorySize) ||
1065          ((garbageTimeOut > 0) && (System.currentTimeMillis() > (lastGarbageTime + garbageTimeOut))))
1066        garbage();
1067    }
1068
1069    /**
1070     * Reports all logged operations on disk.
1071     */
1072    private final void garbage() throws IOException {
1073      long start = System.currentTimeMillis();
1074
1075      if (logmon.isLoggable(BasicLevel.INFO))
1076        logmon.log(BasicLevel.INFO,
1077                   "NTransaction.LogFile.garbage() - begin");
1078
1079      garbageCount += 1;
1080
1081      Operation op = null;
1082      for (Enumeration e = log.elements(); e.hasMoreElements(); ) {
1083        op = (Operation) e.nextElement();
1084
1085        if (op.type == Operation.SAVE) {
1086          if (logmon.isLoggable(BasicLevel.DEBUG))
1087            logmon.log(BasicLevel.DEBUG,
1088                       "NTransaction, LogFile.Save (" + op.dirName + ',' + op.name + ')');
1089
1090          repository.save(op.dirName, op.name, op.value);
1091        } else if (op.type == Operation.DELETE) {
1092          if (logmon.isLoggable(BasicLevel.DEBUG))
1093            logmon.log(BasicLevel.DEBUG,
1094                       "NTransaction, LogFile.Delete (" + op.dirName + ',' + op.name + ')');
1095
1096          repository.delete(op.dirName, op.name);
1097// if (!deleted && file.exists())
1098// logmon.log(BasicLevel.ERROR,
1099// "NTransaction, can't delete " + file.getCanonicalPath());
1100 }
1101        op.free();
1102      }
1103      // Be careful, do not clear log before all modifications are reported
1104 // to disk, in order to avoid load errors.
1105 log.clear();
1106      logMemorySize = 0;
1107
1108      repository.commit();
1109
1110      current = 1;
1111      // Cleans log file
1112 logFile.seek(0);
1113      logFile.write(Operation.END);
1114
1115      long end = System.currentTimeMillis();
1116      lastGarbageTime = end;
1117      garbageTime += end - start;
1118
1119      if (logmon.isLoggable(BasicLevel.INFO))
1120        logmon.log(BasicLevel.INFO,
1121                   "NTransaction.LogFile.garbage() - end: " + (end - start));
1122    }
1123
1124    void stop() {
1125      if (logmon.isLoggable(BasicLevel.INFO))
1126        logmon.log(BasicLevel.INFO, "NTransaction, stops");
1127
1128      try {
1129        garbage();
1130        logFile.close();
1131        repository.close();
1132      } catch (IOException exc) {
1133        logmon.log(BasicLevel.WARN, "NTransaction, can't close logfile", exc);
1134      }
1135
1136      if ((lockFile != null) && (! lockFile.delete())) {
1137        logmon.log(BasicLevel.FATAL,
1138                   "NTransaction, - can't delete lockfile: " +
1139                   lockFile.getAbsolutePath());
1140      }
1141
1142      if (logmon.isLoggable(BasicLevel.INFO))
1143        logmon.log(BasicLevel.INFO, "NTransaction, exits.");
1144    }
1145  }
1146
1147  public static void main(String [] args) throws Exception {
1148    if ("garbage".equals(args[0])) {
1149      NTransaction transaction = new NTransaction();
1150      transaction.init(args[1]);
1151      transaction.stop();
1152    } else if ("list".equals(args[0])) {
1153    } else {
1154      System.err.println("unknown command: " + args[0]);
1155    }
1156  }
1157}
1158
1159final class Operation implements Serializable {
1160  static final int SAVE = 1;
1161  static final int DELETE = 2;
1162  static final int COMMIT = 3;
1163  static final int END = 127;
1164 
1165  int type;
1166  String dirName;
1167  String name;
1168  byte[] value;
1169
1170  private Operation(int type, String dirName, String name, byte[] value) {
1171    this.type = type;
1172    this.dirName = dirName;
1173    this.name = name;
1174    this.value = value;
1175  }
1176
1177  /**
1178   * Returns a string representation for this object.
1179   *
1180   * @return A string representation of this object.
1181   */
1182  public String toString() {
1183    StringBuffer strbuf = new StringBuffer ();
1184
1185    strbuf.append('(').append(super.toString());
1186    strbuf.append(",type=").append(type);
1187    strbuf.append(",dirName=").append(dirName);
1188    strbuf.append(",name=").append(name);
1189    strbuf.append(')');
1190    
1191    return strbuf.toString();
1192  }
1193
1194  private static Pool pool = null;
1195
1196  static {
1197    pool = new Pool("NTransaction$Operation",
1198                    Integer.getInteger("NTLogThresholdOperation",
1199                                       NTransaction.LogThresholdOperation).intValue());
1200  }
1201
1202  static Operation alloc(int type, String dirName, String name) {
1203    return alloc(type, dirName, name, null);
1204  }
1205
1206  static Operation alloc(int type,
1207                         String dirName, String name,
1208                         byte[] value) {
1209    Operation op = null;
1210    
1211    try {
1212      op = (Operation) pool.allocElement();
1213    } catch (Exception exc) {
1214      return new Operation(type, dirName, name, value);
1215    }
1216    op.type = type;
1217    op.dirName = dirName;
1218    op.name = name;
1219    op.value = value;
1220    return op;
1221  }
1222
1223  void free() {
1224    /* to let gc do its work */
1225    dirName = null;
1226    name = null;
1227    value = null;
1228    pool.freeElement(this);
1229  }
1230}
1231
1232final class OperationKey {
1233  static Object newKey(String dirName, String name) {
1234    if (dirName == null) {
1235      return name;
1236    } else {
1237      return new OperationKey(dirName, name);
1238    }
1239  }
1240
1241  private String dirName;
1242  private String name;
1243
1244  private OperationKey(String dirName,
1245                       String name) {
1246    this.dirName = dirName;
1247    this.name = name;
1248  }
1249
1250  public int hashCode() {
1251    // Should compute a specific one.
1252 return dirName.hashCode();
1253  }
1254
1255  public boolean equals(Object obj) {
1256    if (this == obj) return true;
1257    if (obj instanceof OperationKey) {
1258      OperationKey opk = (OperationKey)obj;
1259      if (opk.name.length() != name.length()) return false;
1260      if (opk.dirName.length() != dirName.length()) return false;
1261      if (!opk.dirName.equals(dirName)) return false;
1262      return opk.name.equals(name);
1263    }
1264    return false;
1265  }
1266}
1267

A to Z: JavaDoc & Examples

---

Daily Java News & Articles

---

Open Source Projects

---

Open Source Codes

---

Free Computer Books

---

Remove Frame

---

Popular Tags