accumulo-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ctubb...@apache.org
Subject [04/18] git commit: ACCUMULO-1613 added documentation for conditional writer
Date Wed, 20 Nov 2013 23:33:23 GMT
ACCUMULO-1613 added documentation for conditional writer


Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/73971825
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/73971825
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/73971825

Branch: refs/heads/master
Commit: 739718253bd49dbe5c74e041b5910cfd12edad29
Parents: 2faafcc
Author: Keith Turner <kturner@apache.org>
Authored: Tue Nov 19 17:15:23 2013 -0500
Committer: Keith Turner <kturner@apache.org>
Committed: Tue Nov 19 17:15:23 2013 -0500

----------------------------------------------------------------------
 .../accumulo/core/client/ConditionalWriter.java |   3 +
 .../apache/accumulo/core/data/Condition.java    |  54 ++++
 .../accumulo/core/data/ConditionalMutation.java |   2 +
 .../accumulo_user_manual/chapters/clients.tex   |  27 +-
 .../examples/simple/reservations/ARS.java       | 308 +++++++++++++++++++
 .../resources/docs/examples/README.reservations |  66 ++++
 6 files changed, 459 insertions(+), 1 deletion(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/accumulo/blob/73971825/core/src/main/java/org/apache/accumulo/core/client/ConditionalWriter.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/client/ConditionalWriter.java b/core/src/main/java/org/apache/accumulo/core/client/ConditionalWriter.java
index 4ed4d31..2c24a2e 100644
--- a/core/src/main/java/org/apache/accumulo/core/client/ConditionalWriter.java
+++ b/core/src/main/java/org/apache/accumulo/core/client/ConditionalWriter.java
@@ -23,6 +23,9 @@ import org.apache.accumulo.core.client.impl.thrift.SecurityErrorCode;
 import org.apache.accumulo.core.data.ConditionalMutation;
 
 /**
+ * ConditionalWriter provides the ability to do efficient, atomic read-modify-write operations
on rows. These operations are performed on the tablet server
+ * while a row lock is held.
+ * 
  * @since 1.6.0
  */
 public interface ConditionalWriter {

http://git-wip-us.apache.org/repos/asf/accumulo/blob/73971825/core/src/main/java/org/apache/accumulo/core/data/Condition.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/data/Condition.java b/core/src/main/java/org/apache/accumulo/core/data/Condition.java
index df20682..fc8f2bf 100644
--- a/core/src/main/java/org/apache/accumulo/core/data/Condition.java
+++ b/core/src/main/java/org/apache/accumulo/core/data/Condition.java
@@ -26,6 +26,7 @@ import org.apache.accumulo.core.util.ArgumentChecker;
 import org.apache.hadoop.io.Text;
 
 /**
+ * Conditions that must be met on a particular column in a row.
  * 
  * @since 1.6.0
  */
@@ -76,6 +77,13 @@ public class Condition {
     return cq;
   }
 
+  /**
+   * Sets the version for the column to check. If this is not set then the latest column
will be checked, unless iterators do something different.
+   * 
+   * @param ts
+   * @return returns this
+   */
+
   public Condition setTimestamp(long ts) {
     this.ts = ts;
     return this;
@@ -85,24 +93,53 @@ public class Condition {
     return ts;
   }
 
+  /**
+   * see {@link #setValue(byte[])}
+   * 
+   * @param value
+   * @return returns this
+   */
+
   public Condition setValue(CharSequence value) {
     ArgumentChecker.notNull(value);
     this.val = new ArrayByteSequence(value.toString().getBytes(Constants.UTF8));
     return this;
   }
 
+  /**
+   * This method sets the expected value of a column. Inorder for the condition to pass the
column must exist and have this value. If a value is not set, then
+   * the column must be absent for the condition to pass.
+   * 
+   * @param value
+   * @return returns this
+   */
+
   public Condition setValue(byte[] value) {
     ArgumentChecker.notNull(value);
     this.val = new ArrayByteSequence(value);
     return this;
   }
   
+  /**
+   * see {@link #setValue(byte[])}
+   * 
+   * @param value
+   * @return returns this
+   */
+
   public Condition setValue(Text value) {
     ArgumentChecker.notNull(value);
     this.val = new ArrayByteSequence(value.getBytes(), 0, value.getLength());
     return this;
   }
   
+  /**
+   * see {@link #setValue(byte[])}
+   * 
+   * @param value
+   * @return returns this
+   */
+
   public Condition setValue(ByteSequence value) {
     ArgumentChecker.notNull(value);
     this.val = value;
@@ -113,6 +150,13 @@ public class Condition {
     return val;
   }
 
+  /**
+   * Sets the visibility for the column to check. If not set it defaults to empty visibility.
+   * 
+   * @param cv
+   * @return returns this
+   */
+
   public Condition setVisibility(ColumnVisibility cv) {
     ArgumentChecker.notNull(cv);
     this.cv = new ArrayByteSequence(cv.getExpression());
@@ -123,6 +167,16 @@ public class Condition {
     return cv;
   }
 
+  /**
+   * Set iterators to use when reading the columns value. These iterators will be applied
in addition to the iterators configured for the table. Using iterators
+   * its possible to test other conditions, besides equality and absence, like less than.
On the server side the iterators will be seeked using a range that
+   * covers only the family, qualifier, and visibility (if the timestamp is set then it will
be used to narrow the range). Value equality will be tested using
+   * the first entry returned by the iterator stack.
+   * 
+   * @param iterators
+   * @return returns this
+   */
+
   public Condition setIterators(IteratorSetting... iterators) {
     ArgumentChecker.notNull(iterators);
     

http://git-wip-us.apache.org/repos/asf/accumulo/blob/73971825/core/src/main/java/org/apache/accumulo/core/data/ConditionalMutation.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/accumulo/core/data/ConditionalMutation.java b/core/src/main/java/org/apache/accumulo/core/data/ConditionalMutation.java
index 510396d..c438f6d 100644
--- a/core/src/main/java/org/apache/accumulo/core/data/ConditionalMutation.java
+++ b/core/src/main/java/org/apache/accumulo/core/data/ConditionalMutation.java
@@ -26,6 +26,8 @@ import org.apache.accumulo.core.util.ArgumentChecker;
 import org.apache.hadoop.io.Text;
 
 /**
+ * A Mutation that contains a list of conditions that must all be met before the mutation
is applied.
+ * 
  * @since 1.6.0
  */
 public class ConditionalMutation extends Mutation {

http://git-wip-us.apache.org/repos/asf/accumulo/blob/73971825/docs/src/main/latex/accumulo_user_manual/chapters/clients.tex
----------------------------------------------------------------------
diff --git a/docs/src/main/latex/accumulo_user_manual/chapters/clients.tex b/docs/src/main/latex/accumulo_user_manual/chapters/clients.tex
index 18fbafc..9b35d37 100644
--- a/docs/src/main/latex/accumulo_user_manual/chapters/clients.tex
+++ b/docs/src/main/latex/accumulo_user_manual/chapters/clients.tex
@@ -116,6 +116,31 @@ writer.close();
 An example of using the batch writer can be found at\\
 accumulo/docs/examples/README.batch
 
+\subsection{ConditionalWriter} 
+The ConditionalWriter enables efficient, atomic read-modify-write operations on
+rows.  The ConditionalWriter writes special Mutations which have a list of per
+column conditions that must all be met before the mutation is applied.  The
+conditions are checked in the tablet server while a row lock is
+held\footnote{Mutations written by the BatchWriter will not obtain a row
+lock.}.  The conditions that can be checked for a column are equality and
+absence.  For example a conditional mutation can require that column A is
+absent inorder to be applied.  Iterators can be applied when checking
+conditions.  Using iterators, many other operations besides equality and
+absence can be checked.  For example, using an iterator that converts values
+less than 5 to 0 and everything else to 1, its possible to only apply a
+mutation when a column is less than 5.
+
+In the case when a tablet server dies after a client sent a conditional
+mutation, its not known if the mutation was applied or not.  When this happens
+the ConditionalWriter reports a status of UNKNOWN for the ConditionalMutation.
+In many cases this situation can be dealt with by simply reading the row again
+and possibly sending another conditional mutation.  If this is not sufficient,
+then a higher level of abstraction can be built by storing transactional
+information within a row.
+ 
+An example of using the batch writer can be found at\\
+accumulo/docs/examples/README.reservations
+
 \section{Reading Data}
 
 Accumulo is optimized to quickly retrieve the value associated with a given key, and
@@ -329,4 +354,4 @@ for(KeyValue keyValue : results.getResultsIterator()) {
 
 client.closeScanner(scanner);
 \end{verbatim}
-\normalsize
\ No newline at end of file
+\normalsize

http://git-wip-us.apache.org/repos/asf/accumulo/blob/73971825/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
new file mode 100644
index 0000000..0c51843
--- /dev/null
+++ b/examples/simple/src/main/java/org/apache/accumulo/examples/simple/reservations/ARS.java
@@ -0,0 +1,308 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.accumulo.examples.simple.reservations;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map.Entry;
+
+import jline.console.ConsoleReader;
+
+import org.apache.accumulo.core.client.ConditionalWriter;
+import org.apache.accumulo.core.client.ConditionalWriter.Status;
+import org.apache.accumulo.core.client.ConditionalWriterConfig;
+import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.core.client.IsolatedScanner;
+import org.apache.accumulo.core.client.Scanner;
+import org.apache.accumulo.core.client.ZooKeeperInstance;
+import org.apache.accumulo.core.client.security.tokens.PasswordToken;
+import org.apache.accumulo.core.data.Condition;
+import org.apache.accumulo.core.data.ConditionalMutation;
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.Range;
+import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.security.Authorizations;
+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.
+ */
+
+// 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
+// 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
+
+    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.
+
+    // 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();
+        switch (status) {
+          case ACCEPTED:
+            return result;
+          case REJECTED:
+          case UNKNOWN:
+            // read the row and decide what to do
+            break;
+          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
+        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")) {
+            // EXCERCISE scanning the entire list to find if reserver is already in the list
is inefficient. One possible way to solve this would be to sort the
+            // data differently in Accumulo so that finding the reserver could be done quickly.
+            if (val.equals(who))
+              if (maxReservation == -1)
+                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
+        else
+          result = ReservationResult.WAIT_LISTED;
+      }
+    } 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
+    // 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:
+              // successfully canceled reservation
+              return;
+            case REJECTED:
+            case UNKNOWN:
+              // retry
+              // EXCERCISE exponential backoff 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++) {
+          final int whoIndex = i;
+          Runnable reservationTask = new Runnable() {
+            @Override
+            public void run() {
+              try {
+                reader.println("  " + String.format("%20s", tokens[whoIndex]) + " : " + fars.reserve(tokens[1],
tokens[2], tokens[whoIndex]));
+              } catch (Exception e) {
+                e.printStackTrace();
+              }
+            }
+          };
+
+          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) {
+        List<String> reservations = ars.list(tokens[1], tokens[2]);
+        if (reservations.size() > 0) {
+          reader.println("  Reservation holder : " + reservations.get(0));
+          if (reservations.size() > 1)
+            reader.println("  Wait list : " + reservations.subList(1, reservations.size()));
+        }
+      } else if (tokens[0].equals("quit") && tokens.length == 1) {
+        break;
+      } else if (tokens[0].equals("connect") && tokens.length == 6 && ars
== null) {
+        ZooKeeperInstance zki = new ZooKeeperInstance(tokens[1], tokens[2]);
+        Connector conn = zki.getConnector(tokens[3], new PasswordToken(tokens[4]));
+        if (conn.tableOperations().exists(tokens[5])) {
+          ars = new ARS(conn, tokens[5]);
+          reader.println("  connected");
+        } else
+          reader.println("  No Such Table");
+      } else {
+        System.out.println("  Commands : ");
+        if (ars == null) {
+          reader.println("    connect <instance> <zookeepers> <user> <pass>
<table>");
+        } else {
+          reader.println("    reserve <what> <when> <who> {who}");
+          reader.println("    cancel <what> <when> <who>");
+          reader.println("    list <what> <when>");
+        }
+      }
+    }
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/accumulo/blob/73971825/server/monitor/src/main/resources/docs/examples/README.reservations
----------------------------------------------------------------------
diff --git a/server/monitor/src/main/resources/docs/examples/README.reservations b/server/monitor/src/main/resources/docs/examples/README.reservations
new file mode 100644
index 0000000..7ad9cd3
--- /dev/null
+++ b/server/monitor/src/main/resources/docs/examples/README.reservations
@@ -0,0 +1,66 @@
+Title: Apache Accumulo Isolation Example
+Notice:    Licensed to the Apache Software Foundation (ASF) under one
+           or more contributor license agreements.  See the NOTICE file
+           distributed with this work for additional information
+           regarding copyright ownership.  The ASF licenses this file
+           to you under the Apache License, Version 2.0 (the
+           "License"); you may not use this file except in compliance
+           with the License.  You may obtain a copy of the License at
+           .
+             http://www.apache.org/licenses/LICENSE-2.0
+           .
+           Unless required by applicable law or agreed to in writing,
+           software distributed under the License is distributed on an
+           "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+           KIND, either express or implied.  See the License for the
+           specific language governing permissions and limitations
+           under the License.
+
+This example shows running a simple reservation system implemented using
+conditional mutations.  This system garuntees that only one concurrent user can
+reserve a resource.  The example's reserve command allows multiple users to be
+specified.  When this is done it creates a separate reservation thread for each
+user.  In the example below threads are spun up for alice, bob, eve, mallory,
+and trent to reserve room06 on 20140101.  Bob ends up getting the reservation
+and everyone else is put on a wait list.  The example code will take any string
+for what, when and who.
+
+    $ ./bin/accumulo org.apache.accumulo.examples.simple.reservations.ARS
+    >connect test16 localhost root secret ars
+      connected
+    >
+      Commands : 
+        reserve <what> <when> <who> {who}
+        cancel <what> <when> <who>
+        list <what> <when>
+    >reserve room06 20140101 alice bob eve mallory trent
+                       bob : RESERVED
+                   mallory : WAIT_LISTED
+                     alice : WAIT_LISTED
+                     trent : WAIT_LISTED
+                       eve : WAIT_LISTED
+    >list room06 20140101
+      Reservation holder : bob
+      Wait list : [mallory, alice, trent, eve]
+    >cancel room06 20140101 alice
+    >cancel room06 20140101 bob
+    >list room06 20140101
+      Reservation holder : mallory
+      Wait list : [trent, eve]
+    >quit
+
+Scanning the table in the Accumulo shell after running the example shows the
+following:
+
+    root@test16> table ars
+    root@test16 ars> scan
+    room06:20140101 res:0001 []    mallory
+    room06:20140101 res:0003 []    trent
+    room06:20140101 res:0004 []    eve
+    room06:20140101 tx:seq []    6
+
+The tx:seq column is incremented for each update to the row allowing for
+detection of concurrent changes.  For an update to go through the sequence
+number must not have changed since the data was read.  If it does change then
+the conditional mutation will fail and the example code will retry.
+


Mime
View raw message