Blog Archive

Showing posts with label Tungsten. Show all posts
Showing posts with label Tungsten. Show all posts

Benchmarking Tungsten Parallel Replication

Tungsten parallel apply on slaves, or parallel replication for short, has been available for about a year.   Until recently we did not have many formal benchmarks of its performance.  Fortunately the excellent Percona Live Conference in London accepted my talk on Tungsten parallel replication (slides available here), so Giuseppe Maxia and I finally allocated a block of time for systematic performance testing.

In a nutshell, the results were quite good. In the best cases Tungsten parallel apply out-performs single-threaded native replication by about 4.5 to 1.  Both Giuseppe and I have verified this using slightly different test methodologies, which helps avoid dumb counting mistakes.  Our results also match field tests at a customer site over the previous summer, so we regard them as fairly robust.  In the remainder of this article I would like to expand a bit on the details of the benchmarks as well as the results.  The results shown here are from my tests.

Benchmark Test Design

Both Giuseppe and I used a similar testbed for replication testing:    
  • HP Proliant server, dual Xeon L5520 CPUs with hyper-threading enabled, 72Gb of RAM
  • 1TB HP Smart Array RAID 1+0 
  • Centos 5.6
  • XFS file system
  • MySQL 5.1.57 with InnoDB buffer pool set to 10Gb and using O_DIRECT purge method 
  • Tungsten Replicator 2.0.5 build 347  
For convenience we use MySQL sandbox to set up a master with two slaves, as shown in the following diagram.  It turns out that for measuring replication throughput there is no reason to set up on separate hosts, as the master does little or nothing during the test and we only operate one slave at a time.  



The Tungsten Slave is configured as described in a previous article in this blog, except that there are 30 channels instead of 10.   The exact installation command is given at the end of this article.

The test run uses sysbench to spread transactions evenly across 30 databases of identical size, then measure time to process them.  This is also known as a replication catch-up test. 
  1. Load all MySQL servers with an identical dataset consisting of 30 databases pre-populated with data from sysbench.  Giuseppe wrote a new tool called Large Data Generator that is very helpful for capturing and loading such datasets.  
  2. With the slaves shut down, store the master binlog start position and then run 30 sysbench oltp test processes against the master to update and read from all schemas simultaneously for one hour.  
  3. Start the MySQL slave from the stored master binlog position and measure time to process the sysbench transactions. Shut down the MySQL slave at the end of the test. 
  4. Start the Tungsten slave from the stored master binlog position and measure time to process the sysbench transactions using Tungsten Replicator with 30 channels (i.e. threads). 
Test Results

Database performance is substantially different depending on whether data are fully resident in the buffer pool (cache-resident) or largely read from disk (I/O-bound).  Tungsten parallel replication over 30 databases varies from 1.8 to 4.5 depending on which case you look at, as shown in the following table.   Processing times are in minutes (m).

Test Scenario
Rows/Db
Data Size
MySQL Slave
Tungsten Slave
Ratio
Cache-resident
10K
430Mb
30m 
17m
1.8
I/O-Bound
10M
68Gb
228m
51m
4.5

Let's look at the results in detail.  In the cache-resident test the base dataset is relatively small and loads fully into the buffer cache within a minute or two.  Both MySQL and Tungsten slaves complete in well under an hour.  Here is a graph showing throughput as measured in bytes of binlog processed per 10 second increment.  

Cache-Resident Slave Catch-Up - MySQL vs. Tungsten Replicator, 30 Databases
In the cache-resident case there are virtually no reads from disk as all data are fully resident in the InnoDB buffer pool.  Tungsten Replicator is faster because multiple writes can occur in parallel but the speed-up versus native replication is not especially large.  Note that Tungsten processes around 40Mb every 10 seconds or about 1Gb of binlog every four minutes.

With I/O bound workloads, on the other hand, we see a profound difference in performance.  Tungsten Replicator is at least 6x slower than in the cache-resident case, but still processes updates faster than the master (51 minutes on the slave vs. 60 minutes on the master).   Buffer cache loading is correspondingly fast and Tungsten reaches steady-state performance within about 20 minutes.  MySQL native replication on the other hand is far slower. The slave not only does not catch up, but it would quickly lag far behind  the master under this workload.  It takes about 90 minutes for native replication even to achieve steady state performance after buffer pool loading.
I/O-Bound Slave Catch-Up - MySQL vs. Tungsten Replicator, 30 Databases
Overall we can safely say that single-threaded native replication is likely non-workable in the I/O-bound case without going to some combination of SSDs and/or slave pre-fetch.  


Further Improvements and Caveats

The current results of parallel replication benchmarks on Tungsten are gratifying especially when you consider that two years ago Tungsten Replicator performance was around 10% of the speed of MySQL replication.  Nevertheless, these benchmarks are not the final word.  It is clear there is room for optimization as we observe that Tungsten processes the cache-bound binlog at least 6 times faster than the I/O bound workload.  Much of the difference seems to be time spent reading from disk.   If this could be improved, Tungsten would go even faster.

During the London conference Yoshinori Matsunobu published some excellent performance results using slave pre-fetch, which has encouraged us to build pre-fetch into Tungsten as well.   I am curious to see if we can further boost throughput by adding pre-fetching on each parallel thread, though other people at the conference such as Domas Mituzas were not optimistic.  Either way, I am certain we will improve performance, if not using pre-fetch then with other tricks like batching inserts.

Finally, some caveats.  Our sysbench load is nice because it is evenly distributed across schemas of exactly the same size.  Most application workloads do not behave this way, though some do come very close.  The slides for my talk discuss practical issues in maximizing performance in real applications.  I suspect that a combination of parallelization with pre-fetch will in fact turn out to be a very good solution for a wide variety of workloads.

Fine Print

If you would like to repeat our results (or attack them as fraudulent), here are some parameters that may help.  The database settings in the MySQL sandbox instances are as follows:
default-storage-engine=InnoDB
innodb-additional-mem-pool-size=100M
innodb-flush-method=O_DIRECT
innodb-log-buffer-size=4M
innodb-log-file-size=50M
innodb-thread-concurrency=0
innodb_buffer_pool_size=10G
innodb_file_format=barracuda
innodb_file_per_table=1
innodb_flush_log_at_trx_commit=2
innodb_strict_mode=1log-bin=mysql-binmax-connections=500
max_allowed_packet=48M
skip_slave_start
sync_binlog=0

Next, here is the sysbench command used to generate load on each schema.  We run 30 of these simultaneously varying the database name for each invocation.  This example is for the I/O-bound case.

sysbench --test=oltp --db-driver=mysql --mysql-db=${db} \  --mysql-user=msandbox --mysql-password=msandbox \
  --mysql-host=127.0.0.1 --mysql-port=33306 \
  --oltp-read-only=off --oltp-table-size=10000000 \
  --oltp-index-updates=4 --oltp-non-index-updates=2 \
  --max-requests=200000 \
  --max-time=3600 --num-threads=5 run

The replicator configuration is given in the slides for the talk, but here it is again.   Options in red are required for sandboxes.  Production installations are therefore simpler than what is shown here.

tools/tungsten-installer tools/tungsten-installer --direct -a \
  --service-name=parallel --native-slave-takeover \
  --master-host=127.0.0.1 --master-port=33306  \
  --master-user=msandbox --master-password=msandbox  \
  --slave-host=127.0.0.1 --slave-port=33307  \
  --slave-user=msandbox --slave-password=msandbox  \
  --home-directory=/opt/continuent \
  --property=replicator.store.parallel-queue.maxOfflineInterval=5 \
  --svc-parallelization-type=disk --buffer-size=100 \
  --channels=30 --thl-port=2115 --rmi-port=10010 \
  --skip-validation-check=MySQLPermissionsCheck \
  --skip-validation-check=MySQLApplierServerIDCheck \
  --start-and-report

To equal the results shown above you will also need to assign databases explicitly to channels in the shard.list file.   Otherwise, databases will be assigned  channels using a hashing function, which tends to result in somewhat uneven distributions.  Look in the comments of the shard.list file for instructions on how to do this. 

Finally, all of our tests depend on two excellent tools from Giuseppe Maxia:  MySQL Sandbox and the new Large Data Generator program in the Tungsten Toolbox.  Once you get the hang of them you will become completely addicted as they make test setup both reliable as well as quick.  

Quick Installation of Replication from MySQL to MongoDB

Proof-of-concept Tungsten support for MongoDB arrived last May, when I posted about our hackathon effort to replicate from MySQL to MongoDB.  That code then lay fallow for a few months while we worked on other things like parallel replication, but the period of idleness has ended.  Earlier this week I checked in fixes to Tungsten Replicator to add one-line installation support for MongoDB slaves.

MySQL to MongoDB replication will be officially supported in the Tungsten Replicator 2.0.5 build, which will be available in a few weeks.  However, you can try out MySQL to MongoDB replication right now.  Here is a quick how-to using my lab hosts logos1 for the MySQL master and logos2 for the MongoDB slave. 

1. Download the latest development build of Tungsten Replicator.   See the nightly builds page for S3 URLs.

$ cd /tmp
$ wget --no-check-certificate https://s3.amazonaws.com/files.continuent.com/builds/nightly/tungsten-2.0-snapshots/tungsten-replicator-2.0.5-332.tar.gz

2. Untar and cd into the release. 

$ tar -xzf tungsten-replicator-2.0.5-332.tar.gz
$ cd tungsten-replicator-2.0.5-332

3. Install a MySQL master replicator on a host that has MySQL installed and is configured to use row replication, i.e. binlog_format=row.  Note that you need to enable the colnames and pkey filters.  These add column names to row updates and eliminate update and delete query columns other than those corresponding to the primary key, respectively. Last but not least, ensure strings are converted to Unicode rather than transported as raw bytes, which we have to do in homogeneous MySQL replication to finesse character set issues.  

$ tools/tungsten-installer --master-slave -a \
  --datasource-type=mysql \
  --master-host=logos1  \
  --datasource-user=tungsten  \
  --datasource-password=secret  \
  --service-name=mongodb \
  --home-directory=/opt/continuent \
  --cluster-hosts=logos1 \
  --mysql-use-bytes-for-string=false \
  --svc-extractor-filters=colnames,pkey \
  --svc-parallelization-type=disk --start-and-report

4. Finally, install a MongoDB slave.  Before you do this, ensure mongod 1.8.x is up and running on the host as described in the original blog post on MySQL to MongoDB replication.   My mongod is running on the default port of 27017, so there is no --slave-port option necessary. 

$ tools/tungsten-installer --master-slave -a \
  --datasource-type=mongodb \
  --master-host=logos1  \
  --datasource-user=tungsten  \
  --datasource-password=secret  \
  --service-name=mongodb \
  --home-directory=/opt/continuent \
  --cluster-hosts=logos2 \
  --skip-validation-check=InstallerMasterSlaveCheck \
  --svc-parallelization-type=disk --start-and-report

That's it.  You test replication by logging into MySQL on the master, adding a row to a table, and confirming it reaches the slave.   First the SQL commands: 

$ mysql -utungsten -psecret -hlogos1 test
Welcome to the MySQL monitor.  Commands end with ; or \g.
...
mysql> create table bar(id1 int primary key, data varchar(30));
Query OK, 0 rows affected (0.15 sec)

mysql> insert into bar values(1, 'hello from mysql');
Query OK, 1 row affected (0.00 sec)

Now check the contents of MongoDB:  

$ mongo logos2:27017/test
MongoDB shell version: 1.8.3
connecting to: logos2:27017/test
system.indexes
> db.bar.find()
{ "_id" : ObjectId("4e85269484aef8fcae4b0010"), "id1" : "1", "data" : "hello from mysql" }

Voila!  We may still have bugs, but at least MySQL to MongoDB replication is now easy to install.   

Speaking of bugs, I have been fixing problems as they pop up in testing.  The most significant improvement is a feature I call auto-indexing on MongoDB slaves.  MongoDB materializes collections automatically when you put in the first update, but it does nothing about indexes.  My first TPC-B runs processed less than 100 transactions per second on the MongoDB slave, which is pretty pathetic. The bottleneck is due to MongoDB update operations of the form 'db.account.findAndModify(myquery,mydoc)'.  You must index properties used in the query or things will be very slow.   

Auto-indexing cures the update bottleneck by ensuring that there is an index corresponding to the SQL primary key for any table that we update.  MongoDB makes this logic very easy to implement--you can issue a command like 'db.account.ensureIndex({account_id:1})' to create an index.  What's really cool is that MongoDB will do this even if the collection is not yet materialized--e.g., before you load data.   It seems to be another example of how MongoDB collections materialize whenever you refer to them, which is a very useful feature.  

TPC-B updates into MongoDB are now running at over 1000 transactions per second on my test hosts. I plan to fix more bugs and goose up performance still further over the next few weeks.  Through MongoDB we are unlearning assumptions within Tungsten that are necessary to work with non-relational databases.  It's great preparation for big game hunting next year:  replication to HBase and Cassandra.  

What's Next for Tungsten Replicator

As Giuseppe Maxia recently posted we released Tungsten Replicator 2.0.4 this week.  It has a raft of bug fixes and new features of which one-line installations are the single biggest improvement.  I set up replicators dozens of times a day and having a single command for standard cluster topologies is a huge step forward.  Kudos to Jeff Mace for getting this nailed down.

So what's next?  You can get see what we are up to in general by looking at our issues list.  We cannot do everything at once, but here are the current priorities for Tungsten Replicator 2.0.5.
  • Parallel replication speed and robustness.  I'm currently working on eliminating choke points in performance (like this one) as well as eliminating corner cases that cause the replicator to require manual intervention, such as aging out logs that are still needed by slaves.  
  • Multi-master replication.  This includes better support for system of record architectures, many masters to one slave, and replication between the same databases on different sites.  Stephane Giron nailed a key MyISAM multi-master bug for the last release.  We will continue to polish this as we work through our current projects.   
  • Better installations for more types of databases.  Jeff recently hacked in support for PostgreSQL as well as Oracle slaves, and we are contemplating addition of MongoDB support.  Heterogeneous replication is getting simpler to set up.  
  • Filter usability.  Giuseppe has a list of improvements for filters, which are one of the most powerful Tungsten Replicator features but not as easy for non-developers to use as we would like.  Better installation support is first on the list followed by ability to load and unload dynamically.  
  • Data warehouse loading.  We have a design for fast data warehouse loading that I hope we'll be able to implement in the next few weeks.  Linas Virbalas has also been working on this problem along with a number of other heterogeneous projects for customers.  
This is a lot of work and not everything will necessarily be finished when 2.0.5 goes out.  However, I hope we'll make progress on all of them.  In case you are wondering how we pick things, replicator development is largely driven by customer projects.   If you have something you need in the replicator, please contact Continuent.

After this build we will... Er, let's get 2.0.5 done first.  Suffice it to say we have a long list of useful and interesting features to discuss in future blog articles.

Practical Multi-Master Replication using Shard Filters

Earlier this month I published an article on this blog describing the system of record approach to multi-master replication.  As mentioned in that article my colleagues and I at Continuent have been working on improving Tungsten to make system of record design patterns easier to implement.  This article describes how to set up system of record using Tungsten Replicator shard filters, which are a new feature in Tungsten 2.0.4.  By doing so we will create a multi-master configuration that avoids replication loops and transaction conflicts.  On top of that, it is quite easy to set up.

There are many possible system of record patterns depending on how many schemas are shared and across how many masters.  The following diagram shows three of them.  In contrast to many so-called MySQL multi-master implementations, all masters are live and accept updates.  (By contrast, schemes such as MySQL-MHA make extra masters read-only.  Don't be fooled!)  

For today's exercise we will implement the basic system of record.  Once you understand this you can quickly set up other multi-master scenarios.  

Defining Shard Master Locations

The first step is to tell Tungsten where each shard is mastered.  By mastered we mean it is the one master that receives application updates on that shard, whereas all other masters have copies only or may not even contain the shard at all.  Tungsten uses a variant of CSV (comma-separated format) where the first line contains column names.  You can have any amount of whitespace between entries.  Create a file called shards.map with your favorite editor and type in the following lines. 

shard_id master critical
tungsten_nyc nyc false
tungsten_sjc sjc false
acme         sjc     false
pinnacle        nyc     false

The first column is the name of the shard.  This must be unique--because a shard can only live on one master.  The next column is the "home" master for the shard.  This is the one and only master that should receive shard updates.  The third column defines whether the shard is critical and requires full serialization.  It will be linked to parallel replication in a later release.  

It turns out you do not need to add entries for Tungsten catalog schemas such as tungsten_nyc.  Tungsten Replicator will create them automatically.  They are shown here for completeness only.  

Creating Replication Services

Next we need to define services to replicate bi-directionally between DBMS servers and set options to filter shards using the ShardFilter class, which is new in Tungsten 2.0.4.  The shard filter helps ensure that shards replicate from their home masters only and not from other locations.   If you do not know what replication services are, you can find a description of them in this article.  

Multi-master replication is easy to mis-configure, so to prevent accidents we will tell the shard filter to generate an error any time it processes a shard it has never seen before.  The replication service will immediately fail, which signals that we have to update shard definitions.  This is the safest way to implement system of record or any multi-master configuration for that matter.  It is generally easier to restart replication after correcting the configuration than to mix up data, which can lead to major outages.  

The first step is to set replication services for each master.  These read the binlog and make transactions available to slave replication services.  Here are the commands.   Note that the sjc master is on host logos1, while the nyc master is on logos2.  The remaining examples use these names consistently. 

# Define common master settings. 
COMMON_MASTER_OPTS="--datasource-user=tungsten --datasource-password=secret \
 --home-directory=/opt/continuent --svc-parallelization-type=disk \
 --svc-extractor-filters=shardfilter \
 --property=replicator.filter.shardfilter.unknownShardPolicy=error"

# Set up sjc master. 
tools/tungsten-installer --master-slave -a --master-host=logos1 \ 
--cluster-hosts=logos1 --service-name=sjc $COMMON_MASTER_OPTS --start-and-report 

# Set up nyc master. 
tools/tungsten-installer --master-slave -a --master-host=logos2 \
--cluster-hosts=logos2 --service-name=nyc $COMMON_MASTER_OPTS --start-and-report 

The --svc-extractor-filters option adds shard filtering immediately after event extraction.  The unknownShardPolicy=error setting will cause the masters to die if they process an undefined shard.  

Now we can define the remote slave services for sjc and nyc.  These are special slaves that write transactions onto another master as opposed to a normal slave.  We would like slave services to error out on unknown shards as well.  Also (and this is important) we want them to enforce shard homes.  Here are the commands to create the services and start each one.  

COMMON_SLAVE_OPTS="--release-directory=/opt/continuent/tungsten \
--service-type=remote --allow-bidi-unsafe=true --svc-parallelization-type=disk \
--svc-applier-filters=shardfilter \
--property=replicator.filter.shardfilter.unknownShardPolicy=error \
--property=replicator.filter.shardfilter.enforceHome=true"

# Set up sjc remote slave. 
tools/configure-service -C -a --host=logos2 \
--local-service-name=nyc --role=slave \
--datasource=logos2 --master-host=logos1 $COMMON_SLAVE_OPTS sjc
$trepctl -host logos2 -service sjc start

# Set up nyc remote slave. 
tools/configure-service -C -a --host=logos1 \
--local-service-name=sjc --role=slave \
--datasource=logos1 --master-host=logos2 $COMMON_SLAVE_OPTS nyc
trepctl -host logos1 -service nyc start

The --svc-applier-filters option adds shard filtering before applying to the DBMS.  The unknownShardPolicy=error setting will cause the slaves to die if they process an undefined shard.  Finally, the enforceHome=true option means that each slave will drop any transaction that lives on a different service from that slave's master.  

At the end of this procedure, your services should be online and read to run.  Use 'trepctl services' to make sure.  

Loading Shard Definitions

To make shard definitions take effect, you must load the shard.map contents into each replication service.  You can do this any time the replicator is running but after loading new definitions you must put the replicator online again.  Here are the commands to load the shard maps onto each of the four replication services.   For each replication service, you must delete the old definitions, reload new ones, and get the replicator to go online again. 

#!/bin/bash
MAP=shard.map
for host in logos1 logos2
do
  for service in sjc nyc
  do
    trepctl -host $host -service $service shard -deleteAll
    trepctl -host $host -service $service shard -insert < $MAP
    trepctl -host $host -service $service offline
    trepctl -host $host -service $service wait -state OFFLINE
    trepctl -host $host -service $service online
  done
done

This looks a little clunky and will be reduced to a single command instead of five in a later release.  I put it in a script to make it quicker to run.  The good news is that there is just one shard map that works for all replication services, regardless of location or role.  

Once you finish this step, you can go to any replication service and list the shards it knows about.  Let's pick a service and demonstrate: 

$ trepctl -host logos1 -service sjc shard -list
shard_id master critical
tungsten_nyc nyc false
tungsten_sjc sjc false
acme sjc false
pinnacle nyc false

With this we are ready to start processing some transactions. 

Multi-Master Operation

At this point we have multi-master replication enabled between hosts logos1 and logos2.  You can try it out.  Let's add the acme database to the sjc master on logos1 as an example. 

mysql -utungsten -psecret -hlogos1 
mysql> create database acme;
mysql> use acme
mysql> create table foo (id int);
mysql> insert into foo values(1);

We can see that all of these commands replicate over to the logos2 server quite easily with the following command: 

mysql -utungsten -psecret -hlogos2 -e 'select * from acme.foo'
+------+
| id   |
+------+
|    1 | 
+------+

That seems pretty simple.  In fact it is.  You can go over to logos2 and enter transactions for pinnacle in the same way.  Data replicate back and forth.  There are no replication loops.  There are also no conflicts.  

Adding a New Shard

So what happens when we add a new shard?  The simplest way to see is to create a database using a schema name that does not exist in the shard map.   Let's try to create a database named superior on the nyc master.  

mysql -utungsten -psecret -hlogos2 -e 'create database superior'

Now check the status of the nyc master replication service.  We see it has failed with an error due to the unknown shard.   (Tungsten parses the create database command and assigns it the shard ID "superior.") 

$ trepctl -host logos2 -service nyc status
Processing status command...
NAME                     VALUE
----                     -----
...
pendingError           : Stage task failed: binlog-to-q
pendingErrorCode       : NONE
pendingErrorEventId    : mysql-bin.000157:0000000000002475;1287
pendingErrorSeqno      : 8
pendingExceptionMessage: Rejected event from unknown shard: seqno=8 shard ID=superior
...
state                  : OFFLINE:ERROR...
Finished status command...

This problem is quite easy to fix.  We just open up the shard.map file and add a row for superior so that the file contents look like the following: 

shard_id master critical
tungsten_nyc nyc false
tungsten_sjc sjc false
acme         sjc     false
pinnacle        nyc     false
superior        nyc     false

Reload the shard.map file as shown previously and you will be back in business.  Incidentally, if you do not want the superior database to be replicated to other masters, you can also specify this in the rules.  Just give superior the special master name #LOCAL as in the following example and it will not replicate outside the nyc service. 

superior        #LOCAL  false

In fact, #LOCAL means that any schema named superior will not replicate outside the service in which it is defined.  You can have an unshared schema named superior on every master.  

Where to Next?

The shard support described in this article is now part of Tungsten 2.0.4 and will appear in the official build when it is finally ready.  You can try it out right now using one of our handy nightly builds.  

We plan to build out shard filtering quite a bit from the current base.  One immediate fix is to put in a check so that if an application commits shard updates on the wrong DBMS server, the master replication service on that server will detect it and fail.  This will tell you there's a problem immediately rather than letting you wallow in blissful ignorance while your data become hopelessly mixed up.  We will also simplify the commands to update shards while replicators are online. 

Longer term we will be adding features to propagate shard definitions through replication itself.  Stay tuned for more work in this area.  If you want to help fund work to enable your own applications, please get in contact with me at Continent.  I can think of at least a dozen ways to make our multi-master support better but it's always nicer to spend the effort on features that enable real systems.  In the meantime, I hope you find multi-master with shard filtering useful and look forward to your feedback.  

First the Blog, now the Webinar: Adding Parallel Replication to MySQL in a Hurry

My recent post on setting up Tungsten parallel replication in a hurry got a lot of hits, though to be fair it was probably not the great writing but the fact that at least one popular MySQL blog posted a link to it.  (Thanks, we like you guys too.)  Anyway, I would like to invite anybody who is interested in parallel replication to attend a webinar on Thursday September 1st at 10am PDT to cover installing and using Tungsten.  It's straight-up technical talk to help you start quickly. 

Bringing up Tungsten on an existing MySQL slave only takes a few minutes, so once we have that out of the way I will explain how Tungsten works inside and show you some of the tricks for getting your applications to play nice with parallel replication as well as how to tune performance.  The idea is to minimize fluffy architectural stuff and maximize lab demos that help you bend replication to your will.  The talk will also cover how to get help, log bugs, and even add your own code.  Plus there will be lots of time for questions.

As most readers of this blog know, Tungsten Replicator is open source (GPL V2) and hosted on code.google.com.  If you miss the webinar you may be able to catch up on parallel replication in-person in London toward the end of October.  I just submitted a talk to the next Percona Live and hope it gets accepted.  If so, see you there!

Adding Parallel Replication to MySQL in a Hurry

A previous article on this blog described Tungsten parallel replication using on-disk queues.  On-disk queues are now more or less finished, and I just closed the covering issue for the feature.  The work is bug fixing and performance testing from here on out.  Speaking of performance, that looks fairly good.   A recent on-site test using production workloads showed 3.3X improvement over native MySQL replication while holding resources like memory down to much more reasonable levels than in-memory queues.  We have further optimizations on the way, so this should improve.

Now that parallel replication is working a lot better, what is it good for?  Here is a good start:  assuming your workload is suitable for shard-based replication, Tungsten offers a nice replacement for MySQL slave replication that will give you immediate parallel replication for any MySQL server starting with version 5.0 and up.  We call this native slave takeover.  It looks like the following diagram.

You can migrate MySQL slaves to Tungsten very easily thanks to some great work on the Tungsten installer by Jeff Mace that Giuseppe Maxia also praised in a recent article.  Seeing that Giuseppe is also the Continuent QA director, I guess this means it passed.  :)   Here is how to set up.

First, configure MySQL master/slave replication.  Here are typical commands after you synchronize the slave with a backup from the master.  For these examples, we assume the slave is running on host mercury, and the master is on host saturn.

mysql -utungsten -psecret -hmercury -e "CHANGE MASTER TO \
  MASTER_HOST='saturn', \
  MASTER_USER='repl', \
  MASTER_PASSWORD='s3cr3t', \
  MASTER_LOG_FILE='mysql-bin.000338', \
  MASTER_LOG_POS=4534"
mysql -utungsten -psecret -hmercury -e "START SLAVE"
mysql -utungsten -psecret -hmercury -e "SHOW SLAVE STATUS\G"

Next, download and unpack Tungsten from the latest dev build.  Here are sample commands for build 231.  

wget --no-check-certificate https://s3.amazonaws.com/files.continuent.com/builds/nightly/tungsten-2.0-snapshots/tungsten-replicator-2.0.4-231.tar.gz
tar -xvzf tungsten-replicator-2.0.4-231.tar.gz
cd tungsten-replicator-2.0.4-231

Finally, run the installation using the following handy command.  This will copy the download into its proper location (here /opt/tungsten) and start Tungsten.  Tungsten will stop MySQL replication if it is running and start Tungsten replication from the exact point where native replication left off.
tools/tungsten-installer \
  --direct  \
  --native-slave-takeover \
  --master-host=saturn  \
  --master-user=repl  \
  --master-password=s3cr3t  \
  --slave-host=mercury \
  --slave-user=tungsten  \
  --slave-password=secret  \
  --service-name=takeover \
  --home-directory=/opt/continuent \
  --svc-parallelization-type=disk \
  --channels=10 \
  --start-and-report
If there is a problem with your pre-requisites the installer will print out a message and stop.  Most of these messages should be pretty self-explanatory.   (In some builds there's a problem with OpenJDK failing the install check.  Use --no-validation to get around that.) 

Here is another interesting tip.  If you try Tungsten and want to go back to MySQL replication, you can flip back to MySQL native replication with two commands.  Just take Tungsten offline cleanly, which shuts down all the replicator channels at the same point, and start MySQL replication again: 

trepctl offline
mysql -urepl -ps3cr3t -e 'START SLAVE'

We are currently doing a lot of performance profiling and testing of parallel replication.  I hope to post more about performance results in a future article.  Meanwhile, there are 36 new features (including on-disk queues) and bug fixes that will roll into a final Tungsten 2.0.4 in the next week or so.   Try out the latest builds and see what you think. 

Introducing Tungsten On-Disk Queues for Parallel Replication

Tungsten Replicator has offered shard-based parallel replication to slaves since late 2010.  The initial implementation uses in-memory queues.  Working purely in memory keeps latency low and throughput high.   On the other hand, working in memory consumes valuable RAM.  It also forces us to buffer all in-flight transactions and therefore greatly limits the span of time permissible between the slowest and fastest shard.

Hence our newest improvement:  on-disk parallel queues.  In this article I will cover how parallel replication works in general, how on-disk queues help with parallel replication, and then show how to set up from the latest builds.

First, let's review the basic mechanics.  Parallel replication is really "parallel apply," which means taking a stream of serialized transactions, splitting them into separate streams, and applying them in parallel to a slave DBMS.  In the Tungsten pipeline architecture, we implement this kind of flow using a combination of stages and stores.  One stage reads transactions from the persistent transaction history log (aka THL) to a "parallel store."  The parallel store splits the stream into a set of queues.  The next stage extracts from those queues and applies to the slave.   It looks like the following picture:

From a conceptual point of view the incoming thl-to-q task thread performs an indexing function.  It  guides construction of the queues read by task threads in the q-to-dbms stage.  Within this framework there are many ways to feed events into the parallel queues.    In the case of on-disk queues there are two obvious design options.  
  1. Read data out of the THL and split them into separate transaction logs per parallel queue.  This is very similar to the in-memory approach, except that the queues are now on disk (or SSD or whatever storage you pick).  It can be implemented without adding any extra threads to the parallel store. 
  2. Leave all data in THL.  Add a cursor for each parallel queue that scans the THL and picks only the transactions that belong in that parallel queue.   This requires extra threads to do the scans, hence is more complex to implement.  
Both approaches achieve the primary goal, which is to keep the transactions in storage until we actually need them and thereby minimize memory usage.  This in turn solves a major problem, namely that individual shards can now be many thousands or even millions of transactions apart in the serial history.  Beyond that, it is not completely obvious which approach is better.

For example, option 1 isolates the reads to individual files.  This minimizes overall I/O at the cost of making it more random, since reads and writes are spread over many files.  Option 2 avoids extra writes and keeps I/O sequential, but introduces a bunch of threads doing the equivalent of table scans across the same patch of storage.  Up to a point we can assume that pages are coming out of the OS page cache rather than storage but this assumption will not hold for all operating environments and workloads.  The only way to prove the trade-offs is to implement and test.  (We may end up implementing both.)

After some discussion internally at Continuent as well as with Harrison Fisk from Facebook, we picked option 2 for the initial implementation.  Here is a diagram that shows how it works.


Here is a quick tour of the implementation in Java class THLParallelQueue.  This class maintains an in-memory blocking queue for each channel.  Each queue has a corresponding read thread that scans the THL and places matching events into the queue.  The THLParallelQueue class synchronizes read threads and handles issues like serialization and clean shutdown.  Some memory is therefore consumed, for queues, but they are quite small and amount to far less than keeping all transactions in memory.

So much for the theoretical description. If you would like to test on-disk queues yourself, you can get started in three steps.
  1. Download the latest nightly build of Tungsten Replicator to /tmp. 
  2. Untar and cd into the resulting release directory.  
  3. Install using the new tungsten-installer as described by Guiseppe Maxia.  
Here is an example of set-up commands.  My test system uses logos1 as the master and logos2 as the slave.  The MySQL DBMS build is Percona Server 5.1.54.  Tungsten is installed in /opt/tungsten.

# Download and unpack build.
cd /tmp
wget --no-check-certificate https://s3.amazonaws.com/files.continuent.com/builds/nightly/tungsten-2.0-snapshots/tungsten-replicator-2.0.4-154.tar.gz
# Untar.
tar -xvzf tungsten-replicator-2.0.4-154.tar.gz
cd tungsten-replicator-2.0.4-154
# Set up and start replicators. 
export TUNGSTEN_HOME=/opt/tungsten
/tmp/tungsten-replicator-2.0.4-154/tools/tungsten-installer \
  --master-slave  \
  --master-host=logos1  \
  --datasource-user=tungsten  \
  --datasource-password=secret  \
  --service-name=percona \
  --home-directory=${TUNGSTEN_HOME} \
  --cluster-hosts=logos1,logos2 \
  --relay-directory=${TUNGSTEN_HOME}/relay-logs \
  --datasource-log-directory=/usr/local/percona-5.1.54/data \
  --thl-directory=${TUNGSTEN_HOME}/thl-logs \
  --channels=10 \
  --svc-parallelization-type=disk \
  --start-and-report                                  

Note the bold options to select disk queues--"memory" is the other option--and the number of channels.  You can confirm you have the right queue installed by running the following command against any slave replicator.  You should see the storage class THLParallelQueue in the status output.

$ trepctl -host logos2 status -name stores
Processing status command (stores)...
NAME                VALUE
----                -----
doChecksum        : false
logDir            : /opt/rhodges4/thl-logs/percona
logFileRetention  : 7d
logFileSize       : 100000000
maximumStoredSeqNo: 0
minimumStoredSeqNo: 0
name              : thl
storeClass        : com.continuent.tungsten.replicator.thl.THL
NAME                VALUE
----                -----
criticalPartition : -1
discardCount      : 0
eventCount        : 1
headSeqno         : 0
maxSize           : 10
name              : parallel-queue
queues            : 10
serializationCount: 0
serialized        : false
stopRequested     : false
store.0           : THLParallelReadTask task_id=0 thread_name=store-thl-0 hi_seqno=0 lo_seqno=0 read=1 discarded=1 events=0
store.1           : THLParallelReadTask task_id=1 thread_name=store-thl-1 hi_seqno=0 lo_seqno=0 read=1 discarded=1 events=0
store.2           : THLParallelReadTask task_id=2 thread_name=store-thl-2 hi_seqno=0 lo_seqno=0 read=1 discarded=1 events=0
store.3           : THLParallelReadTask task_id=3 thread_name=store-thl-3 hi_seqno=0 lo_seqno=0 read=1 discarded=1 events=0
store.4           : THLParallelReadTask task_id=4 thread_name=store-thl-4 hi_seqno=0 lo_seqno=0 read=1 discarded=1 events=0
store.5           : THLParallelReadTask task_id=5 thread_name=store-thl-5 hi_seqno=0 lo_seqno=0 read=1 discarded=1 events=0
store.6           : THLParallelReadTask task_id=6 thread_name=store-thl-6 hi_seqno=0 lo_seqno=0 read=1 discarded=1 events=0
store.7           : THLParallelReadTask task_id=7 thread_name=store-thl-7 hi_seqno=0 lo_seqno=0 read=1 discarded=1 events=0
store.8           : THLParallelReadTask task_id=8 thread_name=store-thl-8 hi_seqno=0 lo_seqno=0 read=1 discarded=1 events=0
store.9           : THLParallelReadTask task_id=9 thread_name=store-thl-9 hi_seqno=0 lo_seqno=0 read=1 discarded=0 events=0
storeClass        : com.continuent.tungsten.replicator.thl.THLParallelQueue
syncEnabled       : true
syncInterval      : 2000
Finished status command (stores)...


Now for some fine print.  On-disk queues are implemented but are still undergoing QA.  There are bugs.  The most important problem is performance--the latency is a lot higher than expected on some of our systems, which I suspect is due to an as-yet undiagnosed bug.  If you try them out now you can expect to hit a few problems.  On the other hand, we take any and all feedback quite seriously, so this is your chance provide input and help guide the final implementation.  Please log issues on the Tungsten Replicator issue tracker or bring up questions on the tungsten-discuss mailing list.

Finally, if you would like to learn more about the parallel queue implementation, check out the design documentation on our wiki as well as the source code.  They are both pretty readable.