message passing

Eventual Consistency in MySQL Cluster - implementation part 3

2011-12-22T17:36:00.003Z

As promised, this is the final post in a series looking at eventual consistency with MySQL Cluster asynchronous replication. This time I'll describe the transaction dependency tracking used with NDB$EPOCH_TRANS and review some of the implementation properties.

Transaction based conflict handling with NDB$EPOCH_TRANS

NDB$EPOCH_TRANS is almost exactly the same as NDB$EPOCH, except that when a conflict is detected on a row, the whole user transaction which made the conflicting row change is marked as conflicting, along with any dependent transactions. All of these rejected row operations are then handled using inserts to an exceptions table and realignment operations. This helps avoid the row-shear problems described here.

Including user transaction ids in the Binlog

Ndb Binlog epoch transactions contain row events from all the user transactions which committed in an epoch. However there is no information in the Binlog indicating which user transaction caused each row event. To allow detected conflicts to 'rollback' the other rows modified in the same user transaction, the Slave applying an epoch transaction needs to know which user transaction was responsible for each of the row events in the epoch transaction. This information can now be recorded in the Binlog by using the --ndb-log-transaction-id MySQLD option. Logging Ndb user transaction ids against rows in-turn requires a v2 format RBR Binlog, enabled with the --log-bin-use-v1-row-events=0 option. The mysqlbinlog --verbose tool can be used to see per-row transaction information in the Binlog.

User transaction ids in the Binlog are useful for NDB$EPOCH_TRANS and more. One interesting possibility is to use the user transaction ids and same-row operation dependencies to sort the row events inside an epoch into a partial order. This could enable recovery to a consistent point other than an epoch boundary. A project for a rainy day perhaps?

NDB$EPOCH_TRANS multiple slave passes

Initially, NDB$EPOCH_TRANS proceeds in the same way as NDB$EPOCH, attempting to apply replicated row changes, with interpreted code attached to detect conflicts. If no row conflicts are detected, the epoch transaction is committed as normal with the same minimal overhead as NDB$EPOCH. However if a row conflict is detected, the epoch transaction is rolled back, and reapplied. This is where NDB$EPOCH_TRANS starts to diverge from NDB$EPOCH.

In this second pass, the user transaction ids of rows with detected conflicts are tracked, along with any inter-transaction dependencies detectable from the Binlog. At the end of the second pass, prior to commit, the set of conflicting user transactions is combined with the user transaction dependency data to get a complete set of conflicting user transactions. The epoch transaction initiated in the second pass is then rolled-back and a third pass begins.

In the third pass, only row events for non-conflicting transactions are applied, though these are still applied with conflict detecting interpreted programs attached in case a further conflict has arisen since the second pass. Conflict handling for row events belonging to conflicting transactions is performed in the same way as NDB$EPOCH. Prior to commit, the applied row events are checked for further conflicts. If further conflicts have occurred then the epoch transaction is rolled back again and we return to the second pass. If no further conflicts have occurred then the epoch transaction is committed.

These three passes, and associated rollbacks are only externally visible via new counters added to the MySQLD server. From an external observer's point of view, only non-conflicting transactions are committed, and all row events associated with conflicting transactions are handled as conflicts. As an optimisation, when transactional conflicts have been detected, further epochs are handled with just two passes (second and third) to improve efficiency. Once an epoch transaction with no conflicts has been applied, further epochs are initially handled with the more optimistic and efficient first pass.

Dependency tracking implementation

To build the set of inter-transaction dependencies and conflicts, two hash tables are used. The first is a unique hashmap mapping row event tables and primary keys to transaction ids. If two events for the same table and primary key are found in a single epoch transaction then there is a dependency between those events, specifically the second event depends on the first. If the events belong to different user transactions then there is a dependency between the transactions.

Transaction dependency detection hash :

{Table, Primary keys} -> {Transaction id}

The second hash table is a hashmap of transaction id to an in-conflict marker and a list of dependent user transactions. When transaction dependencies are discovered using the first dependency detection hash, the second hash is modified to reflect the dependency. By the end of processing the epoch transaction, all dependencies detectable from the Binlog are described.

Transaction dependency tracking and conflict marking hash :

{Transaction id} -> {in_conflict, List}

As epoch operations are applied and row conflicts are detected, the operation's user transaction id is marked in the dependency hash as in-conflict. When marking a transaction as in-conflict, all of its dependent transactions must also be transitively marked as in-conflict. This is done by a traverse through the dependency tree of the in-conflict transaction. Due to slave batching, the addition of new dependencies and the marking of conflicting transactions is interleaved, so adding a dependency can result in a sub-tree being marked as in-conflict.

After the second pass is complete, the transaction dependency hash is used as a simple hash for looking up whether a particular transaction id is in conflict or not :

Transaction in-conflict lookup hash :

{Transaction id} -> {in_conflict}

This is used in the third pass to determine whether to apply each row event, or to proceed straight to conflict handling.

The size of these hashes, and the complexity of the dependency graph is bounded by the size of the epoch transaction. There is no need to track dependencies across the boundary of two epoch transactions, as any dependencies will be discovered via conflicts on the data committed by the first epoch transaction when attempting to apply the second epoch transaction.

Event counters

Like the existing conflict detection functions, NDB$EPOCH_TRANS has a row-conflict detection counter called ndb_conflict_epoch_trans.

Additional counters have been added which specifically track the different events associated with transactional conflict detection. These can be seen with the usual SHOW GLOBAL STATUS LIKE syntax, or via the INFORMATION_SCHEMA tables.

ndb_conflict_trans_row_conflict_count
This is essentially the same as ndb_conflict_epoch_trans - the number of row events with conflict detected.
ndb_conflict_trans_row_reject_count
The number of row events which were handled as in-conflict. It will be at least as large as ndb_conflict_trans_row_count, and will be higher if other rows are implicated by being in a conflicting transaction, or being dependent on a row in a conflicting transaction.
A separate ndb_conflict_trans_row_implicated_count could be constructed as ndb_conflict_trans_row_reject_count - ndb_conflict_trans_row_conflict_count
ndb_conflict_trans_reject_count
The number of discrete user transactions detected as in-conflict.
ndb_conflict_trans_conflict_commit_count
The number of epoch transactions which had transactional conflicts detected during application.
ndb_conflict_trans_detect_iter_count
The number of iterations of the three-pass algorithm that have occurred. Each set of passes counts as one. Normally this would be the same as ndb_conflict_trans_conflict_commit_count. Where further conflicts are found on the third pass, another iteration may be required, which would increase this count. So if this count is larger than ndb_conflict_trans_conflict_commit_count then there have been some conflicts generated concurrently with conflict detection, perhaps suggesting a high conflict rate.

Performance properties of NDB$EPOCH and NDB$EPOCH_TRANS

I have tried to avoid getting involved in an explanation of Ndb replication in general which would probably fill a terabyte of posts. Comparing replication using NDB$EPOCH and NDB$EPOCH_TRANS relative to Ndb replication with no conflict detection, what can we can say?

Conflict detection logic is pushed down to data nodes for execution
Minimising extra data transfer + locking
Slave operation batching is preserved
Multiple row events are applied together, saving MySQLD <-> data node round trips, using data node parallelism
For both algorithms, one extra MySQLD <-> data node round-trip is required in the no-conflicts case (best case)
NDB$EPOCH : One extra MySQLD <-> data node round-trip is required per *batch* in the all-conflicts case (worst case)
NDB$EPOCH : Minimal impact to Binlog sizes - one extra row event per epoch.
NDB$EPOCH : Minimal overhead to Slave SQL CPU consumption
NDB$EPOCH_TRANS : One extra MySQLD <-> data node round-trip is required per *batch* per *pass* in the all-conflicts case (worst case)
NDB$EPOCH_TRANS : One round of two passes is required for each conflict newly created since the previous pass.
NDB$EPOCH_TRANS : Small impact to Binlog sizes - one extra row event per epoch plus one user transaction id per row event.
NDB$EPOCH_TRANS : Small overhead to Slave SQL CPU consumption in no-conflict case

Current and intrinsic limitations

These functions support automatic conflict detection and handling without schema or application changes, but there are a number of limitations. Some limitations are due to the current implementation, some are just intrinsic in the asynchronous distributed consistency problem itself.

Intrinsic limitations

Reads from the Secondary are tentative
Data committed on the secondary may later be rolled back. The window of potential rollback is limited, after which Secondary data can be considered stable. This is described in more detail here.
Writes to the Secondary may be rolled back
If this occurs, the fact will be recorded on the Primary. Once a committed write is stable it will not be rolled back.
Out-of-band dependencies between transactions are out-of-scope
For example direct communication between two clients creating a dependency between their committed transactions, not observable from their database footprints.

Current implementation limitations

Detected transaction dependencies are limited to dependencies between binlogged writes (Insert, Update, Delete)
Reads are not currently included.
Delete vs Delete+Insert conflicts risk data divergence
Delete vs Delete conflicts are detected, but currently do not result in conflict handling, so that Delete vs Delete + Insert can result in data divergence.
With NDB$EPOCH_TRANS, unplanned Primary outages may require manual steps to restore Secondary consistency
With pending multiple, time spaced, non-overlapping transactional conflicts, an unexpected failure may need some Binlog processing to ensure consistency.

Want to try it out?

Andrew Morgan has written a great post showing how to setup NDB$EPOCH_TRANS. He's even included non-ascii art. This is probably the easiest way to get started. NDB$EPOCH is slightly easier to get started with as the --ndb-log-transaction-id (and Binlog v2) options are not required.

Edit 23/12/11 : Added index

Eventual consistency in MySQL Cluster - implementation part 2

2011-12-19T13:30:00.001Z

In previous posts I described how row conflicts are detected using epochs. In this post I describe how they are handled.

Row based conflict handling with NDB$EPOCH

Once a row conflict is detected, as well as rejecting the row change, row based conflict handling in the Slave will :

Increment conflict counters
Optionally insert a row into an exceptions table

For NDB$EPOCH, conflict detection and handling operates on one Cluster in an Active-Active pair designated as the Primary. When a Slave MySQLD attached to the Primary Cluster detects a conflict between data stored in the Primary and a replicated event from the Secondary, it needs to realign the Secondary to store the same values for the conflicting data. Realignment involves injecting an event into the Primary Cluster's Binlog which, when applied idempotently on the Secondary Cluster, will force the row on the Secondary Cluster to take the supplied values. This requires either a WRITE_ROW event, with all columns, or a DELETE_ROW event with just the primary key columns. These events can be thought of as compensating events used to revert the original effect of the rejected events.

Conflicts are detected by a Slave MySQLD attached to the Primary Cluster, and realignment events must appear in Binlogs recorded by the same MySQLD and/or other Binlogging MySQLDs attached to the Primary Cluster. This is achieved using a new NdbApi primary key operation type called refreshTuple.

When a refreshTuple operation is executed it will :

Lock the affected row/primary key until transaction commit time, even if it does not exist (much as an Insert would).
Set the affected row's author metacolum to 0
The refresh is logically a local change
On commit
- Row exists case : Set the row's last committed epoch to the current epoch
- Cause a WRITE_ROW (row exists case) or DELETE_ROW (no row exists) event to be generated by attached Binlogging MySQLDs.

Locking the row as part of refreshTuple serialises the conflicting epoch transaction with other potentially conflicting local transactions. Updating the stored epoch and author metacolumns results in the conflicting row conflicting with any further replicated changes occurring while the realignment event is 'in flight'. The compensating row events are effectively new row changes originating at the Primary cluster which need to be monitored for conflicts in the same way as normal row changes.

It is important that the Slave running at the Secondary Cluster where the realignment events will be applied, is running in idempotent mode, so that it can handle the realignment events correctly. If this is not the case then WRITE_ROW realignment events may hit 'Row already exists' errors, and DELETE_ROW realignment events may hit 'Row does not exist' errors.

Observations on conflict windows and consistency

When a conflict is detected, the refresh process results in the row's epoch and author metacolumns being modified so that the window of potential conflict is extended, until the epoch in which the refresh operation was recorded has itself been reflected. If ongoing updates at both clusters continually conflict then refresh operations will continue to be generated, and the conflict window will remain open until a refresh operation manages to propagate with no further conflicts occurring. As with any eventually consistent system, consistency is only guaranteed when the system (or at least the data of interest) is quiescent for a period.

From the Primary cluster's point of view, the conflict window length is the time between committing a local transaction in epoch n, and the attached Slave committing a replicated epoch transaction indicating that epoch n has been applied at the Secondary. Any Secondary-sourced overlapping change applied in this time is in-conflict.

This Cluster conflict window length is comprised of :

Time between commit of transaction, and next Primary Cluster epoch boundary
(Worst = 1 * TimeBetweenEpochs, Best = 0, Avg = 0.5 * TimeBetweenEpochs)
Time required to log event in Primary Cluster's Binlogging MySQLDs Binlog (~negligible)
Time required for Secondary Slave MySQLD IO thread to
- Minimum : Detect new Binlog data - negligible
- Maximum : Consume queued Binlog prior to the new data - unbounded
- Pull new epoch transaction
- Record in Relay log
Time required for Secondary Slave MySQLD SQL thread to
- Minimum : Detect new events in relay log
- Maximum : Consume queued Relay log prior to new data - unbounded
- Read and apply events
- Potentially multiple batches.
- Commit epoch transaction at Secondary
Time between commit of replicated epoch transaction and next Secondary Cluster epoch boundary
(Worst = 1 * TimeBetweenEpochs, Best = 0, Avg = 0.5 * TimeBetweenEpochs)
After this point a Secondary-local commit on the data is possible without conflict
Time required to log event in Secondary Cluster's Binlogging MySQLDs Binlog (~negligible)
Time required for Primary Slave MySQLD IO thread to
- Minimum : Detect new Binlog data
- Maximum : Consume queued Binlog data prior to the new data - unbounded
- Pull new epoch transaction
- Record in Relay log
Time required for Primary Slave MySQLD SQL thread to
- Minimum : Detect new events in relay log
- Maximum : Consume queued Relay log prior to new data - unbounded
- Read and apply events
- Potentially multiple batches.
- For NDB$EPOCH_TRANS, potentially multiple passes
- Commit epoch transaction
- Update max replicated epoch to reflect new maximum.
Further Secondary sourced modifications to the rows are now considered not-in-conflict

From the point of view of an external client with access to both Primary and Secondary clusters, the conflict window only extends from the time transaction commit occurs at the Primary to the time the replicated operations are applied at the Secondary, and its commit time Secondary epoch ends. Changes committed at the Secondary after this will clearly appear to the Primary to have occurred after its epoch was applied on the Secondary and therefore are not in-conflict.

Assuming that both Clusters have the same TimeBetweenEpochs, we can simplify the Cluster conflict window to :

  Cluster_conflict_window_length = EpochDelay +
                                  P_Binlog_lag +
                                  S_Relay_lag +
                                  S_Binlog_lag +
                                  P_Relay_lag

 Where
    EpochDelay minimum is 0
    EpochDelay avg     is TimeBetweenEpochs
    EpochDelay maximum is 2 * TimeBetweenEpochs

Substituting the default value of TimeBetweenEpochs of 100 millis, we get :

     EpochDelay minimum is 0
    EpochDelay avg     is 100 millis
    EpochDelay maximum is 200 millis

Note that TimeBetweenEpochs is an epoch-increment trigger delay. The actual experienced time between epochs can be longer depending on system load. The various Binlog and Relay log delays can vary from close to zero up to infinity. Infinity occurs when replication stops in either direction.

The Cluster conflict window length can be thought of as both

The time taken to detect a conflict with a Primary transaction
The time taken for a committed Secondary transaction to become stable or be reverted

We can define a Client conflict window length as either :

 Primary->Secondary

  Client_conflict_window_length = EpochDelay +
                                  P_Binlog_lag +
                                  S_Relay_lag +
                                  EpochDelay

or

Secondary->Primary

  Client_conflict_window_length = EpochDelay +
                                  S_Binlog_lag +
                                  P_Relay_lag

Where EpochDelay is defined as above.

These definitions are asymmetric. They represent the time taken by the system to determine that a particular change at one cluster definitely happened-before another change at the other cluster. The asymmetry is due to the need for the Secondary part of a Primary->Secondary conflict to be recorded in a different Secondary epoch. The first definition considers an initial change at the Primary cluster, and a following change at the Secondary. The second definition is for the inverse case.

An interesting observation is that for a single pair of near-concurrent updates at different clusters, happened-before depends only on latencies in one direction. For example, an update to the Primary at time Ta, followed by an update to the Secondary at time Tb will not be considered in conflict if:

 Tb - Ta > Client_conflict_window_length(Primary->Secondary)

Client_conflict_window_length(Primary->Secondary) depends on the EpochDelay, the P_Binlog_lag and S_Relay_lag, but not on the S_Binlog_lag or P_Relay_lag. This can mean that high replication latency, or a complete outage in one direction does not always result in increased conflict rates. However, in the case of multiple sequences of near-concurrent updates at different sites, it probably will.

A general property of the NDB$EPOCH family is that the conflict rate has some dependency on the replication latency. Whether two updates to the same row at times Ta and Tb are considered to be in conflict depends on the relationship between those times and the current system replication latencies. This can remove the need for highly synchronised real-time clocks as recommended for NDB$MAX, but can mean that the observed conflict rate increases when the system is lagging. This also implies that more work is required to catch up, which could further affect lag. NDB$MAX requires manual timestamp maintenance, and will not detect incorrect behaviour, but the basic decision on whether two updates are in-conflict is decided at commit time and is independent of the system replication latency.

In summary :

The Client_conflict_window_length in either direction will on average not be less than the EpochDelay (100 millis by default)
Clients racing against replication to update both clusters need only beat the current Client_conflict_window_length to cause a conflict
Replication latencies in either direction are potentially independent
Detected conflict rates partly depend on replication latencies

Stability of reads from the Primary Cluster

In the case of a conflict, the rows at the Primary Cluster will tentatively have replicated operations applied against them by a Slave MySQLD. These conflicting operations will fail prior to commit as their interpreted precondition checks will fail, therefore the conflicting rows will not be modified on the Primary. One effect of this is that a read from the Primary Cluster only ever returns stable data, as conflicting changes are never committed there. In contrast, a read from the Secondary Cluster returns data which has been committed, but may be subject to later 'rollback' via refresh operations from the Primary Cluster.

The same stability of reads observation applies to a row change event stream on the Primary Cluster - events received for a single key will be received in the order they were committed, and no later-to-be-rolled-back events will be observed in the stream.

Stability of reads from the Secondary Cluster

If the Secondary Cluster is also receiving reflected applied epoch information back from the Primary then it will know when it's epoch x has been applied successfully at the Primary. Therefore a read of some row y on the Secondary can be considered tentative while Max_Replicated_Epoch(Secondary) < row_epoch(y), but once Max_Replicated_Epoch(Secondary) >= row_epoch(y) then the read can be considered stable. This is because if the Primary were going to detect a conflict with a Secondary change committed in epoch x, then the refresh events associated with the conflict would be recorded in the same Primary epoch as the notification of the application of epoch x. So if the Secondary observes the notification of epoch x (and updates Max_Replicated_Epoch accordingly), and row y is not modified in the same epoch transaction, then it is stable. The time taken to reach stability after a Secondary Cluster commit will be the Cluster conflict window length.

Perhaps some applications can make better use of the potentially transiently inconsistent Secondary data by categorising their reads from the Secondary as either potentially-inconsistent or stable. To do this, they need to maintain Max_replicated_epoch(Secondary) (By listening to row change events on the ndb_apply_status table) and read the NDB$GCI_64 metacolumn when reading row data. A read from the Secondary is stable if all the NDB$GCI_64 values for all rows read are <= the Secondary's Max_Replicated_Epoch.

In the next post (final post I promise!) I will describe the implementation of the transaction dependency tracking in NDB$EPOCH_TRANS, and review the implementation of both NDB$EPOCH and NDB$EPOCH_TRANS.

Edit 23/12/11 : Added index

Eventual consistency in MySQL Cluster - implementation part 1

2011-12-08T00:20:00.006Z

The last post described MySQL Cluster epochs and why they provide a good basis for conflict detection, with a few enhancements required. This post describes the enhancements.

The following four mechanisms are required to implement conflict detection via epochs :

Slaves should 'reflect' information about replicated epochs they have applied
Applied epoch numbers should be included in the Slave Binlog events returning to the originating cluster, in a Binlog position corresponding to the commit time of the replicated epoch transaction relative to Slave local transactions.
Masters should maintain a maximum replicated epoch
A cluster should use the reflected epoch information to track which of its epochs has been applied by a Slave cluster. This will be the maximum of all epochs applied by the Slave.
Masters should track commit-time epoch per row
To allow per-row detection of conflicts
Masters should track commit-authorship per row
To differentiate recent epochs due to replication or conflicting activity.

'Reflecting' epoch information and maintaining the maximum replicated epoch

Every epoch transaction in the Binlog contains a special WRITE_ROW event on the mysql.ndb_apply_status table which carries the epoch transaction's epoch number. This is designed to give an atomically consistent way to determine a Slave cluster's position relative to a Master cluster. Normally these WRITE_ROW events are applied by the Slave but not logged in the Slave's Binlog, even when --log-slave-updates is ON. A new MySQLD option, --ndb-log-apply-status causes WRITE_ROW events applied to the mysql.ndb_apply_status table to be binlogged at a Slave, even when --log-slave-updates is OFF. These events are logged with the ServerId of the Slave MySQLD, so that they can be applied on the Master, but will not loop infinitely.

Allowing this applied epoch information to propagate through a Slave Cluster has the following effects :

Downstream Clusters become aware of their position relative to all upstream Master clusters, not just their immediate Master cluster.
They gain extra mysql.ndb_apply_status entries for all upstream Masters.
Circularly replicating clusters become aware of which of their epochs, and epoch transactions, have been applied to all clusters in the circle.
They gain extra mysql.ndb_apply_status entries for all Binlogging MySQLDs in the loop

Effect 1 is useful for replication failover with more than two replication-chained clusters where an intermediate cluster is being routed-around (A->B->C) -> (A->C). Cluster C knows the correct Binlog file and position to resume from on A, without consulting B.

Effect 2 could be used to allow clients to wait until their writes have been fully replicated and are globally visible, a kind of synchronous replication. More relevantly, effect 2 allows us to maintain a maximum replicated epoch value for detecting conflicts.

The visible result of using --ndb-log-apply-status on a Slave is that the mysql.ndb_apply_status table on the Master contains extra entries for the Binlogging MySQLDs attached to its Cluster. The maximum replicated epoch is the maximum of these epoch values.

    Cluster 1 Epoch transactions in flight in
         a circular configuration
        (Ignoring Cluster 2 epochs)

                           39       38       37
                     ->---->----->----->----->--
                    /                           \ (Queued epochs 36-26)
          Cluster 1                             Cluster 2
(Queued epochs 23,24) \                           /
                     -<---<------<----<----<----
                          25       26       27

Current epoch = 40
Max replicated epoch = 22

A MySQLD acting as a conflict detecting Slave for a cluster needs to know the attached cluster's maximum replicated epoch for conflict detection. On Slave start, before the Slave starts applying replicated changes to the Ndb storage engine, it scans the mysql.ndb_apply_status table to find the highest reflected epoch value. Rows in mysql.ndb_apply_status with server ids in the CHANGE MASTER TO IGNORE_SERVER_IDS list are considered to be local servers, as well as the Slave's own server id, and the maximum replicated epoch is the maximum epoch value from these rows.

@ Slave start

 max_replicated_epoch = SELECT MAX(epoch)
                          FROM mysql.ndb_apply_status
                         WHERE server_id IN @@IGNORE_SERVER_IDS;

Once the Max_replicated_epoch has been initialised at slave start, it is updated as each reflected epoch event (WRITE_ROW event to mysql.ndb_apply_status) arrives and is processed by the Slave SQL thread. The current Max_replicated_epoch can be seen by issuing the command SHOW STATUS LIKE 'Ndb_slave_max_replicated_epoch';. Note that this is really just a cached copy of the current result of the SELECT MAX(epoch) query from above. One subtlety is that the max_replicated_epoch is only changed when the Slave commits an epoch transaction, as it is only at this point that we know for sure that any event committed on the other cluster before the replicated epoch was applied has been handled.

Per row last-modified epoch storage

Each row stored in Ndb has a built-in hidden metadata column called NDB$GCI64. This columns stores the epoch number at which the row was last modified. For normal system recovery purposes, only the top 32 bits of the 64 bit epoch, called the Global Checkpoint Index or GCI are used. NDB$EPOCH needs further bits to be stored per-row. Epoch values only use a few of the bits in the bottom 32 bits of the epoch, so by default 6 extra bits per row are used to enable a full 64 bit epoch to be stored for each row. The actual number of bits used can be controlled by a parameter to NDB$EPOCH. Where some epoch is not fully expressible in the number of bits available, the bottom 32 bits are saturated, which again errs on the side of safety, potentially causing false conflicts, but ensuring no real conflicts are missed. The ndb_select_all tool has a --gci64 option which shows each row's stored epoch value.

A conflict detecting slave detects conflicts between transactions already committed, whose rows have their commit-time epoch numbers, and incoming operations in an epoch transaction, which are considered to have been committed at the epoch given by the current Maximum Replicated Epoch. An incoming operation is considered to be in-conflict if the row it affects has a last-committed epoch that is greater than the current Maximum Replicated Epoch.

  in_conflict = (ndb_gci64 > max_replicated_epoch)

In other words, at the time the change was committed on the other Cluster, that other Cluster was only aware of our changes as-of our epoch (max_replicated_epoch). Therefore it was unaware of any changes committed in more recent epochs. If the row being changed has been locally modified since that epoch then there have been concurrent modifications and a conflict has been discovered.

Note that this mechanism is purely based on monitoring serialisation of updates to rows. No semantic understanding of row data, or the meaning of applied changes is attempted. Even if both clusters update some row to contain exactly the same value it will be considered to be a conflict, as the updates were not serialised with respect to each other.

Per row hidden Author metacolumn

One advantage of reusing the row's last-modified epoch number for conflict detection is that it is automatically set on every commit. However the downside is that when a replicated modification is found to not be in conflict, and is applied, the row's epoch is automatically set to the current value at commit time as normal. By definition, the current epoch value is always greater than the maximum replicated epoch, and so if a further replicated modification to the same row were to arrive, it would find the row's epoch to be higher than the current maximum replicated epoch, and detect a false conflict.

In theory we could consider the current maximum replicated epoch to be the row's commit time epoch, but as the per-row epoch is used for other more critical DB recovery purposes it's not safe to abuse it in this way. Instead we use the observation that if we found a previous row update from some other cluster to be not-in-conflict, then further updates from it are also not-in-conflict.

To detect this, a new hidden metadata column is introduced called NDB$AUTHOR. This column is set to zero when a row is modified by any unmodified NdbApi client, including MySQLD, but when a row is modified by the MySQLD Slave SQL thread, it is set to one. More generally, NDB$AUTHOR could be set to a non-zero identifier of which other cluster sourced an accepted change. Just setting to one limits us to having one other cluster originating potentially conflicting changes. The ndb_select_all tool has a --author option which shows each row's stored Author value.

By extending the conflict detecting function to examine the NDB$AUTHOR value, we avoid the problem of falsely detecting conflicts when applied consecutive replicated changes.

  in_conflict = (ndb$author != change_author) && (ndb_gci64 > max_replicated_epoch)

We are currently just using 1 to mean 'other author', so this simplifies to :

 in_conflict = (ndb$author != 1) && (ndb_gci64 > max_replicated_epoch)

            = (ndb$author == 0) && (ndb_gci64 > max_replicated_epoch)

This conflict detection function is encoded in an Ndb interpreted program and attached to the replicated DELETE and UPDATE NdbApi operations so that it can be quickly and atomically executed at the Ndb data nodes as a predicate prior to applying the operation.

Ndb binlog row event ordering and false conflicts

The happened-before relationship between reflected epoch events (WRITE_ROW to mysql.ndb_apply_status) and incoming row events is used to determine whether a conflict has occurred. As described in the last post, Ndb offers limited ordering guarantees on the row events within an epoch transaction. The only guarantee is that multiple changes to the same row will be recorded in the order they committed. This implies that the relative ordering of the reflected epoch WRITE_ROW event, on some row in mysql.ndb_apply_status, and other row events on other tables sharing the same epoch transaction is meaningless. The only ordering guarantees between different rows exist at epoch boundaries.

This means that if we see a reflected epoch WRITE_ROW event somewhere in replicated epoch j, then we can only safely assume that this happened before incoming row events in epoch j+1 and later. The row events appearing before and after the reflected epoch WRITE_ROW event in epoch j may have committed before or after the reflected epoch event.

The relaxed relative ordering gives us reduced precision in determining happened-before, and to be safe, we must err on the side of assuming that a conflict exists rather than that it does not. Consider a Master committing a change to row X, recorded in epoch N. This is then applied on the Slave in Slave epoch S. If the Slave then commits a local change affecting the same row X in the same epoch S, this will be returned to the Master in the same Slave epoch transaction, and the Master will be unable to determine whether it occurred before or after it's original write to X, so must assume that it occurred before and is therefore in conflict. If the Slave had committed its change in epoch S+1 or later, the happened-before relationship would be clear and the change would not be considered in conflict.

These potential false conflicts are the price paid here for the lack of fine grained event ordering in the Ndb Binlog.

I'm lost

There's been a lot of information, or at least a lot of words. Let's summarise how NDB$EPOCH and NDB$EPOCH_TRANS detect row conflicts by following

@Cluster A
Transactions modify rows, automatically setting their hidden NDB$GCI64 column to the current epoch and their NDB$AUTHOR column to 0

Binlogging MySQLDs record modified rows in epoch transactions in their Binlogs, together with MySQLD generated mysql.ndb_apply_status WRITE_ROW events
@Cluster B
Slave MySQLDs apply replicated epoch transactions along with their generated mysql.ndb_apply_status WRITE_ROW events

Other clients of Cluster B commit transactions against the same data.

Binlogging MySQLDs 'reflect' the applied-replicated epoch information by recording the mysql.ndb_apply_status WRITE_ROW events in their Binlogs as a result of --ndb-log-apply-status.

Binlogging MySQLDs also record the row changes made by local clients.
@Cluster A
Slave MySQLDs track the incoming reflected epoch mysql.ndb_apply_status WRITE_ROW events to maintain their ndb_slave_max_replicated_epoch variables

Slave MySQLDs attach NdbApi interpreted programs to UPDATE and DELETE operations as they are applied to the database, comparing the row's stored NDB$GCI64 and NDB$AUTHOR columns with constant values supplied in the program.

If there are no conflicts, the UPDATE and DELETE operations are applied, and the row's NDB$AUTHOR columns are set to one indicating a successful Slave modification

If there are conflicts then conflict handling for the conflicting rows begins.

Now does that make any sense? Assuming it does, then next we look at how detected conflicts are handled.

Once again, another wordy endurance test and we're not finished. Surely the end must be near?

Edit 23/12/11 : Added index

Eventual Consistency in MySQL Cluster - using epochs

2011-12-07T14:28:00.007Z

Before getting to the details of how eventual consistency is implemented, we need to look at epochs. Ndb Cluster maintains an internal distributed logical clock known as the epoch, represented as a 64 bit number. This epoch serves a number of internal functions, and is atomically advanced across all data nodes.

Epochs and consistent distributed state

Ndb is a parallel database, with multiple internal transaction coordinator components starting, executing and committing transactions against rows stored in different data nodes. Concurrent transactions only interact where they attempt to lock the same row. This design minimises unnecessary system-wide synchronisation, enabling linear scalability of reads and writes.

The stream of changes made to rows stored at a data node are written to a local Redo log for node and system recovery. The change stream is also published to NdbApi event listeners, including MySQLD servers recording Binlogs. Each node's change stream contains the row changes it was involved in, as committed by multiple transactions, and coordinated by multiple independent transaction coordinators, interleaved in a partial order.

  Incoming independent transactions
 affecting multiple rows

   T3         T4         T7
   T1         T2         T5

    |         |          |
    V         V          V

 --------  --------  --------
 |  1   |  |  2   |  |  3   |
 |  TC  |  |  TC  |  |  TC  |   Data nodes with multiple
 |      |--|      |--|      |   transaction coordinators
 |------|  |------|  |------|   acting on data stored in
 |      |  |      |  |      |       different nodes
 | DATA |  | DATA |  | DATA |
 --------  --------  --------

    |         |          |
    V         V          V

   t4        t4          t3
   t1        t7          t2
   t2        t1          t7
             t5

 Outgoing row change event
  streams by causing
     transaction

These row event streams are generated independently by each data node in a cluster, but to be useful they need to be correlated together. For system recovery from a crash, the data nodes need to recover to a cluster-wide consistent state. A state which contains only whole transactions, and a state which, logically at least, existed at some point in time. This correlation could be done by an analysis of the transaction ids and row dependencies of each recorded row change to determine a valid order for the merged event streams, but this would add significant overhead. Instead, the Cluster uses a distributed logical clock known as the epoch to group large sets of committed transactions together.

Each epoch contains zero or more committed transactions. Each committed transaction is in only one epoch. The epoch clock advances periodically, every 100 milliseconds by default. When it is time for a new epoch to start, a distributed protocol known as the Global Commit Protocol (GCP) results in all of the transaction coordinators in the Cluster agreeing on a point of time in the flow of committing transactions at which to change epoch. This epoch boundary, between the commit of the last transaction in epoch n, and the commit of the first transaction in epoch n+1, is a cluster-wide consistent point in time. Obtaining this consistent point in time requires cluster-wide synchronisation, between all transaction coordinators, but it need only happen periodically.

Furthermore, each node ensures that the all events for epoch n are published before any events for epoch n+1 are published. Effectively the event streams are sorted by epoch number, and the first time a new epoch is encountered signifies a precise epoch boundary.

 Incoming independent transactions

   T3         T4         T7
   T1         T2         T5

    |         |          |
    V         V          V

 --------  --------  --------
 |  1   |  |  2   |  |  3   |
 |  TC  |  |  TC  |  |  TC  |   Data nodes with multiple
 |      |--|      |--|      |   transaction coordinators
 |------|  |------|  |------|   acting on data stored in
 |      |  |      |  |      |      different nodes
 | DATA |  | DATA |  | DATA |
 --------  --------  --------

    |         |          |
    V         V          V

  t4(22)    t4(22)      t3(22)            Epoch 22
  ......    ......      ......
  t1(23)    t7(23)      t2(23)            Epoch 23
  t2(23)    t1(23)      t7(23)
            ......
            t5(24)                        Epoch 24

  Outgoing row change event
  streams by causing transaction
  with epoch numbers in ()

When these independent streams are merge-sorted by epoch number we get a unified change stream. Multiple possible orderings can result.
One Partial ordering is shown here :

      Events      Transactions
               contained in epoch

   t4(22)
   t4(22)      {T4,T3}
   t3(22)

   ......

   t1(23)
   t2(23)
   t7(23)
   t1(23)      {T1, T2, T7}
   t2(23)
   t7(23)

   ......

   t5(24)      {T5}

Note that we can state from this that T4 -> T1 (Happened before), and T1 -> T5. However we cannot say whether T4 -> T3 or T3 -> T4. In epoch 23 we see that the row events resulting from T1, T2 and T7 are interleaved.

Epoch boundaries act as markers in the flow of row events generated by each node, which are then used as consistent points to recover to. Epoch boundaries also allow a single system wide unified transaction log to be generated from each node's row change stream, by merge-sorting the per-node row change streams by epoch number. Note that the order of events within an epoch is still not tightly constrained. As concurrent transactions can only interact via row locks, the order of events on a single row (Table and Primary key value) signifies transaction commit order, but there is by definition no order between transactions affecting independent row sets.

To record a Binlog of Ndb row changes, MySQLD listens to the row change streams arriving from each data node, and merge-sorts them them by epoch into a single, epoch-ordered stream. When all events for a given epoch have been received, MySQLD records a single Binlog transaction containing all row events for that epoch. This Binlog transaction is referred to as an 'Epoch transaction' as it describes all row changes that occurred in an epoch.

Epoch transactions in the Binlog

Epoch transactions in the Binlog have some interesting properties :

Efficiency : They can be considered a kind of Binlog group commit, where multiple user transactions are recorded in one Binlog (epoch) transaction. As an epoch normally contains 100 milliseconds of row changes from a cluster, this is a significant amortisation.
Consistency : Each epoch transaction contains the row operations which occurred when moving the cluster from epoch boundary consistent state A to epoch boundary consistent state B
Therefore, when applied as a transaction by a slave, the slave will atomically move from consistent state A to consistent state B
Inter-epoch ordering : Any row event recorded in epoch n+1 logically happened after every row event in epoch n
Intra-epoch disorder : Any two row events recorded in epoch n, affecting different rows, may have happened in any order.
Intra-epoch key-order : Any two row events recorded in epoch n, affecting the same row, happened in the order they are recorded.

The ordering properties show that epochs give only a partial order, enough to subdivide the row change streams into self-consistent chunks. Within an epoch, row changes may be interleaved in any way, except that multiple changes to the same row will be recorded in the order they were committed.

Each epoch transaction contains the row changes for a particular epoch, and that information is recorded in the epoch transaction itself, as an extra WRITE_ROW event on a system table called mysql.ndb_apply_status. This WRITE_ROW event contains the binlogging MySQLD's server id and the epoch number. This event is added so that it will be atomically applied by the Slave along with the rest of the row changes in the epoch transaction, giving an atomically reliable indicator of the replication 'position' of the Slave relative to the Master Cluster in terms of epoch number. As the epoch number is abstracted from the details of a particular Master MySQLD's binlog files and offsets, it can be used to failover to an alternative Master.

We can visualise a MySQL Cluster Binlog as looking something like this. Each Binlog transaction contains one 'artificially generated' WRITE_ROW event at the start, and then RBR row events for all row changes that occurred in that epoch.

    BEGIN
  WRITE_ROW mysql.ndb_apply_status server_id=4, epoch=6998
  WRITE_ROW ...
  UPDATE_ROW ...
  DELETE_ROW ...
  ...
  COMMIT # Consistent state of the database

  BEGIN
  WRITE_ROW mysql.ndb_apply_status server_id=4, epoch=6999
  ...
  COMMIT # Consistent state of the database

  BEGIN
  WRITE_ROW mysql.ndb_apply_status server_id=4, epoch=7000
  ...
  COMMIT # Consistent state of the database
  ...

A series of epoch transactions, each with a special WRITE_ROW event for recording the epoch on the Slave. You can see this structure using the mysqlbinlog tool with the --verbose option.

Rows tagged with last-commit epoch

Each row in a MySQL Cluster stores a hidden metadata column which contains the epoch at which a write to the row was last committed. This information is used internally by the Cluster during node recovery and other operations. The ndb_select_all tool can be used to see the epoch numbers for rows in a table by supplying the --gci or --gci64 options. Note that the per-row epoch is not a row version, as two updates to a row in reasonably quick succession will have the same commit epoch.

Epochs and eventual consistency

Reviewing epochs from the point of view of my previous posts on eventual consistency we see that :

Epochs provide an incrementing logical clock
Epochs are recorded in the Binlog, and therefore shipped to Slaves
Epoch boundaries imply happened-before relationships between events before and after them in the Binlog

The properties mean that epochs are almost perfect for monitoring conflict windows in an active-active circular replication setup, with only a few enhancements required.

I'll describe these enhancements in the next post.

Edit 23/12/11 : Added index

Speaking at Oracle UK User Group conference

2011-11-25T12:02:00.005Z

I will be speaking in the MySQL track of the UK Oracle User Group conference on 5th December in Birmingham UK. The title of the session is "Building Highly Available and Scalable, Real Time Services with MySQL Cluster" - full details here.

I'm not a regular conference attendee, never mind speaker. However I'm looking forward to meeting current and potential MySQL users, and also attending some of the talks in the MySQL and other tracks. Maybe I can learn something about RAC, or Exadata?

If you are attending and want to talk about MySQL or MySQL Cluster then please track me down and say hello.

Note that this is the first picture I have included in 3 years of posts - maybe I shouldn't wait 3 years for the next one?

Eventual Consistency - detecting conflicts

2011-10-20T01:05:00.006+01:00

In my previous posts I introduced two new conflict detection functions, NDB$EPOCH and NDB$EPOCH_TRANS without explaining how these functions actually detect conflicts? To simplify the explanation I'll initially consider two circularly replicating MySQL Servers, A and B, rather than two replicating Clusters, but the principles are the same.

Commit ordering

Avoiding conflicts requires that data is only modified on one Server at a time. This can be done by defining Master/Slave roles or Active/Passive partitions etc. Where this is not done, and data can be modified anywhere, there can be conflicts. A conflict occurs when the same data is modified at both Servers concurrently, but what does concurrently mean? On a single server, modifications to the same data are serialised by locking or MVCC mechanisms, so that there is a defined order between them. e.g. two modifications MX and MY are committed either in order {MX, MY} or {MY, MX}.

For the purposes of replication, two modifications MX and MY on the same data are concurrent if the order of commit is different at different servers in the system. Each server will choose one order, but if they don't all choose the same order then there is a conflict. Having a different order means that the last modification on each server is different, and therefore the final state of the data can be different on different servers.

One way to avoid conflicts is to get all servers to agree on a commit order before processing an operation - this ensures that all replicas process operations in the same order, waiting if necessary for missing operations to arrive to ensure no commit-order variance.

Note that commit-ordering is only important between modifications affecting the same data - modifications which do not overlap in their data footprint are unrelated and can be committed in any order. A system which totally orders commits may be less efficient than one which only orders conflicting commits.

Happened before

For the NDB$EPOCH asynchronous conflict detection functions, commit orders are monitored to detect when two modifications to the same data have been committed in different orders.

Given two modifications MX and MY to the same data, each server will decide a happened before (denoted ->) relationship between them :

MX -> MY (MX happened before MY)

or
MY -> MX (MY happened before MX)

If all servers agree on order 1, or all servers agree on order 2 then there is no conflict. If there is any disagreement then there is a conflict.

In practice, disagreement arises because the same data is modified at both Server A and Server B before the Server A modification is replicated to B and/or vice-versa.

Sometimes when reading about commit ordering, the reason why commit orders should not diverge is lost - the only reason to care about commit ordering is because it is related to conflicting modifications and the potential for data divergence.

Determining happened before from the Binlog

We assume a steady start state, where both Server A and Server B agree about the state of their data, and no modifications are in-flight. If a client of Server A then commits modification MA1 to row X, then from Server A's point of view, MA1 happened before any future modification to row X.

MA1 -> M*

If a client of Server B commits modification MB1 to row X around the same time (before, or after, or thereabouts), from Server B's point of view, MB1 happened before any future modification to row X.

MB1 -> M*

Both Servers are correct, and content with their world view. Note that in general, when committing a modification Mj, a server naturally asserts that from its point of view the modification happened before any as-yet-unseen modification Mk.

Some time will pass and the replication mechanisms will pull Binlogged changes across and apply them. When Server B pulls and applies Server A's Binlogged changes, modification MA1 will be applied to row X. Server B will then naturally be of the opinion that :

MB1 -> MA1

Independently, Server A will pull Server B's binlogged changes and apply modification MB1 to row X, and will come to the certain opinion that :

MA1 -> MB1

These happened before relationships are contradictory so there is a conflict. If nothing is done then A and B will have diverged, with Server A storing the outcome of MB1, and Server B storing the outcome of MA1.

Note that if the --log-slave-updates server option were on, then Server A's Binlog would have recorded {...MA1...MB1...}, whereas Server B's Binlog would have recorded {...MB1...MA1...}. By recording when the Slave applies replicated updates in the Binlog, we record the commit order of the replicated updates relative to other local updates, and encode the happened before relationship in the relative positions of events in the Binlog.

The Binlog is of course transferred between servers, so in a circular replication setup, Server A can become aware of the happened before information from Server B and vice-versa by examining the received Binlogs. The Slave SQL thread examines Binlogs as it applies them, so can be extended to extract happened before information, and use it to detect conflicts.

Recall that Server A asserts that its committed modification to row X (MA1) happened before any as-yet-unseen replicated modification :

MA1 -> M*

Therefore, to detect a conflict, Server A only needs to detect the case where the incoming Binlog from Server B infers that some modification MB* to row X happened before server A's already committed modification MA1.

If Server B Binlog implies MB* -> MA1 then there has been a conflict

This is in essence how the NDB$EPOCH functions work - the Binlog is used to capture happened before relationships which are checked to determine whether conflicting concurrent modifications have occurred.

Conflict Windows

In the previous example, Server A commits MA1 modifying row X, and Server B commits MB1 also modifying row X. From Server A's point of view, as soon as it commits MA1, there is potential for a replicated modification from B such as MB1 to be found in-conflict with MA1. We say that from Server A's point of view a window of potential conflict on row X has opened when MA1 was committed. Server A monitors Server B's Binlog as it is applied and when it reaches the point where the commit of MA1 at Server B is recorded, Server A knows that any further MB* recorded in Server B's Binlog after this cannot have happened before MA1, therefore the window of potential conflict on row X has closed.

We define the window of potential conflict on a row X as the time between the commit of a modification M1, and the Slave processing of an event in a replicated Binlog indicating that modification M1 has been applied on the other server(s) in the replication loop.

Any incoming replicated modification M2 also affecting row X while it has an open conflict window is in conflict with M1, as it must appear to have happened-before M1 to the server which committed it.

Observations about the window of potential conflict :

It is defined per committed modification per disjoint data set
It can be extended by further modifications to the same data from the same server
The window does not close all further modifications have been fully replicated
Window duration is dependent on the replication round-trip delay
Which can vary greatly
Once it closes, further modifications to the same data from anywhere are safe, but will each open their own window of potential conflict.
From the point of view of one Server, conflicts can occur at any time until the conflict window is closed
From the point of view of one Server, the duration of the window of potential conflict is similar to

Replication Propagation Delay A to B + Replication Propagation Delay B to A

These delays may not be symmetric.
From the point of view of an external observer/actor, the system will detect two modifications MA1 and MB1 committed at times tMA1 and tMA2 as in-conflict if

tMB1 - tMA1 < Replication Propagation Delay A to B

( A before B, but not by enough to avoid conflict )

or

tMA1 - tMB1 < Replication Propagation Delay B to A

( B before A, but not by enough to avoid conflict )
The window of potential conflict can only be as short as the replication propagation delay between systems, which can tend towards, but never reach zero.

Tracking conflict windows with a logical clock

A row's conflict window opens when a modification is committed to it, and closes when the Slave processes an event indicating that the modification was committed on the other server(s). How can we track all of these independent conflict windows? If only we had a database :)

This is solved by maintaining a per-server logical clock, which increments periodically. Each modification to a row sets a hidden metacolumn of the row to the current value of the server's logical clock. This gives each row a kind of coarse logical timestamp. When the logical clock increments, an event is included in the Binlog to record the transition. Further, all row events for modifications with logical clock value X are stored in the Binlog before any row events for modifications with logical clock value X+1.

 Server A Binlog events    ClockVal stored in DB
                         by Modification

...
MA1                       39
MA2                       39
MA3                       39
ClockVal_A = 40
MA4                       40
MA5                       40
ClockVal_A = 41
MA6                       41

When a Slave applies the Binlog, the ClockVal events are passed through into its Binlog, and are then made available to the original server in a circular configuration.

 Server B Binlog events

...
MB1
MB2
ClockVal_A = 40
MB3
MB4
ClockVal_B = 234
MB5
MB6
ClockVal_A = 41
MB7
...

Using the Binlog ordering, we can see that ClockVal_A = 40 happened before MB3 and MB4 at Server B. This implies that MA1, MA2 and MA3 happened before MB3 and MB4 at server B.

When applying Server B's Binlog to Server A, the Slave at Server A maintains a maximum replicated clock value, which increases as it observes its ClockVal_A events returned. When applying a row event originating from Server B, the affected row's stored clock value is first compared to the maximum replicated clock value to determine whether the row event from B conflicts with the latest committed change to the row at Server A.

The two modifications are in conflict if the stored row's clock value is greater than or equal to the maximum replicated clock value.

in_conflict = row_clockval >= maximum_replicated_clockval

Using a logical clock to track conflict windows has the following benefits :

Automatic update on commit of row modification, opening conflict window
Automatic extension of conflict window on further modification on row with open conflict window.
Automatic closure of conflict window on maximum replicated clock value exceeding row's stored value
Efficient storage cost per row - one clock value.
Efficient runtime processing cost - inequality comparison between maximum replicated clock value and row's stored clock value.

As you might have guessed, NDB$EPOCH uses the MySQL Cluster epoch values as a logical clock to detect conflicts. The details of this will have to wait for yet another post. In my first two posts on this subject I thought, 'one more post and I can finish describing this', but here I am at three posts and still not finished. Hopefully the next will get more concrete and finally describe the mysterious workings of NDB$EPOCH. We're getting closer, honest.

Edit 23/12/11 : Added index

Some MySQL projects I think are cool - Shard-Query

2011-10-12T14:00:00.000+01:00

I've already described Justin Swanhart's Flexviews project as something I think is cool. Since then Justin appears to have been working more on Shard-Query which I also think is cool, perhaps even more so than Flexviews.

On the page linked above, Shard-Query is described using the following statements :

"Shard-Query is a distributed parallel query engine for MySQL"

"ShardQuery is a PHP class which is intended to make working with a partitioned dataset easier"

"ParallelPipelining - MPP distributed query engines runs fragments of queries in parallel, combining the results at the end. Like map/reduce except it speaks SQL directly."

The things I like from the above description :

Distributed
Parallel
MySQL
Partitioned
Fragments
Map/Reduce
SQL

The things that scare me :

My fear of PHP is most likely groundless, based on experiences circa 1998. I suspect it runs much of the real money-earning web, and perhaps brings my scribblings to you. However, the applicability of Shard-Query seems so general, that to actually (or apparently) limit it to web-oriented use cases seems a shame. In any case I am not hipster enough to know which language would be better - OCaml? Dart? Only joking. I suppose that if the MySQL Proxy could do something along these lines then the language debate would be moot.

I am likely to fall foul of the lack-of-original-content test if I quote too much from the Shard-Query website, but the How-it-works section seems relevant here.

How it works

The query is parsed using http://code.google.com/p/php-sql-parser
A modified version of the query is executed on each shard.
The queries are executed in parallel using http://gearman.org
The results from each shard are combined together
A version of the original query is then executed over the combined results
All aggregation is done on the slaves (pushed down)
Queries with inlists can be made into parallel queries.
A callback can be used for QueryRouting. You provide a partition column, and a callback which returns information pointing to the correct shard. The most convenient way to do this is with Shard-Key-Mapper

Query rewriting rules

The core of Shard-Query are the query rewriting rules, which Justin introduces in a blog post entitled 8 substitution rules for running SQL in parallel. These transforms and substitutions allow Shard-Query to execute a user supplied query across multiple database shards. A single query (SELECT) can be mapped into a query to be applied to some, or all shards, and further queries to be used to merge the results of the per-shard queries into a final result.

Compared to a single system query, the consistency of the view that the sharded query executes against is less well defined, but this may well be acceptable for some applications.

On a single server

A single MySQL instance offers inter-query parallelism, but currently has very limited intra-query parallelism, Shard-Query can circumvent this by splitting a single query into multiple sharded sub-queries which can run in different MySQLD threads (as they are each submitted by different clients) to give intra-query parallelism. To me this seems more of a cool side effect and proof of reasonable implementation efficiency, than a real reason to use Shard-Query. Perhaps someone out there has the perfect use case for this.

Across multiple servers

The big-name MySQL users scale-out with MySQL, storing subsets of data on separate MySQL instances. Shard-Query allows SQL queries spanning all shards to be executed. This is what scaled-out MySQL has been waiting for.

I don't think it would be a good idea to run heavy traffic through Shard-Query to access a set of sharded MySQL instances yet, but Shard-Query gives a great way to perform occasional queries across all shards. This could be great for reporting and perhaps some light mining for patterns, trends etc. The ability to query across live real time data may be a real gain.

Loose coupling and availability

Scaling out via sharding standalone database servers has many difficulties, but the independence of each shard can benefit availability relative to a more tightly coupled distributed system. The loose coupling of the MySQL instances means that it's far less likely that the failure of one shard will drag others down with it, increasing system availability. Shard-Query can give the loosely coupled shards a smoother facade. The limited set of capabilities that Shard-Query gives over the set of shards may well be more than good enough. Note that 'good enough' is a recurring theme in this 'MySQL projects I think are cool' series. Often 'best' results in expensive or unnecessary compromises as a side-effect of trying to please everybody all the time.

Yet another MySQL sharded scaleout design

Looking at Shard-Query, and the recent MySQL-NoSQL Api developments, it seems like a modern MySQL sharded scaleout design might make use of :

MySQL SQL Apis (PHP, JDBC, ODBC, Ruby, Python, ....)
NoSQL access mechanisms
(HandlerSocket, Memcached(1,2))
ShardQuery for SQL reporting / analysis

Per-instance efficiency can be maximised by using the NoSQL access Apis, single-instance SQL is still available if required for the application, and a global SQL view is also available.

This combination of scalability, efficiency and SQL query-ability could be a sweet spot in the increasingly confusing multi-dimensional space of high throughput distributed databases.

Eventual consistency with transactions

2011-10-10T01:26:00.005+01:00

In my last post I described the motivation for the new NDB$EPOCH conflict detection function in MySQL Cluster. This function detects when a row has been concurrently updated on two asynchronously replicating MySQL Cluster databases, and takes steps to keep the databases in alignment.

With NDB$EPOCH, conflicts are detected and handled on a row granularity, as opposed to column granularity, as this is the granularity of the epoch metadata used to detect conflicts. Dealing with conflicts on a row-by-row basis has implications for schema and application design. The NDB$EPOCH_TRANS function extends NDB$EPOCH, giving stronger consistency guarantees and reducing the impact on applications and schemas.

Concurrency control in a single synchronous system

MySQL Cluster is a relational system. Data is stored in tables with defined schemas of typed columns. As with any relational system, real-world concepts can be modelled in a number of ways with different trade offs. One such consideration is the level of normalisation applied to a data model. Transactions and concurrency control ensure that some data modelled using multiple tables, rows and columns, appears to any external observer to move instantaneously between stable, self consistent states. This is a powerful simplification, and eases the complexity burden on application writers. Each transaction provides the illusion of serialised access to the database. Multiple transactions can execute in parallel, so long as they do not interfere by accessing the same data. Where transactions do interfere, some real serialisation can occur. In practice, applications depend on the serialisation and atomicity guarantees given by transactions, often in ways not fully made explicit or understood by the application designers.

Concurrency control in independent, asynchronously replicated systems

Asynchronously replicating writes between two independent systems erodes the guarantees given by single system concurrency control. Each system maintains its transactional guarantees in parallel, and incorporates modifications from the other system asynchronously, at some time after they were originally committed. Where the same row is modified on both systems concurrently, two versions of the same row are produced, and there is no longer a single history of values for the given row. This can cause replicas to diverge. Note that the window of 'concurrency', or 'potential conflict' is related to the time taken for a committed update to be applied on all replicas. This is similar, or equivalent to the commit delay experienced by a synchronous 2-phase commit system.

Conflicts can be detected using some form of conflict detection. On detecting a conflict, steps can then be taken to avoid divergence, and resolve any unwanted effects of the concurrent writes.

Replica divergence, external effects and cascading impacts

Divergence can be avoided if conflicting writes can be merged in some way. Some write conflicts may be equivalent, associative or otherwise mergeable, especially if the operations are replicated rather than their resulting states. However merging requires specific schema and application knowledge to determine how to merge conflicting writes.

More generally, divergence can be avoided by rejecting one or both conflicting writes. This is the approach we have taken, with handling of rejected writes delegated to the application, where the knowledge exists to handle them via the exceptions table mechanism.

However write conflicts are handled, it is important to consider :

Cascading impacts on dependent operations
Operations based on the results of conflicting operations may themselves require handling to avoid divergence.
Real world / other system effects based on conflicting writes
Maintaining database consistency does not guarantee that real world effects have been correctly compensated.

A database system does not exist in a vacuum. Operations are performed to reflect external world, or external system events. When the effects of an operation are later reverted, the real world effects may also require some compensating actions. These external world compensating actions are beyond the scope of any DBMS system and are application specific. In a real application of this technology, this is probably the most important part of the design.

Any particular conflict originates between two concurrent operations, but once a conflicting operation is committed, other operations can read its results, and commit their own, expanding the impact of the original conflict. Conflicts are discovered asynchronously, some time after the original operations are committed, so there can be a large number of subsequent operations in the replication pipeline which depend on the conflicting operations at the point they are discovered. All of the invalidated subsequent operations must be handled.

Row based conflict detection and data shearing

Using row-based conflict detection and re-alignment can counteract data divergence so that rows become consistent eventually, but this comes at the cost of eroding the atomicity of committed transactions. For example, a committed transaction which writes to three rows may, after conflict handling, have none, one, two or all three row changes reverted.

Within a single system, the two potentially visible states were :

Before transaction (All rows at version 1 : Av1, Bv1, Cv1)
After transaction (All rows at version 2 : Av2, Bv2, Cv2)

With row based conflict detection, ignoring the row variants we're actually conflicting with, there could be :

Before transaction : (All rows at version 1 : Av1, Bv1, Cv1)

After transaction

Av2, Bv2, Cv2 (All rows at version 2)
Av2, Bv2, Cv1 (Cv2 reverted)
Av2, Bv1, Cv2 (Bv2 reverted)
Av2, Bv1, Cv1 (Bv2, Cv2 reverted)
Av1, Bv2, Cv2 (Av2 reverted)
Av1, Bv2, Cv1 (Av2, Cv2 reverted)
Av1, Bv1, Cv2 (Av2, Bv2 reverted)
Av1, Bv1, Cv1 (Av2, Bv2, Cv2 reverted)

Depending on the concept that the distinct rows A,B,C represented, this can vastly increase the complexity of understanding the data. If A, B and C model entirely separate entities, which just happened to be transactionally updated together then there may be no problem if they fare differently in conflict detection. If they model portions of the state of a larger entity then reasoning about the state of that entity becomes complex.

This potential chopping up of changes committed in a transaction can be described as shearing of the data model represented by the schema. In practice, the potential for shearing between rows implies that for tables with conflicts handled on a row basis, cross row consistency is not available. This in turn implies that the schema must be modified to ensure that data items which cannot tolerate relative shear are placed in the same row so that they share the same fate and remain self-consistent. This single-row limit to consistency is native and natural to some NoSQL / key-value / wide column store products, but is a weakening of the normal guarantees in a transactional system.

Requiring that schemas and applications using conflict detection can tolerate shear between any two rows is quite a heavy burden to place on applications, especially those not written with eventual consistency in mind. Is there some way to support optimistic conflict detection without breaking up committed transactions, and shearing rows?

Transaction based conflict detection

One way to avoid inter-row shearing is to perform conflict detection on a row-by-row basis, but on discovering a conflict, take action on a transaction basis. More concretely, when a row conflict is discovered, any other rows written as part of the same transaction should also be considered in-conflict by implication. This reduces the set of stable states back to the original case - all rows at version 1 or all rows at version 2.

Where a row is found to be in-conflict with some replicated row operation, a further replicated row operation on the same row should also be found to be in-conflict, until the conflict condition has been cleared. This property is implicitly implemented in the existing row based conflict detection functions.

When the scope of a conflict is extended to include all row modifications in a transaction, this implies that all following replicated row operations which affect the same rows, must also be in conflict by implication. To avoid row shearing, these implied-in-conflict rows must implicate the other rows in their transactions, and those rows may in-turn implicate other rows. The overall effect is that a single row conflict must cause its transaction, and all dependent transactions to be considered to be in conflict.

From our database centric point of view, transactions can only become dependent on each other through the data they access in the database. If transaction X updates rows A and B, and transaction Y then reads row B and updates row C, then we can say that transaction B has a read-write dependency on transaction A via row B. We cannot tell whether there is some other out-of-band communication between transactions.

By tracking this transaction 'footprint' information, and looking for row overlaps, we can determine transaction dependencies. This is how the new NDB$EPOCH_TRANS function provides transactional conflict detection.

NDB$EPOCH_TRANS conflict detection function

The NDB$EPOCH_TRANS conflict function uses the same mechanism as the NDB$EPOCH function to detect concurrent updates to the same row across two clusters. However, once a row conflict has been detected in an operation which is part of a replicated transaction, all other operations in that replicated transaction are considered to be in conflict. Furthermore, any transactions found to be dependent on that transaction are also considered in conflict. Once the full set of in conflict transactions has been determined, the set of affected rows are handled in the same way as in NDB$EPOCH.

Specifically :

The replicated operations are not applied
The exceptions table(s) are populated with the affected primary keys
The affected row epochs are updated
Realignment Binlog events are generated to (eventually) realign the Secondary cluster

As with NDB$EPOCH, NDB$EPOCH_TRANS is asymmetric, so the Primary Cluster always wins when a conflict is detected. As with NDB$EPOCH, this allows applications needing pessimistic properties to obtain them by accessing the Primary Cluster. Applications which can handle the relaxed consistency of optimism can access either Cluster. With NDB$EPOCH_TRANS, transactions committed on the Secondary Cluster are guaranteed to be atomic, whether or not they are later found to be in conflict. Each committed transaction will either be unaffected by conflict detection, or be completely reverted. There will be no row shear.

This slightly stronger optimistic consistency guarantee may ease the implementation of relaxed consistency / eventually consistent applications. For example, where some concept is modelled by a number of different rows in different tables, any transactional modification will either be atomically applied, or not applied at all, so the relationships between the rows affected by a transaction will preserved. The need to flatten a schema into single-row entities is reduced, although careful design is still required to get a good understanding of transaction boundaries, and the behaviour of the overall system when transactions are reverted.

Transaction dependency tracking

NDB$EPOCH_TRANS is built in to the MySQL Cluster Storage Engine. It is active in the normal MySQL Slave SQL thread, as part of the normal table handler calls made when applying a replicated Binlog. The NDB$EPOCH_TRANS code in the Ndb storage engine tracks transaction dependencies based on the primary keys accessed by row events in the Binlog, and their transaction ids. If two row events have the same table and primary key values, then they affect the same row. If two events affect the same row, and are in different transactions, then the second transaction depends on the first. In this way, a transaction dependency graph is built by the MySQL Cluster Storage Engine as row events are applied by the Slave from a replicated Binlog. This graph is then used to find dependencies when a conflict is detected.

A Binlog only contains WRITE_ROW, UPDATE_ROW and DELETE_ROW events. This means that we only detect dependencies between transactions which write the same rows. We do not currently track dependencies between writers and readers. For example :

Transaction A : {Write row X, Write row Y}
Transaction B : {Read row Y, Write row Z}

Binlog : {{Tx A : Wr X, Wr Y}, {Tx B : Wr Z}}

In this example, the dependency of Transaction B on Transaction A is not recorded in the Binlog, and so the Slave is not aware of it. This would result in the write to row Z not being considered in conflict, when it should be.

A future improvement is to add selective tracking of reads to the Binlog, so that Write -> Read dependencies will implicate reading transactions when a conflict is discovered.

There's more to come

Another long dry post, best consumed with your favourite drink in hand. As I mentioned last time, these functions are pushed, and available in the latest releases of MySQL Cluster. I'd be happy to hear from anyone who wants to try them out and give feedback. I've been deliberately light with implementation details thus far, as I'm saving those for yet another posting. I think that some of the implementation details are interesting from a replication point of view, even if you're not interested in these particular conflict detection algorithms. You may disagree :)

Edit 23/12/11 : Added index

Eventual consistency with MySQL

2011-10-03T13:50:00.005+01:00

tl;dr : New 'automatic' optimistic conflict detection functions available giving the best of both optimistic and pessimistic replication on the same data

MySQL replication supports a number of topologies, and one of the most interesting is an active-active, or master-master topology, where two or more Servers accept read and write traffic, with asynchronous replication between them.

This topology has a number of attractions, including :

Potentially higher availability
Potentially low impact on read/write latency
Service availability insensitive to replication failures
Conceptually simple

However, data consistency is hard to maintain in this environment. Data, and access to it, must usually be partitioned or otherwise controlled, so that the consistency of reads is acceptable, and to avoid lost writes or badly merged concurrent writes. Implementing a distributed data access partitioning scheme which can safely handle communication failures is not simple.

Relaxed read consistency

Relaxed read consistency is a fairly well understood concept, with many Master-Slave topologies deployed where some read traffic is routed to the Slave to offload the Master and get 'read scaling'.
Generally this is acceptable as :

A Read-only Slave's state is self-consistent. It is a state which, at least logically, existed on the Master at some time in the past.
The reading application can tolerate some level of read-staleness w.r.t. the most recently committed writes to the Master

A surprisingly large number of applications can manage with a stale view as long as it is self-consistent.

Read-your-writes

Applications requiring 'read your writes' consistency (or session consistency) must either read from the Master, or wait until the Slave has replicated up to at least the point in time where the application's last write committed on the Master before reading from it. It is simpler and less delay-prone to just read from the Master, but this increases the load on the Master, reducing the ability of a system to read-scale. When the Master is unavailable, some sort of failover is required, and therefore, some sort of recovery process is also required.

Partitioned Active-Active/ Balanced Master-Slave

Rather than treating a whole replica as either Master or Slave, we can have each replica be both a Master and a Slave. The partitioning could be on database level, table level, or some function of the rows contained in tables, perhaps some key prefix which maps to application entities. Balancing the Master/Slave role in this way allows the request load to be balanced, reducing issues with a single system providing the Master 'role' having to do more work.

In this configuration, rather than talking about Master and Slave, it makes more sense to talk about some partition of data being 'Active' on one replica, and 'Backup' on the others. Read requests routed to the Active replica will be guaranteed to get the latest state, whereas the Backup replicas can potentially return stale states. Write requests should always be routed to the Active replica to avoid potential races between concurrent writes.

Implementing a partitioned replicated system like this generally requires application knowledge to choose a partitioning scheme where expected transaction footprints align with the partitioning scheme, and cross-partition transactions are rare/non-existant. Additionally, it requires application modification, or a front-end routing mechanism to ensure that requests are correctly routed. The routing system must also be designed to re-route in cases of communication or system failure, to ensure availability, and avoid data divergence. After a failure, recovery must take care to ensure replicas are resynchronised before restoring Active status to partitions in a recovered replica.

Implementing a partitioned replicated system with request routing, failover and recovery is a complex undertaking. Additionally, it can be considered a pessimistic system. For embarassingly parallel applications, with constrained behaviours, most transactions are non-overlapping in their data footprint in space and (reasonable lengths of) time. Enforced routing of requests to a primary replica adds cost and complexity that is most often unnecessary. Is it possible to take a more optimistic approach?

Optimistic Active-Active replication

An optimistic active-active replication system assumes that conflicting operations are rare, and prefers to handle conflicts after they happen, than to make conflicts impossible, by mapping them to delays or overheads all of the time. The one-time cost of recovering from a conflict after it happens may be higher than the one-time cost of preventing a conflict, but this can be a win if conflicts are rare enough.

Practically, optimistic active-active replication involves allowing transactions to execute and commit at all replicas, and asynchronously propagating their effects between replicas. When applying replicated changes to a replica, checks are made to determine whether any conflicts have occurred.

Benefits of optimism include :

Local reads - low latency, higher availability
Local writes - low latency, higher availability
No need to route requests, failover, recover
Recovery from network failure is the same as for normal async replication - catch up the backlog.

A pessimist is never disappointed, as they always expect the worst, but an optimist is occasionally (often?) disappointed. With active-active replication, this disappointment can include reading stale data, as with relaxed read consistency, or having committed writes later rejected due to a conflict. This is the price of optimism. Not all applications are suited to the slings and arrows inherent in optimism. Some prefer the safety of a pessimistic outlook.

Benefits of pessimism include :

Only durable data returned by reads
Committed writes are durable

MySQL Cluster replication has supported symmetric optimistic conflict detection functions since the 6.3 release. These provide detection of conflicts for optimistic active-active replication, allowing data to be written on any cluster, and write-write conflicts to be detected for handling. The functions use an application defined comparison value to determine when a conflict has occurred, and optionally, which change should 'win'. This is very flexible, but can be difficult to understand, and requires application and schema changes to be made use of.

When presented with an either-or decision, why not ask for both? Is it possible to have the benefits of both optimistic and pessimistic replication? Can we have them both on the same data at the same time?

Asymmetric optimistic Active-Active replication

I have recently been working on new asymmetric conflict detection functions for MySQL Cluster replication. These functions do not require schema or application modifications. They are asymmetric in that one data replica is regarded as the Active replica. However, unlike a pessimistic partitioned replicated system, writes can be made at Active or Backup replicas - they do not have to be routed to the Active replica. Writes made at the Backup replica will asynchronously propagate to the Active replica and be applied, but only if they do not conflict with writes made concurrently at the Active replica.

Having a first class Active replica and a second class Backup replica may seem like a weakness. However, it allows optimistic and pessimistic replication to be mixed, on the same data for different use-cases.

Where a pessimistic approach is required, requests can be routed to the Active replica. At the Active replica, they will be guaranteed to read durable data, and once committed, writes will not be rejected later.

Where an optimistic approach is acceptable, requests can also be routed to the Backup replica. At the Backup replica, committed writes may later be rejected, and reads may return data which will later be rejected. The potential for disappointment is there, and applications must be able to cope with that, but in return, they can read and write locally, with latency and availability independent of network conditions between replicas.

A well understood application and schema can use pessimistic replication, with request routing, where appropriate, and write-anywhere active-active where the application and schema can cope with the relaxed consistency.

New conflict functions - NDB$EPOCH, NDB$EPOCH_TRANS

The new NDB$EPOCH function implements asymmetic conflict detection, on a row basis. One replica of a table is considered Active (or Primary), and the other(s) are Backup (or Secondary). Writes originating from the Backup replica are checked at the Active replica to ensure that they don't conflict with concurrent writes originating at the Active replica. If they do conflict, then they are rejected, and the Backup is realigned to the Active replica's state. In this way, data divergence is avoided, and the replicated system eventually becomes consistent.

The conflict detection, and realignment to give eventual consistency all occur asynchronously as part of the normal MySQL replication mechanisms.

As with the existing conflict detection functions, an exceptions table can be defined which will be populated with the primary keys of rows which have experienced a conflict. This can be used to take application specific actions when a conflict is detected.

Unlike the existing conflict detection functions, no schema changes or application changes are required. However, as with any optimistic replication system, applications must be able to cope with the relaxed consistency on offer. Applications which cannot cope, can still access the data, but should route their requests to Active replicas only, as with a more traditional pessimistic system.

As these functions build on the existing MySQL Cluster asynchronous replication, the existing features are all available :

Slave batching performance optimisations
High availability - redundant replication channels
Transactional replication and progress tracking
Normal MySQL replication features : DDL replication, Binlog, replicate to other engines etc..

Ok, that's long enough for one post - I'll describe NDB$EPOCH_TRANS and its motivations in a follow-up. If you're interested in trying this out, then download the latest versions of MySQL Cluster. If you're interested in the optimistic replication concept in general, I recommend reading Saito and Shapiro's survey.

Edit 23/12/11 : Added index

Some MySQL projects I think are cool - HandlerSocket Plugin

2011-06-23T23:58:00.000+01:00

The HandlerSocket project is described in Yoshinori Matsunobu's blog entry under the title 'Using MySQL as a NoSQL - A story for exceeding 750,000 qps on a commodity server'. It's a great headline and has generated a lot of buzz. Quite a few early commentators were a little confused about what it was - a new NoSQL system using InnoDB? A cache? In memory only? Where does Memcached come in? Does it support the Memcached protocol? If not, why not? Why is it called HandlerSocket?

Inspirations from Memcache may include the focus on simplicity, performance and a simple human readable protocol. As Yoshinori says, Kazuho Oku has already implemented a MySQLD-embedded Memcached server, no need to do it again. What's more, the Memcache protocol offers key-value functionality, whereas implementing a new protocol allows more functionality to be exposed.

The choice of name has come in for some flak. I believe the etymology is that HandlerSocket exposes the existing MySQL Handler interface directly over a separate socket. Perhaps a more exciting name will appear at some point, but looking at the MySQL Handler documentation gives a good background on the basis of the HandlerSocket Api.

HandlerSocket implements more than a Key-Value Api. It supports indexed data access in general including :

Equality search on any index prefix (returning 0, 1 or more rows)
Inequality search on any index prefix (returning 0, 1 or more rows)

This allows far more general use than a simple key-value API. Composite keys can be used and Secondary indexes searched and maintained. This exposes much more of the value of a storage engine like InnoDB.

SQL and non-SQL access to the same data

Yoshinori mentions inspiration from NdbApi for HandlerSocket - they wanted to get fast indexed access performance without excluding the possibility of performing ad-hoc SQL for reports etc. This has been one of the unique benefits of Ndb for some time - the ability to operate on the same underlying data via multiple Apis. Extending this to other MySQL engines (especially InnoDB) is a great idea. What is surprising is the difference a different access layer implementation can make to throughput. Who would have thought that so much performance could be consumed by parsing etc? I suspect that HandlerSocket may create a new benchmark for the MySQL team to optimise parsing towards in future.

As a MySQL daemon plugin, the HandlerSocket plugin gets to create threads running within a MySQLD server instance. These threads can then listen on network sockets and use the Storage Engine Api inside the server to perform primitive operations on Storage engines. From the point of view of the Storage engine, they are just client request handling threads in the Server accessing data concurrently with 'normal' SQL client threads.

Using the Storage Engine Api, HandlerSocket gets :

Storage engine independence
Concurrency control as implemented by the engine
Index maintenance as implemented by the engine
Constraint enforcement as implemented by the engine
Engine features such as online backup, crash recovery, compression, encryption etc.

Not going via the 'SQL layer' means that HandlerSocket misses out on :

SQL functionality (queries, joins, aggregation, UDFs etc.)
Stored procedures
Trigger activation
Query cache
ACL checks
Views

As with NdbApi, this is often a very good trade-off as many applications don't need these features for their heavy-lifting. However they can be very useful for less frequent reporting and administration tasks. Supporting consistent access via SQL or some simpler Api to the same data can avoid the need to split caching from an OLTP database and also potentially the need to split an OLTP and analytic database. Ndb gives more of a 'firewall' between different usage types by physically separating the storage from more complex query processing, but MySQLD could be extended to have more 'workload management' features internally if this were a problem for HandlerSocket.

The impressive published benchmarks use data which fits entirely in memory buffers, so that InnoDb need only write logs and checkpoints. HandlerSocket does not require that all data is in-cache, but the best performance will be achieved if this is the case.

Operation batching/pipelining

As the protocol description states, requests can be pipelined. There is no need for individual clients to make blocking synchronous DB requests. This can vastly increase throughput without excessively straining connection resources.

Often SQL DB access protocols do not make much use of the potential for batching requests on a single connection. JDBC supports batched updates and inserts, and appears to have support (not often talked about) for batched queries, but not every driver implements these and they're often not implemented in the same way. Often the SQL approach seems to be 'If you want different sets of data in one round trip, you need to find a way to get them all in one SQL statement'. This creates a false tension between reducing client-server round trips and avoiding complex joins and unnecessary unions. Decoupling query boundaries from thread blocking / flow of control changes removes this artificial tension and can simplify applications. A recent 'Facebook at Scale' talk describes how they use the MySQL Client's CLIENT_MULTI_STATEMENTS flag to decouple query boundaries and request latency. This pattern is one of the keys to implementing efficient NdbApi clients as well.

Commit grouping

Another HandlerSocket feature possibly inspired by Ndb is the commit grouping. Multiple user writes (it's not clear if/how transaction boundaries can be specified), are combined and committed to the engine together. This amortizes the commit cost across multiple operations. Where the engine performs expensive durability operations (e.g fsync) this can improve write throughput. Writes are group-committed to the Binlog as well, again in a similar way to Ndb. Another shared advantage of merging multiple client 'transactions' into fewer Binlog 'transactions' is that the Slave also gets to benefit from fewer, larger transactions for a given data change rate.

Impact

It's not clear yet what the impact of HandlerSocket will be. It enters a crowded market of SQL, NoSQL, NewSQL, DataGrid and other technologies. Competitors use JSON over HTTP, or Memcached protocols, support richer or simpler Apis, offer transparent sharding, eventually consistent replication or web-scale Erlang distributed Map Reduce. Perhaps HandlerSocket is too old-school?

I think it's a great technology, deserving of success. Coupled with an external sharding layer, it seems to offer a great way to improve the efficiency of MySQL scale out, without losing the ability to perform ad-hoc SQL etc. Time will tell.

Journey upriver to the dark heart of ha_ndbcluster

2011-04-02T01:05:00.007+01:00

Unlike most other MySQL storage engines, Ndb does not perform all of its work in the MySQLD process. The Ndb table handler maps Storage Engine Api calls onto NdbApi calls, which eventually result in communication with data nodes. In terms of layers, we have SQL -> Handler Api -> NdbApi -> Communication. At each of these layer boundaries, the mapping between operations at the upper layer to operations at the lower layer is non trivial, based on runtime state, statistics, optimisations etc.

The MySQL status variables can be used to understand the behaviour of the MySQL Server in terms of user commands processed, and also how these map to some of the Storage Engine Handler Api calls.

Status variables tracking user commands start with 'Com_'

mysql> show status like 'Com\_%';
+---------------------------+-------+
| Variable_name | Value |
+---------------------------+-------+
| Com_admin_commands | 0 |
| Com_assign_to_keycache | 0 |
| Com_alter_db | 0 |
| Com_alter_db_upgrade | 0 |
| Com_alter_event | 0 |
| Com_alter_function | 0 |
| Com_alter_procedure | 0 |
| Com_alter_server | 0 |
| Com_alter_table | 0 |
| Com_alter_tablespace | 0 |
| Com_analyze | 0 |
| Com_backup_table | 0 |
| Com_begin | 0 |
| Com_binlog | 0 |
| Com_call_procedure | 0 |
| Com_change_db | 1 |
| Com_change_master | 0 |
| Com_check | 0 |
| Com_checksum | 0 |
.........
| Com_stmt_reset | 0 |
| Com_stmt_send_long_data | 0 |
| Com_truncate | 0 |
| Com_uninstall_plugin | 0 |
| Com_unlock_tables | 0 |
| Com_update | 1 |
| Com_update_multi | 0 |
| Com_xa_commit | 0 |
| Com_xa_end | 0 |
| Com_xa_prepare | 0 |
| Com_xa_recover | 0 |
| Com_xa_rollback | 0 |
| Com_xa_start | 0 |
+---------------------------+-------+
144 rows in set (0.01 sec)

mysql>

Status variables tracking Handler (Storage engine) Api calls start with 'Handler_'.

mysql> show status like 'Handler\_%';
+----------------------------+-------+
| Variable_name | Value |
+----------------------------+-------+
| Handler_commit | 1 |
| Handler_delete | 0 |
| Handler_discover | 0 |
| Handler_prepare | 0 |
| Handler_read_first | 0 |
| Handler_read_key | 0 |
| Handler_read_next | 0 |
| Handler_read_prev | 0 |
| Handler_read_rnd | 0 |
| Handler_read_rnd_next | 21 |
| Handler_rollback | 0 |
| Handler_savepoint | 0 |
| Handler_savepoint_rollback | 0 |
| Handler_update | 4 |
| Handler_write | 14 |
+----------------------------+-------+
15 rows in set (0.00 sec)

mysql>

The 'Com_%' and 'Handler_%' variables are maintained by the Server for all storage engines. The server maintains these on a per-session, and global basis. By default the session status is shown, but the GLOBAL keyword shows the global view, aggregated across all sessions.

mysql> show global status like 'Handler\_%';
+----------------------------+--------+
| Variable_name | Value |
+----------------------------+--------+
| Handler_commit | 167 |
| Handler_delete | 494041 |
| Handler_discover | 0 |
| Handler_prepare | 0 |
| Handler_read_first | 3 |
| Handler_read_key | 1 |
| Handler_read_next | 0 |
| Handler_read_prev | 0 |
| Handler_read_rnd | 0 |
| Handler_read_rnd_next | 561132 |
| Handler_rollback | 6 |
| Handler_savepoint | 0 |
| Handler_savepoint_rollback | 0 |
| Handler_update | 24 |
| Handler_write | 43442 |
+----------------------------+--------+
15 rows in set (0.00 sec)

mysql>

The SHOW STATUS command is handy for a quick check, but for more interesting analysis, the INFORMATION_SCHEMA tables SESSION_STATUS and GLOBAL_STATUS contain the same data, and support all SQL queries and views.

mysql> select * from information_schema.session_status where Variable_name like 'Handler\_%';
+----------------------------+----------------+
| VARIABLE_NAME | VARIABLE_VALUE |
+----------------------------+----------------+
| HANDLER_COMMIT | 1 |
| HANDLER_DELETE | 0 |
| HANDLER_DISCOVER | 0 |
| HANDLER_PREPARE | 0 |
| HANDLER_READ_FIRST | 0 |
| HANDLER_READ_KEY | 0 |
| HANDLER_READ_NEXT | 0 |
| HANDLER_READ_PREV | 0 |
| HANDLER_READ_RND | 0 |
| HANDLER_READ_RND_NEXT | 85 |
| HANDLER_ROLLBACK | 0 |
| HANDLER_SAVEPOINT | 0 |
| HANDLER_SAVEPOINT_ROLLBACK | 0 |
| HANDLER_UPDATE | 4 |
| HANDLER_WRITE | 89 |
+----------------------------+----------------+
15 rows in set (0.40 sec)

mysql>

Unfortunately the entries in the information_schema are in shouty 1960s CAPITALS. Also note that some Handler calls are made as part of using the information_schema 'database' to fetch the data. This is Heisenberg's principle in action!

To shine some light into the depths of the Ndb storage engine, a set of new status variables has been added to recent cluster-7.0 and cluster-7.1 releases. These status variables track activity at the NdbApi and data node communication layers of the stack. Currently they are divided into 4 subsets :

Global counters
show status like 'ndb_api_%_count';
Session counters
show status like 'ndb_api_%_session';
Slave counters
show status like 'ndb_api_%_slave';
Binlog injector counters
show status like 'ndb_api_%_injector';

Unfortunately the mysql-5.1 server does not allow Storage Engines to differentiate GLOBAL or SESSION status variables, so the Global and Session specific versions of these variables are differentiated by name, and are visible from both GLOBAL and SESSION views of status, so it doesn't matter which you look at.

Global counters
mysql> show status like 'ndb_api_%_count';
+------------------------------------+------------+
| Variable_name | Value |
+------------------------------------+------------+
| Ndb_api_wait_exec_complete_count | 10 |
| Ndb_api_wait_scan_result_count | 19295 |
| Ndb_api_wait_meta_request_count | 67 |
| Ndb_api_wait_nanos_count | 6361966340 |
| Ndb_api_bytes_sent_count | 415704 |
| Ndb_api_bytes_received_count | 116921552 |
| Ndb_api_trans_start_count | 14 |
| Ndb_api_trans_commit_count | 3 |
| Ndb_api_trans_abort_count | 1 |
| Ndb_api_trans_close_count | 14 |
| Ndb_api_pk_op_count | 6 |
| Ndb_api_uk_op_count | 0 |
| Ndb_api_table_scan_count | 11 |
| Ndb_api_range_scan_count | 0 |
| Ndb_api_pruned_scan_count | 0 |
| Ndb_api_scan_batch_count | 25850 |
| Ndb_api_read_row_count | 103371 |
| Ndb_api_trans_local_read_row_count | 51625 |
| Ndb_api_event_data_count | 0 |
| Ndb_api_event_nondata_count | 0 |
| Ndb_api_event_bytes_count | 0 |
+------------------------------------+------------+
21 rows in set (0.00 sec)

mysql>

These counts are aggregated across all MySQL clients in the Server accessing tables in Ndb, since this MySQL Server was started. Note that this *does not* include accesses by clients of other MySQL Servers, or other NdbApi clients.

Session counters
mysql> show status like 'ndb_api_%_session';
+--------------------------------------------+----------+
| Variable_name | Value |
+--------------------------------------------+----------+
| Ndb_api_wait_exec_complete_count_session | 0 |
| Ndb_api_wait_scan_result_count_session | 38 |
| Ndb_api_wait_meta_request_count_session | 2 |
| Ndb_api_wait_nanos_count_session | 11064398 |
| Ndb_api_bytes_sent_count_session | 872 |
| Ndb_api_bytes_received_count_session | 230764 |
| Ndb_api_trans_start_count_session | 1 |
| Ndb_api_trans_commit_count_session | 0 |
| Ndb_api_trans_abort_count_session | 0 |
| Ndb_api_trans_close_count_session | 1 |
| Ndb_api_pk_op_count_session | 0 |
| Ndb_api_uk_op_count_session | 0 |
| Ndb_api_table_scan_count_session | 1 |
| Ndb_api_range_scan_count_session | 0 |
| Ndb_api_pruned_scan_count_session | 0 |
| Ndb_api_scan_batch_count_session | 51 |
| Ndb_api_read_row_count_session | 204 |
| Ndb_api_trans_local_read_row_count_session | 136 |
+--------------------------------------------+----------+
18 rows in set (0.00 sec)

mysql>

These counts are for the current session (MySQL client connection)

Slave counters
mysql> show status like 'ndb_api_%_slave';
+------------------------------------------+-------+
| Variable_name | Value |
+------------------------------------------+-------+
| Ndb_api_wait_exec_complete_count_slave | 0 |
| Ndb_api_wait_scan_result_count_slave | 0 |
| Ndb_api_wait_meta_request_count_slave | 0 |
| Ndb_api_wait_nanos_count_slave | 0 |
| Ndb_api_bytes_sent_count_slave | 0 |
| Ndb_api_bytes_received_count_slave | 0 |
| Ndb_api_trans_start_count_slave | 0 |
| Ndb_api_trans_commit_count_slave | 0 |
| Ndb_api_trans_abort_count_slave | 0 |
| Ndb_api_trans_close_count_slave | 0 |
| Ndb_api_pk_op_count_slave | 0 |
| Ndb_api_uk_op_count_slave | 0 |
| Ndb_api_table_scan_count_slave | 0 |
| Ndb_api_range_scan_count_slave | 0 |
| Ndb_api_pruned_scan_count_slave | 0 |
| Ndb_api_scan_batch_count_slave | 0 |
| Ndb_api_read_row_count_slave | 0 |
| Ndb_api_trans_local_read_row_count_slave | 0 |
+------------------------------------------+-------+
18 rows in set (0.00 sec)

mysql>

Hopefully you are seeing a pattern here. These counters are the NdbApi operations performed by the Slave SQL thread as part of replicating Binlogs into Ndb tables. These counts will only increase from zero if the MySQLD is acting, or has acted as a Slave, and has accessed tables stored in Ndb.

Binlog Injector counters
mysql> show status like 'ndb_api_%_injector';
+--------------------------------------+-------+
| Variable_name | Value |
+--------------------------------------+-------+
| Ndb_api_event_data_count_injector | 0 |
| Ndb_api_event_nondata_count_injector | 0 |
| Ndb_api_event_bytes_count_injector | 0 |
+--------------------------------------+-------+
3 rows in set (0.01 sec)

mysql>

These counts track the data change events received by the Ndb Binlog Injector thread. The Binlog Injector is responsible for recording Cluster changes in the Binlog, but even when Binlogs are not being written, it receives events related to schema changes and other system management functions, so these counts can be non zero on Servers which are not writing a Binlog.

Counter definitions
As you've hopefully noticed, there is naming overlap between each set of status counters. The same events are being counted, and recorded globally, per-session, against the Slave SQL thread and against the Ndb Binlog injector thread.

So what do these different counts actually mean?

Ndb_api_wait_exec_complete_count[_session|_slave]

The number of times a user thread has blocked waiting for some batch of primary key, or secondary unique hash key operations to complete. From the point of view of the MySQL Server, this is idle time, waiting for data nodes to send some response. An alternative name for it could be 'round trip count', and minimising it through operation batching is a good way to reduce response time and increase throughput.

Ndb_api_wait_scan_result_count[_session|_slave]

The number of times a user thread has blocked waiting for some scan operation to complete. It could be waiting for a batch of scan results, or waiting for an acknowledgement of a scan close. In any case, it indicates time spent waiting on communication related to scan processing.

Ndb_api_wait_meta_request_count[_session|_slave]

The number of times a user thread has blocked waiting for some metadata operation to complete. This is quite a catch-all term, which can include DDL (Create/Drop table etc), and some transaction initialisation steps.

Ndb_api_wait_nanos_count[_session|_slave]

The number of nanoseconds a user thread has blocked in one of the three scenarios above. This is kind of an 'IO_WAIT' time for Ndb operations. It tracks how long the thread was blocked waiting for the data nodes to complete their operations. The resolution is nanoseconds, but this requires support from the operating system. On operating systems with lower resolution, this count will be coarser, and some operations may complete with zero observed wait time.

Ndb_api_bytes_sent_count[_session|_slave]

The number of bytes sent to the Ndb data nodes. This includes all request types, rows inserted etc. It does not include regular heartbeating as that generally adds too much noise to make the counters useful.

Ndb_api_bytes_received_count[_session|_slave]

The number of bytes received from the Ndb data nodes. This includes all request types, rows read etc. It does not include regular heartbeating as that generally adds too much noise to make the counters useful.

Ndb_api_trans_start_count[_session|_slave]

The number of NdbApi transactions started. Note that NdbApi transactions are not always immediately on a BEGIN statement as an optimisation.

Ndb_api_trans_commit_count[_session|_slave]

The NdbApi transactions which have been explicitly committed. Not all transactions started are committed or aborted, some are started, and then closed.

Ndb_api_trans_abort_count[_session|_slave]

The NdbApi transactions which have been explicitly aborted.

Ndb_api_trans_close_count[_session|_slave]

The NdbApi transactions which have been closed. It should closely track the number which have been started.

Ndb_api_pk_op_count[_session|_slave]

The number of Primary Key (pk) operations which have been executed. Eack pk operation affects zero or one rows. This includes read, insert, update, write, delete. Note that operations on tables with Blobs can also generate pk and uk operations.

Ndb_api_uk_op_count[_session|_slave]

The number of Unique key (uk) operations which have been executed. Each uk operation affects zero or one rows. This includes read, update, write, delete by unique key.

Ndb_api_table_scan_count[_session|_slave]

The number of table scans which have been started. Table scans can have pushed-down filters, so although they must access all data in a table, they might not return it all to the MySQL Server. Also, started scans may be stopped before all table data is accessed.

Ndb_api_range_scan_count[_session|_slave]

The number of range (Ordered Index) scans which have been started. Range scans take bounds and pushed-down filters, so may access and/or return zero to all rows to the MySQL Server.

Ndb_api_pruned_scan_count[_session|_slave]

The number of scans which have been successfully pruned to one partition of the scanned table/index. See my previous Blog entries for details about scan partition pruning.

Ndb_api_scan_batch_count[_session|_slave]

The number of batches of rows returned to the MySQL Server from scans. Each scanned table/index fragment returns matching rows in batches, whose size is controlled by the batchsize parameters. NdbApi handles one batch from each fragment at a time. Fetching the next batch from a fragment requires a round-trip to the data nodes, although multiple fragments can be asked for their next batch in one trip. Minimising the Ndb_api_scan_batch_count, by increasing batchsize and improving scan selectivity can improve throughput, latency and efficiency.

Ndb_api_read_row_count[_session|_slave]

The number of rows returned to the MySQL Server from Primary Key read, Unique Key reads and Table and Index scans.

Ndb_api_trans_local_read_row_count[_session|_slave]

The number of rows returned to the MySQL Server from the same data node where the reading transaction has its Transaction Coordinator (TC). Where the TC and the data read reside on the same data node, one hop in the data reading control protocols is avoided. This is the goal of transaction hinting and distribution awareness, and its effectiveness can be checked by comparing Ndb_api_trans_local_read_row_count to Ndb_api_trans_read_row_count. The higher the proportion of local reads, the better.

Ndb_api_event_data_count[_injector]

The number of data change events (row insert, delete, update notifications) received from the data nodes. On a Binlogging MySQL Server, this count can give a measure of the rate of data change in a cluster in terms of rows/second.

Ndb_api_event_nondata_count[_injector]

The number of non-data events (table alter/drop notifications etc) received from the data nodes.

Ndb_api_event_bytes_count[_injector]

The total number of bytes of event data (data and nondata events) received from the data nodes. This gives another measure of the Cluster change rate in terms of bytes/second.

At NdbApi level
As the names suggest, all of these counters are reflecting things happening in the NdbApi implementation. The data collection is built into NdbApi, and is therefore also available to any NdbApi client, not just the MySQL Server.

Using the counters
During implementation of these counters, I found it easiest to create a temporary table to store 'base' values for the counters, and define a view containing the difference between the current values and the base values. This made it easy to see the effect on the counters of various different SQL statements, slave operations etc.

Setting up a baseline and a view
Please excuse my schoolboy SQL :

mysql> create table test.counts_base(variable_name varchar(255) primary key, variable_value bigint);
Query OK, 0 rows affected (0.04 sec)

mysql> replace into test.counts_base (variable_name, variable_value) select * from information_schema.session_status where variable_name like 'ndb_api%';
Query OK, 60 rows affected (0.01 sec)
Records: 60 Duplicates: 0 Warnings: 0

mysql> create view test.counts_diff as select test.counts_base.variable_name, information_schema.session_status.variable_value - test.counts_base.variable_value as diff from test.counts_base, information_schema.session_status where test.counts_base.variable_name = information_schema.session_status.variable_name and (information_schema.session_status.variable_value - test.counts_base.variable_value) > 0;
Query OK, 0 rows affected (0.05 sec)

mysql> select * from test.counts_diff where variable_name like 'ndb_api%_session';
Empty set (0.05 sec)

mysql>

Looking at the effects of a SQL statement
First check that the baseline is ok :

mysql> select * from test.counts_diff where variable_name like 'ndb_api%_session';
Empty set (0.05 sec)

Now run the statement :

mysql> select count(1) from demo_table;
+----------+
| count(1) |
+----------+
| 103343 |
+----------+
1 row in set (0.00 sec)

Now look at the difference :

mysql> select * from test.counts_diff where variable_name like 'ndb_api%_session';
+--------------------------------------------+--------+
| variable_name | diff |
+--------------------------------------------+--------+
| NDB_API_WAIT_SCAN_RESULT_COUNT_SESSION | 4 |
| NDB_API_WAIT_META_REQUEST_COUNT_SESSION | 1 |
| NDB_API_WAIT_NANOS_COUNT_SESSION | 619143 |
| NDB_API_BYTES_SENT_COUNT_SESSION | 116 |
| NDB_API_BYTES_RECEIVED_COUNT_SESSION | 268 |
| NDB_API_TRANS_START_COUNT_SESSION | 1 |
| NDB_API_TRANS_CLOSE_COUNT_SESSION | 1 |
| NDB_API_TABLE_SCAN_COUNT_SESSION | 1 |
| NDB_API_SCAN_BATCH_COUNT_SESSION | 2 |
| NDB_API_READ_ROW_COUNT_SESSION | 2 |
| NDB_API_TRANS_LOCAL_READ_ROW_COUNT_SESSION | 1 |
+--------------------------------------------+--------+
11 rows in set (0.05 sec)

mysql>

The select count(1) statement blocked 4 times on scan results, sent 116 bytes and received 268 bytes of data. Two batches of rows were received, and two rows were received in total. One of these rows was from the same node as the transaction's transaction coordinator.
This indicates that select count(1) is optimised in the Ndb handler !

Let's try a more tricky select count. First we must reset the baseline to get a 'clean' difference.

mysql> replace into test.counts_base (variable_name, variable_value) select * from information_schema.session_status where variable_name like 'ndb_api%';
Query OK, 120 rows affected (0.01 sec)
Records: 60 Duplicates: 60 Warnings: 0

mysql> select * from test.counts_diff where variable_name like 'ndb_api%_session';
Empty set (0.05 sec)

mysql> select count(length(string_value) > 10) from demo_table;
+----------------------------------+
| count(length(string_value) > 10) |
+----------------------------------+
| 103343 |
+----------------------------------+
1 row in set (6.77 sec)

mysql> select * from test.counts_diff where variable_name like 'ndb_api%_session';
+--------------------------------------------+------------+
| variable_name | diff |
+--------------------------------------------+------------+
| NDB_API_WAIT_SCAN_RESULT_COUNT_SESSION | 21139 |
| NDB_API_WAIT_NANOS_COUNT_SESSION | 5322402052 |
| NDB_API_BYTES_SENT_COUNT_SESSION | 441628 |
| NDB_API_BYTES_RECEIVED_COUNT_SESSION | 109026900 |
| NDB_API_TRANS_START_COUNT_SESSION | 1 |
| NDB_API_TRANS_CLOSE_COUNT_SESSION | 1 |
| NDB_API_TABLE_SCAN_COUNT_SESSION | 1 |
| NDB_API_SCAN_BATCH_COUNT_SESSION | 25836 |
| NDB_API_READ_ROW_COUNT_SESSION | 103343 |
| NDB_API_TRANS_LOCAL_READ_ROW_COUNT_SESSION | 51735 |
+--------------------------------------------+------------+
10 rows in set (0.05 sec)

mysql>

This select count waited for scan results ~21 thousand times, sent ~440kB of data, and received ~109MB of data from the data nodes. 103,343 rows were read in ~25 thousand scan batches, and roughly half came from the same node as the tranaction coordinator. 5.3 seconds of the 6.77 second runtime were spent waiting for responses from the data nodes.

Other uses
Hopefully this gives some notion of the possibilities with these new counters. Some other ideas :

Debug slow queries
Optimise data distribution and table partitioning
Get real post-execution costs for queries, DML etc.
Understand how data transfer for Blobs is batched
Check bulk insert/deletes are functioning efficiently
Verify Ndb slave batching is in operation
Draw cool graphs

MySQL Cluster online scaling

2011-03-28T00:41:00.003+01:00

Most people looking at a diagram showing the Cluster architecture soon want to know if the system can scale online. Api nodes such as MySQLD processes can be added online, and the storage capacity of existing data nodes can be increased online, but it was not always possible to add new data nodes to the cluster without an initial system restart requiring a backup and restore.

An online add node and data repartitioning feature was finally implemented in MySQL Cluster 7.0. It's not clear how often users actually do scale their Clusters online, but it certainly is a cool thing to be able to do.

There are two parts to the feature :

Online add an empty data node to an existing cluster
Online rebalance existing data across the existing and new data nodes

Adding an empty data node to a cluster sounds trivial, but is actually fairly complex given the cluster's distributed configuration, ring heartbeating etc. Stewart Smith did some preparatory work on this a few years ago, and this was revisited for the feature.

Rebalancing existing table data to make use of the new storage capacity is more challenging. How does this work? More importantly, how does this work online, while transactions are starting and committing, and queries are running?

As an aside, the definition of 'Online' used here is that multiple distributed clients continue to start and commit transactions reading and writing data to the cluster. Some things, like concurrent DDL may be blocked during these operations.

To understand the online data rebalancing mechanism, we need to go into more detail on the native data distribution mechanisms introduced in the last post.

Ndb native partitioning variants

There are currently three variants of Ndb's native partitioning function :

Linear Key
Key
HashMap

Linear Key is used where the table is created with PARTITION BY LINEAR KEY. Key is used where the table is created with PARTITION BY KEY *prior to Cluster 7.0. HashMap is used where the table is created with PARTITION BY KEY in releases starting at Cluster 7.0.

These partitioning functions can be decomposed into two functions, where the first, an MD5 hash of the partition/distribution key, is fixed. MD5 is no longer considered secure, but for the purpose of 'balancing' rows across fragments it is more than adequate.

The variant part of these algorithms is how the MD5 hash of the key is mapped to a partition/fragment number. This has important implications for how repartitioning works.

fragment_num( dist_key_cols ) = mapping_fn( md5 ( dist_key_cols ) )

Linear Key Mapping Function

The Linear Key scheme is part of an old design for online repartitioning. The old design was intended to minimise the amount of data transfer required when repartitioning tables by ensuring that an existing partition could be cleanly 'split in two' so that half of its data could be migrated. This is a good policy, but Linear Key had the downside of requiring a power-of-2 number of partitions.

Where a non power-of-2 number of data nodes existed, this causes problems. Additionally, if repartitioning with this scheme had ever been implemented, it would have required fragments to be 'split in two' to expand the system. The power-of-2 requirement of this scheme was listed as a Cluster limitation in the early days, though this has not been a problem since the non-linear Key scheme became default.

Key Mapping Function

The Key scheme is simpler than Linear Key, and simply divides the hash result modulo the number of fragments of the table to determine which fragment a row resides in. This removes the power-of-2 restriction that Linear Key required, so rows can be evenly balanced across any number of nodes. However, it is not amenable to online reorganisation, as changing the number of table fragments changes the modulo division value, which can result in most of the resulting partition values changing. This means that a reorganisation using this scheme could result in excessive data transfer.

e.g.

mapping_fn_key(x) = x % num_frags

md5(dist_key_cols) = 23

num_frags = 4, 23 % 4 = 3
num_frags = 6, 23 % 6 = 5
num_frags = 8, 23 % 8 = 7

As the system expands, too much data is being moved. This is expensive, slow, requires extra storage etc.

HashMap Mapping Function

In MySQL Cluster 7.0, the HashMap distribution scheme was added and became the default. It is used when PARTITION BY KEY() is explicitly given, or implied if no partitioning specification is given.
The HashMap scheme uses a mapping table from md5 hash result to fragment number. The hash result used is 32-bits, which would require a large lookup table, so we first shrink it down to something more manageable (n) by modulo division by n. n = 240 is the default number, though the implementation supports any modulo value.
The resulting number is then used to lookup a table to get the fragment id which will store the row.

e.g.

mapping_fn_hashmap(x) = lookup_tab [x % mod_val ]

fragment_number = lookup [ md5( dist_key_cols ) % mod_val ]

fragment_number = lookup [ md5( dist_key_cols ) % 240 ]

The lookup table adds another layer of indirection between the hash result (which is fixed for any given key), and the fragment number, whose range can increase over time.

Assuming mod_val is 240, and we start with 4 fragments, then the 240 entry lookup table will have 60 entries with 0, 60 with 1, 60 with 2 and 60 with 3. As an aside, these will be sequenced as 0,1,2,3,0,1,2,3,0.... so that the actual default distribution will be exactly the same as with the KEY scheme.

If we want to spread the table data over 6 fragments, then we can change the table to use a new hashmap lookup table, where 2/6 of the existing 240 values are changed to refer to the new fragment numbers. The other 4/6 are unchanged. Expanding again to 8 fragments, we can change to another new hashmap, where 2/8 of the existing 240 values are changed, and the other 6/8 are unchanged. In each case, the minimum amount of data is affected to maintain balance.

Changing the hashmap is easy, the real work is in moving the data while it's being operated upon, but what the hashmap gives is a way to move only the minimum amount of data required when adding nodes. Only the data that has to move is moved, the rest stays where it is. The data distribution randomisation given by the MD5 function is unaffected, so system balance is maintained.

Choice of HashMap mod_val

As the default mod_val of 240 is significantly higher than common fragment counts, and because it factors well, most configurations will remain well balanced, despite being reorganised.

e.g. Assuming 2-node increments, a minimum with NoOfReplicas=2

240 / 2 = 120
240 / 4 = 60
240 / 6 = 40
240 / 8 = 30
240 / 10 = 24
240 / 12 = 20
240 / 14 = 17.1
240 / 16 = 15
240 / 18 = 13.3
240 / 20 = 12
240 / 22 = 10.9
240 / 24 = 10

240 factors cleanly into whole numbers (of lookup table entries) meaning that data should be well balanced across the table fragments when the data is repartitioned. Where there is not an integer result (e.g. 240/14), we would have most partitions with 17 lookup entries, and two with 18 lookup entries. The imbalance between them would be (18/17)-1 = 6%. If this were problematic, then a different mod_val could be used. A higher mod_val gives smaller partition imbalances, but requires
more memory to store. If necessary, the lookup table could be expanded in size by any integer factor (e.g. 2,3,4..) online to make it large enough to factor better for some desired data node count.

Moving rows online

The HashMap gives fine grained control of data placement, but how does the reorg happen online?

Table Reorganisation is similar Node recovery in some ways, in that the data is copied via fragment scans of the existing fragments, while at the same time, synchronous triggers are used to forward changes made to the existing fragment rows. The triggers and scans only copy data for rows which are to be moved,

e.g. where

new_hashmap_lookup[ md5( dist_key_cols ) % 240 ]
!=
old_hashmap_lookup[ md5( dist_key_cols ) % 240 ]

Other rows are left where they are.

With this mechanism, the new fragments are populated with rows from the existing fragments while read and write transactions continue. Once the fragment scans complete, the new fragments continue to be maintained by the synchronous triggers.

A future GCP boundary is chosed to be the 'cutover' point, and at this GCI, the new HashMap starts getting used for new transaction processing, and the new fragments start being used. Triggers are setup to propagate changes from the new fragments back to the pre-existing fragments, so that any older transactions using the old hashmap definition will see consistent data changes.

Once all transactions using rows from the pre-existing fragments have committed, the synchronous triggers are dropped, and the pre-existing fragments are scanned again, deleting the moved rows. Once this step completes, the reorganisation
is done.

Primary and Unique key operations in Ndb are short lived, and at Hashmap cutover, it doesn't take long until all old operations have committed. However, ordered index and table scans are slower and may not complete for some time. Both old and new row copies are maintained until all scans started using the old distribution have completed, so that ongoing transactions need not be aborted as part of the online reorg.

At the same time as the hashmaps are cutover at a GCI boundary, any NdbApi event subscribers listening to data change events on the table, for example attached MySQLDs recording Binlogs, start receiving events for the moved rows from the new fragments, and stop receiving them from the old fragments.

Transient storage use

When adding data nodes, the reorganisation uses no extra storage space on existing data nodes. On new data nodes, only the space used for the moved data is used. After the reorganisation completes, the space formerly used on pre-existing data nodes can be used for new data, so the system capacity is increased.

Transactional behaviour

Table reorganisation can take some time when there is a lot of data to move. A node or cluster failure during a reorganisation could leave the system in a transient state which would be difficult to recover from. One of the internal infrastructure changes in Cluster 7.0 was making all DDL operations transactional. This means
that they are atomic w.r.t. failures, including node and system failures. This applies to CREATE/DROP/ALTER of TABLE/INDEX/TABLESPACE etc.
This also applies to table reorganisation as it is a form of ALTER TABLE. If the reorganisation fails, or a node fails, or the cluster fails at some point during the reorganisation, then as part of system recovery, the reorganisation will be rolled back, or completed, if it had committed at the time of failure.

So that covers online table reorganisation. I've been meaning to write about it for some time, though somehow these entries always seem to be more like adverts than technical info.

Data distribution in MySQL Cluster

2011-03-26T00:43:00.004Z

MySQL Cluster distributes rows amongst the data nodes in a cluster, and also provides data replication. How does this work? What are the trade offs?

Table fragments

Tables are 'horizontally fragmented' into table fragments each containing a disjoint subset of the rows of the table. The union of rows in all table fragments is the set of rows in the table. Rows are always identified by their primary key. Tables with no primary key are given a hidden primary key by MySQLD.

By default, one table fragment is created for each data node in the cluster at the time the table is created.

Node groups and Fragment replicas

The data nodes in a cluster are logically divided into Node groups. The size of each Node group is controlled by the NoOfReplicas parameter. All data nodes in a Node group store the same data. In other words, where the NoOfReplicas parameter is two or greater, each table fragment has a number of replicas, stored on multiple separate data nodes in the same nodegroup for availability.

One replica of each fragment is considered primary, and the other(s) are considered backup replicas. Normally, each node contains a mix of primary and backup fragments for every table, which encourages system balance.

Which replica to use?

The primary fragment replica is used to serialise locking between transactions concurrently accessing the same row. Write operations update all fragment replicas synchronously, ensuring no committed data loss on node failure. Read operations normally access the primary fragment replica, ensuring consistency. Reads with a special lock mode can access the backup fragment replicas.

Primary key read protocol

When an NdbApi client (for example a MySQLD process) wants to read a row by primary key, it sends a read request to a data node acting as a Transaction Coordinator (TC).
The TC node will determine which fragment the row would be stored in from the primary key, decide which replica to access (usually the primary), and send a read request to the data node containing that fragment replica. The data node containing the fragment replica then sends the row's data (if present) directly back to the requesting NdbApi client, and also sends a read acknowledgement or failure notification back to the TC node, which also propagates it back to the NdbApi client.

Minimising inter data node hops

The 'critical path' for this protocol in terms of potential inter-data-node hops is four hops :

Client -> TC -> Fragment -> TC -> Client

To minimise remote client experienced latency, ideally two inter-node hops can be avoided by having the TC node and the Fragment replica(s) on the same node. This requires controlling the choice of node for TC based on the primary key of the data which will be read. Where a transaction only reads rows stored on the same node as its TC, this can improve latency and system efficiency.

Distribution awareness

From NdbApi, users can specify a table and key when starting a transaction. The transaction will then choose a TC data node based on where the corresponding row's primary fragment replica is located in the system. This mechanism is sometimes referred to as 'transaction hinting'.

The Ndb handler in MySQLD generally waits for the first primary key lookup in a user session before starting an NdbApi transaction, so that it can choose a TC node based on this. This is a best-effort attempt at having the data node acting as TC colocated with the accessed data. This feature is usually referred to as 'Distribution Awareness'.

Write operations also benefit from distribution awareness, but not to the same extent in systems with NoOfReplicas > 1. Write operations must update all fragment replicas, which must be stored on different nodes, in the same nodegroup, so for NoOfReplicas > 1, distribution awareness avoids inter-node-group communication, and some intra-node-group communication, but some inter-data-node communication is always required. In a system with good data partitioning and distribution awareness, most read transactions will access only one data node, and write transactions will result in messaging between the data nodes of a single node group. Messaging between node groups will be minimal.

Distribution keys

By default, the whole of a table's primary key is used to determine which fragment replica will store a row. However, any subset of the columns in the primary key can be used. The key columns used to determine the row distribution are called the 'distribution key'.

Where a table's primary key contains only one column, the distribution key must be the full primary key. Where the primary key has more than one column, the distribution key can be different to (a subset of) the primary key.

From MySQLD, a distribution key can be set using the normal PARTITION BY KEY() syntax. The effect of using a distribution key which is a subset of the primary key is that rows with different primary key values, but the same distribution key values are guaranteed to be stored in the same table fragment.

For example, if we create a table :


CREATE TABLE user_accounts (user_id               BIGINT,
                                                        account_type     VARCHAR(255),
                                                        username             VARCHAR(60),
                                                        state          INT,
                           PRIMARY KEY (user_id, account_type))
                              engine = ndb partition by key (user_id);

Then insert some rows :


INSERT INTO user_accounts VALUES (22, "Twitter", "Bader", 2),
                                                                  (22, "Facebook", "Bd77", 2),
                                                                  (22, "Flickr", "BadB", 3),
                                                                  (23, "Facebook", "JJ", 2);

Then we know that all rows with the same value(s) for the distribution key (user_id), will be stored on the same fragment. If we know that individual transactions are likely to access rows with the same distribution key value then this will increase the effectiveness of distribution awareness. Many schemas are 'partitionable' like this, though not all.

Note that partitioning is a performance hint in Ndb - correctness is not affected in any way, and transactions can always span table fragments on the same or different data nodes. This allows applications to take advantage of the performance advantages of distribution awareness without requiring that all transactions affect only one node etc as required by simpler 'sharding' mechanisms.

Correlated distribution keys across tables

A further guarantee from Ndb is that two tables with the same number of fragments, and the same number and type of distribution keys will have rows distributed in the same way.

For example, if we add another table :


CREATE TABLE user_prefs (user_id         BIGINT,
                                                  type               VARCHAR(60),
                                                  value    VARCHAR(255),
                        PRIMARY KEY (user_id, type))
                           engine = ndb partition by key (user_id);

Then insert some rows :


INSERT INTO user_prefs VALUES (22, "Coffee",  "Milk + 6 sugars"),
                                                            (22, "Eggs",    "Over easy"),
                                                            (23, "Custard", "With skin");

Then we know that the rows with the same user_id in the user_prefs and user_accounts tables will be stored on the same data node. Again, this helps with distribution awareness. In this example, we are ensuring that rows related to a single user, as identified by a common user_id, will be located on one data node, maximising system efficiency, and minimising latency.

Ordered index scan pruning

MySQL Cluster supports arbitrary ordered indexes. Ordered indexes are defined on one or more columns and support range scan operations. Range scans are defined by supplying optional lower and upper bounds. All rows between these bounds are returned.

Each Ndb ordered index is implemented as a number of in memory tree structures (index fragments), distributed with the fragments of the indexed table. Each index fragment contains the index entries for the local table fragment. Having ordered indexes local to the table fragments makes index maintenance more efficient, but means that there may not be much locality to exploit when scanning as rows in a range may be spread across all index fragments of an index.

The only case where an ordered index scan does not require to scan all index fragments is where it is known that all rows in the range will be found in one table fragment.
This is the case where both :

The ordered index has all of the table's distribution keys as a prefix
The range is contained within one value of the table's distribution keys

NdbApi detects this case when a range scan is defined, and 'prunes' the scan to one index fragment (and therefore one data node). For all other cases, all index fragments must be scanned.

Continuing the example above, assuming an ordered index on the primary key, the following ordered index scans can be pruned :


 SELECT * FROM user_accounts WHERE user_id = 22;
 SELECT * FROM user_accounts WHERE user_id = 22 AND account_type LIKE 'F%';

However, the following ordered index scans cannot be pruned, as matching rows are not guaranteed to be stored in one table fragment :


 SELECT * FROM user_accounts WHERE account_type = "Facebook";
 SELECT * FROM user_accounts WHERE user_id > 20 AND user_id < 30;

MySQLD partitioning variants and manually controlling distribution

Since MySQL 5.1, table partitioning has been supported. Tables can be partitioned based on functions of the distribution keys such as :

KEY
LINEAR KEY
HASH
RANGE
LIST

For engines other than Ndb, partitioning is implemented in the Server, with each partition implemented as a separate table in the Storage engine. Ndb implements these partition functions natively, using them to control data distribution across table fragments in a single table.

From Ndb's point of view, KEY and LINEAR KEY are native partitioning functions. Ndb knows how to determine which table fragment to use for a row from a table's distribution key, based on an MD5 hash of the distribution key.

HASH, RANGE and LIST are not natively supported by Ndb. When accessing tables defined using these functions, MySQLD must supply information to NdbApi to indicate which fragments to access. For example before primary key insert, update, delete and read operations, the table fragment to perform the operation on must be supplied. From MySQLD, the partitioning layer supplies this information.

Any NdbApi application can use the same mechanisms to manually control data distribution across table fragments. At the NdbApi level this is referred to as 'User Defined' partitioning. This feature is rarely used. One downside of using User Defined partitioning is that online data redistribution is not supported. I'll discuss Online data redistribution in a future post here.

Edited on 12/10/11 to fix formatting imbalance

Low latency distributed parallel joins

2011-01-26T22:51:00.005Z

When MySQL AB bought Sun Microsystems in 2008 (or did Sun buy MySQL?), most of the MySQL team merged with the existing Database Technology Group (DBTG) within Sun. The DBTG group had been busy working on JavaDB, Postgres and other DB related projects as well as 'High Availability DB' (HADB), which was Sun's name for the database formerly known as Clustra.

Clustra originated as a University research project which spun out into a startup company and was then acquired by Sun around the era of dot-com. A number of technical papers describing aspects of Clustra's design and history can be found online, and it is in many ways similar to Ndb Cluster, not just in their shared Scandinavian roots. Both are shared-nothing parallel databases originally aimed at the Telecoms market, supporting high availability and horizontal scalability. Clustra has an impressive feature set and many years of development behind it, but limited exposure to general purpose use.

At the time of the MySQL acquisition, HADB/Clustra was embedded in a number of Sun products as a session store and metadata repository, but was not available for external customers for general purpose use. Shortly afterwards, a decision was made to move HADB into a 'sustaining' model, and most of the ex-HADB team then became available to work on other projects. MySQL has greatly benefited from the injection of skills and enthusiasm from the Sun DBTG across a number of different teams, which is maybe not well known to those outside the company.

In the Cluster team, one project which has really benefited is the SPJ (Select Project Join) project, which couldn't have happened without the expertise and energy of the ex-Clustra/HADB team working on it.

The SPJ project started around the time of the last MySQL Developers conference in Riga in September 2008. The intention at the time was to look at ways of efficiently supporting more complex queries, specifically involving table joins, reducing unnecessary data transfer, communication latencies and context switches and increasing parallelism.

The main insights at the start of the project included :

Join mechanisms should be based on linking existing NdbApi single-table access primitives
Join result sets need not and should not be fully materialised at the data nodes
Join mechanisms need not be fully general or fully capable initially, as full generality/capability is already available with the existing Apis
Small joins should be targeted (Number of rows, number of tables)

These points have simplified the project scope greatly, and allowed many painful and costly detours to be avoided.

SQL execution in MySQL Cluster

As described in a previous post, Ndb Cluster was originally designed to quickly execute small queries with a high update rate at low latency. Larger more complex queries were executed by a separate query processor. The emphasis in this design is that complex queries are possible, but not necessarily fast or efficient. A main goal is that complex queries do not adversely affect the properties of the high volume, low latency requests.

All data access in MySQL Cluster is via the NdbApi interface. The NdbApi gives access to data stored in tables in the Cluster via four primitives :

Read row by primary key
Read row by secondary unique key
Scan a range of rows in an ordered index with optional conditions
Scan all rows in a table with optional conditions

Each of these primitives operates on a single table, and any table joins must be done by the NdbApi user. Different primary key/unique key operations and individual scan operations run in parallel across the data nodes in a cluster. NdbApi allows different operations and scan requests to be batched together to minimise latency due to communication delays, and it is essential to use this batching to get minimal latencies with Ndb.

In MySQL Cluster, attached MySQL Servers act as query processors, and MySQL's SQL execution engine breaks complex queries down into calls to its generic Storage Engine (SE) Api, which also deals with data access one table at a time.

The Ndb storage engine then further decomposes these SE Api calls into NdbApi primitive operations on individual tables.

MySQL supports SQL queries by performing the SE Api calls to read data, then comparing and matching results, sorting, buffering and reformatting. This works very well and gives MySQL Cluster great SQL functionality and compatibility, although users may find that the latency of their individual queries is not as low as with other engines such as MyISAM and InnoDB, which do not have to perform inter-process communication to implement their SE Api calls.

For minimum latency, Ndb requires that the MySQL Server makes efficient requests for data, requesting as much data as possible at once, and not using the data until it is essential to make forward progress - e.g. when there is a real data dependency.

MySQL features such as Insert, Update and Delete batching, and Batched Key Access minimise the MySQLD to Data node round trips required to execute certain types of operations, but they are unable to help when there are real data dependencies in a query. For example when the server needs to read some value from table t1 to know which rows to read from table t2 then there is no alternative but to read the t1 rows into memory before issuing any reads from t2.

Linked Operations

To reduce the need for extra Api to Data node round trips for every data dependency, we must allow operations to be linked. If we can describe the data dependency as a link between NdbApi operations, then it can be resolved amongst the data nodes. For example, rather than stating :


 SQL > select t1.b, t2.c from t1,t2 where t1.pk=22 and t1.b=t2.pk;
        ndbapi > read column b from t1 where pk = 22;
           

                   [round trip]
           

                   (b = 15)
        ndbapi > read column c from t2 where pk = 15;
          

                   [round trip]
          

                   (c = 30)
        [ return b = 15, c = 30 ]

We would state the join/operation linkage at the ndbapi level :


        ndbapi > read column @b:=b from t1 where pk = 22;
                 read column c from t2 where pk=@b;
          

                   [round trip]
          

                   (b = 15, c = 30)
        [ return b = 15, c = 30 ]

We allow read operations to be parameterised on the results of previous operations, and have the linking of the operations, the flow of results into parameters, handled by the data nodes. The data dependency still results in some execution serialisation at the data node layer, but not at the api layer, so data dependencies within queries needn't result in extra round trips between the MySQL server and the data nodes. Where the dependent data happens to be on the same data node, the dependency can be resolved with no inter-process communication at all.

Viewing the database software as a stack, the execution of the join is being 'pushed down' the stack, to a lower layer. For this reason, the SPJ functionality is also sometimes referred to as pushed-down joins or just pushed joins. Pushing functionality closer to the data can result in improved performance due to lower latency, reduced data transfer etc. In the case of MySQL Cluster, it can avoid inter-process communication, as well as enable parallelism across the data nodes.

In theory, linking can occur between any of the 4 primitive operation types :

Primary key access (PK)
Unique key access (UK)
Ordered index range scan (OI)
(Range bounds and optional conditions parameterised)
Table scan (TS)
(Optional conditions parameterised)

In practice, linking the cardinality (0|1) operations (Primary key, Unique key) together is simpler than linking with the scans. In turn, linking a scan to a cardinality (0|1) operation is simpler than linking a scan to another scan.

Linking a table scan to a table scan results in a cross-join and is probably going to be unpleasantly expensive for anything other than small tables.

The initial SPJ implementation supports combinations of Primary/Unique key operations linked together with at most one ordered index scan.

A future implementation will support multiple ordered index scans in a single request. This is more complex to handle due to the buffering required of the different scan result sets, and the resulting result ordering versus efficiency tradeoffs.

The SPJ Api is implemented as an extension to the existing NdbApi, with similar primitive concepts, but with the addition of the means to link the primitives together. As with the existing NdbApi, the usage pattern is along the lines :

Define operation(s)
Define further linked operation(s)
Execute() // One round trip to the data nodes
Examine results

In terms of batching, a tree of linked operations, with one const-parameterised root operation, and one or more child operations, is considered to be a single operation. Multiple SPJ operations, each actually a tree of primitive operations, can be executed simultaneously in a batch, along with other 'basic' NdbApi operations.

Where a scan is included, the scan can be advanced using the normal nextResult() mechanism, which also advances the results returned by any cardinality (0|1) child operations.

NoJoins - Not only Joins

While the SPJ extensions are described here in terms of joins, at the NdbApi level they are really 'linked operations'. One design goal which is not completely aligned with the join concept was to allow scans of multiple different tables to be parameterised on a single root operation. For example :

read @eid:= entity_id from map_table where username="jan";
scan blog_titles from blog_posts where entity_id=@eid;
scan latest_tweets from twitter_feed where entitiy_id = @eid;
scan share_prices from stock_feed where entity_id = @eid;
....

Here there is a data dependency between the first lookup and n peer child scans. I want to read all of this data in one round trip, but I don't necessarily want to have to express this in a single 'join' query. If we had a more relational/SQL oriented Api we might have had to create some unholy union of the different results, with masses of repeated values or nulls, or repeat the first lookup for each of n two-way joins.

With the linked operation concept, we can clearly state that the child scans are parameterised by the first lookup, without having to introduce some further unnatural coupling between the rows returned by each scan, which are otherwise independent.

So although SPJ is named after and described as supporting joins, it doesn't mean that you have to be 'join-oriented' or a SQL Samurai to benefit from it. It may be quite useful for efficiently traversing graphs, hierarchies and other links between rows where the concept of a 'join' is quite alien.

Check it out

A mysql-5.1-cluster-7.1 source tree with the SPJ enhancements can be downloaded from here. You can see the NdbApi extensions in the storage/ndb/include/ndbapi directory of the source tree. This source also includes extensions to the MySQL Ndb handler to make use of the new SPJ Api for SQL queries, which I hope to describe a little next time. If you want to download and try out SPJ then see some of the other blog posts about how to get started with it.

Some MySQL projects I think are cool - OpenQuery Graph Engine (OQG)

2010-09-27T11:45:00.004+01:00

This project was announced a year or so ago by Antony Curtis who used to work for MySQL AB. Having met Antony a few times I was intrigued to see what he was up to. The quote on the OpenQuery website describes it well :

The Open Query GRAPH engine (OQGRAPH) is a computation engine allowing hierarchies and more complex graph structures to be handled in a relational fashion. In a nutshell, tree structures and friend-of-a-friend style searches can now be done using standard SQL syntax, and results joined onto other tables.

That sounds cool, and it's the first time I've heard of a MySQL 'Computation engine' plugin. Delving further into the manual gives some insight, and there's some unexpected twists there :

OQG is a storage engine, but data stored is not persistent w.r.t. server crashes.
All tables have the same schema, storing details of graph 'edges'.
The fixed schema has a magic column called 'latch'
Depending on the constant value of latch used in a SELECT statement on the table, the engine will return different 'pseudo results'.

The last fact is the coolest one. As far as I understand it :

SELECT where latch = NULL AND ... allows queries on the graph as though it were a list of edges (as the data was entered).
SELECT where latch = 0 [AND ...] allows queries on the graph as though it were a list of nodes.
SELECT where latch = 1 [AND ...] allows Dijkstra's shortest path algorithm to be applied to the graph
SELECT where latch = 2 [AND ...] allows a breadth-first search to be applied to the graph

This is a superb hack! I imagine the OQG engine internally has an in-memory graph structure which is maintained as edges are added via the INSERT Api. The SELECT Api then gives access to different views of the underlying graph and even allows complex parameterised functions to be applied to the graph, giving results as a set of rows which can be decoded into the required result. It's not pretty, but it's an extremely pragmatic approach to embedding graph access and operations within a database.

It's also undeniable that the use of magic numbers and the 'latch' column adds a certain arcane wackiness that charms this reader. It's definitely a MySQL-style solution, continuing the tradition of MyISAM, Blackhole, Federated etc where good-enough gets to market before best, and 20% of the implementation effort delivers 80% of the functionality.

Once again I'd be interested to hear about how this is actually being used, and what sort of difference it is making.

Each of these three cool projects enable new solutions individually and expand the dimensions of what is possible using MySQL. In combination they open up a vast expanse of potential. One of the best things about them is that they all happened outside the confines of MySQL / Sun / Oracle. Hopefully they will get the success they deserve so that we can have more cool new projects in future.

Some MySQL projects I think are cool - Spider Storage Engine

2010-09-27T11:39:00.002+01:00

One thing that has puzzled me about MySQL Server is that it became famous for sharded scale-out deployments in well known web sites and yet has no visible support for such deployments. The MySQL killer feature for some time has been built-in asynchronous replication and gigabytes of blogs have been written about how to setup, use, debug and optimise replication, but when it comes to 'sharding' there is nothing built in. Perhaps to have attempted to implement something would have artificially constrained user's imaginations, whereas having no support at all has allowed 1,000 solutions to sprout? Perhaps there just wasn't MySQL developer bandwidth available, or perhaps it just wasn't the best use of the available time. In any case, it remains unclaimed territory to this day.

On first hearing of the Federated storage engine some years ago, I mistakenly assumed that this could be the basis of some MySQL scale-out solution. Perhaps a layer of front end 'proxy' MySQLDs could federate tables from a layer of backend MySQLDs giving some level of distribution transparency to sharded data. However as I found out, the Federated engine was not designed with such a scenario in mind. It has a certain internal elegance and simplicity, but unfortunately it is a little too simple for anything other than light duties.

The Spider storage engine extends the Federated concept of a table definition being a 'link' to a table on a remote MySQL server. However, it also integrates with the table partitioning features of MySQL 5.1, allowing each partition of a table to be specified as a 'link' to a table on a remote MySQL server. This effectively allows the built-in partitioning mechanisms of MySQLD (PARTITION BY RANGE/LIST/HASH) to be used to shard/partition rows across multiple MySQL servers transparently.

One of the major drawbacks of the Federated engine was that it had very little support for 'pushing conditions' to the MySQLD instance storing the source tables. This meant that well behaved selective queries issued on the 'front-end' MySQLD instance could result in non-selective queries being issued to the 'back-end' MySQLD instances, and large volumes of data being unnecessarily transferred back to the 'front-end' MySQLD where query processing then discarded it.

Spider attempts to improve this situation by pushing conditions down to the MySQLDs containing the source data. Combined with the partition pruning available from the MySQLD partitioning engine this should significantly reduce the amount of redundant data transferred in some cases.

So I think Spider is a pretty cool project. Like MySQL Cluster, it bears the burden of making MySQLD more data-distribution-aware and I think they're doing great work. It'd be great to hear stories about how Spider is being used, especially if anyone is using it *with* MySQL Cluster.

Some MySQL projects I think are cool - Flexviews

2010-09-27T11:34:00.002+01:00

Most of the time we think of SQL queries as being executed at a point in time and generating a single definitive result, but huge efficiency gains are available when data changes are tracked and derived views are partially updated as needed rather than being fully recomputed periodically. MySQL has support for views on tables, but there is currently no support for materialized views. While thinking about this topic I decided to have another look at Justin Swanhart's Flexviews tool and it's definitely a cool MySQL based project.

Flexviews is an open source set of non-intrusive addons to MySQL enabling materialized views to be defined and maintained as the underlying tables are changed. If you're not sure what a materialized view is or why they can be useful then I recommend reading the intro on the Flexviews site. I was particularly impressed by the documented support for GROUP BY, aggregates and joins.

I have a vague recollection of reading a blog post about an early version of Flexviews which used MySQL triggers to collect data changes on underlying tables, and feeling that it was probably a little flaky. However I now read that the 'Change Data Capture' in recent versions has been factored out into a separate tool called FlexCDC which 'mines' Row-based Binlog entries. This is a far more promising approach, and useful for many other applications. From my own point of view, it makes Flexviews potentially useful for maintaining materialized views of data stored in MySQL Cluster, where Binlog is the only centralised record of change. It also got me thinking that MySQL Cluster already has code inside the Server listening to data changes for writing the Binlog, which could be extended to capture data changes into some other tables if it were a useful feature.

The fact that Flexviews is implemented without Server changes or hooks is impressive and it's a great example of a MySQL 'ecosystem' project. It would be great to read some blog entries about how people are using it and what it is doing for them.

ACID tradeoffs, modularity, plugins, Drizzle

2010-03-25T09:48:00.009Z

Most software people are aware of the ACID acronym coined by Jim Gray. With the growth of the web and open source, the scaling and complexity constraints imposed on DBMS implementations supporting ACID are more visible, and new (or at least new terms for known) compromises and tradeoffs are being discussed widely. The better known NoSQL systems are giving insight by example into particular choices of tradeoffs.

Working at MySQL, I have often been surprised at the variety of potential alternatives when implementing a DBMS, and the number of applications which don't need the full set of ACID letters in the strictest form. The original MySQL storage engine, MyISAM is one of the first and most successful examples of an 'ACID remix'. The people drawn to DBMS development work often have a perfectionist streak, which can cause them to tend to prefer 'nothing' over 'imperfect'. MyISAM was and still is a flag-bearer for 'good enough'. Perhaps we should be less modest and call it 'more than good enough'.

One seldom discussed benefit of MySQL's storage engine architecture is that pressure to make 'The One True Storage Engine' is reduced. DBMS products with one fixed database engine need to optimise for all supported use cases. This is a great engineering challenge, but increases design effort, requirements for configuration and auto-tuning, constraints on any design change or reoptimisation etc. With MySQL, there are multiple existing storage engines, each with a (sub)set of target use-cases in mind. A single MySQL server can maintain and access tables in different storage engines, each tuned as closely as possible to the use-case for the data, without adding complexity to unrelated engines. Engines can be wildly optimised for a narrow use case as there are plausible alternative engines available for other use cases.

I understand that one aim of the Drizzle project is to extend the modularity of the MySQL Server on multiple axes, allowing diversity to flourish. As a one-time Java coder, who enjoyed the pleasures of design-by-interface, I can see the attraction. While the effort is guided by an actual need for modularity and real examples of alternative plugins, it can be a great force multiplier. There is always the risk of modularity for its own sake - a branch of Architecture Astronautics. Sure symptoms, which I may have suffered from in the past, include the class names FactoryFactory..., PolicyPolicy, or [Anything]Broker).

Another good vibe from Drizzle is the microkernel concept, although would say that there's some terminological abuse occurring here! Perhaps it could more reasonably be said that MySQL has a TeraKernel and Drizzle has a MegaKernel? In any case the motivations are good. Decoupling the huge chunks of functionality glued together inside MySQLD is great for long term software integrity, understanding dependencies, finding (and introducing) bugs, and might make it easier to start adding functionality again. Replication seems especially ripe ground for alternative plugins. User authentication is another often requested 'chunk'. It will take longer to crystalise interfaces for more deeply embedded areas like the query Optimizer/Executor, but if these interfaces are arising from a real need then that can drive the API design.

One aspect of storage engine modularity that is not often mentioned is that some MySQL storage engines also moonlight with other products. The Berkeley database (BDB) is probably the oldest and most promiscuous, embedded in DNS daemons, LDAP servers and all sorts of other places. Ndb is unusual in that it can be used from separate MySQLD and other NdbApi processes at the same time. InnoDB has also recently added an embedded variant. This trend will accelerate, especially when some of the distributed NoSQL systems start supporting 'pluggable local storage' APIs. I imagine that a NoSQL local storage engine API could be somewhat simpler to implement than the MySQL SE API, at least to start with!

Ndb software architecture

2009-09-28T22:43:00.004+01:00

I'm sure that someone else can describe the actual history of Ndb development much better, but here's my limited and vague understanding.

Ndb is developed in an environment (Ericsson AXE telecoms switch) where Ericsson's PLEX is the language of choice
PLEX supports multiple state machines (known as blocks) sending messages (known as signals) between them with some system-level conventions for starting up, restart and message classes. Blocks maintain internal state and define signal handling routines for different signal types. Very little abstraction within a block beyond subroutines is supported. (I'd love to hear some more detail on PLEX and how it has evolved). This architecture maps directly to the AXE processor design (APZ) which is unusual in having signal buffers implemented directly in silicon rather than software. This hard-coding drove Ndb's initial max supported signal size of 25 x 32-bit words.

An emulated PLEX environment (VM) is made available on Unix systems, written in C++
The VM runs as a Unix process. PLEX code for blocks is interpreted. Signals are routed between blocks by the VM. This allows development and deployment of PLEX based systems on standard Unix systems. It also allows Plex based systems to easily interact with Unix software. Each VM instance is a single threaded process routing incoming signals to the signal handling functions in each block class.

A PLEX to C++ translation system is designed
Blocks are mapped to large C++ classes with signal handling methods and per-block global state mapped to member variables. The limited labelling and abstraction encoded in the PLEX source are mapped to C style code within C++ classes.

The VM environment is 'branched' from the original PLEX/AXE environment and starts to evolve independently as a base for Ndb.
It offers access to more OS services such as communication, disk IO etc. Plex interpretation functionality is removed as all relevant Plex code has been mapped to native C++. VM instances can communicate with each other over various channels and form a distributed system.

At some point in the timeline around here the Ndb team and product leave Ericsson

Over time, common block functionality is abstracted into base and utility classes.
Hardware and system-convention sourced constraints are eased, the level of abstraction is raised. New blocks are designed and implemented without a Plex heritage making use of C++ abstraction facilities. Existing blocks are refactored.

Multi-threaded Ndbd (ndbmtd) is introduced, with groups of block instances running on different threads.
Rather than being a radical design, it's a move back towards the original PLEX design point of 1 block instance per processor.

Today, Ndb executes a blocks communicating via signals model. Signals are no longer limited to 25 words. In single threaded Ndb (ndbd), all blocks share a single thread, with separate threads used for inter-VM communication setup and disk IO. In multi threaded Ndb (ndbmtd), block instances are grouped, and different functional groups share threads. In all cases, each block instance remains single-threaded, although the thread may be shared with other blocks.

The blocks and signals model is reminiscent of Erlang and Hoare's CSP – where concurrency is modelled as serial (or sequential) processes communicating with explicit messages, as opposed to a shared-memory model where communication occurs via memory with correctness controlled by locks, memory barriers and atomic instructions. It can also be considered similar to MPI and the Active object / Actor model.

Using explicit messaging for synchronisation/communication has costs – at runtime a given algorithm may require more data copying. At design time, potential concurrency must be explicitly designed-in with messaging and state changes. Mapping sequential algorithms to message passing state machines may require a bigger code transformation than mapping to a naive multithread safe shared-memory and locks implementation.

However I believe that these costs are generally paid off by the benefit of improved code clarity. Inter state-machine synchronisation becomes clearly visible, making synchronisation costs easier to visualise and understand. With explicit messaging as the main mechanism for inter-thread and inter-process communication, there is only a small kernel of multithreaded code to be implemented, proved correct and optimised. The bulk of the code can be implemented in single threaded style. There is no need for diverse libraries of multithread-optimised data structures. Processor and system architecture specific code and tradeoffs are minimised.

Internally, Ndb's VM supports only asynchronous messages between blocks. Using an asynchonous message passing style has many benefits. As the sending thread does not block awaiting a response to a message sent, it can work on other jobs, perhaps including the message just sent. This allows it to make the best use of warm instruction and data caches, reduces voluntary context switches and can reduce the likelihood of deadlock. Blocking IO (network, disk) is outsourced to a pool of threads. The signal processing thread(s) never block, except when no signals are available to process. The responsiveness of the system can be ensured by using prioritised job queues to determine the job to execute next and minimising the time spent processing individual jobs. From a formal point of view the number of possible multithreaded interactions is vastly reduced as thread-interleaving is only significant at signal processing boundaries. These limitations can make it easier to reason about the correctness and timing properties of the system.

However, coding in this asynchronous, event-driven style can be demanding. Any blocking operations (disk access, blocking communications, requests to other threads or processes etc.) must be implemented as an asynchronous request and response pair. This style can have an abstraction-dissolving property as many published data structures and algorithms are implemented assuming a synchronous model and making much use of the caller's stack for state storage and managing control flow. It can be difficult to design abstractions for the asynchronous style which don't leak so much messy detail as to be pointless. Additionally, the asynchronous style tends to flatten a system – as the need to return control to the lowest-level call point whenever concurrency is possible acts as a force against deep layers of abstraction. Side effects of this can include a tendency for error handling code to be non-localised to the source of the error. However, that is part of the charm of working on the system. The C++ environment gives a wide set of tools for designing such abstractions, and each improvement made simplifies future work.

Comments, corrections?

MySQL Cluster development

2009-09-10T16:15:00.013+01:00

MySQL Cluster is the name given to one or more MySQL Server processes, connected to an Ndb Cluster database. From the point of view of the MySQL Server processes, the Ndb Cluster is a Storage Engine, implementing transactional storage of tables containing rows. From the point of view of the Ndb Cluster database, the MySQL Server processes are API nodes, performing DDL and DML transactions on tables stored in the cluster. Both exist independently – Ndb Cluster can be used without attached MySQL Server processes, but almost all users of Ndb Cluster connect at least one MySQL Server for DDL and administration.

Ndb stands for Network DataBase. This is a telecoms phrase where Network usually refers to a fixed or wireless telephone network, rather than the database topology definition of the term. Ndb was originally designed as a platform for implementing databases required to operate telecoms networks - HLR, VLR, Number Portability, Fraud Detection etc. At the time Ndb was first designed, Network Databases were generally implemented in-house on exotic 'switch' hardware by telecoms equipment vendors, often with hard-coded schemas and very inflexible query capabilities. These databases were expensive to develop and maintain, but had superb reliability and exceptional performance on minimal spec. hardware. The aim of the original Ndb design was to couple these desirable properties with more general purpose database functionality and deliver the result on a more standard hardware and OS stack.

I first discovered Ndb Cluster around 2001, when looking at potential designs for the next generation of an existing HLR database. I read the paper by Mikael Ronström in Ericsson Review (No 4,1997) which gives a good overview of the Ndb functionality. This paper describes functionality in the current tense when in fact some of the features described are yet to be implemented in 2009! This sort of optimism and vision has helped Ndb to survive and thrive over the years. The Ericsson Review paper was written while Ndb was one of multiple telecoms-database projects at Ericsson. Since then the Ndb product and team were spun out as a separate company, before being sold to MySQL AB in 2003 as a result of the dot com affair.

Ndb was originally designed for :

High throughput – sustaining tens to hundreds of thousands of transactions per second
Low latency – bounded transactions latencies which can be reliably factored into end-to-end latency budgets, implying main-memory storage
High update to read ratio – 50/50 as the norm
Transactional properties : Atomicity, Consistency, Isolation, Durability
Fault tolerance + HA – No single point of failure, automatic failover and recovery with minimal user or application involvement. Online upgrade. N-way synchronous and asynchronous replication. Fail-fast fault isolation.
Persistence – disk checkpointing and logging with automated recovery
Scalability – Parallel query execution. Distributed system can utilise > 1 system's resources. Capacity can be expanded horizontally with extra systems.

In the original Ndb design, high volume low latency transactions are submitted directly to the cluster using simple access primitives on the client. More complex queries are submitted to a separate query processor which itself uses combinations of the simpler primitives to access the cluster. An early example of a higher-level query processor was created by Martin Sköld who extended an Object Oriented query processor to create 'QDB' which could perform queries against data stored in Ndb. Numerous high level front-end processors have been implemented since.

Using MySQLD as a higher-level query processing front end we come to the architecture of MySQL Cluster, with MySQLD providing SQL based access to data stored in the cluster. In this sense MySQLD and Ndb cluster are a perfect fit and were designed for each other before they first met! Despite MySQLD being the default and most prominent front end to Ndb cluster, a number of others exist including several open and closed-source LDAP servers (OpenLDAP, OpenDS), several Java APIs and an Apache module giving HTTP access to data stored in Ndb.

The separation of low level, simple, fast access and higher level, more flexible access allows MySQL Cluster to offer many benefits of a full RDBMS without always incurring the drawback of over-generality. This fits well with many large transaction processing systems, where most heavy transaction processing does not require the full flexibility of the RDBMS, but some less frequent analysis does. Separating the central database engine (which in Ndb is referred to as the kernel ) from the query processing layer can also help with workload management – even the most complex queries are subdivided into manageable components and resources can be shared fairly.

The original Ndb design was not aimed at :

Disk resident storage
Where data larger-than-aggregate-system-memory-capacity can be stored on disk. This functionality was later added in the MySQL 5.1 timeframe
Complex query processing
Where multiple tables are joined. This was always possible, but not always efficient. Improving the efficiency of MySQL and Ndb on complex query processing is ongoing work - as it is in all actively developed RDBMS, for some definition of complex :).
Storing large rows
Ndb currently has a per-row size limit of around 8kB, ignoring Blob and Text column types.
One size fits all
Being a drop-in replacement for an existing MySQL engine such as MyISAM or InnoDB
Many initial users were not aware of the history of Ndb, and expected it to be (MySQL + InnoDB/MyISAM) + 'Clustering'. Issuing 'ALTER TABLE xxx ENGINE=ndbcluster;' appeared to be all that was required to gain fault tolerance, but the performance of queries on the resulting tables was not always as expected!

Since the initial integration of Ndb Cluster with MySQLD in 2003+, there have been many improvements to bring Ndb closer in behaviour to the most popular MySQL engines, and to optimise MySQLD for Ndb's strengths, including :

Support for Autoincrement and primary key-less tables
Synchronisation of schemas across connected MySQLD instances
Support for MySQL character sets and collations
Storage and retrieval of Blob and Text columns
Support for pushed-down filter conditions
Support for batching of operations
Integration with MySQL asynchronous replication
'Distribution awareness' in MySQLD for efficiency

These improvements have required work in the Ndb table handler - the code which maps MySQL storage engine API calls from the generic SQL layer to the underlying storage engine. Some improvements have also required enhancements in the storage engine API and Server, for example a new API to expose conditions (WHERE or HAVING clause predicates) to the storage engine, enabling it to perform more efficient filtering. These changes add complexity to MySQLD and the storage engine API, but as they are implemented generically, they can be reused by other engines. The pushed conditions API is now being used by the Spider engine for similar reasons to Ndb – e.g. to push filtering functionality as close to the data as possible. The Batched Key Access (BKA) improvements made to the MySQLD join executor benefit Ndb, but also benefit MyISAM and InnoDB to a lesser extent. This Functionality push-down pattern – increasing the granularity and complexity of work items which can be passed to the storage engine - will continue and benefit all storage engines.

The next large step to be taken by the MySQL Server team in this direction is referred to as Query Fragment Pushdown, where MySQLD can pass parts of queries to a storage engine for execution. Storage engines which support SQL natively could perhaps use their own implementation-aware optimisation and execution engines to efficiently evaluate query fragments. For Ndb, we are designing composite primitives at the NdbApi level for evaluating query fragments more efficiently - in parallel and closer to the data. This work will increase the number of query types that Ndb can handle efficiently, increasing the number of applications where Ndb is a good fit.

For an in-depth description of the original Ndb requirements, design approach and some specific design solutions, Mikael's phD thesis is the place to go. This is probably the best source of information on the design philosophy of Ndb Cluster. However as it is a frozen document it does not reflect the current state of the system, and as it is an academic paper, it does not describe the lower level, more software engineering oriented aspects of the system implementation.

I hope to cover some of these aspects in a future post.

Latency hiding patterns in CPUs

2009-04-28T00:51:00.005+01:00

Latency is a major factor in CPU design. Most general purpose CPUs execute a sequence of instructions with the logical convention that one instruction completes before the next starts. However memory access latency and even the latency of pure computation bottleneck the achievable processing throughput. To circumvent this a number of techniques are used to parallelise computation and communication.

Conductors have capacitance and resistance and therefore take time to switch between voltage levels. This manifests as a propagation delay proportional to the length of the conductor. As CPU clock speeds increase, the time between each rising clock edge reduces and the maximum length of conductor that can be charged in that time shrinks. CPUs and other data-path designs must be designed to ensure that no signal propagation path is near the tolerance for propagation delay at the designed maximum clock speed.

In practice this means that there is a trade-off between clock frequency and circuit size. Large circuits can accomplish more per clock cycle, but cannot be clocked as high as smaller circuits. To continue increasing clock rates with a fixed feature size, a circuit must be broken up into smaller and smaller sub-circuits with synchronous boundaries. Even this is not enough in some cases as the clock signals themselves are skewed by the propagation latencies across the chip. In some cases this is solved by having islands of synchronous logic which communicate asynchronously as no synchronous clock can span the whole circuit.

So even within a chip, there can be a large, and growing difference between local and remote communication latency. Ignoring this simplifies designs but forces lowest common denominator performance. This trade-off between complexity and latency awareness and tolerance is repeated at many layers of the system stack.

Even within a synchronous logic island on a chip, there is latency to be hidden. Instruction fetch, decode and execute takes time and pipelining used in CPU designs to maximise throughput despite irreducible latency. The various steps of instruction decode, register selection, ALU activation etc. are split out with latches between so that the maximum propagation delay in any stage of the pipleine can be minimised. For example, a datapath with a 400 gate worst-case propagation delay could theoretically be split into 4 stages, each with a 100 gate worst case propagation delay, allowing a 4 times faster clock speed.

With pipelining in a CPU, the first instruction to be processed takes at least the same amount of time to be completed, but after that, a new instruction can be completed on every clock cycle - theoretically allowing 4 times as many instructions to be completed per unit time. In practice memory latency and bandwidth limits and dependencies between instructions mean that the pipeline can stall, or bubbles can form, reducing the achieved throughput. However, these problems can often be alleviated and pipelines have been very successful in CPU design with instruction fetch-execute-retire pipelines comprising as many as 40 or more stages with multiple instances of stages in super-scalar designs

So pipelining in general allows repetitive serial task processing to be parallelised with potential benefits :

Increased parallelism between instances of serial tasks
Allowing greater throughput
Benefits of task-specificity (instruction and data caching benefits)
Potentially improving efficiency

and the following caveats :

Dependencies between tasks must be understood and honoured
Sometimes this reduces or eliminates parallelism
Pipeline stalls or bubbles due to job starvation or dependencies will reduce throughput
The pipeline must be kept fed.
Achievable throughput is bottlenecked by the slowest stage
As with a production line, the slowest worker sets the pace.
As with a production line, slow workers can be duplicated to improve throughput.
Individual task latency can increase

Pipelining as a general technique can apply to any system that processes tasks of a regular form which can be split into stages. The SEDA system from Harvard is a general purpose server framework for implementing server processes as pipelined stages. The software setting allows more flexible and dynamic tradeoffs to be made between pipeline lengths and widths. It also offers a flexible way to separate asynchronous and synchronous steps.

Other interesting latency hiding techniques seen in CPUs are mostly associated with hiding memory access latency.

Latency hiding patterns

2009-04-28T00:33:00.003+01:00

I want to write an entry about latency hiding patterns. Unfortunately my previous attempts became too long and boring even for me. This time I'm going to try something smaller and less rich to get the blog-flow more regular.

I'm interested in latency hiding patterns as I repeatedly see them being implemented at all levels of systems from the silicon to the top of the application stack. Often latency hiding techniques are what deliver better-than-Moore's law performance improvements and can have great impacts to usability as well as throughput and system efficiency.

I would describe latency hiding techniques as things that :

Maximise the value of communication
Maximise the concurrency of communication and computation

Communication value is maximised by avoiding communication, and when it cannot be avoided, minimising the overheads.
Communication and computation concurrency is maximised by maximising the independence of communication and computation. This requires understanding the dependencies between computation and communication.

Right, all very abstract. Let's hope some examples are more interesting.

Protel II

2009-04-16T23:13:00.003+01:00

In the first post I described the basic Protel language. In the mid 1990s it was extended with object oriented capabilities. This was done in a number of phases and was tied into the development of a project at Nortel called Generic Services Framework (GSF). This was an object oriented reimplementation of 'call processing' on the DMS with wide scope and a huge development team. Nortel even created an 'Object Center' somewhere in Canada with a helpline that confused designers could call to get OO Protel advice. I suspect that today it would be a 'Center of Object Excellence'. I heard that the GSF project was not an unqualified success, but it did drive the evolution of Protel-2 which was used later in other products. In retrospect it seems that GSF and Protel-2 were more motivated by the Objects-with-everything zeitgeist than any particularly compelling benefits.

Protel-2 supports single inheritance with a common root class called $OBJECT. Methods can be explicitly declared to be overridable in base classes, similarly to C++'s virtual keyword. It does not support operator overloading, or overloading method names with different signatures. It supports fully abstract classes.
Like Eiffel, Protel-2 allows parameterless methods which return a value to be called without parentheses. This allows data members to be refactored to be read directly or via an accessor function method without changing the callers. I'm not sure how valuable this is in practice, especially as it does not affect assignment. Perhaps some Eiffel practitioners have experience of finding this useful? Protel-2 methods can be declared to be read-only with respect to the object instances they operate on, in a similar way to specifying const-ness in C++.

The GENERIC keyword in Protel-2 allows class definitions to be parameterised by type. This allows the creation of type-safe generic collection classes and datastructures. The type parameterisation is similar to the Generics mechanism in Java, in that it is effectively a compile-time-only mechanism. The compiler generates a single underlying class implementation and checks type-correctness at compile time. A side-effect is that all access to the parameterised class must be made via a pointer. This fits well with the requirement for online load-replacing implementation etc, but it offers much reduced power compared to C++'s code-generation style templating mechanisms.

As with most C++ implementations, Protel-2 objects contain a vtbl pointer in the first word of their data followed by data members of superclasses and the class instance. On XA-Core, this sometimes presented a problem in that the normal transactional memory ownership mechanism could create a bottleneck when used for the vtbl ptr, but often data in the rest of the class should use the transactional memory mechanism. To deal with this, special libraries were created to allow the object header to be stored in WRITEBLOCKING memory and the rest of the object to be stored in BLOCKING memory.

That's scraping the bottom of the barrel on Protel-2 information in my head. I thought there was more there (or maybe just something more interesting). At least I've written it down.

Protel I

2009-04-12T23:11:00.005+01:00

Protel is the PRocedure Oriented Type Enforcing Language used for most DMS software. Currently there's not much information available about it online. An early paper describing the language is referenced here, but is hidden behind a subscription-only portal. Wikipedia offers a very minimal definition of the term but little else. I think there's some good stuff that should be recorded so I'll attempt to describe what I found interesting.

So what is Protel like? I'm told that it's similar to Modula-2 and even Modula-3 and it's true that it shares explicit BEGIN / END block syntax with Pascal, and all code is divided into modules.
One of the most basic differences between Protel and these languages is its use of the composite symbol '->' or 'Gozinta' (Goes into) for assignment. This eliminates any confusion between assignment and equality testing. This 'removal of ambiguity' is a key pattern in the design of Protel. Similar manifestations of the pattern are the rules :

No operator precedence rules
Unlike C, Protel assigns no built-in relative precedence to various operators. All expressions are evaluated left-to-right, and the programmer must use brackets explicitly to specify non l2r evaluation.
No preprocessor
There is no standard preprocessor and therefore no macro language. The compiler supports some limited compile time expression evaluation including sizeof() for types etc. This avoids context specific semantics for source code and non-visible code expansion
No support for pointer arithmetic
Where a pointer is to be treated as referring to some element of an array, the descriptior mechanism, which includes bounds checking, is to be used rather than pointer arithmetic.
Control structure specific end-of-block keywords
Rather than a single END keyword, Protel employs ENDBLOCK, ENDPROC, ENDWHILE, etc. to aid code readability.
No 'keystroke reduction mechanisms'
C's ++, += etc.

These rules can make life harder when writing code and increase verbosity, but can aid readability and reduce the amount of non-local knowledge needed to understand code.

Protel modules contain one or more source code files which can export definitions for use by other modules. Different source files within a module can be arranged into trees which control compile order within a module. Modules often have multiple levels of interface source files - most public and general APIs in the top level, more private and specific APIs in the lower levels. Access to definitions in each interface file can be controlled independently if required.

Basic Protel supports built-in and user defined types, pointers, arrays, an 'array slice' descriptor mechanism, and a novel extensible fixed-size type called an Area as well as some type-reflection capabilities.

A Protel DESCriptor is used to refer to a range of elements in an array of . It is used in the same way an array - with a subsript as an lvalue or rvalue to an expression (though the usual meaning of those terms is confused by the Gozinta operator!). The compiler is aware of the of the slice being DESCribed, and by inference the size of the elements. In the storage allocated to the DESC itself, it stores a pointer to the zeroth element and an upperbound in terms of elements. In this way it can provide bounds checking on accesses through the DESC. When an out-of-bounds exception is hit, the actual upperbound and the supplied subscript are available in the exception report, often allowing debugging straight from the trace. The array slice abstraction can be a nice way to deal with zero-copy in a protocol stack.

Protel offers the BIND keyword which can be used to define a local-scope alias to some variable instance. It's use is encouraged to reduce keystrokes, and it is also useful for indicating to the reader and the compiler that some dereferencing operations need only be performed once even though the referenced value is used multiple times. A side effect of its use is that it reduces the tension between short, easily typed and long, descriptive variable names, allowing long descriptive names to be shortened in use when necessary. Of course this can add to ambiguity and confusion.

Protel supports typed procedure pointers with an explicit type classifier - PROCVAR. I suspect that this type classifier exists to improve code readability rather than for any syntactic necessity as PTR to PROC can be used similarly. PROCVARs are used heavily in SOS code to allow applications to override behaviours and extend OS behaviour. SOS has unique terms for the use of procedure pointers :

GATE
SOS name for a Procedure Variable that is expected to be set by some other module. It is a 'gate' to some other implementation. Usually gates are defined in lower-level modules.
TARGET
SOS name for a procedure implementation referenced from a GATE. This is the 'target' of a call to a 'gate'. Usually targets are defined in higher-level modules
ASPECT
SOS name for a structure containing a number of procedure variables. This can be thought of as an 'interface' in the Java sense - a set of method signatures. Often a lower-level module would provide an API for a higher-level API to add some functionality including some data and perhaps an 'aspect' of procedure variables.

As well as supporting standard composite structures similar to a C struct, Protel supports the concept of an Area which can be used to implement a form of type-inheritance. An AREA is similar to a struct, but contains as part of its definition a storage size in bits, as well as zero or more members. At compile time, it is checked that the declared member's types fit within the size given, and instances will be created with the size given. Other modules can declare further areas which REFINE this area, and add definitions to the original AREA. The compiler will check that the complete set of member definitions continue to fit in the bit size of the original AREA. This mechanism can be used to create trees of hierarchically related data types which is very useful for code modularity and extensibility as well as more basic optionality similar to a C union. Putting procedure pointers into the Area gives a rather rough extensible virtual-method mechanism in-language. However, most DMS software designers used AREA refinements for hierarchically varying data rather than allowing control flow overrides.

Protel offers some reflection capability via the TYPEDESC operator. It is applied to a type and returns a structure which can be used to determine type names, bit offsets etc. SOS supports an online extensible data dictionary which uses TYPEDESC to track types and their relationships. It is version aware and is used by Table Control and for data reformatting between software releases.

Combining PROCVARs and REFINEd AREAs allowed extensible systems to be built fairly easily without OO techniques built into the language. However, the explicit nature of the PROCVARs, the requirement to define up-front bitsizes for refinable areas and the general micromanagement required to define, initialise and use 'object' hierarchies made from these components discouraged most designers from using them in this way. Providing tools at this atomic level encouraged each designer to try their own combination of hard coding, ProcVars, extensible areas, pointers-to-extension-structs, pointers-to-data-structs-with-procvars etc. More manual visualisation effort was required to grasp these mechanisms than would be required for an equivalent language with OO extensions.

I think PROTEL was fairly state-of-the-art when it was introduced for DMS software. Especially considering its planned use for telecoms switching equipment, it is a very general purpose language, not visibly oriented towards telecoms. It has a fairly clean split between language features and runtime libraries. Perhaps if it had been more widely known of beyond the confines of Nortel/BNR then it could have enjoyed some life of its own? I believe it is still being actively used - these days SOS images run in virtualised environments with code written by outsourced employees paid by a broken company, but I imagine there must still be patches getting written. However the outlook looks bleak. With Nortel on the rocks and apparently no interesting information about Protel available on the web (except this blog of course :) ), it looks like it could vanish after 30 years.

In the mid to late nineties, Nortel added object orientation to Protel, I'll talk briefly about that in a future post.

What is SOS? Part III

2009-03-28T23:43:00.003Z

Part 1 Part 2

Online code patching and extension

A SOS system is comprised of modules, which contain code and various data segments and are vaguely similar to shared libraries or DLLs. Each function in a module has a function pointer stored at some known offset in a code header segment. The actual machine code for the function is stored in a different segment of the module. This indirection requires that all procedure calls involve a pointer dereference, but gives the flexibility to change the implementation of any procedure at any time. The code and data segments in a module also include limited 'spare' space, so that a number of extra global data variables and functions can be added to a module online. This, coupled with the ability to load completely new modules with arbitrary code and data makes a SOS system completly runtime-patchable, with all behaviours modifiable online. Online upgrade of a running module is referred to as load-replacement. In development it is used to test and debug code, and in deployment it is used to patch code, and to add small new features.

Run-time code modification is made managable by the cooperation of the standard source code control system (PLS), the Protel language compiler and linker, and SOS. At compile and link time, metadata about the header contents and sizes, and the version of source compiled is included in the module file. When SOS is asked to load-replace the module, it compares the new module with the existing module and will only allow the load-replace if it can be done safely. When a module is replaced, SOS updates its module metadata with the new module's version etc.

SOS also includes a patch management system which tracks the state of applied patches. Patches use the basic module load-replacement system in a controlled and automated way to :

Load-replace existing modules to add hooks into existing code
Load new modules to contain modified functionality and state storage space
Execute patch application and removal steps

SOS tracks patch dependencies and can generally unapply and reapply patches at runtime. It tracks inter-patch dependencies, and allows different deployments to run different sets of patches. This is especially useful when patches are used to implement features and functionality specific to a single user.

Writing SOS patches is quite an art, and whole teams that write nothing but patches existed in Nortel's good times. Often the patch specialists were very technically capable and innovative, being aware of the innards of SOS and able to deal with the extra dimensions of design visualisation required to consider patch application and later removal. However, extended exposure to writing patches in the convoluted style required for safe application and removal tends to corrode a designer's sense of elegance and abstraction.

'Relational' data access system

SOS includes a table based data access layer called 'Table control'. This layer supports interactive and batch access to tables. Tables include key columns and non-key columns with a flexible type system, including inheritance. Tables are statically defined with a good deal of flexibility in the implementation of the mapping onto the underlying data source. Table control supports separate data representations at 'External', 'Logical', 'Data' and 'Physical' layers. These abstractions give great freedom to decouple the external, user visible view of the data from the internal constraint optimised storage of the data. Table control was initially designed to give a standard way to store and retrieve DMS configuration information, but over time in different products is used to give standardised access to huge databases of mobile subscriber information etc. The Table Control API was later built on to implement external data provisioning and management systems, and is a large part of the online software upgrade process.

Despite being table and column oriented, table control is only 'relational' in a limited sense. There is no SQL-style declarative language for querying data stored in tables, and no standard way to 'join' tables. However, foreign key constraints can be enforced, and DMS supports a basic scripting language which can be used to write scripts to automate cross-table analysis and maintenance.

User model

SOS supports multiple interactive user sessions, connected via telnet or older technologies. Users can have various permissions with respect to commands, table access etc, and all user activity can be logged.

Online software upgrade

Theoretically, module load-replacement can be used to upgrade software, but in practice it is only used for bug fixes and small features with economic or time-pressure reasons for in-release delivery. Writing all code to be online replacable against old code adds an excessive burden to the design and test cycles.

SOS supports online upgrade of duplex systems via the ONP process (One Night(mare) Process). What happens is :

Hardware sync-match is dropped, splitting the system into two separate systems, one active, the other inactive.
The inactive side is rebooted with the new software
Bulk personality and state data from the active side is transferred across to the inactive side, potentially involving data reformats for changed or extended schemas.
Once all bulk data is transferred, up to date state data transfer starts
Once all components agree that state transfer is reasonably up-to-date
- New Active side activity is stopped
- All remaining state is transferred
- Inactive side becomes active side (SWitch of ACTivity). IO systems are reconnected to new active side.
Users perform acceptance testing on the new system for a limited period
If users decide to revert :
- Modified state is transferred back to newly inactive side
- SWACT is performed in reverse
If users decide to continue :
- Newly inactive side is dropped and hardware sync-match is restored.

This upgrade mechanism is complex and error prone, but it offers online upgrade with minimal service outage (of the order of 4 seconds) at the cost of a temporary loss of redundancy.

From the application designer's point of view, they need to consider :

Data reformats
Table control can automate some conversions which map into type-promotions. More complex conversions can be performed in user-code callbacks.
State transfer
Essential state can be transferred around SWACT using user-code callbacks
Protocol compatibility
Newer software versions must support old protocol versions until all parties can deal with newer versions.
Upgrade-abort implications
If upgrade is reverted then data and states which only exist in the new version must be avoided or dealt with.

SOS applications generally support direct upgrade over 3 versions. In a DMS system comprising a number of smaller SOS based systems, generally the upgrades start at the leaves of the tree of systems (peripherals), and work back towards the Computing Module (CM). This implies that each system must be willing to accept old-version protocol interactions from systems higher in the tree than it, but need not worry about protocol versions for systems lower in the tree (Assuming the usual computer-science leaves-at-the-bottom tree layout). Given that individual system complexity increases as you go 'up' the tree to the root, this is a good arrangement.

Online multithreaded breakpoint capable debugger

SOS supports interactive use and one application users can use is an interactive breakpoint and tracepoint capable debugger. This tool allows a running system to be inspected, and code and data to be modified on the fly. Break and tracepoints can be made data-conditional and can use thread (process) ids to be thread conditional. The debugger also has some knowledge of symbols and offsets within modules.
Full debugging access with breakpoints is not usually made available for deployed systems as the risk of accidental damage is too great.

Well that's been a quick tour of SOS. It's an interesting system with very little documentation outside of Nortel. My own memories of it are fading fast, so please excuse mistakes and the lack of detail here. I don't intend to blog about the system in-general any more, but may cover some specific details that are of interest.

(Well of interest to me, as no-one else seems interested so far :) )

What is SOS? part II

2009-01-06T23:37:00.006Z

Part 1 Part 3

Resource Ownership Model

SOS implements an extensible resource ownership module. System resources such as memory, IPC 'mailboxes', semaphores and any user-defined resource are owned by either a loaded module or a running process. Process death results in process owned resources being freed. Module unload results in module owned resources being freed. Processes are owned by Modules.

Since all processes in SOS share memory by default (more like threads in other OS), a mechanism for cleaning up resources on process death is very useful.

Applications can arrange to have their own resources owned by a process or a module using a mechanism confusingly called 'Events'. This mechanism guarantees a callback on process death or module unload allowing arbitrary cleanup. This is essential as processes can exit at any point with no stack cleanup due to exceptions (divide by zero, bus error etc.).

Prioritised proportional scheduling

SOS processes are each placed in a process class. Each class has a guaranteed share of CPU where all process class shares sum to 100% of available CPU. Additionally, all process classes have a relative priority.

The scheduler chooses which process to run next based on answering the question "What is the highest priority class with both a runnable process and time left in it's CPU share?".

The guaranteed share mechanism is used to ensure that all process classes regularly get some share of CPU time even in a heavily loaded system, while still providing priority to high urgency tasks. This allows the system to avoid throughput or latency degradation as load increases.

The placement of processes into process classes is generally fixed at design time. The set of process class guaranteed shares (called a scheduler template) is also generally fixed, but is changed between restarts and normal operation.

Application control of scheduler pre-emption

The SOS scheduler implements pre-emptive multitasking on a single CPU core, changing the running process periodically. Processes can temporarily stop pre-emption occurring via the rather wordy setunpreemptable() call. This can be used to demarcate critical regions in code, where atomic actions are performed. All other processes can only observe the state of memory before the call to setunpreemptable(), or after a call to setpreemptable().

Since all other processes are excluded from running while the current process is unpreemptable, it is important that only bounded amounts of computation are performed while unpreemptable. To enforce this, SOS implements an unpreemptable timer, of the order of tens of milliseconds which stops the running process with an exception if it remains unpreemptable for longer than this. For this reason, activities like blocking IO and dynamic memory allocation cannot reliably be performed while unpreemptable.

Most applications are collections of message driven state machines, and run unpre-emptably in bursts, processing messages, updating state and sending responses. As well as giving low-overhead mutual exclusion, making applications non-blocking with control over pre-emption maximises the instruction cache hit rate and improves data cache locality.

That's enough for now, I'm boring myself !

What is SOS?

2008-12-12T16:36:00.011Z

Part 2 Part 3

SOS is either the Switch, Support or Service Operating System which runs on a number of the components making up a Digital Multiplex Switch (DMS).
Work began on the system around 1979. It is mostly written in and highly coupled to the PROTEL language and the PLS (Product Library System) SCCM tool.
It is a pre-emptive multitasking operating system with some bullet-pointable features :

Comprised entirely of runtime reloadable modules
Multiple memory pools with different durability + protection characteristics
Multiple levels of system restart with restart escalation
Strong and extensible resource ownership model
Prioritised proportional scheduling
Online code patching and extension
Built-in relational style database system
Support for multiuser interactive use
Support for online upgrade to new version
Contains online multi threaded trace/breakpoint debugger

One of the most retro features is having no per-process memory protection. All processes run in a shared address space which makes them similar to modern day threads within a single process. Chaos is somewhat contained by the support for write-protected memory. One advantage of not having per-process address spaces is that processor caches and TLBs do not need to be flushed when context switching.

Module based
The PROTEL language allows a system to be split into modules, each with multiple source code files. All definitions in the files are contained in the scope of the module. Each source file can be marked as either public interface, private interface or implementation. Modules import definitions from other module's public and permitted private interfaces. The Module concept provides a component-level encapsulation, independent of OO or other abstraction mechanisms used in the code itself.

SOS allows modules to be loaded at runtime. SOS also allows modules to be 'replaced' at runtime. This involves overwriting the object code of the module while only making safe modifications to the module's exported procedure entry points and global data. This is the basis of the online code patch system which allows any object code to be replaced while processes execute over it.

Each module can define an entry procedure. This is called when the system is performing a restart and allows the module to take different initialisation actions depending on the restart type.

A SOS system is comprised of a set of modules and an initialisation order. At the various restarts, the SOS system iterates through the modules in initialisation order, calling their entry procedures.

To allow different types of systems sharing the same source modules to be easily defined, sets of modules, and their dependencies can be grouped together to form larger components. A system can then be specified in terms of these larger components. The inter-component and inter-module dependencies are then used together with some hints to compute the module initialisation order and build a system image.

Multiple memory types
The DMS model is unusual in that memory is expected to provide sufficient persistence for most data, with disk based recovery only occasionally required. This is a reasonable assumption given fault tolerant redundant memory with redundant power supplies, arrays of lead acid batteries etc.
A number of basic memory types are defined by SOS, with a number of variants for special purposes.

PSPROT
Program Store, protected. Used for object code. Write protected. Loaded from and Saved to a SOS image on disk/tape.
DSSAVE
Data Store. Not initialised by operating system reboot or restart. Not part of a SOS image.
DSPROT
Data Store, write protected. Used for configuration or otherwise slow changing data. Loaded from and Saved to a SOS image on disk/tape.
DSPERM
Data Store, permanently allocated, wiped by some restarts. Not part of a SOS image.
DSTEMP
Data Store, temporarily allocated, wiped by most restarts. Not part of a SOS image.

The DSSAVE memory is limited in size but is useful for tracking system debugging state across multiple OS reboots. Most applications have no need for it.
DSPROT is written to by transiently removing write protection during the write. If a write is attempted while write protection is active, the writing process gets an exception. Special handling is required while DSPROT is being backed up to ensure a consistent snapshot is taken.
DSPERM remains allocated across all restart types but is reset on some (see below). This gives it the interesting property that a pointer to allocated DSPERM must be stored in DSPROT memory to ensure that the allocated memory can be 'found' again after a restart.
DSTEMP is deallocated and reset across all restart types.

The memory types tie into the set of restart types supported by the operating system (below).
One of the main benefits of this system is that it orients application designers towards thinking of their application in terms of multiple levels of state, and the benefits of throwing state away to recover from error situations.

Multiple levels of system restart
SOS defines three levels of restart :

Initial Program Load (IPL)
This is performed only once when a module is initially loaded
Reload restart
This is the most severe restart type and occurs as part of a reboot, or when an assertion failure or user request demands it.
DSPERM memory is reset, DSTEMP memory is deallocated and reset. All modules' entry procedures are called.
Cold restart
This is second-most-severe restart type and occurs when requested by the user, or when a number of Warm restarts have failed to clear a problem. DSTEMP memory is deallocated and reset.
Warm restart
This is the least severe restart type and occurs when requested by the user or when the system determines that a number of failure indicators suggest ill health. DSTEMP memory is deallocated and reset.

By placing different parts of a module's state in different memory types, and reallocating/reinitialising the state in the module's entry procedure, Applications can cooperate with the system's restart escalation mechanism. One of the Call Processing (CALLP) applications written on SOS uses Warm Restart to drop connecting calls, but keep connected calls, and Cold restart to drop all calls.

Low level modules in SOS monitor system health indicators (number of process deaths, exceptions while in a critical region, system load etc.) and if there is a perceived problem will trigger a warm restart of the system.

If the warm restart fails, or the system does not recover correctly after a number of warm restarts, the restart type is escalated to a cold restart. Modules are generally designed to re-initialise more state during a cold restart (which as a result, generally takes longer to accomplish).

If multiple cold restarts fail, the system escalates to a Reload restart, which, again, reinitialises more state, taking longer.

If all attempts to restart the running system fail, a reboot can be attempted which reloads the system image from disk and performs a reload restart on it.

If this fails, previously backed up images are tried.

In this way, the system automatically escalates recovery efforts, resetting more and more state each time, eventually trying previous images. The driving philosophy is to *never* give up trying to recover. Never wait for a friendly user to press a key, or intervene.

What makes this different?
SOS is curious in the ways it differs from the Operating Systems in common use today but it is also similar in a number of ways :

Written in high level language
Written for general purpose CPU and memory model
Multitasking
Supports interactive use

These features are not particularly noteworthy for a modern general purpose OS, but for one designed in 1979 for a telecoms switch they are unusual. Other telecoms software at the time tended to be more :

Written in assembly language and/or
Written in telecoms-specific DSL with severe expressivity limitations
Designed for telecoms specific CPUs and hardware
Cooperatively scheduled
Very limited interactivity

I believe SOS was ahead of its time in being fairly general purpose, powerful and flexible.

Well done if you got this far, I'll continue boring on about SOS in another post...

Hardware Transactional Memory II

2008-12-09T00:49:00.005Z

In my last entry, I introduced Nortel's XA-Core platform which I believe was one of the first commercially successful HTM machines. This time I want to talk about the hardware.

Modular Architecture
An XA-Core system is comprised of various card types including Processing Elements (PEs), Shared Memory cards (SM) and IO processors (IOP). These components are connected by a 'Gigabit Interconnect' (GI) which in practice is a set of point-to-point optical links with agreeable 'hot pluggable' and optical isolation properties. All of these card types exist in various versions and live in a standardish DMS rack with redundant power and cooling.

Processing Element
The PE card has a number of large chips and a few smaller ones (probably most circuit boards do :)) :

Two lockstep PowerPC CPUs, initially PPC603.
A 'Hippo' or 'Rhino' chip acting as the CPU -> Memory interface.
One custom 'PIGI' chip interfacing the PE with the GI

The processors run in lockstep with a fairly standard comparator mechanism to check them. From each processor's perspective, the Hippo/Rhino chip looks like main memory. Among other things, the Hippo/Rhino and PIGI chips provide :

Lock-step comparison of CPU outputs
Mapping of PPC bus requests to GI protocol requests, including transaction identifiers etc.
Mapping of 32-bit PE address space to 40-bit Shared Memory Address space.

The PE board additionally has some local memory, referred to as 'Scratch' which applications can use as a non-persistent workspace. While the PE has lockstepped CPUs, it is not possible to run the PE with only one CPU functioning. If either CPU fails, the whole PE is isolated. This avoids the requirement for a post-mismatch fault detection algorithm.

The minimum configuration is two PE cards, giving tolerance of one failure. Extra PE cards can be configured to give different n+m fault tolerance configurations.

When PE cards are inserted, they perform a self test, a number of initialisation steps are performed and then they begin executing the SOS scheduler loop, taking work. When a PE card is hot-pulled or fails, any outstanding memory transaction is rolled back. Some other PE can then pick up the aborted work from wherever in shared memory the original PE found the work.

Shared Memory card
The SM cards contain some fairly fast memory accessed via a custom 'SMOAC' (Shared Memory Ownership and Access Controller) chip.

The SMOAC chip maintains the memory ownership information that is necessary to enforce the transactional semantics of memory access. Ownership information is maintained for every 32-bytes (PPC cache line) of memory in the system using ownership information sent with cache-line read and write requests from the PE.

To support rollback of unwanted memory transactions, every cache line is duplicated within an SM card. Every cache line has an Active copy (last committed) and an Update copy (dirty, yet to be committed). This doubles the amount of memory required, although it probably simplifies the hardware design and theoretically allows arbitrarily large transactions.

Logical memory is mapped onto the SM cards in 32MByte blocks. The normal configuration is that every block is mapped onto at least two SM cards, and sometimes three. This allows for one or two SM card failures to be tolerated. Combined with the two copies of memory required for the transaction mechanism, this means that each byte of logical shared memory requires four to six bytes of physical memory.

When SM cards are inserted, SOS decides which blocks should be copied to the new card, and begins a background task to copy the blocks across.

Input/Output Processor
The IO processor cards are used to connect the XA-Core to the outside world, including terminals, disk + tape and the rest of the DMS system. The cards themselves contain single PPC CPUs (no lockstepped redundancy here) and ASICs to interface to the GI and provide some DMA capability. IOPs are deployed in pairs so that no IO facility is completely lost due to an IOP failure.

Weird / Cool things

Fault tolerant, single system image, shared memory multiprocessing
I don't think many examples exist where a single-system-image SMP can handle an arbitrary processor failure without a crash.
Existing correct code runs correctly in parallel
But may be serialised due to contention on shared memory access
Even IO is transactional
This puts pressure on IO latencies, and requires good batching to minimise IO overheads.
Easier identification of transient CPU faults
When a mismatch is detected within a PE, the failing operation can be safely rolled back and retried on the same PE multiple times. This can be used to help diagnose hard faults from transient / temporal faults.
Online System split is possible for upgrade
To support online software and data upgrade, the system can de-duplicate memory, assign a PE to the 'other' half of the memory and boot it from a system image on disk. This gives 2 systems running on one machine. At cutover time, most PEs and IOPs are quickly migrated from the old to the new side. Eventually memory can be re-duplicated from the new half.
No 'standard-SMP' cache-coherency glue logic required

Not so cool things

Four to Six times memory hardware overhead
Perhaps some trade-off between maximum transaction size and hardware complexity could have been made?
Expensive memory required to contain latency
Large, complex custom ASICs required
Pushes out time-to-market, reduces time-in-market for modifications. Expensive.
OS cooperation required to assist with transaction demarcation, ensuring forward progress, IO handling, bringing PEs, IOPs and SMs on + offline.
Requires cooperation from OS owners.

So that's my tour of the XA-Core hardware. Please comment if you have corrections or further questions, I may be able to dredge up some more details.
Next time I'll talk about some of the modifications made to the SOS operating system to make it run on this platform.

Hardware Transactional Memory I

2008-12-05T19:26:00.005Z

The first multiprocessor I worked on was Nortel's XA-Core platform. This exotic platform was a replacement for the 'Computing Module' (CM) of their DMS telecoms switching platform.

Background
Previous CM generations are built on a pair of CPUs (Motorola 88k, 68k, BNR NT40) run in lockstep through a comparator for fault tolerance. The software running on these includes a multitasking OS (SOS) and a huge amount of call processing, database, hardware support and other telecoms code written in the proprietary PROTEL language, starting around 1979. SOS supports write-protectable memory, but not per-process memory protection, so the memory map resembles a heavily multithreaded process. Shared data is commonly used with an assumption of a strictly ordered memory model. Heavy use is made of a single-global-lock to enforce mutual exclusion between processes to the extent that the bulk of the computation time is spent with a single process holding the global lock in 'jumbo' timeslices of tens of milliseconds. Much of the large code base is > 10 years old and in a 'frozen' state where changes are not possible,

The problem
How to increase CM computation capacity beyond the incremental improvements available from successive generations of CPUs without a huge software rewriting and revalidation effort and while maintaining CPU and memory fault tolerance?

The solution ( XA-Core patent)
Create a fault tolerant SMP platform with replicated hardware transactional memory. Modify the OS so that a process claiming the 'single global lock' implictly sets the boundaries on a memory transaction. Handle inter-process memory access contention by rolling back one of the contenders. Handle CPU failure by rolling back in-progress memory transactions.
The achievable level of parallelism is then limited by the memory access patterns of the concurrently running processes at the cache-line level.
Code can still be written using the 'single CPU multitasking OS with big-global-lock' approach. Incremental improvements to available parallelism can be made by changing the data access patterns of the parallel processes. Tools exist to monitor contention between competing processes and map it to stack traces and/or data structures.

The interesting details and issues
The actual hardware used, the transaction handling in the operating system, handling IO, application modifications required etc.

In the spirit of actually completing some blog entries, I'll continue this post later.

To be continued...

Design fossils

2008-06-10T15:21:00.010+01:00

Mobile telephone networks have a kind of gritty Service Oriented Architecture. The Services are defined in GSM, UMTS and other standards. The protocols have hard unfashionable names like SS7, TCAP and MAP, and the bytes call themselves octets and don't waste their entropy transporting 'markup'. Even so, the basic SOA design and benefits are still visible and these benefits have enabled multi{national, operator, vendor, protocol} mobile communication to become a basic utility.

The scalability, reliability and success of the GSM network design and implementation seems to me to be an under-celebrated achievement of software and systems engineering. Perhaps the price of success is to become invisible. As the pace of technical development accelerates, major achievements of the past lose their significance and appear to be inevitable and mundane increments on the road to now.

Mobile telephony network technology is reasonably mature, and most software and system design effort goes into reducing the cost per subscriber and adding speculative features. However there is a huge resource of distilled design expertise, proven design patterns and traces of design evolution captured in these telephony systems as they have existed, and as they exist now. Unfortunately much of this information is encoded in odd and underknown languages, sealed in peculiar and fruity source code control systems and fading in the minds of ageing and modestly eccentric employees. Fortunately, this adds to the charm of investigating.

I intend to write up some of the things I have found interesting before I too forget....

Intention

2008-06-09T15:48:00.003+01:00

To periodically post about systems I have worked on, am working on and am interested in. To share references to interesting blog entries, articles, papers, books, designs, patterns and systems. Hopefully not to share too much detail of my daily habits and emotional state.

We shall see...