You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@accumulo.apache.org by ec...@apache.org on 2014/01/02 20:46:40 UTC
[2/3] git commit: ACCUMULO-2124 minor updates to comments
ACCUMULO-2124 minor updates to comments
Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/9bfef59f
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/9bfef59f
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/9bfef59f
Branch: refs/heads/master
Commit: 9bfef59ffe75ccea174222f6b000b6009e611f65
Parents: a95831a
Author: Eric Newton <er...@gmail.com>
Authored: Thu Jan 2 14:46:37 2014 -0500
Committer: Eric Newton <er...@gmail.com>
Committed: Thu Jan 2 14:46:37 2014 -0500
----------------------------------------------------------------------
.../examples/simple/reservations/ARS.java | 126 +++++++++----------
1 file changed, 63 insertions(+), 63 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/accumulo/blob/9bfef59f/examples/simple/src/main/java/org/apache/accumulo/examples/simple/reservations/ARS.java
----------------------------------------------------------------------
diff --git a/examples/simple/src/main/java/org/apache/accumulo/examples/simple/reservations/ARS.java b/examples/simple/src/main/java/org/apache/accumulo/examples/simple/reservations/ARS.java
index 509a674..5e71acf 100644
--- a/examples/simple/src/main/java/org/apache/accumulo/examples/simple/reservations/ARS.java
+++ b/examples/simple/src/main/java/org/apache/accumulo/examples/simple/reservations/ARS.java
@@ -41,51 +41,51 @@ import org.apache.hadoop.io.Text;
/**
* Accumulo Reservation System : An example reservation system using Accumulo. Supports atomic reservations of a resource at a date. Wait list are also
- * supported. Inorder to keep the example simple no checking is done of the date. Also the code is inefficient, if interested in improving it take a look at the
- * EXCERCISE comments.
+ * supported. In order to keep the example simple, no checking is done of the date. Also the code is inefficient, if interested in improving it take a look at
+ * the EXCERCISE comments.
*/
// EXCERCISE create a test that verifies correctness under concurrency. For example, have M threads making reservations against N resources. Each thread could
-// randomly reserver and cancel resources for a single user. When each thread finishes it know what the state of its single user should be. When all threads
-// finish collect their expected state and verify the status of all users and resources. For extra credit run the test on a IAAS provider using 10 nodes and
+// randomly reserve and cancel resources for a single user. When each thread finishes, it know what the state of its single user should be. When all threads
+// finish, collect their expected state and verify the status of all users and resources. For extra credit run the test on a IAAS provider using 10 nodes and
// 10 threads per node.
public class ARS {
-
+
private Connector conn;
private String rTable;
-
+
public enum ReservationResult {
RESERVED, WAIT_LISTED
}
-
+
public ARS(Connector conn, String rTable) {
this.conn = conn;
this.rTable = rTable;
}
-
+
public List<String> setCapacity(String what, String when, int count) {
- // EXCERCISE implement this method which atomically sets a capacity and returns anyone who was moved to the waitlist if the capacity was decreased
-
+ // EXCERCISE implement this method which atomically sets a capacity and returns anyone who was moved to the wait list if the capacity was decreased
+
throw new UnsupportedOperationException();
}
-
+
public ReservationResult reserve(String what, String when, String who) throws Exception {
-
+
String row = what + ":" + when;
-
+
// EXCERCISE This code assumes there is no reservation and tries to create one. If a reservation exist then the update will fail. This is a good strategy
- // when its expected there are usually no reservations. Could modify the code to scan first.
-
+ // when it is expected there are usually no reservations. Could modify the code to scan first.
+
// The following mutation requires that the column tx:seq does not exist and will fail if it does.
ConditionalMutation update = new ConditionalMutation(row, new Condition("tx", "seq"));
update.put("tx", "seq", "0");
update.put("res", String.format("%04d", 0), who);
-
+
ReservationResult result = ReservationResult.RESERVED;
-
+
ConditionalWriter cwriter = conn.createConditionalWriter(rTable, new ConditionalWriterConfig());
-
+
try {
while (true) {
Status status = cwriter.write(update).getStatus();
@@ -99,24 +99,24 @@ public class ARS {
default:
throw new RuntimeException("Unexpected status " + status);
}
-
- // EXCERCISE in the case of many threads trying to reserve a slot this approach of immediately retrying is inefficient. Exponential backoff is good
- // general solution to solve contention problems like this. However in this particular case exponential backoff could penalize the earliest threads that
- // attempted to make a reservation putting them later in the list. A more complex solution could involve having independent sub-queues within the row
- // that approximately maintain arrival order and use exponential back off to fairly merge the sub-queues into the main queue.
-
- // its important to use an isolated scanner so that only whole mutations are seen
+
+ // EXCERCISE in the case of many threads trying to reserve a slot, this approach of immediately retrying is inefficient. Exponential back-off is good
+ // general solution to solve contention problems like this. However in this particular case, exponential back-off could penalize the earliest threads
+ // that attempted to make a reservation by putting them later in the list. A more complex solution could involve having independent sub-queues within
+ // the row that approximately maintain arrival order and use exponential back off to fairly merge the sub-queues into the main queue.
+
+ // it is important to use an isolated scanner so that only whole mutations are seen
Scanner scanner = new IsolatedScanner(conn.createScanner(rTable, Authorizations.EMPTY));
scanner.setRange(new Range(row));
-
+
int seq = -1;
int maxReservation = -1;
-
+
for (Entry<Key,Value> entry : scanner) {
String cf = entry.getKey().getColumnFamilyData().toString();
String cq = entry.getKey().getColumnQualifierData().toString();
String val = entry.getValue().toString();
-
+
if (cf.equals("tx") && cq.equals("seq")) {
seq = Integer.parseInt(val);
} else if (cf.equals("res")) {
@@ -127,21 +127,21 @@ public class ARS {
return ReservationResult.RESERVED; // already have the first reservation
else
return ReservationResult.WAIT_LISTED; // already on wait list
-
+
// EXCERCISE the way this code finds the max reservation is very inefficient.... it would be better if it did not have to scan the entire row.
// One possibility is to just use the sequence number. Could also consider sorting the data in another way and/or using an iterator.
maxReservation = Integer.parseInt(cq);
}
}
-
+
Condition condition = new Condition("tx", "seq");
if (seq >= 0)
condition.setValue(seq + ""); // only expect a seq # if one was seen
-
+
update = new ConditionalMutation(row, condition);
update.put("tx", "seq", (seq + 1) + "");
update.put("res", String.format("%04d", maxReservation + 1), who);
-
+
// EXCERCISE if set capacity is implemented, then result should take capacity into account
if (maxReservation == -1)
result = ReservationResult.RESERVED; // if successful, will be first reservation
@@ -151,48 +151,48 @@ public class ARS {
} finally {
cwriter.close();
}
-
+
}
-
+
public void cancel(String what, String when, String who) throws Exception {
-
+
String row = what + ":" + when;
-
- // Even though this method is only deleting a column, its important to use a conditional writer. By updating the seq # when deleting a reservation it
+
+ // Even though this method is only deleting a column, its important to use a conditional writer. By updating the seq # when deleting a reservation, it
// will cause any concurrent reservations to retry. If this delete were done using a batch writer, then a concurrent reservation could report WAIT_LISTED
// when it actually got the reservation.
-
+
ConditionalWriter cwriter = conn.createConditionalWriter(rTable, new ConditionalWriterConfig());
-
+
try {
while (true) {
-
+
// its important to use an isolated scanner so that only whole mutations are seen
Scanner scanner = new IsolatedScanner(conn.createScanner(rTable, Authorizations.EMPTY));
scanner.setRange(new Range(row));
-
+
int seq = -1;
String reservation = null;
-
+
for (Entry<Key,Value> entry : scanner) {
String cf = entry.getKey().getColumnFamilyData().toString();
String cq = entry.getKey().getColumnQualifierData().toString();
String val = entry.getValue().toString();
-
+
// EXCERCISE avoid linear scan
-
+
if (cf.equals("tx") && cq.equals("seq")) {
seq = Integer.parseInt(val);
} else if (cf.equals("res") && val.equals(who)) {
reservation = cq;
}
}
-
+
if (reservation != null) {
ConditionalMutation update = new ConditionalMutation(row, new Condition("tx", "seq").setValue(seq + ""));
update.putDelete("res", reservation);
update.put("tx", "seq", (seq + 1) + "");
-
+
Status status = cwriter.write(update).getStatus();
switch (status) {
case ACCEPTED:
@@ -201,55 +201,55 @@ public class ARS {
case REJECTED:
case UNKNOWN:
// retry
- // EXCERCISE exponential backoff could be used here
+ // EXCERCISE exponential back-off could be used here
break;
default:
throw new RuntimeException("Unexpected status " + status);
}
-
+
} else {
// not reserved, nothing to do
break;
}
-
+
}
} finally {
cwriter.close();
}
}
-
+
public List<String> list(String what, String when) throws Exception {
String row = what + ":" + when;
-
+
// its important to use an isolated scanner so that only whole mutations are seen
Scanner scanner = new IsolatedScanner(conn.createScanner(rTable, Authorizations.EMPTY));
scanner.setRange(new Range(row));
scanner.fetchColumnFamily(new Text("res"));
-
+
List<String> reservations = new ArrayList<String>();
-
+
for (Entry<Key,Value> entry : scanner) {
String val = entry.getValue().toString();
reservations.add(val);
}
-
+
return reservations;
}
-
+
public static void main(String[] args) throws Exception {
final ConsoleReader reader = new ConsoleReader();
ARS ars = null;
-
+
while (true) {
String line = reader.readLine(">");
if (line == null)
break;
-
+
final String[] tokens = line.split("\\s+");
-
+
if (tokens[0].equals("reserve") && tokens.length >= 4 && ars != null) {
// start up multiple threads all trying to reserve the same resource, no more than one should succeed
-
+
final ARS fars = ars;
ArrayList<Thread> threads = new ArrayList<Thread>();
for (int i = 3; i < tokens.length; i++) {
@@ -264,16 +264,16 @@ public class ARS {
}
}
};
-
+
threads.add(new Thread(reservationTask));
}
-
+
for (Thread thread : threads)
thread.start();
-
+
for (Thread thread : threads)
thread.join();
-
+
} else if (tokens[0].equals("cancel") && tokens.length == 4 && ars != null) {
ars.cancel(tokens[1], tokens[2], tokens[3]);
} else if (tokens[0].equals("list") && tokens.length == 3 && ars != null) {