|
|
The following Java code provides a fully functional example of a multi-threaded transactional JE application. The example opens an environment and database, and then creates 5 threads, each of which writes 500 records to the database. The keys used for these writes are pre-determined strings, while the data is a class that contains randomly generated data. This means that the actual data is arbitrary and therefore uninteresting; we picked it only because it requires minimum code to implement and therefore will stay out of the way of the main points of this example.
Each thread writes 10 records under a single transaction before committing and writing another 10 (this is repeated 50 times). At the end of each transaction, but before committing, each thread calls a function that uses a cursor to read every record in the database. We do this in order to make some points about database reads in a transactional environment.
Of course, each writer thread performs deadlock detection as described in this manual. In addition, normal recovery is performed when the environment is opened.
To implement this example, we need three classes:
TxnGuide.java
This is the main class for the application. It performs environment and database management, spawns threads, and creates the data that is placed in the database. See TxnGuide.java for implementation details.
DBWriter.java
This class extends java.lang.Thread, and as such it is our thread implementation. It is responsible for actually reading and writing to the database. It also performs all of our transaction management. See DBWriter.java for implementation details.
PayloadData.java
This is a data class used to encapsulate several data fields. It is fairly uninteresting, except that the usage of a class means that we have to use the bind APIs to serialize it for storage in the database. See PayloadData.java for implementation details.
The main class in our example application is used to open and close our environment and database. It also spawns all the threads that we need. We start with the normal series of Java package and import statements, followed by our class declaration:
// File TxnGuide.java
package db.txn;
import com.sleepycat.bind.serial.StoredClassCatalog;
import com.sleepycat.db.Database;
import com.sleepycat.db.DatabaseConfig;
import com.sleepycat.db.DatabaseException;
import com.sleepycat.db.DatabaseType;
import com.sleepycat.db.LockDetectMode;
import com.sleepycat.db.Environment;
import com.sleepycat.db.EnvironmentConfig;
import java.io.File;
import java.io.FileNotFoundException;
public class TxnGuide {
Next we declare our class' private data members. Mostly these are used for constants such as the name of the database that we are opening and the number of threads that we are spawning. However, we also declare our environment and database handles here.
private static String myEnvPath = "./";
private static String dbName = "mydb.db";
private static String cdbName = "myclassdb.db";
// DB handles
private static Database myDb = null;
private static Database myClassDb = null;
private static Environment myEnv = null;
private static final int NUMTHREADS = 5;
Next, we implement our usage() method. This application optionally accepts a single command line argument which is used to identify the environment home directory.
private static void usage() {
System.out.println("TxnGuide [-h <env directory>]");
System.exit(-1);
}
Now we implement our main() method. This method simply calls the methods to parse the command line arguments and open the environment and database. It also creates the stored class catalog that we use for serializing the data that we want to store in our database. Finally, it creates and then joins the database writer threads.
public static void main(String args[]) {
try {
// Parse the arguments list
parseArgs(args);
// Open the environment and databases
openEnv();
// Get our class catalog (used to serialize objects)
StoredClassCatalog classCatalog =
new StoredClassCatalog(myClassDb);
// Start the threads
DBWriter[] threadArray;
threadArray = new DBWriter[NUMTHREADS];
for (int i = 0; i < NUMTHREADS; i++) {
threadArray[i] = new DBWriter(myEnv, myDb, classCatalog);
threadArray[i].start();
}
// Join the threads. That is, wait for each thread to
// complete before exiting the application.
for (int i = 0; i < NUMTHREADS; i++) {
threadArray[i].join();
}
} catch (Exception e) {
System.err.println("TxnGuide: " + e.toString());
e.printStackTrace();
} finally {
closeEnv();
}
System.out.println("All done.");
}
Next we implement openEnv(). This method is used to open the environment and then a database in that environment. Along the way, we make sure that every handle is free-threaded, and that the transactional subsystem is correctly initialized. Because this is a concurrent application, we also declare how we want deadlock detection to be performed. In this case, we use JE's internal block detector to determine whether a deadlock has occurred when a thread attempts to acquire a lock. We also indicate that we want the deadlocked thread with the youngest lock to receive deadlock notification.
Notice that we also cause normal recovery to be run when we open the environment. This is the standard and recommended thing to do whenever you start up a transactional application.
For the database open, notice that we open the database such that it supports duplicate records. This is required purely by the data that we are writing to the database, and it is only necessary if you run the application more than once without first deleting the environment.
Finally, notice that we open the database such that it supports uncommitted reads. We do this so that some cursor activity later in this example can read uncommitted data. If we did not do this, then our countRecords() method described later in this example would cause our thread to self-deadlock. This is because the cursor could not be opened to support uncommitted reads (that flag on the cursor open would, in fact, be silently ignored).
private static void openEnv() throws DatabaseException {
System.out.println("opening env");
// Set up the environment.
EnvironmentConfig myEnvConfig = new EnvironmentConfig();
myEnvConfig.setAllowCreate(true);
myEnvConfig.setInitializeCache(true);
myEnvConfig.setInitializeLocking(true);
myEnvConfig.setInitializeLogging(true);
myEnvConfig.setRunRecovery(true);
myEnvConfig.setTransactional(true);
// EnvironmentConfig.setThreaded(true) is the default behavior
// in Java, so we do not have to do anything to cause the
// environment handle to be free-threaded.
// Indicate that we want db to internally perform deadlock
// detection. Also indicate that the transaction that has
// performed the least amount of write activity to
// receive the deadlock notification, if any.
myEnvConfig.setLockDetectMode(LockDetectMode.MINWRITE);
// Set up the database
DatabaseConfig myDbConfig = new DatabaseConfig();
myDbConfig.setType(DatabaseType.BTREE);
myDbConfig.setAllowCreate(true);
myDbConfig.setTransactional(true);
myDbConfig.setSortedDuplicates(true);
myDbConfig.setReadUncomitted(true);
// no DatabaseConfig.setThreaded() method available.
// db handles in java are free-threaded so long as the
// env is also free-threaded.
try {
// Open the environment
myEnv = new Environment(new File(myEnvPath), // Env home
myEnvConfig);
// Open the database. Do not provide a txn handle. This open
// is auto committed because DatabaseConfig.setTransactional()
// is true.
myDb = myEnv.openDatabase(null, // txn handle
dbName, // Database file name
null, // Database name
myDbConfig);
// Used by the bind API for serializing objects
// Class database must not support duplicates
myDbConfig.setSortedDuplicates(false);
myClassDb = myEnv.openDatabase(null, // txn handle
cdbName, // Database file name
null, // Database name,
myDbConfig);
} catch (FileNotFoundException fnfe) {
System.err.println("openEnv: " + fnfe.toString());
System.exit(-1);
}
}
Finally, we implement the methods used to close our environment and databases, parse the command line arguments, and provide our class constructor. This is fairly standard code and it is mostly uninteresting from the perspective of this manual. We include it here purely for the purpose of completeness.
private static void closeEnv() {
System.out.println("Closing env and databases");
if (myDb != null ) {
try {
myDb.close();
} catch (DatabaseException e) {
System.err.println("closeEnv: myDb: " +
e.toString());
e.printStackTrace();
}
}
if (myClassDb != null ) {
try {
myClassDb.close();
} catch (DatabaseException e) {
System.err.println("closeEnv: myClassDb: " +
e.toString());
e.printStackTrace();
}
}
if (myEnv != null ) {
try {
myEnv.close();
} catch (DatabaseException e) {
System.err.println("closeEnv: " + e.toString());
e.printStackTrace();
}
}
}
private TxnGuide() {}
private static void parseArgs(String args[]) {
int nArgs = args.length;
for(int i = 0; i < args.length; ++i) {
if (args[i].startsWith("-")) {
switch(args[i].charAt(1)) {
case 'h':
if (i < nArgs - 1) {
myEnvPath = new String(args[++i]);
}
break;
default:
usage();
}
}
}
}
}
Before we show the implementation of the database writer thread, we need to show the class that we will be placing into the database. This class is fairly minimal. It simply allows you to store and retrieve an int, a String, and a double. We will be using the Sleepycat bind API from within the writer thread to serialize instances of this class and place them into our database.
package db.txn;
import java.io.Serializable;
public class PayloadData implements Serializable {
private int oID;
private String threadName;
private double doubleData;
PayloadData(int id, String name, double data) {
oID = id;
threadName = name;
doubleData = data;
}
public double getDoubleData() { return doubleData; }
public int getID() { return oID; }
public String getThreadName() { return threadName; }
}
DBWriter.java provides the implementation for our database writer thread. It is responsible for:
All transaction management.
Responding to deadlock exceptions.
Providing data to be stored into the database.
Serializing and then writing the data to the database.
In order to show off some of the ACID properties provided by JE's transactional support, DBWriter.java does some things in a less efficient way than you would probably decide to use in a true production application. First, it groups 10 database writes together in a single transaction when you could just as easily perform one write for each transaction. If you did this, you could use auto commit for the individual database writes, which means your code would be slightly simpler and you would run a much smaller chance of encountering blocked and deadlocked operations. However, by doing things this way, we are able to show transactional atomicity, as well as deadlock handling.
At the end of each transaction, DBWriter.java runs a cursor over the entire database by way of counting the number of records currently existing in the database. There are better ways to discover this information, but in this case we want to make some points regarding cursors, transactional applications, and deadlocking (we get into this in more detail later in this section).
To begin, we provide the usual package and import statements, and we declare our class:
package db.txn;
import com.sleepycat.bind.EntryBinding;
import com.sleepycat.bind.serial.StoredClassCatalog;
import com.sleepycat.bind.serial.SerialBinding;
import com.sleepycat.bind.tuple.StringBinding;
import com.sleepycat.db.Cursor;
import com.sleepycat.db.CursorConfig;
import com.sleepycat.db.Database;
import com.sleepycat.db.DatabaseEntry;
import com.sleepycat.db.DatabaseException;
import com.sleepycat.db.DeadlockException;
import com.sleepycat.db.Environment;
import com.sleepycat.db.LockMode;
import com.sleepycat.db.OperationStatus;
import com.sleepycat.db.Transaction;
import java.io.UnsupportedEncodingException;
import java.util.Random;
public class DBWriter extends Thread
{
Next we declare our private data members. Notice that we get handles for the environment and the database. We also obtain a handle for an EntryBinding. We will use this to serialize PayloadData class instances (see PayloadData.java) for storage in the database. The random number generator that we instatiate is used to generate unique data for storage in the database. The MAX_RETRY variable is used to define how many times we will retry a transaction in the face of a deadlock. And, finally, keys is a String array that holds the keys used for our database entries.
private Database myDb = null;
private Environment myEnv = null;
private EntryBinding dataBinding = null;
private Random generator = new Random();
private static final int MAX_RETRY = 20;
private static String[] keys = {"key 1", "key 2", "key 3",
"key 4", "key 5", "key 6",
"key 7", "key 8", "key 9",
"key 10"};
Next we implement our class constructor. The most interesting thing we do here is instantiate a serial binding for serializing PayloadData instances.
// Constructor. Get our DB handles from here
DBWriter(Environment env, Database db, StoredClassCatalog scc)
throws DatabaseException {
myDb = db;
myEnv = env;
dataBinding = new SerialBinding(scc, PayloadData.class);
}
Now we implement our thread's run() method. This is the method that is run when DBWriter threads are started in the main program (see TxnGuide.java).
// Thread method that writes a series of records
// to the database using transaction protection.
// Deadlock handling is demonstrated here.
public void run () {
The first thing we do is get a null transaction handle before going into our main loop.
Transaction txn = null;
// Perform 50 transactions
for (int i=0; i<50; i++) {
Next we declare a retry variable. This is used to determine whether a deadlock should result in our retrying the operation. We also declare a retry_count variable that is used to make sure we do not retry a transaction forever in the unlikely event that the thread is unable to ever get a necessary lock. (The only thing that might cause this is if some other thread dies while holding an important lock. This is the only code that we have to guard against that because the simplicity of this application makes it highly unlikely that it will ever occur.)
boolean retry = true;
int retry_count = 0;
// while loop is used for deadlock retries
while (retry) {
Now we go into the try block that we use for deadlock detection. We also begin our transaction here.
// try block used for deadlock detection and
// general db exception handling
try {
// Get a transaction
txn = myEnv.beginTransaction(null, null);
Now we write 10 records under the transaction that we have just begun. By combining multiple writes together under a single transaction, we increase the likelihood that a deadlock will occur. Normally, you want to reduce the potential for a deadlock and in this case the way to do that is to perform a single write per transaction. In other words, we should be using auto commit to write to our database for this workload.
However, we want to show deadlock handling and by performing multiple writes per transaction we can actually observe deadlocks occurring. We also want to underscore the idea that you can combing multiple database operations together in a single atomic unit of work. So for our example, we do the (slightly) wrong thing.
Further, notice that we store our key into a DatabaseEntry using com.sleepycat.bind.tuple.StringBinding to perform the serialization. Also, when we instantiate the PayloadData object, we call getName() which gives us the string representation of this thread's name, as well as Random.nextDouble() which gives us a random double value. This latter value is used so as to avoid duplicate records in the database.
// Write 10 records to the db
// for each transaction
for (int j = 0; j < 10; j++) {
// Get the key
DatabaseEntry key = new DatabaseEntry();
StringBinding.stringToEntry(keys[j], key);
// Get the data
PayloadData pd = new PayloadData(i+j, getName(),
generator.nextDouble());
DatabaseEntry data = new DatabaseEntry();
dataBinding.objectToEntry(pd, data);
// Do the put
myDb.put(txn, key, data);
}
Having completed the inner database write loop, we could simply commit the transaction and continue on to the next block of 10 writes. However, we want to first illustrate a few points about transactional processing so instead we call our countRecords() method before calling the transaction commit. countRecords() uses a cursor to read every record in the database and return a count of the number of records that it found.
Because countRecords() reads every record in the database, if used incorrectly the thread will self-deadlock. The writer thread has just written 500 records to the database, but because the transaction used for that write has not yet been committed, each of those 500 records are still locked by the thread's transaction. If we then simply run a non-transactional cursor over the database from within the same thread that has locked those 500 records, the cursor will block when it tries to read one of those transactional protected records. The thread immediately stops operation at that point while the cursor waits for the read lock it has requested. Because that read lock will never be released (the thread can never make any forward progress), this represents a self-deadlock for the the thread.
There are three ways to prevent this self-deadlock:
We can move the call to countRecords() to a point after the thread's transaction has committed.
We can allow countRecords() to operate under the same transaction as all of the writes were performed.
We can reduce our isolation guarantee for the application by allowing uncommitted reads.
For this example, we choose to use option 3 (uncommitted reads) to avoid the deadlock. This means that we have to open our database such that it supports uncommitted reads, and we have to open our cursor handle so that it knows to perform uncommitted reads.
Note that in In-Memory Transaction Example, we simply perform the cursor operation using the same transaction as is used for the thread's writes.
// commit
System.out.println(getName() + " : committing txn : "
+ i);
// Using uncommitted reads to avoid the deadlock, so null
// is passed for the transaction here.
System.out.println(getName() + " : Found " +
countRecords(null) + " records in the database.");
Having performed this somewhat inelegant counting of the records in the database, we can now commit the transaction.
try {
txn.commit();
txn = null;
} catch (DatabaseException e) {
System.err.println("Error on txn commit: " +
e.toString());
}
retry = false;
If all goes well with the commit, we are done and we can move on to the next batch of 10 records to add to the database. However, in the event of an error, we must handle our exceptions correctly. The first of these is a deadlock exception. In the event of a deadlock, we want to abort and retry the transaction, provided that we have not already exceeded our retry limit for this transaction.
} catch (DeadlockException de) {
System.out.println("################# " + getName() +
" : caught deadlock");
// retry if necessary
if (retry_count < MAX_RETRY) {
System.err.println(getName() +
" : Retrying operation.");
retry = true;
retry_count++;
} else {
System.err.println(getName() +
" : out of retries. Giving up.");
retry = false;
}
In the event of a standard, non-specific database exception, we simply log the exception and then give up (the transaction is not retried).
} catch (DatabaseException e) {
// abort and don't retry
retry = false;
System.err.println(getName() +
" : caught exception: " + e.toString());
System.err.println(getName() +
" : errno: " + e.getErrno());
e.printStackTrace();
And, finally, we always abort the transaction if the transaction handle is not null. Note that immediately after committing our transaction, we set the transaction handle to null to guard against aborting a transaction that has already been committed.
} finally {
if (txn != null) {
try {
txn.abort();
} catch (Exception e) {
System.err.println("Error aborting txn: " +
e.toString());
e.printStackTrace();
}
}
}
}
}
}
The final piece of our DBWriter class is the countRecords() implementation. Notice how in this example we open the cursor such that it performs uncommitted reads:
// A method that counts every record in the database.
// Note that this method exists only for illustrative purposes.
// A more straight-forward way to count the number of records in
// a database is to use the Database.getStats() method.
private int countRecords(Transaction txn) throws DatabaseException {
DatabaseEntry key = new DatabaseEntry();
DatabaseEntry data = new DatabaseEntry();
int count = 0;
Cursor cursor = null;
try {
// Get the cursor
CursorConfig cc = new CursorConfig();
cc.setReadUncomitted(true);
cursor = myDb.openCursor(txn, cc);
while (cursor.getNext(key, data, LockMode.DEFAULT) ==
OperationStatus.SUCCESS) {
count++;
}
} finally {
if (cursor != null) {
cursor.close();
}
}
return count;
}
}
This completes our transactional example. If you would like to experiment with this code, you can find the example in the following location in your JE distribution:
DB_INSTALL/examples_java/src/db/txn