High Scalability - High Scalability - Comet - An Example of the New Key-Code Databases

High Scalability - High Scalability - Comet - An Example of the New Key-Code Databases

Comet is an active distributed key-value store built at the University of Washington. The paper describing Comet is Comet: An active distributed key-value store, there are also slides, and a MP3 of a presentation given at OSDI '10. Here's a succinct overview of Comet:

Today's cloud storage services, such as Amazon S3 or peer-to-peer DHTs, are highly inflexible and impose a variety of constraints on their clients: specific replication and consistency schemes, fixed data timeouts, limited logging, etc. We witnessed such inflexibility first-hand as part of our Vanish work, where we used a DHT to store encryption keys temporarily. To address this issue, we built Comet, an extensible storage service that allows clients to inject snippets of code that control their data's behavior inside the storage service.

I found this paper quite interesting because it takes the initial steps of collocating code with a key-value store, which turns it into what might called a key-code store. This is something I've been exploring as a way of moving behavior to data in order to overcome network limitations in the cloud and provide other benefits. An innovator in this area is the Alchemy Database, which has already combined Redis and Lua. A good platform for this sort of thing might be Node.js integrated with V8. This would allow complex Javascript programs to run in an efficient evented container. There are a lot of implications of this sort of architecture, more about that later, but the Comet paper describes a very interesting start.

From the abstract and conclusion:

This paper described Comet, an active distributed key value store. Comet enables clients to customize a distributed storage system in application-specific ways using Comet’s active storage objects. By supporting ASOs, Comet allows multiple applications with diverse requirements to share a common storage system. We implemented Comet on the Vuze DHT using a severely restricted Lua language sandbox for handler programming. Our measurements and experience demonstrate that a broad range of behaviors and customizations are possible in a safe, but active, storage environment.
Distributed key-value storage systems are widely used incorporations and across the Internet. Our research seeks to greatly expand the application space for key-value storage systems through application-specific customization. We designed and implemented Comet, an extensible, distributed key-value store. Each Comet node stores a collection of active storage objects (ASOs) that consist of a key, a value, and a set of handlers. Comet handlers run as a result of timers or storage operations, such as get or put, allowing an ASO to take dynamic, application-specific actions to customize its behavior. Handlers are written in a simple sandboxed extension language, providing properties of safety and isolation.
We implemented a Comet prototype for the Vuze DHT, deployed Comet nodes on Vuze from PlanetLab, and built and evaluated over a dozen Comet applications. Our experience demonstrates that simple, safe, and restricted extensibility can significantly increase the power and range of applications that can run on distributed active storage systems. This approach facilitates the sharing of a single storage system by applications with diverse needs, allowing them to reap the consolidation benefits inherent in today’s massive clouds.

Related Articles
Roxana Geambasu, Ph.D. Candidate
Scriptable Object Cache by Rohit Karlupia

Riak's Bitcask - A Log-Structured Hash Table for Fast Key/Value Data

Riak's Bitcask - A Log-Structured Hash Table for Fast Key/Value Data: "


How would you implement a key-value storage system if you were starting from scratch? The approach Basho settled on with Bitcask, their new backend for Riak, is an interesting combination of using RAM to store a hash map of file pointers to values and a log-structured file system for efficient writes. In this excellent Changelog interview, some folks from Basho describe Bitcask in more detail.

The essential Bitcask:

"

Design — Sheepdog Project

Design — Sheepdog Project

The architecture of Sheepdog is fully symmetric; there is no central node such as a meta-data server. This design enables following features.

• Linear scalability in performance and capacity
  When more performance or capacity is needed, Sheepdog can be grown linearly by simply adding new machines to the cluster.
• No single point of failure
  Even if a machine fails, the data is still accessible through other machines.
• Easy administration
  There is no config file about cluster’s role. When administrators launch Sheepdog programs at the newly added machine, Sheepdog automatically detects the added machine and begins to configure it as a member of the cluster.

Architecture

Sheepdog is a storage system that provides a simple key-value interface to Sheepdog client (qemu block driver). Sheepdog is consists of multiple nodes.

Sheepdog consists of only one server (we call collie) and patched QEMU/KVM.

Virtual Disk Image (VDI)

A Sheepdog client divides a VM image into fixed-size objects (4 MB by default) and store them on the distributed storage system. Each object is identified by globally unique 64 bit id, and replicated to multiple nodes.

Object

Sheepdog objects are grouped into two types.

• VDI Object: A VDI object contains metadata for a VM image such as image name, disk size, creation time, etc.
• Data Object: A VM images is divided into a data object. Sheepdog client generally access this object.

Sheepdog uses consistent hashing to decide where objects store. Consistent hashing is a scheme that provides hash table functionality, and the addition or removal of nodes does not significantly change the mapping of objects. I/O load is balanced across the nodes by features of hash table. A mechanism of distributing the data not randomly but intelligently is a future work.

Each node is placed on consistent hashing ring based on its own id. To determine where to store the object, Sheepdog client gets the object id, finds the corresponding point on the ring, and walk clockwise to determine the target nodes.

VDI Operation

In most cases, Sheepdog clients can access their images independently because we do not allow for clients to access the same image at the same time. But some VDI operations (e.g. cloning VDI, locking VDI) must be done exclusively because the operations updating global information. To implement this in the highly available system, we use a group communication system (GCS). Group communication systems provide specific guarantees such as total ordering of messages. We use corosync, one of most famous GCS.

Using Object Database db4o as Storage Provider in Voldemort

Using Object Database db4o as Storage Provider in Voldemort: "

Abstract: In this article I will show you how easy it is to add support for Versant’s object database, db4o in project Voldemort, an Apache licensed distributed key-value storage system used at LinkedIn which is useful for certain high-scalability storage problems where simple functional partitioning is not sufficient (Note: Voldemort borrows heavily from Amazon’s Dynamo, if you’re interested in this technology all the available papers about Dynamo should be useful).

Voldemort’s storage layer is completely mockable so development and unit testing can be done against a throw-away in-memory storage system without needing a real cluster or, as we will show next, even a real storage engine like db4o for simple testing.

Note: All db4o related files that are part of this project are publicly available on ☞ GitHub

This is an article contributed by German Viscuso from db4objects (a division of Versant Corporation).

Content:

1. Voldemort in a Nutshell
2. 
   Key/Value Storage System
   
   
   • Pros and Cons
   
   
3. Logical Architecture
4. db4o API in a Nutshell
5. db4o as Key/Value Storage Provider
6. db4o and BerkeleyDB side by side
7. Conclusion

Voldemort in a Nutshell

In one sentence Voldemort is basically a big, distributed, persistent, fault-tolerant hash table designed to tradeoff consistency for availability.

It’s implemented as a highly available key-value storage system for services that need to provide an “always-on” experience. To achieve this level of availability, Voldemort sacrifices consistency under certain failure scenarios. It makes extensive use of object versioning and application-assisted conflict resolution.

Voldemort targets applications that operate with weak/eventual consistency (the “C” in ACID) if this results in high availability even under occasional network failures. It does not provide any isolation guarantees and permits only single key updates. It is well known that when dealing with the possibility of network failures, strong consistency and high data availability cannot be achieved simultaneously.

The system uses a series of known techniques to achieve scalability and availability: data is partitioned and replicated using consistent hashing, and consistency is facilitated by object versioning. The consistency among replicas during updates is maintained by a quorum-like technique and a decentralized replica synchronization protocol. It employs a gossip based distributed failure detection and membership protocol and is a completely decentralized system with minimal need for manual administration. Storage nodes can be added and removed from without requiring any manual partitioning or redistribution.

Voldemort is not a relational database, it does not attempt to satisfy arbitrary relations while satisfying ACID properties. Nor is it an object database that attempts to transparently map object reference graphs. Nor does it introduce a new abstraction such as document-orientation.

However, given the current complex scenario of persistence solutions in the industry we’ll see that different breeds can be combined to offer solutions that become the right tool for the job at hand. In this case we’ll see how a highly scalable NoSQL solution such as Voldemort can use a fast Java based embedded object database (Versant’s db4o) as the underlying persistence mechanism. Voldemort provides the scalability while db4o provides fast key/value pair persistence.

Key/Value Storage System

To enable high performance and availability Voldemort allows only very simple key-value data access. Both keys and values can be complex compound objects including lists or maps, but none-the-less the only supported queries are effectively the following:

value = store.get( key )
store.put( key, value )
store.delete( key )

This is clearly not good enough for all storage problems, there are a variety of trade-offs:

Cons:
no complex query filters
all joins must be done in code
no foreign key constraints
no triggers/callbacks

Pros:

• only efficient queries are possible, very predictable performance
• easy to distribute across a cluster
• service-orientation often disallows foreign key constraints and forces joins to be done in code anyway (because key refers to data maintained by another service)
• using a relational db you need a caching layer to scale reads, the caching layer typically forces you into key-value storage anyway
• often end up with xml or other denormalized blobs for performance anyway
• clean separation of storage and logic (SQL encourages mixing business logic with storage operations for efficiency)
• no object-relational miss-match, no mapping

Having a three operation interface makes it possible to transparently mock out the entire storage layer and unit test using a mock-storage implementation that is little more than a HashMap. This makes unit testing outside of a particular container or environment much more practical and was certainly instrumental in simplifying the db4o storage engine implementation.

Logical Architecture

In Voldemort, each storage node has three main software component groups: request coordination, membership and failure detection, and a local persistence engine. All these components are implemented in Java.

If we take a closer look we’ll see that each layer in the code implements a simple storage interface that does put, get, and delete. Each of these layers is responsible for performing one function such as TCP/IP network communication, serialization, version reconciliation, inter-node routing, etc. For example the routing layer is responsible for taking an operation, say a PUT, and delegating it to all the N storage replicas in parallel, while handling any failures.

Voldemort’s local persistence component allows for different storage engines to be plugged in. Engines that are in use are Oracle’s Berkeley Database (BDB), Oracle’s MySQL, and an in-memory buffer with persistent backing store. The main reason for designing a pluggable persistence component is to choose the storage engine best suited for an application’s access patterns. For instance, BDB can handle objects typically in the order of tens of kilobytes whereas MySQL can handle objects of larger sizes. Applications choose Voldemort’s local persistence engine based on their object size distribution. The majority of Voldemort’s production instances currently use BDB.

In this article we’ll provide details about the creation of a new storage engine (yellow box layer on the image above) that uses db4o instead of Oracle’s Berkeley DB (BDB), Oracle’s MySQL or just memory. Moreover we’ll show that db4o’s performance is on-par with BDB (if not better) while introducing a significant reduction in the implementation complexity.

db4o API in a Nutshell

db4o’s basic API is not far away from Voldemort’s API in terms of simplicity but it’s not limited to the storage of key/value pairs. db4o is a general purpose object persistence engine that can deal with the peculiarities of native POJO persistence that exhibit arbitrarily complex object references.

The very basic API looks like this:

container.store( object )
container.delete( object )
objectSet = container.queryByExample( prototype )

Two more advanced querying mechanisms are available and we will introduce one of them (SODA) on the following sections. As you can see, in order to provide an interface for key/value storage (what Voldemort expects) we’ll have to provide an intermediate layer which will take requests via the basic Voldemort API and will translate that into db4o API calls.

db4o as Key/Value Storage Provider

db4o fits seamlessly into the Voldemort picture as a storage provider because:

• db4o allows free form storage of objects with no special requirements on object serialization which results in a highly versatile storage solution which can easily be configured to store from low level objects (eg byte arrays) to objects of arbitrarily complexity (both for keys and values on key/value pairs)
• db4o’s b-tree based indexing capabilities allow for quick retrieval of keys while acting as a key/value pair storage provider
• db4o’s simple query system allows the retrieval of key/value pairs with one-line of code
• db4o can persist versioned objects directly which frees the developer from having to use versioning wrappers on stored values

db4o can act as a simpler but powerful replacement to Oracle’s Berkley DB (BDB) because the db4o complexity of the db4o storage provider is closer to Voldemort’s “in memory” storage provider than the included BDB provider which results in faster test and development cycles (remember Voldemort is work in progress).

As a first step in the implementation we provide a generic class to expose db4o as a key/value storage provider which is not dependant on Voldemort’s API (also useful to db4o developers who need this functionality in any application). The class is voldemort.store.db4o.Db4oKeyValueProvider and relies on Java generics so it can be used with any sort of key/value objects.

This class implements the following methods which are more or less self explanatory considering they operate on a store of key/value pairs:

voldemort.store.db4o.Db4oKeyValueProvider.keyIterator()
voldemort.store.db4o.Db4oKeyValueProvider.getKeys()
voldemort.store.db4o.Db4oKeyValueProvider.getValues(Key)
voldemort.store.db4o.Db4oKeyValueProvider.pairIterator()
voldemort.store.db4o.Db4oKeyValueProvider.get(Key)
voldemort.store.db4o.Db4oKeyValueProvider.delete(Key)
voldemort.store.db4o.Db4oKeyValueProvider.delete(Key, Value)
voldemort.store.db4o.Db4oKeyValueProvider.delete(Db4oKeyValuePair)
voldemort.store.db4o.Db4oKeyValueProvider.put(Key, Value)
voldemort.store.db4o.Db4oKeyValueProvider.put(Db4oKeyValuePair)
voldemort.store.db4o.Db4oKeyValueProvider.getAll()
voldemort.store.db4o.Db4oKeyValueProvider.getAll(Iterable)
voldemort.store.db4o.Db4oKeyValueProvider.truncate()
voldemort.store.db4o.Db4oKeyValueProvider.commit()
voldemort.store.db4o.Db4oKeyValueProvider.rollback()
voldemort.store.db4o.Db4oKeyValueProvider.close()
voldemort.store.db4o.Db4oKeyValueProvider.isClosed()

Second, we implement a db4o StorageProvider following Voldemort’s API, namely voldemort.store.db4o.Db4oStorageEngine which define the key (K) as ByteArray and the value (V) as byte[] for key/value classes.

Since this class inherits from abstract class voldemort.store.StorageEngine it must (and it does) implement the following methods:

voldemort.store.db4o.Db4oStorageEngine.getVersions(K)
voldemort.store.db4o.Db4oStorageEngine.get(K)
voldemort.store.db4o.Db4oStorageEngine.getAll(Iterable)
voldemort.store.db4o.Db4oStorageEngine.put(K, Versioned)
voldemort.store.db4o.Db4oStorageEngine.delete(K, Version)

We also implement the class that handles the db4o storage configuration: voldemort.store.db4o.Db4oStorageConfiguration
This class is responsible for providing all the initialization parameters when the db4o database is created (eg. index creation on the key (K) field for fast retrieval of key/value pairs)

Third and last we provide three support classes:

• 
  voldemort.store.db4o.Db4oKeyValuePair : a generic key/value pair class. Instances of this class will be what ultimately gets stored in the db4o database.
• 
  voldemort.store.db4o.Db4oKeysIterator : an Iterator implementaion that allows to iterate over the keys (K).
• 
  voldemort.store.db4o.Db4oEntriesIterator: an Iterator implementaion that allows to iterate over the values (V).

and a few test classes: voldemort.CatDb4oStore, voldemort.store.db4o.Db4oStorageEngineTest (this is the main test class), voldemort.store.db4o.Db4oSplitStorageEngineTest.

db4o and BerkeleyDB side by side

Let’s take a look at db4o’s simplicity by comparing the matching methods side by side with BDB (now both can act as Voldemort’s low level storage provider).

Constructors: BdbStorageEngine() vs Db4oStorageEngine()

The db4o constructor get rids of the serializer object because it can store objects of any complexity with no transformation. The code:

this.versionSerializer = new Serializer() {
public byte[] toBytes(Version object) {
return ((VectorClock) object).toBytes();
}
public Version toObject(byte[] bytes) {
return versionedSerializer.getVersion(bytes);
}
};

is no longer necessary and, of course, this impacts all storage operations because db4o requires no conversions “to bytes” and “to object” back and forth:

Fetch by Key

BDB (fetch by key):

DatabaseEntry keyEntry = new DatabaseEntry(key.get());
DatabaseEntry valueEntry = new DatabaseEntry();
List results = Lists.newArrayList();

for(OpStatus status = cursor.getSearchKey(keyEntry, valueEntry, lockMode);
status == OperationStatus.SUCCESS;
status = cursor.getNextDup(keyEntry, valueEntry, lockMode)) {

results.add(serializer.toObject(valueEntry.getData()));
};
return results;

db4o (fetch by key)

Query query = getContainer().query();
query.constrain(Db4oKeyValuePair.class);
query.descend('key').constrain(key);
return query.execute();

in this example we use “SODA”, a low level and powerful graph based query system where you basically build a query tree and pass it to db4o for execution.

Store Key/Value Pairs

BDB (store key/value pair):

DatabaseEntry keyEntry = new DatabaseEntry(key.get());
boolean succeeded = false;
transaction = this.environment.beginTransaction(null, null);
// Check existing values
// if there is a version obsoleted by this value delete it
// if there is a version later than this one, throw an exception
DatabaseEntry valueEntry  = new DatabaseEntry();
cursor = getBdbDatabase().openCursor(transaction, null);

for(OpStatus status=cursor.getSearchKey(keyEntry,valueEntry,LockMode.RMW);
status == OperationStatus.SUCCESS;
status = cursor.getNextDup(keyEntry, valueEntry, LockMode.RMW)) {

VectorClock clock = new VectorClock(valueEntry.getData());
Occured occured = value.getVersion().compare(clock);

if(occured == Occured.BEFORE) throw new ObsoleteVersionException();
else if(occured == Occured.AFTER) // best effort delete of obsolete previous value!
cursor.delete();
}

// Okay so we cleaned up all the prior stuff, so now we are good to
// insert the new thing
valueEntry = new DatabaseEntry(versionedSerializer.toBytes(value));
OperationStatus status = cursor.put(keyEntry, valueEntry);
if(status != OperationStatus.SUCCESS)
throw new PersistenceFailException('Put operation failed:' + status);
succeeded = true;

db4o (store key/value pair):

boolean succeeded = false;
candidates = keyValueProvider.get(key);

for(Db4oKeyValuePair> pair: candidates) {
Occured occured = value.getVersion().compare(pair.getValue().getVersion());
if(occured == Occured.BEFORE) throw new ObsoleteVersionException();
else if(occured == Occured.AFTER) // best effort delete of obsolete previous value!
keyValueProvider.delete(pair);
}
// Okay so we cleaned up all the prior stuff, so we can now insert
try {
keyValueProvider.put(key, value);
} catch(Db4oException de) {
throw new PersistenceFailException("Put op failed:" + de.getMessage()); succeeded = true;
}

Status of Unit Tests

Let’s take a look at the current status of unit tests in for the db4o storage provider (Db4oStorageEngineTest):

1. 
   testPersistence: single key/value pair storage and retrieval with storage engine close operation in between (it takes some time because this is the first test and the Voldemort system is kick started).
2. 
   testEquals: test that retrieval of storage engine instance by name gets you the same instance.
3. 
   testNullConstructorParameters: null constructors for storage engine instantiation are illegal. Arguments are storage engine name and database configuration.
4. 
   testSimultaneousIterationAndModification: threaded test of simultaneous puts (inserts), deletes and iteration over pairs (150 puts and 150 deletes before start of iteration).
5. 
   testGetNoEntries: test that empty storage empty returns zero pairs.
6. 
   testGetNoKeys: test that empty storage empty returns zero keys.
7. 
   testKeyIterationWithSerialization: test storage and key retrieval of 5 pairs serialized as
   Strings.
8. 
   testIterationWithSerialization: test storage and retrieval of 5 pairs serialized as Strings.
9. 
   testPruneOnWrite: stores 3 versions for one key and tests prune (overwriting of previous versions should happen).
10. 
    testTruncate: stores 3 pairs and issues a truncate database operation. Then verifies db is empty.
11. 
    testEmptyByteArray: stores 1 pair with zeroed key and tests correct retrieval.
12. 
    testNullKeys: test that basic operations (get, put, getAll, delete,etc) fail with a null key parameter (5 operations on pairs total).
13. 
    testPutNullValue: test that put operation works correctly with a null value (2 operations on pairs total, put and get).
14. 
    testGetAndDeleteNonExistentKey: test that key of non persistent pair doesn’t return a value.
15. 
    testFetchedEqualsPut: stores 1 pair with complex version and makes sure only one entry is stored and retrieved.
16. 
    testVersionedPut: tests that obsolete or equal versions of a value can’t be stored. Test correct storage of incremented version (13 operations on pairs total).
17. 
    testDelete: puts 2 pairs with conflicting versions and deletes one (7 operations on pairs total).
18. 
    testGetVersions: Test retrieval of different versions of pair (3 operations on pairs total).
19. 
    testGetAll: puts 10 pairs and tests if all can be correctly retrieved at once (getAll)
20. 
    testGetAllWithAbsentKeys: same as before but with non persistent keys (getAll returns 0 pairs).
21. 
    testCloseIsIdempotent: tests that a second close does not result in error.

Conclusion

db4o provides a low level implementation of a key/value storage engine that is both simpler than BDB and on-par (if not better) in performance. Moreover, the implementation shows that different persistence solutions such as the typical highly scalable, eventually consistent NoSQL engine and object databases can be combined to provide a solution that becomes the right tool for the job at hand which makes typical “versus” arguments obsolete.

Finally the short implementation path taken to make db4o act as a reliable key/value storage provider show the power and versatility of native embedded object databases.


Original title and link for this post: Using Object Database db4o as Storage Provider in Voldemort (published on the NoSQL blog: myNoSQL)

"