1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
|
package persist.txn;
import java.util.Random;
import com.sleepycat.je.CursorConfig;
import com.sleepycat.je.DatabaseException;
import com.sleepycat.je.DeadlockException;
import com.sleepycat.je.Environment;
import com.sleepycat.je.Transaction;
import com.sleepycat.persist.EntityCursor;
import com.sleepycat.persist.EntityStore;
import com.sleepycat.persist.PrimaryIndex;
public class StoreWriter extends Thread
{
private EntityStore myStore = null;
private Environment myEnv = null;
private PrimaryIndex<Integer,PayloadDataEntity> pdIndex;
private Random generator = new Random();
private boolean passTxn = false;
private static final int MAX_RETRY = 20;
// Constructor. Get our handles from here
StoreWriter(Environment env, EntityStore store)
throws DatabaseException {
myStore = store;
myEnv = env;
// Open the data accessor. This is used to store persistent
// objects.
pdIndex = myStore.getPrimaryIndex(Integer.class,
PayloadDataEntity.class);
}
// Thread method that writes a series of objects
// to the store using transaction protection.
// Deadlock handling is demonstrated here.
public void run () {
Transaction txn = null;
// Perform 50 transactions
for (int i=0; i<50; i++) {
boolean retry = true;
int retry_count = 0;
// while loop is used for deadlock retries
while (retry) {
// try block used for deadlock detection and
// general exception handling
try {
// Get a transaction
txn = myEnv.beginTransaction(null, null);
// Write 10 PayloadDataEntity objects to the
// store for each transaction
for (int j = 0; j < 10; j++) {
// Instantiate an object
PayloadDataEntity pd = new PayloadDataEntity();
// Set the Object ID. This is used as the primary key.
pd.setID(i + j);
// The thread name is used as a secondary key, and
// it is retrieved by this class's getName() method.
pd.setThreadName(getName());
// The last bit of data that we use is a double
// that we generate randomly. This data is not
// indexed.
pd.setDoubleData(generator.nextDouble());
// Do the put
pdIndex.put(txn, pd);
}
// commit
System.out.println(getName() + " : committing txn : " + i);
System.out.println(getName() + " : Found " +
countObjects(null) + " 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");
// 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;
}
} catch (DatabaseException e) {
// abort and don't retry
retry = false;
System.err.println(getName() +
" : caught exception: " + e.toString());
System.err.println(getName() +
" : errno: " + e.toString());
e.printStackTrace();
} finally {
if (txn != null) {
try {
txn.abort();
} catch (Exception e) {
System.err.println("Error aborting transaction: " +
e.toString());
e.printStackTrace();
}
}
}
}
}
}
// This simply counts the number of objects contained in the
// store and returns the result. You can use this method
// in three ways:
//
// First call it with an active txn handle.
//
// Secondly, configure the cursor for dirty reads
//
// Third, call countObjects AFTER the writer has committed
// its transaction.
//
// If you do none of these things, the writer thread will
// self-deadlock.
private int countObjects(Transaction txn) throws DatabaseException {
int count = 0;
CursorConfig cc = new CursorConfig();
// This is ignored if the store is not opened with uncommitted read
// support.
cc.setReadUncommitted(true);
EntityCursor<PayloadDataEntity> cursor = pdIndex.entities(txn, cc);
try {
for (PayloadDataEntity pdi : cursor) {
count++;
}
} finally {
if (cursor != null) {
cursor.close();
}
}
return count;
}
}
|
