package persist.txn;
import com.sleepycat.db.CursorConfig;
import com.sleepycat.db.DatabaseException;
import com.sleepycat.db.DeadlockException;
import com.sleepycat.db.Environment;
import com.sleepycat.db.Transaction;
import com.sleepycat.persist.EntityCursor;
import com.sleepycat.persist.EntityStore;
import com.sleepycat.persist.PrimaryIndex;
import java.util.Iterator;
import java.util.Random;
import java.io.UnsupportedEncodingException;
public class StoreWriter extends Thread
{
private EntityStore myStore = null;
private Environment myEnv = null;
private PrimaryIndex<Integer,PayloadDataEntity> pdKey;
private Random generator = new Random();
private boolean passTxn = false;
private static final int MAX_RETRY = 20;
StoreWriter(Environment env, EntityStore store)
throws DatabaseException {
myStore = store;
myEnv = env;
pdKey = myStore.getPrimaryIndex(Integer.class,
PayloadDataEntity.class);
}
public void run () {
Transaction txn = null;
for (int i=0; i<50; i++) {
boolean retry = true;
int retry_count = 0;
while (retry) {
try {
txn = myEnv.beginTransaction(null, null);
for (int j = 0; j < 10; j++) {
PayloadDataEntity pd = new PayloadDataEntity();
pd.setID(i + j);
pd.setThreadName(getName());
pd.setDoubleData(generator.nextDouble());
pdKey.put(txn, pd);
}
System.out.println(getName() + " : committing txn : " + i);
System.out.println(getName() + " : Found " +
countObjects(txn) + " objects in the store.");
try {
txn.commit();
txn = null;
} catch (DatabaseException e) {
System.err.println("Error on txn commit: " +
e.toString());
}
retry = false;
} catch (DeadlockException de) {
System.out.println("################# " + getName() +
" : caught deadlock");
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;
}
} catch (DatabaseException e) {
retry = false;
System.err.println(getName() +
" : caught exception: " + e.toString());
System.err.println(getName() +
" : errno: " + e.getErrno());
e.printStackTrace();
} finally {
if (txn != null) {
try {
txn.abort();
} catch (Exception e) {
System.err.println("Error aborting transaction: " +
e.toString());
e.printStackTrace();
}
}
}
}
}
}
private int countObjects(Transaction txn) throws DatabaseException {
int count = 0;
CursorConfig cc = new CursorConfig();
cc.setReadUncommitted(true);
EntityCursor<PayloadDataEntity> cursor = pdKey.entities(txn, cc);
try {
for (PayloadDataEntity pdi : cursor) {
count++;
}
} finally {
if (cursor != null) {
cursor.close();
}
}
return count;
}
}