Java > Open Source Codes > org > jboss > cache > loader > bdbje > BdbjeCacheLoader


1 package org.jboss.cache.loader.bdbje;
2
3 import com.sleepycat.bind.serial.SerialBinding;
4 import com.sleepycat.bind.serial.StoredClassCatalog;
5 import com.sleepycat.bind.tuple.TupleBinding;
6 import com.sleepycat.bind.tuple.TupleInput;
7 import com.sleepycat.bind.tuple.TupleOutput;
8 import com.sleepycat.je.Cursor;
9 import com.sleepycat.je.Database;
10 import com.sleepycat.je.DatabaseConfig;
11 import com.sleepycat.je.DatabaseEntry;
12 import com.sleepycat.je.DeadlockException;
13 import com.sleepycat.je.Environment;
14 import com.sleepycat.je.EnvironmentConfig;
15 import com.sleepycat.je.JEVersion;
16 import com.sleepycat.je.LockMode;
17 import com.sleepycat.je.OperationStatus;
18 import com.sleepycat.je.Transaction;
19 import org.apache.commons.logging.Log;
20 import org.apache.commons.logging.LogFactory;
21 import org.jboss.cache.CacheSPI;
22 import org.jboss.cache.Fqn;
23 import org.jboss.cache.Modification;
24 import org.jboss.cache.config.CacheLoaderConfig.IndividualCacheLoaderConfig;
25 import org.jboss.cache.loader.AbstractCacheLoader;
26
27 import java.io.File ;
28 import java.io.IOException ;
29 import java.io.Serializable ;
30 import java.util.Collections ;
31 import java.util.HashMap ;
32 import java.util.HashSet ;
33 import java.util.List ;
34 import java.util.Map ;
35 import java.util.Set ;
36 import java.util.concurrent.ConcurrentHashMap ;
37
38
39 /**
40  * A persistent <code>CacheLoader</code> based on Berkeley DB Java Edition.
41  * <p/>
42  * <p>The configuration string format is:</p>
43  * <pre>environmentDirectoryName[#databaseName]</pre>
44  * <p>where databaseName, if omitted, defaults to the ClusterName property
45  * of the CacheImpl.</p>
46  * <p/>
47  * <p>A je.properties file may optionally be placed in the JE environment
48  * directory and used to customize the default JE configuration.</p>
49  *
50  * @author Mark Hayes May 16, 2004
51  * @author Bela Ban
52  * @version $Id: BdbjeCacheLoader.java,v 1.25 2006/12/30 19:48:50 msurtani Exp $
53  */
54 public class BdbjeCacheLoader extends AbstractCacheLoader
55 {
56
57    private static final int MAX_TXN_RETRIES = 10;
58    private static final char LOWEST_UTF_CHAR = '\u0001';
59
60    private static final Log log = LogFactory.getLog(BdbjeCacheLoader.class);
61
62    private BdbjeCacheLoaderConfig config;
63    private Environment env;
64    private String cacheDbName;
65    private String catalogDbName;
66    private Database cacheDb;
67    private Database catalogDb;
68    private StoredClassCatalog catalog;
69    private SerialBinding serialBinding;
70    private Map <Object , Transaction> txnMap;
71    private boolean transactional;
72
73    /*
74     * Service implementation -- lifecycle methods.
75     * Note that setConfig() and setCache() are called before create().
76     */
77
78    /**
79     * Does nothing since start() does all the work.
80     */
81    public void create() throws Exception
82    {
83       String license = "\n*************************************************************************************\n" +
84               "Berkeley DB Java Edition version: " + JEVersion.CURRENT_VERSION.toString() + "\n" +
85               "JBoss Cache can use Berkeley DB Java Edition from Oracle \n" +
86               "(http://www.oracle.com/database/berkeley-db/je/index.html)\n" +
87               "for persistent, reliable and transaction-protected data storage.\n" +
88               "If you choose to use Berkeley DB Java Edition with JBoss Cache, you must comply with the terms\n" +
89               "of Oracle's public license, included in the file LICENSE.txt.\n" +
90               "If you prefer not to release the source code for your own application in order to comply\n" +
91               "with the Oracle public license, you may purchase a different license for use of\n" +
92               "Berkeley DB Java Edition with JBoss Cache.\n" +
93               "See http://www.oracle.com/database/berkeley-db/je/index.html for pricing and license terms\n" +
94               "*************************************************************************************";
95       System.out.println(license);
96
97       log.trace("Creating BdbjeCacheLoader instance.");
98       checkNotOpen();
99    }
100
101    /**
102     * Does nothing since stop() does all the work.
103     */
104    public void destroy()
105    {
106    }
107
108    /**
109     * Opens the JE environment and the database specified by the configuration
110     * string. The environment and databases are created if necessary.
111     */
112    public void start()
113            throws Exception
114    {
115
116       log.trace("Starting BdbjeCacheLoader instance.");
117       checkNotOpen();
118
119       if (cache == null)
120       {
121          throw new IllegalStateException (
122                  "A non-null Cache property (CacheSPI object) is required");
123       }
124       String configStr = config.getLocation();
125       if (config.getLocation() == null)
126       {
127          configStr = System.getProperty("java.io.tmpdir");
128          config.setLocation(configStr);
129       }
130
131       // test location
132 File location = new File (configStr);
133       if (!location.exists())
134       {
135          boolean created = location.mkdirs();
136          if (!created) throw new IOException ("Unable to create cache loader location " + location);
137
138       }
139       if (!location.isDirectory())
140       {
141          throw new IOException ("Cache loader location [" + location + "] is not a directory!");
142       }
143
144       /* Parse config string. */
145       File homeDir;
146       int offset = configStr.indexOf('#');
147       if (offset >= 0 && offset < configStr.length() - 1)
148       {
149          homeDir = new File (configStr.substring(0, offset));
150          cacheDbName = configStr.substring(offset + 1);
151       }
152       else
153       {
154          homeDir = new File (configStr);
155          cacheDbName = cache.getClusterName();
156       }
157       catalogDbName = cacheDbName + "_class_catalog";
158
159       /*
160        * If the CacheImpl is transactional, we will create transactional
161        * databases. However, we always create a transactional environment
162        * since it may be shared by transactional and non-transactional caches.
163        */
164       transactional = cache.getTransactionManager() != null;
165
166       try
167       {
168          /* Open the environment, creating it if it doesn't exist. */
169          EnvironmentConfig envConfig = new EnvironmentConfig();
170          envConfig.setAllowCreate(true);
171          envConfig.setTransactional(true);
172          if (log.isTraceEnabled()) log.trace("Creating JE environment with home dir " + homeDir);
173          env = new Environment(homeDir, envConfig);
174          if (log.isDebugEnabled()) log.debug("Created JE environment " + env + " for cache loader " + this);
175          /* Open cache and catalog databases. */
176          openDatabases();
177       }
178       catch (Exception e)
179       {
180          destroy();
181          throw e;
182       }
183    }
184
185    /**
186     * Opens all databases and initializes database related information.
187     */
188    private void openDatabases()
189            throws Exception
190    {
191
192       /* Use a generic database config, with no duplicates allowed. */
193       DatabaseConfig dbConfig = new DatabaseConfig();
194       dbConfig.setAllowCreate(true);
195       dbConfig.setTransactional(transactional);
196
197       /* Create/open the cache database and associated catalog database. */
198       cacheDb = env.openDatabase(null, cacheDbName, dbConfig);
199       catalogDb = env.openDatabase(null, catalogDbName, dbConfig);
200
201       /* Use the catalog for the serial binding. */
202       catalog = new StoredClassCatalog(catalogDb);
203       serialBinding = new SerialBinding(catalog, null);
204
205       /* Start with a fresh transaction map. */
206       txnMap = new ConcurrentHashMap <Object , Transaction>();
207    }
208
209    /**
210     * Closes all databases, ignoring exceptions, and nulls references to all
211     * database related information.
212     */
213    private void closeDatabases()
214    {
215
216       if (cacheDb != null)
217       {
218          try
219          {
220             cacheDb.close();
221          }
222          catch (Exception shouldNotOccur)
223          {
224             log.warn("Caught unexpected exception", shouldNotOccur);
225          }
226       }
227       if (catalogDb != null)
228       {
229          try
230          {
231             catalogDb.close();
232          }
233          catch (Exception shouldNotOccur)
234          {
235             log.warn("Caught unexpected exception", shouldNotOccur);
236          }
237       }
238       cacheDb = null;
239       catalogDb = null;
240       catalog = null;
241       serialBinding = null;
242       txnMap = null;
243    }
244
245    /**
246     * Closes the JE databases and environment, and nulls references to them.
247     * The environment and databases are not removed from the file system.
248     * Exceptions during close are ignored.
249     */
250    public void stop()
251    {
252
253       closeDatabases();
254
255       if (env != null)
256       {
257          try
258          {
259             env.close();
260          }
261          catch (Exception shouldNotOccur)
262          {
263             log.warn("Unexpected exception", shouldNotOccur);
264          }
265       }
266       env = null;
267    }
268
269    /*
270     * CacheLoader implementation.
271     */
272
273    /**
274     * Sets the configuration string for this cache loader.
275     */
276    public void setConfig(IndividualCacheLoaderConfig base)
277    {
278       checkNotOpen();
279
280       if (base instanceof BdbjeCacheLoaderConfig)
281       {
282          this.config = (BdbjeCacheLoaderConfig) base;
283       }
284       else
285       {
286          config = new BdbjeCacheLoaderConfig(base);
287       }
288
289       if (log.isTraceEnabled()) log.trace("Configuring cache loader with location = " + config.getLocation());
290    }
291
292    public IndividualCacheLoaderConfig getConfig()
293    {
294       return config;
295    }
296
297    /**
298     * Sets the CacheImpl owner of this cache loader.
299     */
300    public void setCache(CacheSPI c)
301    {
302       super.setCache(c);
303       checkNotOpen();
304    }
305
306    /**
307     * Returns an unmodifiable set of relative children names (strings), or
308     * returns null if the parent node is not found or if no children are found.
309     * This is a fairly expensive operation, and is assumed to be performed by
310     * browser applications. Calling this method as part of a run-time
311     * transaction is not recommended.
312     */
313    public Set <String > getChildrenNames(Fqn name)
314            throws Exception
315    {
316
317       checkOpen();
318       checkNonNull(name, "name");
319
320       DatabaseEntry prefixEntry = makeKeyEntry(name);
321       DatabaseEntry dataEntry = new DatabaseEntry();
322       dataEntry.setPartial(0, 0, true);
323
324       String namePart = "";
325       int namePartIndex = name.size();
326       Set <String > set = null;
327
328       Cursor cursor = cacheDb.openCursor(null, null);
329       try
330       {
331          while (true)
332          {
333             DatabaseEntry keyEntry = makeKeyEntry(prefixEntry, namePart);
334             OperationStatus status =
335                     cursor.getSearchKeyRange(keyEntry, dataEntry, null);
336             if (status != OperationStatus.SUCCESS ||
337                     !startsWith(keyEntry, prefixEntry))
338             {
339                break;
340             }
341             if (set == null)
342             {
343                set = new HashSet <String >();
344             }
345             Fqn childName = makeKeyObject(keyEntry);
346             namePart = childName.get(namePartIndex).toString();
347             set.add(namePart);
348             namePart += LOWEST_UTF_CHAR;
349          }
350       }
351       finally
352       {
353          cursor.close();
354       }
355       if (set != null)
356       {
357          return Collections.unmodifiableSet(set);
358       }
359       else
360       {
361          return null;
362       }
363    }
364
365    /**
366     * Returns a map containing all key-value pairs for the given FQN, or null
367     * if the node is not present.
368     * This operation is always non-transactional, even in a transactional
369     * environment.
370     */
371    public Map get(Fqn name)
372            throws Exception
373    {
374
375       checkOpen();
376       checkNonNull(name, "name");
377
378       DatabaseEntry keyEntry = makeKeyEntry(name);
379       DatabaseEntry foundData = new DatabaseEntry();
380       OperationStatus status = cacheDb.get(null, keyEntry, foundData, null);
381       if (status == OperationStatus.SUCCESS)
382       {
383          // changed createIfNull param to true
384 // See http://jira.jboss.com/jira/browse/JBCACHE-118
385 return makeDataObject(foundData, true);
386       }
387       else
388       {
389          return null;
390       }
391    }
392
393    // See http://jira.jboss.com/jira/browse/JBCACHE-118 for why this is commented out.
394
395    /**
396     * Returns the data object stored under the given FQN and key, or null if
397     * the FQN and key are not present.
398     * This operation is always non-transactional, even in a transactional
399     * environment.
400     */
401 // public Object get(Fqn name, Object key)
402 // throws Exception {
403 //
404 // Map map = get(name);
405 // if (map != null) {
406 // return map.get(key);
407 // } else {
408 // return null;
409 // }
410 // }
411
412    /**
413     * Returns whether the given node exists.
414     */
415    public boolean exists(Fqn name)
416            throws Exception
417    {
418
419       checkOpen();
420       checkNonNull(name, "name");
421
422       DatabaseEntry keyEntry = makeKeyEntry(name);
423       DatabaseEntry foundData = new DatabaseEntry();
424       foundData.setPartial(0, 0, true);
425       OperationStatus status = cacheDb.get(null, keyEntry, foundData, null);
426       return (status == OperationStatus.SUCCESS);
427    }
428
429
430    /**
431     * Stores a single FQN-key-value record.
432     * Intended to be used in a non-transactional environment, but will use
433     * auto-commit in a transactional environment.
434     */
435    public Object put(Fqn name, Object key, Object value) throws Exception
436    {
437
438       checkOpen();
439       checkNonNull(name, "name");
440
441       Object oldVal;
442       if (transactional)
443       {
444          Modification mod =
445                  new Modification(Modification.ModificationType.PUT_KEY_VALUE, name, key, value);
446          commitModification(mod);
447          oldVal = mod.getOldValue();
448       }
449       else
450       {
451          oldVal = doPut(null, name, key, value);
452       }
453       return oldVal;
454    }
455
456
457    /**
458     * Internal version of store(String,Object,Object) that allows passing a
459     * transaction.
460     */
461    private Object doPut(Transaction txn, Fqn name, Object key, Object value)
462            throws Exception
463    {
464
465       Object oldVal = null;
466       /* To update-or-insert, try putNoOverwrite first, then a RMW cycle. */
467       Map <Object , Object > map = new HashMap <Object , Object >();
468       map.put(key, value);
469       DatabaseEntry dataEntry = makeDataEntry(map);
470       DatabaseEntry keyEntry = makeKeyEntry(name);
471       Cursor cursor = cacheDb.openCursor(txn, null);
472       try
473       {
474          OperationStatus status = cursor.putNoOverwrite(keyEntry, dataEntry);
475          if (status == OperationStatus.SUCCESS)
476          {
477             createParentNodes(cursor, name);
478          }
479          else
480          {
481             DatabaseEntry foundData = new DatabaseEntry();
482             status = cursor.getSearchKey(keyEntry, foundData, LockMode.RMW);
483             if (status == OperationStatus.SUCCESS)
484             {
485                map = makeDataObject(foundData, true);
486                oldVal = map.put(key, value);
487                cursor.putCurrent(makeDataEntry(map));
488             }
489          }
490       }
491       finally
492       {
493          cursor.close();
494       }
495       return oldVal;
496    }
497
498
499    /**
500     * Stores a map of key-values for a given FQN, but does not delete existing
501     * key-value pairs (that is, it does not erase).
502     * Intended to be used in a non-transactional environment, but will use
503     * auto-commit in a transactional environment.
504     */
505    public void put(Fqn name, Map values)
506            throws Exception
507    {
508
509       checkOpen();
510       checkNonNull(name, "name");
511
512       if (transactional)
513       {
514          commitModification(
515                  new Modification(Modification.ModificationType.PUT_DATA, name, values));
516       }
517       else
518       {
519          doPut(null, name, values);
520       }
521    }
522
523
524    /**
525     * Internal version of put(Fqn,Map) that allows passing a
526     * transaction.
527     */
528    private void doPut(Transaction txn, Fqn name, Map values)
529            throws Exception
530    {
531
532       // JBCACHE-769 -- make a defensive copy
533 values = (values == null ? null : new HashMap (values));
534
535       /* To update-or-insert, try putNoOverwrite first, then a RMW cycle. */
536       DatabaseEntry dataEntry = makeDataEntry(values);
537       DatabaseEntry keyEntry = makeKeyEntry(name);
538       Cursor cursor = cacheDb.openCursor(txn, null);
539       try
540       {
541          OperationStatus status = cursor.putNoOverwrite(keyEntry, dataEntry);
542          if (status == OperationStatus.SUCCESS)
543          {
544             createParentNodes(cursor, name);
545          }
546          else
547          {
548             DatabaseEntry foundData = new DatabaseEntry();
549             status = cursor.getSearchKey(keyEntry, foundData, LockMode.RMW);
550             if (status == OperationStatus.SUCCESS)
551             {
552                Map map = makeDataObject(foundData, true);
553                if (values != null)
554                {
555                   map.putAll(values);
556                }
557                cursor.putCurrent(makeDataEntry(map));
558             }
559          }
560       }
561       finally
562       {
563          cursor.close();
564       }
565    }
566
567    /**
568     * Internal version of put(Fqn,Map) that allows passing a
569     * transaction and erases existing data.
570     */
571    private void doPutErase(Transaction txn, Fqn name, Map values)
572            throws Exception
573    {
574
575       // JBCACHE-769 -- make a defensive copy
576 values = (values == null ? null : new HashMap (values));
577
578       DatabaseEntry dataEntry = makeDataEntry(values);
579       DatabaseEntry keyEntry = makeKeyEntry(name);
580       Cursor cursor = cacheDb.openCursor(txn, null);
581       try
582       {
583          cursor.put(keyEntry, dataEntry);
584          createParentNodes(cursor, name);
585       }
586       finally
587       {
588          cursor.close();
589       }
590    }
591
592    /**
593     * Applies the given modifications.
594     * Intended to be used in a non-transactional environment, but will use
595     * auto-commit in a transactional environment.
596     */
597    public void put(List <Modification> modifications) throws Exception
598    {
599       checkOpen();
600       checkNonNull(modifications, "modifications");
601
602       if (transactional)
603       {
604          commitModifications(modifications);
605       }
606       else
607       {
608          doPut(null, modifications);
609       }
610    }
611
612    /**
613     * Internal version of put(List) that allows passing a transaction.
614     */
615    private void doPut(Transaction txn, List <Modification> modifications)
616            throws Exception
617    {
618
619       /* This could be optimized by grouping modifications by Fqn, and
620        * performing a single database operation for each Fqn (record). */
621
622       for (Modification mod : modifications)
623       {
624          Fqn name = mod.getFqn();
625          Object oldVal;
626          switch (mod.getType())
627          {
628             case PUT_KEY_VALUE:
629                oldVal = doPut(txn, name, mod.getKey(), mod.getValue());
630                mod.setOldValue(oldVal);
631                break;
632             case PUT_DATA:
633                doPut(txn, name, mod.getData());
634                break;
635             case PUT_DATA_ERASE:
636                doPutErase(txn, name, mod.getData());
637                break;
638             case REMOVE_KEY_VALUE:
639                oldVal = doRemove(txn, name, mod.getKey());
640                mod.setOldValue(oldVal);
641                break;
642             case REMOVE_NODE:
643                doRemove(txn, name);
644                break;
645             case REMOVE_DATA:
646                doRemoveData(txn, name);
647                break;
648             default:
649                throw new IllegalArgumentException (
650                        "Unknown Modification type: " + mod.getType());
651          }
652       }
653    }
654
655    /**
656     * Creates parent nodes of the given Fqn, moving upward until an existing
657     * node is found.
658     */
659    private void createParentNodes(Cursor cursor, Fqn name)
660            throws Exception
661    {
662
663       DatabaseEntry dataEntry = makeDataEntry(null);
664       for (int nParts = name.size() - 1; nParts >= 1; nParts -= 1)
665       {
666          DatabaseEntry keyEntry = makeKeyEntry(name, nParts);
667          OperationStatus status = cursor.putNoOverwrite(keyEntry, dataEntry);
668          if (status != OperationStatus.SUCCESS)
669          {
670             break;
671          }
672       }
673    }
674
675    /**
676     * Deletes the node for a given FQN and all its descendent nodes.
677     * Intended to be used in a non-transactional environment, but will use
678     * auto-commit in a transactional environment.
679     */
680    public void remove(Fqn name)
681            throws Exception
682    {
683
684       checkOpen();
685       checkNonNull(name, "name");
686
687       if (transactional)
688       {
689          commitModification(
690                  new Modification(Modification.ModificationType.REMOVE_NODE, name));
691       }
692       else
693       {
694          doRemove(null, name);
695       }
696    }
697
698    /**
699     * Internal version of remove(Fqn) that allows passing a transaction.
700     */
701    private void doRemove(Transaction txn, Fqn name)
702            throws Exception
703    {
704
705       DatabaseEntry keyEntry = makeKeyEntry(name);
706       DatabaseEntry foundKey = new DatabaseEntry();
707       DatabaseEntry foundData = new DatabaseEntry();
708       foundData.setPartial(0, 0, true);
709       Cursor cursor = cacheDb.openCursor(txn, null);
710       try
711       {
712          OperationStatus status =
713                  cursor.getSearchKey(keyEntry, foundData, LockMode.RMW);
714          while (status == OperationStatus.SUCCESS)
715          {
716             cursor.delete();
717             status = cursor.getNext(foundKey, foundData, LockMode.RMW);
718             if (status == OperationStatus.SUCCESS &&
719                     !startsWith(foundKey, keyEntry))
720             {
721                status = OperationStatus.NOTFOUND;
722             }
723          }
724       }
725       finally
726       {
727          cursor.close();
728       }
729    }
730
731    /**
732     * Deletes a single FQN-key-value record.
733     * Intended to be used in a non-transactional environment, but will use
734     * auto-commit in a transactional environment.
735     */
736    public Object remove(Fqn name, Object key)
737            throws Exception
738    {
739
740       checkOpen();
741       checkNonNull(name, "name");
742
743       Object oldVal;
744       if (transactional)
745       {
746          Modification mod =
747                  new Modification(Modification.ModificationType.REMOVE_KEY_VALUE, name, key);
748          commitModification(mod);
749          oldVal = mod.getOldValue();
750       }
751       else
752       {
753          oldVal = doRemove(null, name, key);
754       }
755       return oldVal;
756    }
757
758    /**
759     * Internal version of remove(String,Object) that allows passing a
760     * transaction.
761     */
762    private Object doRemove(Transaction txn, Fqn name, Object key)
763            throws Exception
764    {
765
766       Object oldVal = null;
767       DatabaseEntry keyEntry = makeKeyEntry(name);
768       DatabaseEntry foundData = new DatabaseEntry();
769       Cursor cursor = cacheDb.openCursor(txn, null);
770       try
771       {
772          OperationStatus status =
773                  cursor.getSearchKey(keyEntry, foundData, LockMode.RMW);
774          if (status == OperationStatus.SUCCESS)
775          {
776             Map map = makeDataObject(foundData, true);
777             oldVal = map.remove(key);
778             cursor.putCurrent(makeDataEntry(map));
779          }
780       }
781       finally
782       {
783          cursor.close();
784       }
785       return oldVal;
786    }
787
788    /**
789     * Clears the map for the given node, but does not remove the node.
790     */
791    public void removeData(Fqn name)
792            throws Exception
793    {
794
795       checkOpen();
796       checkNonNull(name, "name");
797
798       if (transactional)
799       {
800          commitModification(
801                  new Modification(Modification.ModificationType.REMOVE_DATA, name));
802       }
803       else
804       {
805          doRemoveData(null, name);
806       }
807    }
808
809    /**
810     * Internal version of removeData(Fqn) that allows passing a transaction.
811     */
812    private void doRemoveData(Transaction txn, Fqn name)
813            throws Exception
814    {
815
816       DatabaseEntry dataEntry = new DatabaseEntry();
817       dataEntry.setPartial(0, 0, true);
818       DatabaseEntry keyEntry = makeKeyEntry(name);
819       Cursor cursor = cacheDb.openCursor(txn, null);
820       try
821       {
822          OperationStatus status =
823                  cursor.getSearchKey(keyEntry, dataEntry, LockMode.RMW);
824          if (status == OperationStatus.SUCCESS)
825          {
826             cursor.putCurrent(makeDataEntry(null));
827          }
828       }
829       finally
830       {
831          cursor.close();
832       }
833    }
834
835    /**
836     * Begins a transaction and applies the given modifications.
837     * <p/>
838     * <p>If onePhase is true, commits the transaction; otherwise, associates
839     * the txn value with the transaction and expects commit() or rollback() to
840     * be called later with the same tx value. Performs retries if necessary to
841     * resolve deadlocks.</p>
842     */
843    public void prepare(Object tx, List <Modification> modifications, boolean onePhase) throws Exception
844    {
845       checkOpen();
846       checkNonNull(modifications, "modifications");
847       if (!onePhase)
848       {
849          checkNonNull(tx, "tx");
850       }
851       if (!transactional)
852       {
853          throw new UnsupportedOperationException (
854                  "prepare() not allowed with a non-transactional cache loader");
855       }
856       Transaction txn = performTransaction(modifications);
857       if (onePhase)
858       {
859          txn.commit();
860       }
861       else
862       {
863          txnMap.put(tx, txn);
864       }
865    }
866
867    /**
868     * Performs and commits a single modification. The loader must be
869     * transactional. Commits the transaction if successful, or aborts the
870     * transaction and throws an exception if not successful.
871     */
872    private void commitModification(Modification mod)
873            throws Exception
874    {
875
876       commitModifications(Collections.singletonList(mod));
877    }
878
879    /**
880     * Performs and commits a list of modifications. The loader must be
881     * transactional. Commits the transaction if successful, or aborts the
882     * transaction and throws an exception if not successful.
883     */
884    private void commitModifications(List <Modification> mods)
885            throws Exception
886    {
887
888       if (!transactional) throw new IllegalStateException ();
889       Transaction txn = performTransaction(mods);
890       txn.commit();
891    }
892
893    /**
894     * Performs the given operation, starting a transaction and performing
895     * retries. Returns the transaction if successful; aborts the transaction
896     * and throws an exception if not successful.
897     */
898    private Transaction performTransaction(List <Modification> modifications)
899            throws Exception
900    {
901
902       /*
903        * Note that we can't use TransactionRunner here since if onePhase=false
904        * in the call to prepare(), we do not want to commit. TransactionRunner
905        * always commits or aborts.
906        */
907
908       int retries = MAX_TXN_RETRIES;
909       while (true)
910       {
911          Transaction txn = env.beginTransaction(null, null);
912          try
913          {
914             doPut(txn, modifications);
915             return txn;
916          }
917          catch (Exception e)
918          {
919             txn.abort();
920             if (e instanceof DeadlockException && retries > 0)
921             {
922                retries -= 1;
923             }
924             else
925             {
926                throw e;
927             }
928          }
929       }
930    }
931
932    /**
933     * Commits the given transaction, or throws IllegalArgumentException if the
934     * given key is not associated with an uncommited transaction.
935     */
936    public void commit(Object tx)
937            throws Exception
938    {
939
940       checkOpen();
941       checkNonNull(tx, "tx");
942
943       Transaction txn = txnMap.remove(tx);
944       if (txn != null)
945       {
946          txn.commit();
947       }
948       else if (transactional)
949       {
950          throw new IllegalArgumentException ("Unknown txn key: " + tx);
951       }
952    }
953
954    /**
955     * Commits the given transaction, or throws IllegalArgumentException if the
956     * given key is not associated with an uncommited transaction.
957     */
958    public void rollback(Object tx)
959    {
960
961       checkOpen();
962       checkNonNull(tx, "tx");
963
964       Transaction txn = txnMap.remove(tx);
965       if (txn != null)
966       {
967          try
968          {
969             txn.abort();
970          }
971          catch (Exception ignored)
972          {
973          }
974       }
975       else if (transactional)
976       {
977          throw new IllegalArgumentException ("Unknown txn key: " + tx);
978       }
979    }
980
981    /**
982     * Returns whether the given entry starts with the given prefix bytes.
983     * Used to determine whether a database key starts with a given FQN.
984     */
985    private boolean startsWith(DatabaseEntry entry,
986                               DatabaseEntry prefix)
987    {
988       int size = prefix.getSize();
989       if (size > entry.getSize())
990       {
991          return false;
992       }
993       byte[] d1 = entry.getData();
994       byte[] d2 = prefix.getData();
995       int o1 = entry.getOffset();
996       int o2 = prefix.getOffset();
997       for (int i = 0; i < size; i += 1)
998       {
999          if (d1[o1 + i] != d2[o2 + i])
1000         {
1001            return false;
1002         }
1003      }
1004      return true;
1005   }
1006
1007   /**
1008    * Converts a database entry to an Fqn.
1009    */
1010   private Fqn makeKeyObject(DatabaseEntry entry)
1011   {
1012
1013      Fqn name = Fqn.ROOT;
1014      TupleInput tupleInput = TupleBinding.entryToInput(entry);
1015      while (tupleInput.available() > 0)
1016      {
1017         String part = tupleInput.readString();
1018         name = new Fqn(name, part);
1019      }
1020      return name;
1021   }
1022
1023   /**
1024    * Converts an Fqn to a database entry.
1025    */
1026   private DatabaseEntry makeKeyEntry(Fqn name)
1027   {
1028      return makeKeyEntry(name, name.size());
1029   }
1030
1031   /**
1032    * Converts an Fqn to a database entry, outputing the given number of name
1033    * parts.
1034    */
1035   private DatabaseEntry makeKeyEntry(Fqn name, int nParts)
1036   {
1037      /* Write the sequence of name parts. */
1038      TupleOutput tupleOutput = new TupleOutput();
1039      for (int i = 0; i < nParts; i += 1)
1040      {
1041         tupleOutput.writeString(name.get(i).toString());
1042      }
1043
1044      /* Return the tuple as an entry. */
1045      DatabaseEntry entry = new DatabaseEntry();
1046      TupleBinding.outputToEntry(tupleOutput, entry);
1047      return entry;
1048   }
1049
1050   /**
1051    * Creates a key database entry from a parent database entry (prefix) and
1052    * a child name part.
1053    */
1054   private DatabaseEntry makeKeyEntry(DatabaseEntry prefix, String namePart)
1055   {
1056
1057      /* Write the bytes of the prefix followed by the child name. */
1058      TupleOutput tupleOutput = new TupleOutput();
1059      tupleOutput.writeFast(prefix.getData(),
1060              prefix.getOffset(),
1061              prefix.getSize());
1062      tupleOutput.writeString(namePart);
1063
1064      /* Return the tuple as an entry. */
1065      DatabaseEntry entry = new DatabaseEntry();
1066      TupleBinding.outputToEntry(tupleOutput, entry);
1067      return entry;
1068   }
1069
1070   /**
1071    * Converts a database entry to a Map.
1072    */
1073   private Map <Object , Object > makeDataObject(DatabaseEntry entry, boolean createIfNull)
1074   {
1075      Map <Object , Object > map = (Map <Object , Object >) serialBinding.entryToObject(entry);
1076      if (createIfNull && map == null)
1077      {
1078         map = new HashMap <Object , Object >();
1079      }
1080      return map;
1081   }
1082
1083   /**
1084    * Converts a Map to a database entry.
1085    */
1086   private DatabaseEntry makeDataEntry(Map map)
1087   {
1088
1089      if (map != null)
1090      {
1091         if (map.size() == 0)
1092         {
1093            map = null;
1094         }
1095         else if (!(map instanceof Serializable ))
1096         {
1097            map = new HashMap (map);
1098         }
1099      }
1100      DatabaseEntry entry = new DatabaseEntry();
1101      serialBinding.objectToEntry(map, entry);
1102      return entry;
1103   }
1104
1105   /**
1106    * Throws an exception if the environment is not open.
1107    */
1108   private void checkOpen()
1109   {
1110      if (env == null)
1111      {
1112         throw new IllegalStateException (
1113                 "Operation not allowed before calling create()");
1114      }
1115   }
1116
1117   /**
1118    * Throws an exception if the environment is not open.
1119    */
1120   private void checkNotOpen()
1121   {
1122      if (env != null)
1123      {
1124         throw new IllegalStateException (
1125                 "Operation not allowed after calling create()");
1126      }
1127   }
1128
1129   /**
1130    * Throws an exception if the parameter is null.
1131    */
1132   private void checkNonNull(Object param, String paramName)
1133   {
1134      if (param == null)
1135      {
1136         throw new NullPointerException (
1137                 "Parameter must not be null: " + paramName);
1138      }
1139   }
1140}
1141

A to Z: JavaDoc & Examples

---

Daily Java News & Articles

---

Open Source Projects

---

Open Source Codes

---

Free Computer Books

---

Remove Frame

---

Popular Tags