Nexedi Enterprise Objects
Presents:●●NEO components
Load balancing &redundancy●Locking scheme●Disaster recovery

Support NEO !
www.neoppod.org© Nexedi SA 2010 – All rights reserved – GNU Free Documentation License

Scalability & fault tolerance

Replication

Non-native
invalidations
Space 1

Tolerance 0Space 1

Space 1

Tolerance S-1Tolerance S-1© Nexedi SA 2010 – All rights reserved – GNU Free Documentation License

0 ≤ R ≤ S-1
Space S-R

Tolerance RComponents & exchanges

Parallelism

Partition table & node state© Nexedi SA 2010 – All rights reserved – GNU Free Documentation License

LockingZEO, relStorage,...
T1
T2

Database-levellockNEO
T1T2© Nexedi SA 2010 – All rights reserved – GNU Free Documentation License

Object-level locks

Locking effects
Parallel changes to a single object© Nexedi SA 2010 – All rights reserved – GNU Free Documentation License

Parallel changes to different object

© Nexedi SA 2010 – All rights reserved – GNU Free Documentation License

Moving to NEO© Nexedi SA 2010 – All rights reserved – GNU Free Documentation License

TODO- Implementation performance (better client-side caching,"zero-copy" protocol module, ...)
- Better admin tools (monitoring, control)- Backend-independent backups- Replication node
- Existing ZODB features : BLOBs, readCurrent API, ...
- Contributions to ZODB : transaction garbage collection API- NEO2 : Long-distance NEO

Support NEO !
www.neoppod.org© Nexedi SA 2010 – All rights reserved – GNU Free Documentation License

Extra slides© Nexedi SA 2010 – All rights reserved – GNU Free Documentation License© Nexedi SA 2010 – All rights reserved – GNU Free Documentation LicenseTwo-phase commit : Phase 2ClientStorage nodes© Nexedi SA 2010 – All rights reserved – GNU Free Documentation License

MasterNEO Overview

1

Nexedi Enterprise Objects
Presents:●●NEO components
Load balancing &redundancy●Locking scheme●Disaster recovery

Support NEO !
www.neoppod.org© Nexedi SA 2010 – All rights reserved – GNU Free Documentation License

NEO (Nexedi Enterprise Objects) is a ZODB.Storage implementation aimed at scalability:- high data volumes (petabyte scale)- high availability (fault tolerance)- high performances (clustering)
It is developped by Nexedi in collaboration with LIP6 and LIPN laboratories whose task is to formally
prove NEO design, Pilot Systems, MINES ParisTech and Dakar "Cheikh Anta Diop" University whosetask is to stress-test NEO and develop applications taking advantage of NEO design.

This presentation was shown at DZUG 2010 in Dresden by Vincent Pelletier from Nexedi.

NEO Overview

2What makes NEO different from other storage implementations, and what needs it addresses.
Why not bare FileStorage ? Bare FileStorage doesn't allow concurrent accesses from multiple processes,so (Python) application cannot take advantage of multiple cores/CPUs, even less multiple machines.
Why not ZEO, which addresses this problem ? ZEO doesn't allow data distribution over multiple disks, soapplication cannot use more storage space than what is available to a single computer, or get better
bandwidth.
Why not ZEORaid ? ZEORaid does not allow scaling in data size terms, as it does RAID-1-ish replication(each underlying storage contains the whole database).
Why not relStorage ? relStorage has scalability issues because Relational DataBases do not send
invalidation messages, and it must emulate them. Also, depending on the underlying RDB, it might notsupport master-master replication. And also, it suffers from the same problem as ZEORaid for data size
scalability.
NEO's design addresses these problems:
NEO allows tweaking desired fault-tolerance and space efficiency, by providing it with a number of
replicates ("R" in above figure), achieving data distribution among machines. Master-master replication isnatively implemented in NEO, as well as invalidation handling.

NEO Overview

3The client part implements ZODB.Storage API, allowing NEO to be used as a drop-in replacement in anyapplication using the Zope Object DataBase, including low-level client-side data caching. It uses theMulti-Queue caching algorythm, which improves memory usage efficiency when used as a 2 nd-level cache
as is the case with ZODB.
A master node is a process providing functions of sequential number generation (OIDs and TIDs),
transaction control including object cache invalidation notifications for clients, and storage nodes loadbalancing decisions. Due to its function, it is important for good performance for master node to bequickly reachable from any other node, limiting the distance between them.
A storage node is a process providing persistent storage space. This storage is used for database objects
and associated transaction meta-data, and also contains NEO meta-data: current partition table andnumber generators state.
An admin node is a process listening for notifications from a master node and archiving them to provide
administrator with cluster health monitoring. It also serves as an administrative command relay.
Neoctl is a comman-line tool to allow querying cluster state and issuing commands, such as adding ordropping a storage from cluster.
It is important to note that object data never goes through any node but storage and client.

NEO Overview

4NEO objectives are being scalable and resilient to hardware failures. To achieve this, it involves
parallelism by scattering multiple instances of each kind of node over different machines:
It is obviously possible to access the same cluster from multiple client nodes (ie, Zope processes),
allowing processing power scalability.
Master nodes are also possible to scatter on multiple machines, although at any time only one will beactually use (the "primary master"), while others are just spares ("secondary masters"), ready to take overin case of primary master failure.
Storage nodes parallelism allows two kind of distribution: load balancing, by containing only a part of theentire object database, and redundancy by storing any database part in multiple storage nodes and making
the cluster resilient to hardware failures – as long as at least one copy of every database partition isavailable.
Admin node redundancy is trivial and can be likened to client node redundancy.

NEO Overview

5The partition table is known to all nodes in a NEO cluster, and updated by the primary master node. It
provides two indirection levels.
The first indirection is between an object and a partition number. This is achieved by a simple hash on
persistent object identifier (OID), which is known whether we want to load or store an object. A partition
is identified by its number.
The second indirection is between partition and nodes which contain it (we then reffer to cells as apartition on a given). A partition can exist in multiple places in a cluster, which is how fault tolerance isachieved. This level of indirection contains a partition state property, which can have several values:
Out of date: a node in such state might be completely unavailable or in the process of replicating partition
data from other storage nodes. It cannot be used for reading data, but data can be written to it if node isup.
Up to date: this node has a complete copy of partition content, which includes all objects and theirrevisions if any.
Feeding: this node is busy providing data to another storage node. This is a variant of "Up to date" state,
but nodes should refrain reading from this node, so replication can operate at maximum speed.
Once the node is chosen, it can be accessed by it's listening ip/port present in the node table.

NEO Overview

6In usual implementations of the ZODB.Storage API, at least one storage-level lock is used to serialise
commits. In given example, T1 delays T2 commit by the duration of its 2-phases commit 2 nd phase(implementation-dependent), while each transaction actually modify unrelated objects.
The induced delays grows as the number of applications accessing the database grows, and puts a limit on
the benefit applications can expect from multi-processing / clustering.
To remove this limit, NEO uses a different locking scheme: in a way similar to relational databaseengines using row-level locking rather than table-level, NEO uses object-level locks.

NEO Overview

7

Object-level locking scheme doesn't alter conflict detection and resolution: modifying the same object in
two concurrent transactions will result in the latter waiting for the former to commit. When the former isdone committing, conflict resolution will happen if object class implements it.
When different objects are being altered in parallel, they do not block each other. This is the desired effectof finer locking granularity, to make commit scale better by being able to run in parallel.
Note: this locking scheme has not been proven correct yet. It is believed that it should not harm existing
applicatoins, because the only application-visible effect is that database view does not depend on
transaction commit wall-clock start time order. This means that the result from a transaction T2
committing later than a transaction T1 might be visible to a third transaction before T1's result. In above
example, this is shown by having B1 visible while A0 is still visible.

NEO Overview

8High-availability challenges disaster recovery mechanisms, because restoring a lot of data should not taketoo much time.
The first disaster recovery mechanism is backup. In NEO, each storage node can handle its own backups
independently to then recover from failures which involve few enough nodes so service is preserved (ie,there is still at least one copy of each object and each transaction in the database). Restoring data fromnode's backup and starting storage process will trigger a replication between surviving node(s) assigned topartition(s) assigned to failed node, and it will catch up with what was missing in its backup. This is justthe regular mechanism used when any storage node gets disconnected from a running cluster.
This way, the amount of time needed to restore data does not matter for cluster availability, only the timeneeded to replicate missed data is to some extent (it increases cluster load without interrupting normalfunction).

NEO Overview

9Preserving high-availability when a broader disaster occurs (power loss, fire, ...) is the most difficult part,
because it becomes limited by the time needed to restore a node (multiple node could be restored inparallel).
So the solution here is to have a second cluster in a different place (as different as needed to be unlikely tosuffer from the same problem as the main datacenter) and updated periodically from main cluster (to beconsistent at any point in time).
This is achieved by having a replication node periodically build the list of actions secondary cluster must
do to reach a newer consistent database state, and sending them to it. This reduces the impact of increased
network latency, which prevents joining both cluster into a single one.

NEO Overview

10

It is already possible to migrate an existing database to NEO, and out of NEO again if needed, via theneomigrate command.
There are currently two possible setups for the NEO cluster side. One is a single-cluster setup, the other isa multi-clutser setup. The former suits single-ZODB applications, while the second would suite
applications which already span over multiple ZODBs via, for example Zope mount-points.
The problem with a multi-clutser setup is that it involves a significant overhead: duplicatedconfigurations, duplicated processes. As there is currently no known reliable way of merging two
databases into one, the remaining possibility is so-called "multi-export": a single NEO cluster hosting
multiple independent ZODB. Work to support this has already been done, but is not finished as of thiswriting.

NEO Overview

11TODO- Implementation performance (better client-side caching,"zero-copy" protocol module, ...)
- Better admin tools (monitoring, control)- Backend-independent backups- Replication node
- Existing ZODB features : BLOBs, readCurrent API, ...
- Contributions to ZODB : transaction garbage collection API- NEO2 : Long-distance NEO

Support NEO !
www.neoppod.org© Nexedi SA 2010 – All rights reserved – GNU Free Documentation License

NEO is not mature yet. It is not ready to be used in production, and lacks some functionalities.
First of all, its implementation has not been tweaked for performance. An example of this is the current
protocol implementation, which tends to copy data over and over before using it or sending it.
Also, the MultiQueue implementation used in client nodes has been shown to take a huge benefit frombeing rewritten in a compiled, statically-typed language.
Administration tools are currently quite rough, and doesn't prevent an absent-minded admin fromdestroying his database.
Backup creation and restoration currently depends on storage back-end own backup implementation. As
NEO should be able to move to different back-ends (KyotoCabinet support is being considered), it shouldimplement this.
Replication node presented in page on Global disaster recovery is not implemented yet.
NEO lacks some ZODB features, such as BLOB support and readCurrent API.
Also, NEO could contribute back code to ZODB, such as transaction garbage collection for a morescalable pack implementation on storage implementations such as NEO, where transaction metadata is notnecessarily on the same place as object data.
And finally, a long-term goal would be to develop a long-distance-capable NEO architecture, allowing tospread nodes around the planet while preserving ZODB semantics.

NEO Overview

12Extra slides© Nexedi SA 2010 – All rights reserved – GNU Free Documentation License

The following pages were shown as a support for questions asked after the DZUG 2010 presentation.

NEO Overview

13The two-phase commit starts by asking the primary master node to generate a TID, unless one is provided
by client application, in which case there is nothing to do at network level.
Then object data can be sent to storage nodes.
Sending an object has the effect of taking a write lock on that object, which is local to each storage node.
If that lock is already taken when the node receives the request, the locking TID is checked: if it's later
than the one trying to get the lock, a conflict is notified to the client. As object is already being modified
by a later transaction, no conflict resolution can happen, so client will always raise a conflict error in suchcase. Otherwise, it will delay the store operation until the lock is released.
This locking scheme can lead to deadlocks if multiple client nodes send the same object to multiple
storage nodes. These deadlocks are recovered by a time-out mechanism: if a client doesn't get theresponse for a store request after some time, it asks storage node if the store request is being delayed due
to a write lock, and if so aborts the transaction.
When the lock is taken for an operation, storage node checks for conflicts on stored object. Those
conflicts are detected by comparing the base revision of that object (the revision client got from databasewhen it started modifying it) matches its current committed revision. If those revisions are equal, storage
sends a acknowledgement to the client node. Otherwise, storage node doesn't take the write lock for thatobject, and responds to client with a conflict notification.
Vote ends this commit phase by just waiting for all pending responses from storage nodes.
If transaction must be aborted, it is enough to ask all storage to release locks held by that transaction.

NEO Overview

14

When client node is asked to finish commit, it sends a single requests to primary master node. That node
will then take a read-lock for all objects involved in the transaction. When storage nodes take that lock,they write received objects to a temporary location, where they will survive a node restart but are notaccessible by other nodes.
Once all storage acknowledged the lock take, primary master node asks them to release all locks ontransaction objects. When storage nodes release this lock, they also move previously written objects to thedefinitive location, where they become available as the current revision of given object to other nodes.
Also, primary master node answers committing client that commit is over, and sends cache invalidations
to other client nodes. In both cases, this is used to invalidate client caches, with the only difference thatcommitting client should store committed data in its local cache rather than just flushing.
This locking scheme achieves a "network barrier": it prevents being able to read simultaneously different
versions of the same object while storage nodes are independently committing data.