Completed the implementation and test cases for Redis API.

Summary:
Completed the implementation for the Redis API for Lists.
The Redis API uses rocksdb as a backend to persistently
store maps from key->list. It supports basic operations
for appending, inserting, pushing, popping, and accessing
a list, given its key.

Test Plan:
  - Compile with: make redis_test
  - Test with: ./redis_test
  - Run all unit tests (for all rocksdb) with: make all check
  - To use an interactive REDIS client use: ./redis_test -m
  - To clean the database before use:       ./redis_test -m -d

Reviewers: haobo, dhruba, zshao

Reviewed By: haobo

CC: leveldb

Differential Revision: https://reviews.facebook.net/D10833

Makefile

@@ -62,7 +62,8 @@ TESTS = \
 	auto_roll_logger_test \
 	filelock_test \
 	merge_test \
-	stringappend_test
+	stringappend_test \
+	redis_test
 
 TOOLS = \
         sst_dump \
@@ -183,6 +184,9 @@ coding_test: util/coding_test.o $(LIBOBJECTS) $(TESTHARNESS)
 stringappend_test: utilities/merge_operators/string_append/stringappend_test.o $(LIBOBJECTS) $(TESTHARNESS)
 	$(CXX) utilities/merge_operators/string_append/stringappend_test.o $(LIBOBJECTS) $(TESTHARNESS) $(EXEC_LDFLAGS) -o $@ $(LDFLAGS)
 
+redis_test: utilities/redis/redis_lists_test.o $(LIBOBJECTS) $(TESTHARNESS)
+	$(CXX) utilities/redis/redis_lists_test.o $(LIBOBJECTS) $(TESTHARNESS) $(EXEC_LDFLAGS) -o $@ $(LDFLAGS)
+
 histogram_test: util/histogram_test.o $(LIBOBJECTS) $(TESTHARNESS)
 	$(CXX) util/histogram_test.o $(LIBOBJECTS) $(TESTHARNESS) $(EXEC_LDFLAGS) -o$@ $(LDFLAGS)
 

utilities/merge_operators/string_append/stringappend.cc

@@ -8,10 +8,10 @@
 
 #include <memory>
 #include <assert.h>
+
 #include "leveldb/slice.h"
 #include "leveldb/merge_operator.h"
 #include "utilities/merge_operators.h"
-#include <iostream>
 
 namespace leveldb {
 

utilities/redis/README

@@ -0,0 +1,14 @@
+This folder defines a REDIS-style interface for Rocksdb.
+Right now it is written as a simple tag-on in the leveldb::RedisLists class.
+It implements Redis Lists, and supports only the "non-blocking operations".
+
+Internally, the set of lists are stored in a rocksdb database, mapping keys to
+values. Each "value" is the list itself, storing a sequence of "elements".
+Each element is stored as a 32-bit-integer, followed by a sequence of bytes.
+The 32-bit-integer represents the length of the element (that is, the number
+of bytes that follow). And then that many bytes follow.
+
+
+NOTE: This README file may be old. See the actual redis_lists.cc file for
+definitive details on the implementation. There should be a header at the top
+of that file, explaining a bit of the implementation details.

utilities/redis/redis_list_exception.h

@@ -0,0 +1,24 @@
+/**
+ * A simple structure for exceptions in RedisLists.
+ *
+ * @author Deon Nicholas (dnicholas@fb.com)
+ * Copyright 2013 Facebook
+ */
+
+#ifndef LEVELDB_REDIS_LIST_EXCEPTION_H
+#define LEVELDB_REDIS_LIST_EXCEPTION_H
+
+#include <exception>
+
+namespace leveldb {
+
+class RedisListException: public std::exception {
+ public:
+  const char* what() const throw() {
+    return "Invalid operation or corrupt data in Redis List.";
+  }
+};
+
+} // namespace leveldb
+
+#endif // LEVELDB_REDIS_LIST_EXCEPTION_H

utilities/redis/redis_list_iterator.h

@@ -0,0 +1,306 @@
+/**
+ * RedisListIterator:
+ * An abstraction over the "list" concept (e.g.: for redis lists).
+ * Provides functionality to read, traverse, edit, and write these lists.
+ *
+ * Upon construction, the RedisListIterator is given a block of list data.
+ * Internally, it stores a pointer to the data and a pointer to current item.
+ * It also stores a "result" list that will be mutated over time.
+ *
+ * Traversal and mutation are done by "forward iteration".
+ * The Push() and Skip() methods will advance the iterator to the next item.
+ * However, Push() will also "write the current item to the result".
+ * Skip() will simply move to next item, causing current item to be dropped.
+ *
+ * Upon completion, the result (accessible by WriteResult()) will be saved.
+ * All "skipped" items will be gone; all "pushed" items will remain.
+ *
+ * @throws Any of the operations may throw a RedisListException if an invalid
+ *          operation is performed or if the data is found to be corrupt.
+ *
+ * @notes By default, if WriteResult() is called part-way through iteration,
+ *        it will automatically advance the iterator to the end, and Keep()
+ *        all items that haven't been traversed yet. This may be subject
+ *        to review.
+ *
+ * @notes Can access the "current" item via GetCurrent(), and other
+ *        list-specific information such as Length().
+ *
+ * @notes The internal representation is due to change at any time. Presently,
+ *        the list is represented as follows:
+ *          - 32-bit integer header: the number of items in the list
+ *          - For each item:
+ *              - 32-bit int (n): the number of bytes representing this item
+ *              - n bytes of data: the actual data.
+ *
+ * @author Deon Nicholas (dnicholas@fb.com)
+ * Copyright 2013 Facebook
+ */
+
+#include <string>
+
+#include "redis_list_exception.h"
+#include "leveldb/slice.h"
+#include "util/coding.h"
+
+namespace leveldb {
+
+/// An abstraction over the "list" concept.
+/// All operations may throw a RedisListException
+class RedisListIterator {
+ public:
+  /// Construct a redis-list-iterator based on data.
+  /// If the data is non-empty, it must formatted according to @notes above.
+  ///
+  /// If the data is valid, we can assume the following invariant(s):
+  ///  a) length_, num_bytes_ are set correctly.
+  ///  b) cur_byte_ always refers to the start of the current element,
+  ///       just before the bytes that specify element length.
+  ///  c) cur_elem_ is always the index of the current element.
+  ///  d) cur_elem_length_ is always the number of bytes in current element,
+  ///       excluding the 4-byte header itself.
+  ///  e) result_ will always contain data_[0..cur_byte_) and a header
+  ///  f) Whenever corrupt data is encountered or an invalid operation is
+  ///      attempted, a RedisListException will immediately be thrown.
+  RedisListIterator(const std::string& list_data)
+      : data_(list_data.data()),
+        num_bytes_(list_data.size()),
+        cur_byte_(0),
+        cur_elem_(0),
+        cur_elem_length_(0),
+        length_(0),
+        result_() {
+
+    // Initialize the result_ (reserve enough space for header)
+    InitializeResult();
+
+    // Parse the data only if it is not empty.
+    if (num_bytes_ == 0) {
+      return;
+    }
+
+    // If non-empty, but less than 4 bytes, data must be corrupt
+    if (num_bytes_ < sizeof(length_)) {
+      ThrowError("Corrupt header.");    // Will break control flow
+    }
+
+    // Good. The first bytes specify the number of elements
+    length_ = DecodeFixed32(data_);
+    cur_byte_ = sizeof(length_);
+
+    // If we have at least one element, point to that element.
+    // Also, read the first integer of the element (specifying the size),
+    //   if possible.
+    if (length_ > 0) {
+      if (cur_byte_ + sizeof(cur_elem_length_) <= num_bytes_) {
+        cur_elem_length_ = DecodeFixed32(data_+cur_byte_);
+      } else {
+        ThrowError("Corrupt data for first element.");
+      }
+    }
+
+    // At this point, we are fully set-up.
+    // The invariants described in the header should now be true.
+  }
+
+  /// Reserve some space for the result_.
+  /// Equivalent to result_.reserve(bytes).
+  void Reserve(int bytes) {
+    result_.reserve(bytes);
+  }
+
+  /// Go to next element in data file.
+  /// Also writes the current element to result_.
+  RedisListIterator& Push() {
+    WriteCurrentElement();
+    MoveNext();
+    return *this;
+  }
+
+  /// Go to next element in data file.
+  /// Drops/skips the current element. It will not be written to result_.
+  RedisListIterator& Skip() {
+    MoveNext();
+    --length_;          // One less item
+    --cur_elem_;        // We moved one forward, but index did not change
+    return *this;
+  }
+
+  /// Insert elem into the result_ (just BEFORE the current element / byte)
+  /// Note: if Done() (i.e.: iterator points to end), this will append elem.
+  void InsertElement(const Slice& elem) {
+    // Ensure we are in a valid state
+    CheckErrors();
+
+    const int kOrigSize = result_.size();
+    result_.resize(kOrigSize + SizeOf(elem));
+    EncodeFixed32(result_.data() + kOrigSize, elem.size());
+    memcpy(result_.data() + kOrigSize + sizeof(uint32_t),
+           elem.data(),
+           elem.size());
+    ++length_;
+    ++cur_elem_;
+  }
+
+  /// Access the current element, and save the result into *curElem
+  void GetCurrent(Slice* curElem) {
+    // Ensure we are in a valid state
+    CheckErrors();
+
+    // Ensure that we are not past the last element.
+    if (Done()) {
+      ThrowError("Invalid dereferencing.");
+    }
+
+    // Dereference the element
+    *curElem = Slice(data_+cur_byte_+sizeof(cur_elem_length_),
+                     cur_elem_length_);
+  }
+
+  // Number of elements
+  int Length() const {
+    return length_;
+  }
+
+  // Number of bytes in the final representation (i.e: WriteResult().size())
+  int Size() const {
+    // result_ holds the currently written data
+    // data_[cur_byte..num_bytes-1] is the remainder of the data
+    return result_.size() + (num_bytes_ - cur_byte_);
+  }
+
+  // Reached the end?
+  bool Done() const {
+    return cur_byte_ >= num_bytes_ || cur_elem_ >= length_;
+  }
+
+  /// Returns a string representing the final, edited, data.
+  /// Assumes that all bytes of data_ in the range [0,cur_byte_) have been read
+  ///  and that result_ contains this data.
+  /// The rest of the data must still be written.
+  /// So, this method ADVANCES THE ITERATOR TO THE END before writing.
+  Slice WriteResult() {
+    CheckErrors();
+
+    // The header should currently be filled with dummy data (0's)
+    // Correctly update the header.
+    // Note, this is safe since result_ is a vector (guaranteed contiguous)
+    EncodeFixed32(&result_[0],length_);
+
+    // Append the remainder of the data to the result.
+    result_.insert(result_.end(),data_+cur_byte_, data_ +num_bytes_);
+
+    // Seek to end of file
+    cur_byte_ = num_bytes_;
+    cur_elem_ = length_;
+    cur_elem_length_ = 0;
+
+    // Return the result
+    return Slice(result_.data(),result_.size());
+  }
+
+ public: // Static public functions
+
+  /// An upper-bound on the amount of bytes needed to store this element.
+  /// This is used to hide representation information from the client.
+  /// E.G. This can be used to compute the bytes we want to Reserve().
+  static uint32_t SizeOf(const Slice& elem) {
+    // [Integer Length . Data]
+    return sizeof(uint32_t) + elem.size();
+  }
+
+ private: // Private functions
+
+  /// Initializes the result_ string.
+  /// It will fill the first few bytes with 0's so that there is
+  ///  enough space for header information when we need to write later.
+  /// Currently, "header information" means: the length (number of elements)
+  /// Assumes that result_ is empty to begin with
+  void InitializeResult() {
+    assert(result_.empty());            // Should always be true.
+    result_.resize(sizeof(uint32_t),0); // Put a block of 0's as the header
+  }
+
+  /// Go to the next element (used in Push() and Skip())
+  void MoveNext() {
+    CheckErrors();
+
+    // Check to make sure we are not already in a finished state
+    if (Done()) {
+      ThrowError("Attempting to iterate past end of list.");
+    }
+
+    // Move forward one element.
+    cur_byte_ += sizeof(cur_elem_length_) + cur_elem_length_;
+    ++cur_elem_;
+
+    // If we are at the end, finish
+    if (Done()) {
+      cur_elem_length_ = 0;
+      return;
+    }
+
+    // Otherwise, we should be able to read the new element's length
+    if (cur_byte_ + sizeof(cur_elem_length_) > num_bytes_) {
+      ThrowError("Corrupt element data.");
+    }
+
+    // Set the new element's length
+    cur_elem_length_ = DecodeFixed32(data_+cur_byte_);
+
+    return;
+  }
+
+  /// Append the current element (pointed to by cur_byte_) to result_
+  /// Assumes result_ has already been reserved appropriately.
+  void WriteCurrentElement() {
+    // First verify that the iterator is still valid.
+    CheckErrors();
+    if (Done()) {
+      ThrowError("Attempting to write invalid element.");
+    }
+
+    // Append the cur element.
+    result_.insert(result_.end(),
+                   data_+cur_byte_,
+                   data_+cur_byte_+ sizeof(uint32_t) + cur_elem_length_);
+  }
+
+  /// Will ThrowError() if neccessary.
+  /// Checks for common/ubiquitous errors that can arise after most operations.
+  /// This method should be called before any reading operation.
+  /// If this function succeeds, then we are guaranteed to be in a valid state.
+  /// Other member functions should check for errors and ThrowError() also
+  ///  if an error occurs that is specific to it even while in a valid state.
+  void CheckErrors() {
+    // Check if any crazy thing has happened recently
+    if ((cur_elem_ > length_) ||                              // Bad index
+        (cur_byte_ > num_bytes_) ||                           // No more bytes
+        (cur_byte_ + cur_elem_length_ > num_bytes_) ||        // Item too large
+        (cur_byte_ == num_bytes_ && cur_elem_ != length_) ||  // Too many items
+        (cur_elem_ == length_ && cur_byte_ != num_bytes_)) {  // Too many bytes
+      ThrowError("Corrupt data.");
+    }
+  }
+
+  /// Will throw an exception based on the passed-in message.
+  /// This function is guaranteed to STOP THE CONTROL-FLOW.
+  /// (i.e.: you do not have to call "return" after calling ThrowError)
+  void ThrowError(const char* const msg = NULL) {
+    // TODO: For now we ignore the msg parameter. This can be expanded later.
+    throw RedisListException();
+  }
+
+ private:
+  const char* const data_;      // A pointer to the data (the first byte)
+  const uint32_t num_bytes_;    // The number of bytes in this list
+
+  uint32_t cur_byte_;           // The current byte being read
+  uint32_t cur_elem_;           // The current element being read
+  uint32_t cur_elem_length_;    // The number of bytes in current element
+
+  uint32_t length_;             // The number of elements in this list
+  std::vector<char> result_;    // The output data
+};
+
+} // namespace leveldb

utilities/redis/redis_lists.cc

@@ -0,0 +1,551 @@
+/**
+ * A (persistent) Redis API built using the rocksdb backend.
+ * Implements Redis Lists as described on: http://redis.io/commands#list
+ *
+ * @throws All functions may throw a RedisListException on error/corruption.
+ *
+ * @notes Internally, the set of lists is stored in a rocksdb database,
+ *        mapping keys to values. Each "value" is the list itself, storing
+ *        some kind of internal representation of the data. All the
+ *        representation details are handled by the RedisListIterator class.
+ *        The present file should be oblivious to the representation details,
+ *        handling only the client (Redis) API, and the calls to rocksdb.
+ *
+ * @TODO  Presently, all operations take at least O(NV) time where
+ *        N is the number of elements in the list, and V is the average
+ *        number of bytes per value in the list. So maybe, with merge operator
+ *        we can improve this to an optimal O(V) amortized time, since we
+ *        wouldn't have to read and re-write the entire list.
+ *
+ * @author Deon Nicholas (dnicholas@fb.com)
+ * Copyright 2013 Facebook
+ */
+
+#include "redis_lists.h"
+
+#include <iostream>
+#include <memory>
+#include <cmath>
+
+#include "leveldb/slice.h"
+#include "util/coding.h"
+
+namespace leveldb
+{
+
+/// Constructors
+
+RedisLists::RedisLists(const std::string& db_path,
+                       Options options, bool destructive)
+    : put_option_(),
+      get_option_() {
+
+  // Store the name of the database
+  db_name_ = db_path;
+
+  // If destructive, destroy the DB before re-opening it.
+  if (destructive) {
+    DestroyDB(db_name_, Options());
+  }
+
+  // Now open and deal with the db
+  DB* db;
+  Status s = DB::Open(options, db_name_, &db);
+  if (!s.ok()) {
+    std::cerr << "ERROR " << s.ToString() << std::endl;
+    assert(false);
+  }
+
+  db_ = std::unique_ptr<DB>(db);
+}
+
+
+/// Accessors
+
+// Number of elements in the list associated with key
+//   : throws RedisListException
+int RedisLists::Length(const std::string& key) {
+  // Extract the string data representing the list.
+  std::string data;
+  db_->Get(get_option_, key, &data);
+
+  // Return the length
+  RedisListIterator it(data);
+  return it.Length();
+}
+
+// Get the element at the specified index in the (list: key)
+// Returns <empty> ("") on out-of-bounds
+//   : throws RedisListException
+bool RedisLists::Index(const std::string& key, int32_t index,
+                       std::string* result) {
+  // Extract the string data representing the list.
+  std::string data;
+  db_->Get(get_option_, key, &data);
+
+  // Handle REDIS negative indices (from the end); fast iff Length() takes O(1)
+  if (index < 0) {
+    index = Length(key) - (-index);  //replace (-i) with (N-i).
+  }
+
+  // Iterate through the list until the desired index is found.
+  int curIndex = 0;
+  RedisListIterator it(data);
+  while(curIndex < index && !it.Done()) {
+    ++curIndex;
+    it.Skip();
+  }
+
+  // If we actually found the index
+  if (curIndex == index && !it.Done()) {
+    Slice elem;
+    it.GetCurrent(&elem);
+    if (result != NULL) {
+      *result = elem.ToString();
+    }
+
+    return true;
+  } else {
+    return false;
+  }
+}
+
+// Return a truncated version of the list.
+// First, negative values for first/last are interpreted as "end of list".
+// So, if first == -1, then it is re-set to index: (Length(key) - 1)
+// Then, return exactly those indices i such that first <= i <= last.
+//   : throws RedisListException
+std::vector<std::string> RedisLists::Range(const std::string& key,
+                                           int32_t first, int32_t last) {
+  // Extract the string data representing the list.
+  std::string data;
+  db_->Get(get_option_, key, &data);
+
+  // Handle negative bounds (-1 means last element, etc.)
+  int listLen = Length(key);
+  if (first < 0) {
+    first = listLen - (-first);           // Replace (-x) with (N-x)
+  }
+  if (last < 0) {
+    last = listLen - (-last);
+  }
+
+  // Verify bounds (and truncate the range so that it is valid)
+  first = std::max(first, 0);
+  last = std::min(last, listLen-1);
+  int len = std::max(last-first+1, 0);
+
+  // Initialize the resulting list
+  std::vector<std::string> result(len);
+
+  // Traverse the list and update the vector
+  int curIdx = 0;
+  Slice elem;
+  for (RedisListIterator it(data); !it.Done() && curIdx<=last; it.Skip()) {
+    if (first <= curIdx && curIdx <= last) {
+      it.GetCurrent(&elem);
+      result[curIdx-first].assign(elem.data(),elem.size());
+    }
+
+    ++curIdx;
+  }
+
+  // Return the result. Might be empty
+  return result;
+}
+
+// Print the (list: key) out to stdout. For debugging mostly. Public for now.
+void RedisLists::Print(const std::string& key) {
+  // Extract the string data representing the list.
+  std::string data;
+  db_->Get(get_option_, key, &data);
+
+  // Iterate through the list and print the items
+  Slice elem;
+  for (RedisListIterator it(data); !it.Done(); it.Skip()) {
+    it.GetCurrent(&elem);
+    std::cout << "ITEM " << elem.ToString() << std::endl;
+  }
+
+  //Now print the byte data
+  RedisListIterator it(data);
+  std::cout << "==Printing data==" << std::endl;
+  std::cout << data.size() << std::endl;
+  std::cout << it.Size() << " " << it.Length() << std::endl;
+  Slice result = it.WriteResult();
+  std::cout << result.data() << std::endl;
+  if (true) {
+    std::cout << "size: " << result.size() << std::endl;
+    const char* val = result.data();
+    for(int i=0; i<(int)result.size(); ++i) {
+      std::cout << (int)val[i] << " " << (val[i]>=32?val[i]:' ') << std::endl;
+    }
+    std::cout << std::endl;
+  }
+}
+
+/// Insert/Update Functions
+/// Note: The "real" insert function is private. See below.
+
+// InsertBefore and InsertAfter are simply wrappers around the Insert function.
+int RedisLists::InsertBefore(const std::string& key, const std::string& pivot,
+                             const std::string& value) {
+  return Insert(key, pivot, value, false);
+}
+
+int RedisLists::InsertAfter(const std::string& key, const std::string& pivot,
+                            const std::string& value) {
+  return Insert(key, pivot, value, true);
+}
+
+// Prepend value onto beginning of (list: key)
+//   : throws RedisListException
+int RedisLists::PushLeft(const std::string& key, const std::string& value) {
+  // Get the original list data
+  std::string data;
+  db_->Get(get_option_, key, &data);
+
+  // Construct the result
+  RedisListIterator it(data);
+  it.Reserve(it.Size() + it.SizeOf(value));
+  it.InsertElement(value);
+
+  // Push the data back to the db and return the length
+  db_->Put(put_option_, key, it.WriteResult());
+  return it.Length();
+}
+
+// Append value onto end of (list: key)
+// TODO: Make this O(1) time. Might require MergeOperator.
+//   : throws RedisListException
+int RedisLists::PushRight(const std::string& key, const std::string& value) {
+  // Get the original list data
+  std::string data;
+  db_->Get(get_option_, key, &data);
+
+  // Create an iterator to the data and seek to the end.
+  RedisListIterator it(data);
+  it.Reserve(it.Size() + it.SizeOf(value));
+  while (!it.Done()) {
+    it.Push();    // Write each element as we go
+  }
+
+  // Insert the new element at the current position (the end)
+  it.InsertElement(value);
+
+  // Push it back to the db, and return length
+  db_->Put(put_option_, key, it.WriteResult());
+  return it.Length();
+}
+
+// Set (list: key)[idx] = val. Return true on success, false on fail.
+//   : throws RedisListException
+bool RedisLists::Set(const std::string& key, int32_t index,
+                     const std::string& value) {
+  // Get the original list data
+  std::string data;
+  db_->Get(get_option_, key, &data);
+
+  // Handle negative index for REDIS (meaning -index from end of list)
+  if (index < 0) {
+    index = Length(key) - (-index);
+  }
+
+  // Iterate through the list until we find the element we want
+  int curIndex = 0;
+  RedisListIterator it(data);
+  it.Reserve(it.Size() + it.SizeOf(value));  // Over-estimate is fine
+  while(curIndex < index && !it.Done()) {
+    it.Push();
+    ++curIndex;
+  }
+
+  // If not found, return false (this occurs when index was invalid)
+  if (it.Done() || curIndex != index) {
+    return false;
+  }
+
+  // Write the new element value, and drop the previous element value
+  it.InsertElement(value);
+  it.Skip();
+
+  // Write the data to the database
+  // Check status, since it needs to return true/false guarantee
+  Status s = db_->Put(put_option_, key, it.WriteResult());
+
+  // Success
+  return s.ok();
+}
+
+/// Delete / Remove / Pop functions
+
+// Trim (list: key) so that it will only contain the indices from start..stop
+//  Invalid indices will not generate an error, just empty,
+//  or the portion of the list that fits in this interval
+//   : throws RedisListException
+bool RedisLists::Trim(const std::string& key, int32_t start, int32_t stop) {
+  // Get the original list data
+  std::string data;
+  db_->Get(get_option_, key, &data);
+
+  // Handle negative indices in REDIS
+  int listLen = Length(key);
+  if (start < 0) {
+    start = listLen - (-start);
+  }
+  if (stop < 0) {
+    stop = listLen - (-stop);
+  }
+
+  // Truncate bounds to only fit in the list
+  start = std::max(start, 0);
+  stop = std::min(stop, listLen-1);
+
+  // Construct an iterator for the list. Drop all undesired elements.
+  int curIndex = 0;
+  RedisListIterator it(data);
+  it.Reserve(it.Size());          // Over-estimate
+  while(!it.Done()) {
+    // If not within the range, just skip the item (drop it).
+    // Otherwise, continue as usual.
+    if (start <= curIndex && curIndex <= stop) {
+      it.Push();
+    } else {
+      it.Skip();
+    }
+
+    // Increment the current index
+    ++curIndex;
+  }
+
+  // Write the (possibly empty) result to the database
+  Status s = db_->Put(put_option_, key, it.WriteResult());
+
+  // Return true as long as the write succeeded
+  return s.ok();
+}
+
+// Return and remove the first element in the list (or "" if empty)
+//   : throws RedisListException
+bool RedisLists::PopLeft(const std::string& key, std::string* result) {
+  // Get the original list data
+  std::string data;
+  db_->Get(get_option_, key, &data);
+
+  // Point to first element in the list (if it exists), and get its value/size
+  RedisListIterator it(data);
+  if (it.Length() > 0) {            // Proceed only if list is non-empty
+    Slice elem;
+    it.GetCurrent(&elem);           // Store the value of the first element
+    it.Reserve(it.Size() - it.SizeOf(elem));
+    it.Skip();                      // DROP the first item and move to next
+
+    // Update the db
+    db_->Put(put_option_, key, it.WriteResult());
+
+    // Return the value
+    if (result != NULL) {
+      *result = elem.ToString();
+    }
+    return true;
+  } else {
+    return false;
+  }
+}
+
+// Remove and return the last element in the list (or "" if empty)
+// TODO: Make this O(1). Might require MergeOperator.
+//   : throws RedisListException
+bool RedisLists::PopRight(const std::string& key, std::string* result) {
+  // Extract the original list data
+  std::string data;
+  db_->Get(get_option_, key, &data);
+
+  // Construct an iterator to the data and move to last element
+  RedisListIterator it(data);
+  it.Reserve(it.Size());
+  int len = it.Length();
+  int curIndex = 0;
+  while(curIndex < (len-1) && !it.Done()) {
+    it.Push();
+    ++curIndex;
+  }
+
+  // Extract and drop/skip the last element
+  if (curIndex == len-1) {
+    assert(!it.Done());         // Sanity check. Should not have ended here.
+
+    // Extract and pop the element
+    Slice elem;
+    it.GetCurrent(&elem);       // Save value of element.
+    it.Skip();                  // Skip the element
+
+    // Write the result to the database
+    db_->Put(put_option_, key, it.WriteResult());
+
+    // Return the value
+    if (result != NULL) {
+      *result = elem.ToString();
+    }
+    return true;
+  } else {
+    // Must have been an empty list
+    assert(it.Done() && len==0 && curIndex == 0);
+    return false;
+  }
+}
+
+// Remove the (first or last) "num" occurrences of value in (list: key)
+//   : throws RedisListException
+int RedisLists::Remove(const std::string& key, int32_t num,
+                       const std::string& value) {
+  // Negative num ==> RemoveLast; Positive num ==> Remove First
+  if (num < 0) {
+    return RemoveLast(key, -num, value);
+  } else if (num > 0) {
+    return RemoveFirst(key, num, value);
+  } else {
+    return RemoveFirst(key, Length(key), value);
+  }
+}
+
+// Remove the first "num" occurrences of value in (list: key).
+//   : throws RedisListException
+int RedisLists::RemoveFirst(const std::string& key, int32_t num,
+                            const std::string& value) {
+  // Ensure that the number is positive
+  assert(num >= 0);
+
+  // Extract the original list data
+  std::string data;
+  db_->Get(get_option_, key, &data);
+
+  // Traverse the list, appending all but the desired occurrences of value
+  int numSkipped = 0;         // Keep track of the number of times value is seen
+  Slice elem;
+  RedisListIterator it(data);
+  it.Reserve(it.Size());
+  while (!it.Done()) {
+    it.GetCurrent(&elem);
+
+    if (elem == value && numSkipped < num) {
+      // Drop this item if desired
+      it.Skip();
+      ++numSkipped;
+    } else {
+      // Otherwise keep the item and proceed as normal
+      it.Push();
+    }
+  }
+
+  // Put the result back to the database
+  db_->Put(put_option_, key, it.WriteResult());
+
+  // Return the number of elements removed
+  return numSkipped;
+}
+
+
+// Remove the last "num" occurrences of value in (list: key).
+// TODO: I traverse the list 2x. Make faster. Might require MergeOperator.
+//   : throws RedisListException
+int RedisLists::RemoveLast(const std::string& key, int32_t num,
+                           const std::string& value) {
+  // Ensure that the number is positive
+  assert(num >= 0);
+
+  // Extract the original list data
+  std::string data;
+  db_->Get(get_option_, key, &data);
+
+  // Temporary variable to hold the "current element" in the blocks below
+  Slice elem;
+
+  // Count the total number of occurrences of value
+  int totalOccs = 0;
+  for (RedisListIterator it(data); !it.Done(); it.Skip()) {
+    it.GetCurrent(&elem);
+    if (elem == value) {
+      ++totalOccs;
+    }
+  }
+
+  // Construct an iterator to the data. Reserve enough space for the result.
+  RedisListIterator it(data);
+  int bytesRemoved = std::min(num,totalOccs)*it.SizeOf(value);
+  it.Reserve(it.Size() - bytesRemoved);
+
+  // Traverse the list, appending all but the desired occurrences of value.
+  // Note: "Drop the last k occurrences" is equivalent to
+  //  "keep only the first n-k occurrences", where n is total occurrences.
+  int numKept = 0;          // Keep track of the number of times value is kept
+  while(!it.Done()) {
+    it.GetCurrent(&elem);
+
+    // If we are within the deletion range and equal to value, drop it.
+    // Otherwise, append/keep/push it.
+    if (elem == value) {
+      if (numKept < totalOccs - num) {
+        it.Push();
+        ++numKept;
+      } else {
+        it.Skip();
+      }
+    } else {
+      // Always append the others
+      it.Push();
+    }
+  }
+
+  // Put the result back to the database
+  db_->Put(put_option_, key, it.WriteResult());
+
+  // Return the number of elements removed
+  return totalOccs - numKept;
+}
+
+/// Private functions
+
+// Insert element value into (list: key), right before/after
+//  the first occurrence of pivot
+//   : throws RedisListException
+int RedisLists::Insert(const std::string& key, const std::string& pivot,
+                       const std::string& value, bool insert_after) {
+  // Get the original list data
+  std::string data;
+  db_->Get(get_option_, key, &data);
+
+  // Construct an iterator to the data and reserve enough space for result.
+  RedisListIterator it(data);
+  it.Reserve(it.Size() + it.SizeOf(value));
+
+  // Iterate through the list until we find the element we want
+  Slice elem;
+  bool found = false;
+  while(!it.Done() && !found) {
+    it.GetCurrent(&elem);
+
+    // When we find the element, insert the element and mark found
+    if (elem == pivot) {                // Found it!
+      found = true;
+      if (insert_after == true) {       // Skip one more, if inserting after it
+        it.Push();
+      }
+      it.InsertElement(value);
+    } else {
+      it.Push();
+    }
+
+  }
+
+  // Put the data (string) into the database
+  if (found) {
+    db_->Put(put_option_, key, it.WriteResult());
+  }
+
+  // Returns the new (possibly unchanged) length of the list
+  return it.Length();
+}
+
+
+}

utilities/redis/redis_lists.h

@@ -0,0 +1,104 @@
+/**
+ * A (persistent) Redis API built using the rocksdb backend.
+ * Implements Redis Lists as described on: http://redis.io/commands#list
+ *
+ * @throws All functions may throw a RedisListException
+ *
+ * @author Deon Nicholas (dnicholas@fb.com)
+ * Copyright 2013 Facebook
+ */
+
+#include <string>
+#include "leveldb/db.h"
+#include "redis_list_iterator.h"
+#include "redis_list_exception.h"
+
+namespace leveldb {
+
+/// The Redis functionality (see http://redis.io/commands#list)
+/// All functions may THROW a RedisListException
+class RedisLists {
+ public: // Constructors / Destructors
+  /// Construct a new RedisLists database, with name/path of db.
+  /// Will clear the database on open iff destructive is true (default false).
+  /// Otherwise, it will restore saved changes.
+  /// May throw RedisListException
+  RedisLists(const std::string& db_path,
+             Options options, bool destructive = false);
+
+ public:  // Accessors
+  /// The number of items in (list: key)
+  int Length(const std::string& key);
+
+  /// Search the list for the (index)'th item (0-based) in (list:key)
+  /// A negative index indicates: "from end-of-list"
+  /// If index is within range: return true, and return the value in *result.
+  /// If (index < -length OR index>=length), then index is out of range:
+  ///   return false (and *result is left unchanged)
+  /// May throw RedisListException
+  bool Index(const std::string& key, int32_t index,
+             std::string* result);
+
+  /// Return (list: key)[first..last] (inclusive)
+  /// May throw RedisListException
+  std::vector<std::string> Range(const std::string& key,
+                                 int32_t first, int32_t last);
+
+  /// Prints the entire (list: key), for debugging.
+  void Print(const std::string& key);
+
+ public: // Insert/Update
+  /// Insert value before/after pivot in (list: key). Return the length.
+  /// May throw RedisListException
+  int InsertBefore(const std::string& key, const std::string& pivot,
+                   const std::string& value);
+  int InsertAfter(const std::string& key, const std::string& pivot,
+                  const std::string& value);
+
+  /// Push / Insert value at beginning/end of the list. Return the length.
+  /// May throw RedisListException
+  int PushLeft(const std::string& key, const std::string& value);
+  int PushRight(const std::string& key, const std::string& value);
+
+  /// Set (list: key)[idx] = val. Return true on success, false on fail
+  /// May throw RedisListException
+  bool Set(const std::string& key, int32_t index, const std::string& value);
+
+ public: // Delete / Remove / Pop / Trim
+  /// Trim (list: key) so that it will only contain the indices from start..stop
+  /// Returns true on success
+  /// May throw RedisListException
+  bool Trim(const std::string& key, int32_t start, int32_t stop);
+
+  /// If list is empty, return false and leave *result unchanged.
+  /// Else, remove the first/last elem, store it in *result, and return true
+  bool PopLeft(const std::string& key, std::string* result);  // First
+  bool PopRight(const std::string& key, std::string* result); // Last
+
+  /// Remove the first (or last) num occurrences of value from the list (key)
+  /// Return the number of elements removed.
+  /// May throw RedisListException
+  int Remove(const std::string& key, int32_t num,
+             const std::string& value);
+  int RemoveFirst(const std::string& key, int32_t num,
+                  const std::string& value);
+  int RemoveLast(const std::string& key, int32_t num,
+                 const std::string& value);
+
+ private: // Private Functions
+  /// Calls InsertBefore or InsertAfter
+  int Insert(const std::string& key, const std::string& pivot,
+             const std::string& value, bool insert_after);
+ private:
+  std::string db_name_;       // The actual database name/path
+  WriteOptions put_option_;
+  ReadOptions get_option_;
+
+  /// The backend rocksdb database.
+  /// Map : key --> list
+  ///       where a list is a sequence of elements
+  ///       and an element is a 4-byte integer (n), followed by n bytes of data
+  std::unique_ptr<DB> db_;
+};
+
+} // namespace leveldb

utilities/redis/redis_lists_test.cc

@@ -0,0 +1,900 @@
+/**
+ * A test harness for the Redis API built on rocksdb.
+ *
+ * USAGE: Build with: "make redis_test" (in rocksdb directory).
+ *        Run unit tests with: "./redis_test"
+ *        Manual/Interactive user testing: "./redis_test -m"
+ *        Manual user testing + restart database: "./redis_test -m -d"
+ *
+ * TODO:  Add LARGE random test cases to verify efficiency and scalability
+ *
+ * @author Deon Nicholas (dnicholas@fb.com)
+ * Copyright 2013 Facebook
+ */
+
+
+#include <iostream>
+#include <cctype>
+
+#include "redis_lists.h"
+#include "util/testharness.h"
+#include "util/random.h"
+
+using namespace leveldb;
+using namespace std;
+
+namespace leveldb {
+
+class RedisListsTest {
+ public:
+  static const string kDefaultDbName;
+  static Options options;
+
+  RedisListsTest() {
+    options.create_if_missing = true;
+  }
+};
+
+const string RedisListsTest::kDefaultDbName = "/tmp/redisdefaultdb/";
+Options RedisListsTest::options = Options();
+
+// operator== and operator<< are defined below for vectors (lists)
+// Needed for ASSERT_EQ
+
+// Compare two lists for equality.
+bool operator==(const std::vector<std::string>& a,
+                const std::vector<std::string>& b) {
+  if (a.size() != b.size()) {
+    return false;
+  }
+
+  int n = a.size();
+  for (int i=0; i<n; ++i) {
+    if (a[i] != b[i]) {
+      return false;
+    }
+  }
+
+  return true;
+}
+
+// Print out a list
+ostream& operator<<(ostream& out, const std::vector<std::string>& vec) {
+  out << "[";
+  int n = vec.size();
+  for(int i=0; i<n; ++i) {
+    if (i > 0) {
+      out << ", ";
+    }
+    out << vec[i];
+  }
+  out << "]";
+  return out;
+}
+
+/// THE TEST CASES BEGIN HERE
+
+// PushRight, Length, Index, Range
+TEST(RedisListsTest, SimpleTest) {
+  RedisLists redis(kDefaultDbName, options, true);   // Destructive
+
+  string tempv; // Used below for all Index(), PopRight(), PopLeft()
+
+  // Simple PushRight (should return the new length each time)
+  ASSERT_EQ(redis.PushRight("k1", "v1"), 1);
+  ASSERT_EQ(redis.PushRight("k1", "v2"), 2);
+  ASSERT_EQ(redis.PushRight("k1", "v3"), 3);
+
+  // Check Length and Index() functions
+  ASSERT_EQ(redis.Length("k1"), 3);        // Check length
+  ASSERT_TRUE(redis.Index("k1", 0, &tempv));
+  ASSERT_EQ(tempv, "v1");   // Check valid indices
+  ASSERT_TRUE(redis.Index("k1", 1, &tempv));
+  ASSERT_EQ(tempv, "v2");
+  ASSERT_TRUE(redis.Index("k1", 2, &tempv));
+  ASSERT_EQ(tempv, "v3");
+
+  // Check range function and vectors
+  std::vector<std::string> result = redis.Range("k1", 0, 2);   // Get the list
+  std::vector<std::string> expected_result(3);
+  expected_result[0] = "v1";
+  expected_result[1] = "v2";
+  expected_result[2] = "v3";
+  ASSERT_EQ(result, expected_result);  // Uses my overloaded operator==() above
+}
+
+// PushLeft, Length, Index, Range
+TEST(RedisListsTest, SimpleTest2) {
+  RedisLists redis(kDefaultDbName, options, true);   // Destructive
+
+  string tempv; // Used below for all Index(), PopRight(), PopLeft()
+
+  // Simple PushRight
+  ASSERT_EQ(redis.PushLeft("k1", "v3"), 1);
+  ASSERT_EQ(redis.PushLeft("k1", "v2"), 2);
+  ASSERT_EQ(redis.PushLeft("k1", "v1"), 3);
+
+  // Check Length and Index() functions
+  ASSERT_EQ(redis.Length("k1"), 3);        // Check length
+  ASSERT_TRUE(redis.Index("k1", 0, &tempv));
+  ASSERT_EQ(tempv, "v1");   // Check valid indices
+  ASSERT_TRUE(redis.Index("k1", 1, &tempv));
+  ASSERT_EQ(tempv, "v2");
+  ASSERT_TRUE(redis.Index("k1", 2, &tempv));
+  ASSERT_EQ(tempv, "v3");
+
+  // Check range function and vectors
+  std::vector<std::string> result = redis.Range("k1", 0, 2);   // Get the list
+  std::vector<std::string> expected_result(3);
+  expected_result[0] = "v1";
+  expected_result[1] = "v2";
+  expected_result[2] = "v3";
+  ASSERT_EQ(result, expected_result);  // Uses my overloaded operator==() above
+}
+
+// Exhaustive test of the Index() function
+TEST(RedisListsTest, IndexTest) {
+  RedisLists redis(kDefaultDbName, options, true);   // Destructive
+
+  string tempv; // Used below for all Index(), PopRight(), PopLeft()
+
+  // Empty Index check (return empty and should not crash or edit tempv)
+  tempv = "yo";
+  ASSERT_TRUE(!redis.Index("k1", 0, &tempv));
+  ASSERT_EQ(tempv, "yo");
+  ASSERT_TRUE(!redis.Index("fda", 3, &tempv));
+  ASSERT_EQ(tempv, "yo");
+  ASSERT_TRUE(!redis.Index("random", -12391, &tempv));
+  ASSERT_EQ(tempv, "yo");
+
+  // Simple Pushes (will yield: [v6, v4, v4, v1, v2, v3]
+  redis.PushRight("k1", "v1");
+  redis.PushRight("k1", "v2");
+  redis.PushRight("k1", "v3");
+  redis.PushLeft("k1", "v4");
+  redis.PushLeft("k1", "v4");
+  redis.PushLeft("k1", "v6");
+
+  // Simple, non-negative indices
+  ASSERT_TRUE(redis.Index("k1", 0, &tempv));
+  ASSERT_EQ(tempv, "v6");
+  ASSERT_TRUE(redis.Index("k1", 1, &tempv));
+  ASSERT_EQ(tempv, "v4");
+  ASSERT_TRUE(redis.Index("k1", 2, &tempv));
+  ASSERT_EQ(tempv, "v4");
+  ASSERT_TRUE(redis.Index("k1", 3, &tempv));
+  ASSERT_EQ(tempv, "v1");
+  ASSERT_TRUE(redis.Index("k1", 4, &tempv));
+  ASSERT_EQ(tempv, "v2");
+  ASSERT_TRUE(redis.Index("k1", 5, &tempv));
+  ASSERT_EQ(tempv, "v3");
+
+  // Negative indices
+  ASSERT_TRUE(redis.Index("k1", -6, &tempv));
+  ASSERT_EQ(tempv, "v6");
+  ASSERT_TRUE(redis.Index("k1", -5, &tempv));
+  ASSERT_EQ(tempv, "v4");
+  ASSERT_TRUE(redis.Index("k1", -4, &tempv));
+  ASSERT_EQ(tempv, "v4");
+  ASSERT_TRUE(redis.Index("k1", -3, &tempv));
+  ASSERT_EQ(tempv, "v1");
+  ASSERT_TRUE(redis.Index("k1", -2, &tempv));
+  ASSERT_EQ(tempv, "v2");
+  ASSERT_TRUE(redis.Index("k1", -1, &tempv));
+  ASSERT_EQ(tempv, "v3");
+
+  // Out of bounds (return empty, no crash)
+  ASSERT_TRUE(!redis.Index("k1", 6, &tempv));
+  ASSERT_TRUE(!redis.Index("k1", 123219, &tempv));
+  ASSERT_TRUE(!redis.Index("k1", -7, &tempv));
+  ASSERT_TRUE(!redis.Index("k1", -129, &tempv));
+}
+
+
+// Exhaustive test of the Range() function
+TEST(RedisListsTest, RangeTest) {
+  RedisLists redis(kDefaultDbName, options, true);   // Destructive
+
+  string tempv; // Used below for all Index(), PopRight(), PopLeft()
+
+  // Simple Pushes (will yield: [v6, v4, v4, v1, v2, v3])
+  redis.PushRight("k1", "v1");
+  redis.PushRight("k1", "v2");
+  redis.PushRight("k1", "v3");
+  redis.PushLeft("k1", "v4");
+  redis.PushLeft("k1", "v4");
+  redis.PushLeft("k1", "v6");
+
+  // Sanity check (check the length;  make sure it's 6)
+  ASSERT_EQ(redis.Length("k1"), 6);
+
+  // Simple range
+  std::vector<std::string> res = redis.Range("k1", 1, 4);
+  ASSERT_EQ((int)res.size(), 4);
+  ASSERT_EQ(res[0], "v4");
+  ASSERT_EQ(res[1], "v4");
+  ASSERT_EQ(res[2], "v1");
+  ASSERT_EQ(res[3], "v2");
+
+  // Negative indices (i.e.: measured from the end)
+  res = redis.Range("k1", 2, -1);
+  ASSERT_EQ((int)res.size(), 4);
+  ASSERT_EQ(res[0], "v4");
+  ASSERT_EQ(res[1], "v1");
+  ASSERT_EQ(res[2], "v2");
+  ASSERT_EQ(res[3], "v3");
+
+  res = redis.Range("k1", -6, -4);
+  ASSERT_EQ((int)res.size(), 3);
+  ASSERT_EQ(res[0], "v6");
+  ASSERT_EQ(res[1], "v4");
+  ASSERT_EQ(res[2], "v4");
+
+  res = redis.Range("k1", -1, 5);
+  ASSERT_EQ((int)res.size(), 1);
+  ASSERT_EQ(res[0], "v3");
+
+  // Partial / Broken indices
+  res = redis.Range("k1", -3, 1000000);
+  ASSERT_EQ((int)res.size(), 3);
+  ASSERT_EQ(res[0], "v1");
+  ASSERT_EQ(res[1], "v2");
+  ASSERT_EQ(res[2], "v3");
+
+  res = redis.Range("k1", -1000000, 1);
+  ASSERT_EQ((int)res.size(), 2);
+  ASSERT_EQ(res[0], "v6");
+  ASSERT_EQ(res[1], "v4");
+
+  // Invalid indices
+  res = redis.Range("k1", 7, 9);
+  ASSERT_EQ((int)res.size(), 0);
+
+  res = redis.Range("k1", -8, -7);
+  ASSERT_EQ((int)res.size(), 0);
+
+  res = redis.Range("k1", 3, 2);
+  ASSERT_EQ((int)res.size(), 0);
+
+  res = redis.Range("k1", 5, -2);
+  ASSERT_EQ((int)res.size(), 0);
+
+  // Range matches Index
+  res = redis.Range("k1", -6, -4);
+  ASSERT_TRUE(redis.Index("k1", -6, &tempv));
+  ASSERT_EQ(tempv, res[0]);
+  ASSERT_TRUE(redis.Index("k1", -5, &tempv));
+  ASSERT_EQ(tempv, res[1]);
+  ASSERT_TRUE(redis.Index("k1", -4, &tempv));
+  ASSERT_EQ(tempv, res[2]);
+
+  // Last check
+  res = redis.Range("k1", 0, -6);
+  ASSERT_EQ((int)res.size(), 1);
+  ASSERT_EQ(res[0], "v6");
+}
+
+// Exhaustive test for InsertBefore(), and InsertAfter()
+TEST(RedisListsTest, InsertTest) {
+  RedisLists redis(kDefaultDbName, options, true);
+
+  string tempv; // Used below for all Index(), PopRight(), PopLeft()
+
+  // Insert on empty list (return 0, and do not crash)
+  ASSERT_EQ(redis.InsertBefore("k1", "non-exist", "a"), 0);
+  ASSERT_EQ(redis.InsertAfter("k1", "other-non-exist", "c"), 0);
+  ASSERT_EQ(redis.Length("k1"), 0);
+
+  // Push some preliminary stuff [g, f, e, d, c, b, a]
+  redis.PushLeft("k1", "a");
+  redis.PushLeft("k1", "b");
+  redis.PushLeft("k1", "c");
+  redis.PushLeft("k1", "d");
+  redis.PushLeft("k1", "e");
+  redis.PushLeft("k1", "f");
+  redis.PushLeft("k1", "g");
+  ASSERT_EQ(redis.Length("k1"), 7);
+
+  // Test InsertBefore
+  int newLength = redis.InsertBefore("k1", "e", "hello");
+  ASSERT_EQ(newLength, 8);
+  ASSERT_EQ(redis.Length("k1"), newLength);
+  ASSERT_TRUE(redis.Index("k1", 1, &tempv));
+  ASSERT_EQ(tempv, "f");
+  ASSERT_TRUE(redis.Index("k1", 3, &tempv));
+  ASSERT_EQ(tempv, "e");
+  ASSERT_TRUE(redis.Index("k1", 2, &tempv));
+  ASSERT_EQ(tempv, "hello");
+
+  // Test InsertAfter
+  newLength =  redis.InsertAfter("k1", "c", "bye");
+  ASSERT_EQ(newLength, 9);
+  ASSERT_EQ(redis.Length("k1"), newLength);
+  ASSERT_TRUE(redis.Index("k1", 6, &tempv));
+  ASSERT_EQ(tempv, "bye");
+
+  // Test bad value on InsertBefore
+  newLength = redis.InsertBefore("k1", "yo", "x");
+  ASSERT_EQ(newLength, 9);
+  ASSERT_EQ(redis.Length("k1"), newLength);
+
+  // Test bad value on InsertAfter
+  newLength = redis.InsertAfter("k1", "xxxx", "y");
+  ASSERT_EQ(newLength, 9);
+  ASSERT_EQ(redis.Length("k1"), newLength);
+
+  // Test InsertBefore beginning
+  newLength = redis.InsertBefore("k1", "g", "begggggggggggggggg");
+  ASSERT_EQ(newLength, 10);
+  ASSERT_EQ(redis.Length("k1"), newLength);
+
+  // Test InsertAfter end
+  newLength = redis.InsertAfter("k1", "a", "enddd");
+  ASSERT_EQ(newLength, 11);
+  ASSERT_EQ(redis.Length("k1"), newLength);
+
+  // Make sure nothing weird happened.
+  ASSERT_TRUE(redis.Index("k1", 0, &tempv));
+  ASSERT_EQ(tempv, "begggggggggggggggg");
+  ASSERT_TRUE(redis.Index("k1", 1, &tempv));
+  ASSERT_EQ(tempv, "g");
+  ASSERT_TRUE(redis.Index("k1", 2, &tempv));
+  ASSERT_EQ(tempv, "f");
+  ASSERT_TRUE(redis.Index("k1", 3, &tempv));
+  ASSERT_EQ(tempv, "hello");
+  ASSERT_TRUE(redis.Index("k1", 4, &tempv));
+  ASSERT_EQ(tempv, "e");
+  ASSERT_TRUE(redis.Index("k1", 5, &tempv));
+  ASSERT_EQ(tempv, "d");
+  ASSERT_TRUE(redis.Index("k1", 6, &tempv));
+  ASSERT_EQ(tempv, "c");
+  ASSERT_TRUE(redis.Index("k1", 7, &tempv));
+  ASSERT_EQ(tempv, "bye");
+  ASSERT_TRUE(redis.Index("k1", 8, &tempv));
+  ASSERT_EQ(tempv, "b");
+  ASSERT_TRUE(redis.Index("k1", 9, &tempv));
+  ASSERT_EQ(tempv, "a");
+  ASSERT_TRUE(redis.Index("k1", 10, &tempv));
+  ASSERT_EQ(tempv, "enddd");
+}
+
+// Exhaustive test of Set function
+TEST(RedisListsTest, SetTest) {
+  RedisLists redis(kDefaultDbName, options, true);
+
+  string tempv; // Used below for all Index(), PopRight(), PopLeft()
+
+  // Set on empty list (return false, and do not crash)
+  ASSERT_EQ(redis.Set("k1", 7, "a"), false);
+  ASSERT_EQ(redis.Set("k1", 0, "a"), false);
+  ASSERT_EQ(redis.Set("k1", -49, "cx"), false);
+  ASSERT_EQ(redis.Length("k1"), 0);
+
+  // Push some preliminary stuff [g, f, e, d, c, b, a]
+  redis.PushLeft("k1", "a");
+  redis.PushLeft("k1", "b");
+  redis.PushLeft("k1", "c");
+  redis.PushLeft("k1", "d");
+  redis.PushLeft("k1", "e");
+  redis.PushLeft("k1", "f");
+  redis.PushLeft("k1", "g");
+  ASSERT_EQ(redis.Length("k1"), 7);
+
+  // Test Regular Set
+  ASSERT_TRUE(redis.Set("k1", 0, "0"));
+  ASSERT_TRUE(redis.Set("k1", 3, "3"));
+  ASSERT_TRUE(redis.Set("k1", 6, "6"));
+  ASSERT_TRUE(redis.Set("k1", 2, "2"));
+  ASSERT_TRUE(redis.Set("k1", 5, "5"));
+  ASSERT_TRUE(redis.Set("k1", 1, "1"));
+  ASSERT_TRUE(redis.Set("k1", 4, "4"));
+
+  ASSERT_EQ(redis.Length("k1"), 7); // Size should not change
+  ASSERT_TRUE(redis.Index("k1", 0, &tempv));
+  ASSERT_EQ(tempv, "0");
+  ASSERT_TRUE(redis.Index("k1", 1, &tempv));
+  ASSERT_EQ(tempv, "1");
+  ASSERT_TRUE(redis.Index("k1", 2, &tempv));
+  ASSERT_EQ(tempv, "2");
+  ASSERT_TRUE(redis.Index("k1", 3, &tempv));
+  ASSERT_EQ(tempv, "3");
+  ASSERT_TRUE(redis.Index("k1", 4, &tempv));
+  ASSERT_EQ(tempv, "4");
+  ASSERT_TRUE(redis.Index("k1", 5, &tempv));
+  ASSERT_EQ(tempv, "5");
+  ASSERT_TRUE(redis.Index("k1", 6, &tempv));
+  ASSERT_EQ(tempv, "6");
+
+  // Set with negative indices
+  ASSERT_TRUE(redis.Set("k1", -7, "a"));
+  ASSERT_TRUE(redis.Set("k1", -4, "d"));
+  ASSERT_TRUE(redis.Set("k1", -1, "g"));
+  ASSERT_TRUE(redis.Set("k1", -5, "c"));
+  ASSERT_TRUE(redis.Set("k1", -2, "f"));
+  ASSERT_TRUE(redis.Set("k1", -6, "b"));
+  ASSERT_TRUE(redis.Set("k1", -3, "e"));
+
+  ASSERT_EQ(redis.Length("k1"), 7); // Size should not change
+  ASSERT_TRUE(redis.Index("k1", 0, &tempv));
+  ASSERT_EQ(tempv, "a");
+  ASSERT_TRUE(redis.Index("k1", 1, &tempv));
+  ASSERT_EQ(tempv, "b");
+  ASSERT_TRUE(redis.Index("k1", 2, &tempv));
+  ASSERT_EQ(tempv, "c");
+  ASSERT_TRUE(redis.Index("k1", 3, &tempv));
+  ASSERT_EQ(tempv, "d");
+  ASSERT_TRUE(redis.Index("k1", 4, &tempv));
+  ASSERT_EQ(tempv, "e");
+  ASSERT_TRUE(redis.Index("k1", 5, &tempv));
+  ASSERT_EQ(tempv, "f");
+  ASSERT_TRUE(redis.Index("k1", 6, &tempv));
+  ASSERT_EQ(tempv, "g");
+
+  // Bad indices (just out-of-bounds / off-by-one check)
+  ASSERT_EQ(redis.Set("k1", -8, "off-by-one in negative index"), false);
+  ASSERT_EQ(redis.Set("k1", 7, "off-by-one-error in positive index"), false);
+  ASSERT_EQ(redis.Set("k1", 43892, "big random index should fail"), false);
+  ASSERT_EQ(redis.Set("k1", -21391, "large negative index should fail"), false);
+
+  // One last check (to make sure nothing weird happened)
+  ASSERT_EQ(redis.Length("k1"), 7); // Size should not change
+  ASSERT_TRUE(redis.Index("k1", 0, &tempv));
+  ASSERT_EQ(tempv, "a");
+  ASSERT_TRUE(redis.Index("k1", 1, &tempv));
+  ASSERT_EQ(tempv, "b");
+  ASSERT_TRUE(redis.Index("k1", 2, &tempv));
+  ASSERT_EQ(tempv, "c");
+  ASSERT_TRUE(redis.Index("k1", 3, &tempv));
+  ASSERT_EQ(tempv, "d");
+  ASSERT_TRUE(redis.Index("k1", 4, &tempv));
+  ASSERT_EQ(tempv, "e");
+  ASSERT_TRUE(redis.Index("k1", 5, &tempv));
+  ASSERT_EQ(tempv, "f");
+  ASSERT_TRUE(redis.Index("k1", 6, &tempv));
+  ASSERT_EQ(tempv, "g");
+}
+
+// Testing Insert, Push, and Set, in a mixed environment
+TEST(RedisListsTest, InsertPushSetTest) {
+  RedisLists redis(kDefaultDbName, options, true);   // Destructive
+
+  string tempv; // Used below for all Index(), PopRight(), PopLeft()
+
+  // A series of pushes and insertions
+  // Will result in [newbegin, z, a, aftera, x, newend]
+  // Also, check the return value sometimes (should return length)
+  int lengthCheck;
+  lengthCheck = redis.PushLeft("k1", "a");
+  ASSERT_EQ(lengthCheck, 1);
+  redis.PushLeft("k1", "z");
+  redis.PushRight("k1", "x");
+  lengthCheck = redis.InsertAfter("k1", "a", "aftera");
+  ASSERT_EQ(lengthCheck , 4);
+  redis.InsertBefore("k1", "z", "newbegin");  // InsertBefore beginning of list
+  redis.InsertAfter("k1", "x", "newend");     // InsertAfter end of list
+
+  // Check
+  std::vector<std::string> res = redis.Range("k1", 0, -1); // Get the list
+  ASSERT_EQ((int)res.size(), 6);
+  ASSERT_EQ(res[0], "newbegin");
+  ASSERT_EQ(res[5], "newend");
+  ASSERT_EQ(res[3], "aftera");
+
+  // Testing duplicate values/pivots (multiple occurrences of 'a')
+  ASSERT_TRUE(redis.Set("k1", 0, "a"));     // [a, z, a, aftera, x, newend]
+  redis.InsertAfter("k1", "a", "happy");    // [a, happy, z, a, aftera, ...]
+  ASSERT_TRUE(redis.Index("k1", 1, &tempv));
+  ASSERT_EQ(tempv, "happy");
+  redis.InsertBefore("k1", "a", "sad");     // [sad, a, happy, z, a, aftera, ...]
+  ASSERT_TRUE(redis.Index("k1", 0, &tempv));
+  ASSERT_EQ(tempv, "sad");
+  ASSERT_TRUE(redis.Index("k1", 2, &tempv));
+  ASSERT_EQ(tempv, "happy");
+  ASSERT_TRUE(redis.Index("k1", 5, &tempv));
+  ASSERT_EQ(tempv, "aftera");
+  redis.InsertAfter("k1", "a", "zz");         // [sad, a, zz, happy, z, a, aftera, ...]
+  ASSERT_TRUE(redis.Index("k1", 2, &tempv));
+  ASSERT_EQ(tempv, "zz");
+  ASSERT_TRUE(redis.Index("k1", 6, &tempv));
+  ASSERT_EQ(tempv, "aftera");
+  ASSERT_TRUE(redis.Set("k1", 1, "nota"));    // [sad, nota, zz, happy, z, a, ...]
+  redis.InsertBefore("k1", "a", "ba");        // [sad, nota, zz, happy, z, ba, a, ...]
+  ASSERT_TRUE(redis.Index("k1", 4, &tempv));
+  ASSERT_EQ(tempv, "z");
+  ASSERT_TRUE(redis.Index("k1", 5, &tempv));
+  ASSERT_EQ(tempv, "ba");
+  ASSERT_TRUE(redis.Index("k1", 6, &tempv));
+  ASSERT_EQ(tempv, "a");
+
+  // We currently have: [sad, nota, zz, happy, z, ba, a, aftera, x, newend]
+  // redis.Print("k1");   // manually check
+
+  // Test Inserting before/after non-existent values
+  lengthCheck = redis.Length("k1"); // Ensure that the length doesnt change
+  ASSERT_EQ(lengthCheck, 10);
+  ASSERT_EQ(redis.InsertBefore("k1", "non-exist", "randval"), lengthCheck);
+  ASSERT_EQ(redis.InsertAfter("k1", "nothing", "a"), lengthCheck);
+  ASSERT_EQ(redis.InsertAfter("randKey", "randVal", "ranValue"), 0); // Empty
+  ASSERT_EQ(redis.Length("k1"), lengthCheck); // The length should not change
+
+  // Simply Test the Set() function
+  redis.Set("k1", 5, "ba2");
+  redis.InsertBefore("k1", "ba2", "beforeba2");
+  ASSERT_TRUE(redis.Index("k1", 4, &tempv));
+  ASSERT_EQ(tempv, "z");
+  ASSERT_TRUE(redis.Index("k1", 5, &tempv));
+  ASSERT_EQ(tempv, "beforeba2");
+  ASSERT_TRUE(redis.Index("k1", 6, &tempv));
+  ASSERT_EQ(tempv, "ba2");
+  ASSERT_TRUE(redis.Index("k1", 7, &tempv));
+  ASSERT_EQ(tempv, "a");
+
+  // We have: [sad, nota, zz, happy, z, beforeba2, ba2, a, aftera, x, newend]
+
+  // Set() with negative indices
+  redis.Set("k1", -1, "endprank");
+  ASSERT_TRUE(!redis.Index("k1", 11, &tempv));
+  ASSERT_TRUE(redis.Index("k1", 10, &tempv));
+  ASSERT_EQ(tempv, "endprank"); // Ensure Set worked correctly
+  redis.Set("k1", -11, "t");
+  ASSERT_TRUE(redis.Index("k1", 0, &tempv));
+  ASSERT_EQ(tempv, "t");
+
+  // Test out of bounds Set
+  ASSERT_EQ(redis.Set("k1", -12, "ssd"), false);
+  ASSERT_EQ(redis.Set("k1", 11, "sasd"), false);
+  ASSERT_EQ(redis.Set("k1", 1200, "big"), false);
+}
+
+// Testing Trim, Pop
+TEST(RedisListsTest, TrimPopTest) {
+  RedisLists redis(kDefaultDbName, options, true);   // Destructive
+
+  string tempv; // Used below for all Index(), PopRight(), PopLeft()
+
+  // A series of pushes and insertions
+  // Will result in [newbegin, z, a, aftera, x, newend]
+  redis.PushLeft("k1", "a");
+  redis.PushLeft("k1", "z");
+  redis.PushRight("k1", "x");
+  redis.InsertBefore("k1", "z", "newbegin");    // InsertBefore start of list
+  redis.InsertAfter("k1", "x", "newend");       // InsertAfter end of list
+  redis.InsertAfter("k1", "a", "aftera");
+
+  // Simple PopLeft/Right test
+  ASSERT_TRUE(redis.PopLeft("k1", &tempv));
+  ASSERT_EQ(tempv, "newbegin");
+  ASSERT_EQ(redis.Length("k1"), 5);
+  ASSERT_TRUE(redis.Index("k1", 0, &tempv));
+  ASSERT_EQ(tempv, "z");
+  ASSERT_TRUE(redis.PopRight("k1", &tempv));
+  ASSERT_EQ(tempv, "newend");
+  ASSERT_EQ(redis.Length("k1"), 4);
+  ASSERT_TRUE(redis.Index("k1", -1, &tempv));
+  ASSERT_EQ(tempv, "x");
+
+  // Now have: [z, a, aftera, x]
+
+  // Test Trim
+  ASSERT_TRUE(redis.Trim("k1", 0, -1));       // [z, a, aftera, x] (do nothing)
+  ASSERT_EQ(redis.Length("k1"), 4);
+  ASSERT_TRUE(redis.Trim("k1", 0, 2));                     // [z, a, aftera]
+  ASSERT_EQ(redis.Length("k1"), 3);
+  ASSERT_TRUE(redis.Index("k1", -1, &tempv));
+  ASSERT_EQ(tempv, "aftera");
+  ASSERT_TRUE(redis.Trim("k1", 1, 1));                     // [a]
+  ASSERT_EQ(redis.Length("k1"), 1);
+  ASSERT_TRUE(redis.Index("k1", 0, &tempv));
+  ASSERT_EQ(tempv, "a");
+
+  // Test out of bounds (empty) trim
+  ASSERT_TRUE(redis.Trim("k1", 1, 0));
+  ASSERT_EQ(redis.Length("k1"), 0);
+
+  // Popping with empty list (return empty without error)
+  ASSERT_TRUE(!redis.PopLeft("k1", &tempv));
+  ASSERT_TRUE(!redis.PopRight("k1", &tempv));
+  ASSERT_TRUE(redis.Trim("k1", 0, 5));
+
+  // Exhaustive Trim test (negative and invalid indices)
+  // Will start in [newbegin, z, a, aftera, x, newend]
+  redis.PushLeft("k1", "a");
+  redis.PushLeft("k1", "z");
+  redis.PushRight("k1", "x");
+  redis.InsertBefore("k1", "z", "newbegin");    // InsertBefore start of list
+  redis.InsertAfter("k1", "x", "newend");       // InsertAfter end of list
+  redis.InsertAfter("k1", "a", "aftera");
+  ASSERT_TRUE(redis.Trim("k1", -6, -1));                     // Should do nothing
+  ASSERT_EQ(redis.Length("k1"), 6);
+  ASSERT_TRUE(redis.Trim("k1", 1, -2));
+  ASSERT_TRUE(redis.Index("k1", 0, &tempv));
+  ASSERT_EQ(tempv, "z");
+  ASSERT_TRUE(redis.Index("k1", 3, &tempv));
+  ASSERT_EQ(tempv, "x");
+  ASSERT_EQ(redis.Length("k1"), 4);
+  ASSERT_TRUE(redis.Trim("k1", -3, -2));
+  ASSERT_EQ(redis.Length("k1"), 2);
+}
+
+// Testing Remove, RemoveFirst, RemoveLast
+TEST(RedisListsTest, RemoveTest) {
+  RedisLists redis(kDefaultDbName, options, true);   // Destructive
+
+  string tempv; // Used below for all Index(), PopRight(), PopLeft()
+
+  // A series of pushes and insertions
+  // Will result in [newbegin, z, a, aftera, x, newend, a, a]
+  redis.PushLeft("k1", "a");
+  redis.PushLeft("k1", "z");
+  redis.PushRight("k1", "x");
+  redis.InsertBefore("k1", "z", "newbegin");    // InsertBefore start of list
+  redis.InsertAfter("k1", "x", "newend");       // InsertAfter end of list
+  redis.InsertAfter("k1", "a", "aftera");
+  redis.PushRight("k1", "a");
+  redis.PushRight("k1", "a");
+
+  // Verify
+  ASSERT_TRUE(redis.Index("k1", 0, &tempv));
+  ASSERT_EQ(tempv, "newbegin");
+  ASSERT_TRUE(redis.Index("k1", -1, &tempv));
+  ASSERT_EQ(tempv, "a");
+
+  // Check RemoveFirst (Remove the first two 'a')
+  // Results in [newbegin, z, aftera, x, newend, a]
+  int numRemoved = redis.Remove("k1", 2, "a");
+  ASSERT_EQ(numRemoved, 2);
+  ASSERT_TRUE(redis.Index("k1", 0, &tempv));
+  ASSERT_EQ(tempv, "newbegin");
+  ASSERT_TRUE(redis.Index("k1", 1, &tempv));
+  ASSERT_EQ(tempv, "z");
+  ASSERT_TRUE(redis.Index("k1", 4, &tempv));
+  ASSERT_EQ(tempv, "newend");
+  ASSERT_TRUE(redis.Index("k1", 5, &tempv));
+  ASSERT_EQ(tempv, "a");
+  ASSERT_EQ(redis.Length("k1"), 6);
+
+  // Repopulate some stuff
+  // Results in: [x, x, x, x, x, newbegin, z, x, aftera, x, newend, a, x]
+  redis.PushLeft("k1", "x");
+  redis.PushLeft("k1", "x");
+  redis.PushLeft("k1", "x");
+  redis.PushLeft("k1", "x");
+  redis.PushLeft("k1", "x");
+  redis.PushRight("k1", "x");
+  redis.InsertAfter("k1", "z", "x");
+
+  // Test removal from end
+  numRemoved = redis.Remove("k1", -2, "x");
+  ASSERT_EQ(numRemoved, 2);
+  ASSERT_TRUE(redis.Index("k1", 8, &tempv));
+  ASSERT_EQ(tempv, "aftera");
+  ASSERT_TRUE(redis.Index("k1", 9, &tempv));
+  ASSERT_EQ(tempv, "newend");
+  ASSERT_TRUE(redis.Index("k1", 10, &tempv));
+  ASSERT_EQ(tempv, "a");
+  ASSERT_TRUE(!redis.Index("k1", 11, &tempv));
+  numRemoved = redis.Remove("k1", -2, "x");
+  ASSERT_EQ(numRemoved, 2);
+  ASSERT_TRUE(redis.Index("k1", 4, &tempv));
+  ASSERT_EQ(tempv, "newbegin");
+  ASSERT_TRUE(redis.Index("k1", 6, &tempv));
+  ASSERT_EQ(tempv, "aftera");
+
+  // We now have: [x, x, x, x, newbegin, z, aftera, newend, a]
+  ASSERT_EQ(redis.Length("k1"), 9);
+  ASSERT_TRUE(redis.Index("k1", -1, &tempv));
+  ASSERT_EQ(tempv, "a");
+  ASSERT_TRUE(redis.Index("k1", 0, &tempv));
+  ASSERT_EQ(tempv, "x");
+
+  // Test over-shooting (removing more than there exists)
+  numRemoved = redis.Remove("k1", -9000, "x");
+  ASSERT_EQ(numRemoved , 4);    // Only really removed 4
+  ASSERT_EQ(redis.Length("k1"), 5);
+  ASSERT_TRUE(redis.Index("k1", 0, &tempv));
+  ASSERT_EQ(tempv, "newbegin");
+  numRemoved = redis.Remove("k1", 1, "x");
+  ASSERT_EQ(numRemoved, 0);
+
+  // Try removing ALL!
+  numRemoved = redis.Remove("k1", 0, "newbegin");   // REMOVE 0 will remove all!
+  ASSERT_EQ(numRemoved, 1);
+
+  // Removal from an empty-list
+  ASSERT_TRUE(redis.Trim("k1", 1, 0));
+  numRemoved = redis.Remove("k1", 1, "z");
+  ASSERT_EQ(numRemoved, 0);
+}
+
+
+// Test Multiple keys and Persistence
+TEST(RedisListsTest, PersistenceMultiKeyTest) {
+
+  string tempv; // Used below for all Index(), PopRight(), PopLeft()
+
+  // Block one: populate a single key in the database
+  {
+    RedisLists redis(kDefaultDbName, options, true);   // Destructive
+
+    // A series of pushes and insertions
+    // Will result in [newbegin, z, a, aftera, x, newend, a, a]
+    redis.PushLeft("k1", "a");
+    redis.PushLeft("k1", "z");
+    redis.PushRight("k1", "x");
+    redis.InsertBefore("k1", "z", "newbegin");    // InsertBefore start of list
+    redis.InsertAfter("k1", "x", "newend");       // InsertAfter end of list
+    redis.InsertAfter("k1", "a", "aftera");
+    redis.PushRight("k1", "a");
+    redis.PushRight("k1", "a");
+
+    ASSERT_TRUE(redis.Index("k1", 3, &tempv));
+    ASSERT_EQ(tempv, "aftera");
+  }
+
+  // Block two: make sure changes were saved and add some other key
+  {
+    RedisLists redis(kDefaultDbName, options, false); // Persistent, non-destructive
+
+    // Check
+    ASSERT_EQ(redis.Length("k1"), 8);
+    ASSERT_TRUE(redis.Index("k1", 3, &tempv));
+    ASSERT_EQ(tempv, "aftera");
+
+    redis.PushRight("k2", "randomkey");
+    redis.PushLeft("k2", "sas");
+
+    redis.PopLeft("k1", &tempv);
+  }
+
+  // Block three: Verify the changes from block 2
+  {
+    RedisLists redis(kDefaultDbName, options, false); // Presistent, non-destructive
+
+    // Check
+    ASSERT_EQ(redis.Length("k1"), 7);
+    ASSERT_EQ(redis.Length("k2"), 2);
+    ASSERT_TRUE(redis.Index("k1", 0, &tempv));
+    ASSERT_EQ(tempv, "z");
+    ASSERT_TRUE(redis.Index("k2", -2, &tempv));
+    ASSERT_EQ(tempv, "sas");
+  }
+}
+
+/// THE manual REDIS TEST begins here
+/// THIS WILL ONLY OCCUR IF YOU RUN: ./redis_test -m
+
+void MakeUpper(std::string* const s) {
+  int len = s->length();
+  for(int i=0; i<len; ++i) {
+    (*s)[i] = toupper((*s)[i]); // C-version defined in <ctype.h>
+  }
+}
+
+/// Allows the user to enter in REDIS commands into the command-line.
+/// This is useful for manual / interacticve testing / debugging.
+///  Use destructive=true to clean the database before use.
+///  Use destructive=false to remember the previous state (i.e.: persistent)
+/// Should be called from main function.
+int manual_redis_test(bool destructive){
+  RedisLists redis(RedisListsTest::kDefaultDbName,
+                   RedisListsTest::options,
+                   destructive);
+
+  // TODO: Right now, please use spaces to separate each word.
+  //  In actual redis, you can use quotes to specify compound values
+  //  Example: RPUSH mylist "this is a compound value"
+
+  std::string command;
+  while(true) {
+    cin >> command;
+    MakeUpper(&command);
+
+    if (command == "LINSERT") {
+      std::string k, t, p, v;
+      cin >> k >> t >> p >> v;
+      MakeUpper(&t);
+      if (t=="BEFORE") {
+        std::cout << redis.InsertBefore(k, p, v) << std::endl;
+      } else if (t=="AFTER") {
+        std::cout << redis.InsertAfter(k, p, v) << std::endl;
+      }
+    } else if (command == "LPUSH") {
+      std::string k, v;
+      std::cin >> k >> v;
+      redis.PushLeft(k, v);
+    } else if (command == "RPUSH") {
+      std::string k, v;
+      std::cin >> k >> v;
+      redis.PushRight(k, v);
+    } else if (command == "LPOP") {
+      std::string k;
+      std::cin >> k;
+      string res;
+      redis.PopLeft(k, &res);
+      std::cout << res << std::endl;
+    } else if (command == "RPOP") {
+      std::string k;
+      std::cin >> k;
+      string res;
+      redis.PopRight(k, &res);
+      std::cout << res << std::endl;
+    } else if (command == "LREM") {
+      std::string k;
+      int amt;
+      std::string v;
+
+      std::cin >> k >> amt >> v;
+      std::cout << redis.Remove(k, amt, v) << std::endl;
+    } else if (command == "LLEN") {
+      std::string k;
+      std::cin >> k;
+      std::cout << redis.Length(k) << std::endl;
+    } else if (command == "LRANGE") {
+      std::string k;
+      int i, j;
+      std::cin >> k >> i >> j;
+      std::vector<std::string> res = redis.Range(k, i, j);
+      for (auto it = res.begin(); it != res.end(); ++it) {
+        std::cout << " " << (*it);
+      }
+      std::cout << std::endl;
+    } else if (command == "LTRIM") {
+      std::string k;
+      int i, j;
+      std::cin >> k >> i >> j;
+      redis.Trim(k, i, j);
+    } else if (command == "LSET") {
+      std::string k;
+      int idx;
+      std::string v;
+      cin >> k >> idx >> v;
+      redis.Set(k, idx, v);
+    } else if (command == "LINDEX") {
+      std::string k;
+      int idx;
+      std::cin >> k >> idx;
+      string res;
+      redis.Index(k, idx, &res);
+      std::cout << res << std::endl;
+    } else if (command == "PRINT") {      // Added by Deon
+      std::string k;
+      cin >> k;
+      redis.Print(k);
+    } else if (command == "QUIT") {
+      return 0;
+    } else {
+      std::cout << "unknown command: " << command << std::endl;
+    }
+  }
+}
+
+} // namespace leveldb
+
+
+// USAGE: "./redis_test" for default (unit tests)
+//        "./redis_test -m" for manual testing (redis command api)
+//        "./redis_test -m -d" for destructive manual test (erase db before use)
+
+
+// Check for "want" argument in the argument list
+bool found_arg(int argc, char* argv[], const char* want){
+  for(int i=1; i<argc; ++i){
+    if (strcmp(argv[i], want) == 0) {
+      return true;
+    }
+  }
+  return false;
+}
+
+// Will run unit tests.
+// However, if -m is specified, it will do user manual/interactive testing
+// -m -d is manual and destructive (will clear the database before use)
+int main(int argc, char* argv[]) {
+  if (found_arg(argc, argv, "-m")) {
+    bool destructive = found_arg(argc, argv, "-d");
+    return leveldb::manual_redis_test(destructive);
+  } else {
+    return leveldb::test::RunAllTests();
+  }
+}
+