Drizzle Replication - The Transaction Log

Jay Pipes

Tuesday, October 27. 2009

Drizzle Replication - The Transaction Log

In this installment of my Drizzle Replication blog series, I'll be talking about the Transaction Log. Before reading this entry, you may want to first read up on the Transaction Message, which is a central concept to this blog entry.

The transaction log is just one component of Drizzle's default replication services, but it also serves as a generalized log of atomic data changes to a particular server. In this way, it is only partially related to replication. The transaction log is used by components of the replication services to store changes made to a server's data. However, there is nothing that mandates that this particular transaction log be a required feature for Drizzle replication systems. For instance, Eric Lambert is currently working on a Gearman-based replication service which, while following the same APIs, does not require the transaction log to function. Furthermore, other, non-replication-related modules may use the transaction log themselves. For instance, a future Recovery and/or Backup module may just as easily use the transaction log for its own purposes as well.

Before we get into the details, it's worth noting the general goals we've had for the transaction log, as these goals may help explain some of the design choices made. In short, the goals for the transaction log are:

  • Introduce no global contention points (mutexes/locks)
  • Once written, the transaction log may not be modified
  • The transaction log should be easily readable in multiple programming languages

Overview of the Transaction Log Structure

The format of the transaction log is simple and straightforward. It is a single file that contains log entries, one after another. These log entries have a type associated with them. Currently, there are only two types of entries that can go in the transaction log: a Transaction message entry and a BLOB entry. We will only cover the Transaction message entry in this article, as I'll leave how to deal with BLOBs for a separate article entirely.

Each entry in the transaction log is preceded by a 4 bytes containing an integer code identifying the type of entry to follow. The bytes which follow this type header are interpreted based on the type of entry. For entries of type Transaction message, the graphics here show the layout of the entry in the log. First, a 4 byte length header is written, then the serialized Transaction message, then a 4 byte checksum of the serialized Transaction message.


Details of the TransactionLog::apply() Method

For those interested in how the transaction log is written to, I'm going to detail the apply() method of the TransactionLog class in /plugin/transaction_log/transaction_log.cc. The TransactionLog class is simply a subclass of plugin::TransactionApplier and therefore must implement the single pure virtual apply method of that class interface.

The TransactionLog class has a private drizzled::atomic<off_t> called log_offset which is an offset into the transaction log file that is incremented with each atomic write to the log file. You will notice in the code below that this atomic off_t is stored locally, then incremented by the total length of the log entry to be written. A buffer is then written to the log file using pwrite() at the original offset. In this way, we completely avoid calling pthread_mutex_lock() or similar when writing to the log file, which should increase scalability of the transaction log.

  1. void TransactionLog::apply(const message::Transaction &to_apply)
  2. {
  3.   uint8_t *buffer; /* Buffer we will write serialized header,
  4.                       message and trailing checksum to */
  5.   uint8_t *orig_buffer;
  6.  
  7.   int error_code;
  8.   size_t message_byte_length= to_apply.ByteSize();
  9.   ssize_t written;
  10.   off_t cur_offset;
  11.   size_t total_envelope_length= HEADER_TRAILER_BYTES + message_byte_length;
  12.  
  13.   /*
  14.    * Attempt allocation of raw memory buffer for the header,
  15.    * message and trailing checksum bytes.
  16.    */
  17.   buffer= static_cast<uint8_t *>(malloc(total_envelope_length));
  18.   if (buffer == NULL)
  19.   {
  20.     errmsg_printf(ERRMSG_LVL_ERROR,
  21.                   _("Failed to allocate enough memory to buffer header, transaction message, and trailing checksum bytes. Tried to allocate %" PRId64
  22.                     " bytes.  Error: %s\n"),
  23.                   static_cast<int64_t>(total_envelope_length),
  24.                   strerror(errno));
  25.     state= CRASHED;
  26.     deactivate();
  27.     return;
  28.   }
  29.   else
  30.     orig_buffer= buffer; /* We will free() orig_buffer, as buffer is moved during write */
  31.  
  32.   /*
  33.    * Do an atomic increment on the offset of the log file position
  34.    */
  35.   cur_offset= log_offset.fetch_and_add(static_cast<off_t>(total_envelope_length));
  36.  
  37.   /*
  38.    * We adjust cur_offset back to the original log_offset before
  39.    * the increment above...
  40.    */
  41.   cur_offset-= static_cast<off_t>((total_envelope_length));
  42.  
  43.   /*
  44.    * Write the header information, which is the message type and
  45.    * the length of the transaction message into the buffer
  46.    */
  47.   buffer= protobuf::io::CodedOutputStream::WriteLittleEndian32ToArray(static_cast<uint32_t>(ReplicationServices::TRANSACTION), buffer);
  48.   buffer= protobuf::io::CodedOutputStream::WriteLittleEndian32ToArray(static_cast<uint32_t>(message_byte_length), buffer);
  49.  
  50.   /*
  51.    * Now write the serialized transaction message, followed
  52.    * by the optional checksum into the buffer.
  53.    */
  54.   buffer= to_apply.SerializeWithCachedSizesToArray(buffer);
  55.  
  56.   uint32_t checksum= 0;
  57.   if (do_checksum)
  58.   {
  59.     checksum= drizzled::hash::crc32(reinterpret_cast<char *>(buffer) - message_byte_length, message_byte_length);
  60.   }
  61.  
  62.   /* We always write in network byte order */
  63.   buffer= protobuf::io::CodedOutputStream::WriteLittleEndian32ToArray(checksum, buffer);
  64.  
  65.   /*
  66.    * Quick safety...if an error occurs above in another writer, the log
  67.    * file will be in a crashed state.
  68.    */
  69.   if (unlikely(state == CRASHED))
  70.   {
  71.     /*
  72.      * Reset the log's offset in case we want to produce a decent error message including
  73.      * the original offset where an error occurred.
  74.      */
  75.     log_offset= cur_offset;
  76.     free(orig_buffer);
  77.     return;
  78.   }
  79.  
  80.   /* Write the full buffer in one swoop */
  81.   do
  82.   {
  83.     written= pwrite(log_file, orig_buffer, total_envelope_length, cur_offset);
  84.   }
  85.   while (written == -1 && errno == EINTR); /* Just retry the write when interrupted by a signal... */
  86.  
  87.   if (unlikely(written != static_cast<ssize_t>(total_envelope_length)))
  88.   {
  89.     errmsg_printf(ERRMSG_LVL_ERROR,
  90.                   _("Failed to write full size of transaction.  Tried to write %" PRId64
  91.                     " bytes at offset %" PRId64 ", but only wrote %" PRId32 " bytes.  Error: %s\n"),
  92.                   static_cast<int64_t>(total_envelope_length),
  93.                   static_cast<int64_t>(cur_offset),
  94.                   static_cast<int64_t>(written),
  95.                   strerror(errno));
  96.     state= CRASHED;
  97.     /*
  98.      * Reset the log's offset in case we want to produce a decent error message including
  99.      * the original offset where an error occurred.
  100.      */
  101.     log_offset= cur_offset;
  102.     deactivate();
  103.   }
  104.   free(orig_buffer);
  105.  
  106.   error_code= my_sync(log_file, 0);
  107.  
  108.   if (unlikely(error_code != 0))
  109.   {
  110.     errmsg_printf(ERRMSG_LVL_ERROR,
  111.                   _("Failed to sync log file. Got error: %s\n"),
  112.                   strerror(errno));
  113.   }
  114. }

Reading the Transaction Log

OK, so the above code shows how the transaction log is written. What about reading the log file? Well, it's pretty simple. There is an example program in /drizzle/message/transaction_reader.cc which has code showing how to do this. Here's a snippet from that program:

  1. int main(int argc, char* argv[])
  2. {
  3. ...
  4.   message::Transaction transaction;
  5.  
  6.   file= open(argv[1], O_RDONLY);
  7.   if (file == -1)
  8.   {
  9.     fprintf(stderr, _("Cannot open file: %s\n"), argv[1]);
  10.     return -1;
  11.   }
  12. ...
  13.   protobuf::io::ZeroCopyInputStream *raw_input=
  14.     new protobuf::io::FileInputStream(file);
  15.   protobuf::io::CodedInputStream *coded_input=
  16.     new protobuf::io::CodedInputStream(raw_input);
  17.  
  18.   char *buffer= NULL;
  19.   char *temp_buffer= NULL;
  20.   uint32_t length= 0;
  21.   uint32_t previous_length= 0;
  22.   uint32_t checksum= 0;
  23.   bool result= true;
  24.   uint32_t message_type= 0;
  25.  
  26.   /* Read in the length of the command */
  27.   while (result == true &&
  28.          coded_input->ReadLittleEndian32(&message_type) == true &&
  29.          coded_input->ReadLittleEndian32(&length) == true)
  30.   {
  31.     if (message_type != ReplicationServices::TRANSACTION)
  32.     {
  33.       fprintf(stderr, _("Found a non-transaction message "
  34.                         "in log.  Currently, not supported.\n"));
  35.       exit(1);
  36.     }
  37.  
  38.     if (length > INT_MAX)
  39.     {
  40.       fprintf(stderr, _("Attempted to read record bigger than INT_MAX\n"));
  41.       exit(1);
  42.     }
  43.  
  44.     if (buffer == NULL)
  45.     {
  46.       /*
  47.        * First time around...just malloc the length.  This block gets rid
  48.        * of a GCC warning about uninitialized temp_buffer.
  49.        */
  50.       temp_buffer= (char *) malloc(static_cast<size_t>(length));
  51.     }
  52.     /* No need to allocate if we have a buffer big enough... */
  53.     else if (length > previous_length)
  54.     {
  55.       temp_buffer= (char *) realloc(buffer, static_cast<size_t>(length));
  56.     }
  57.  
  58.     if (temp_buffer == NULL)
  59.     {
  60.       fprintf(stderr, _("Memory allocation failure trying to "
  61.                         "allocate %" PRIu64 " bytes.\n"),
  62.               static_cast<uint64_t>(length));
  63.       break;
  64.     }
  65.     else
  66.       buffer= temp_buffer;
  67.  
  68.     /* Read the Command */
  69.     result= coded_input->ReadRaw(buffer, (int) length);
  70.     if (result == false)
  71.     {
  72.       fprintf(stderr, _("Could not read transaction message.\n"));
  73.       fprintf(stderr, _("GPB ERROR: %s.\n"), strerror(errno));
  74.       fprintf(stderr, _("Raw buffer read: %s.\n"), buffer);
  75.       break;
  76.     }
  77.  
  78.     result= transaction.ParseFromArray(buffer, static_cast<int32_t>(length));
  79.     if (result == false)
  80.     {
  81.       fprintf(stderr, _("Unable to parse command. Got error: %s.\n"),
  82.               transaction.InitializationErrorString().c_str());
  83.       if (buffer != NULL)
  84.         fprintf(stderr, _("BUFFER: %s\n"), buffer);
  85.       break;
  86.     }
  87.  
  88.     /* Print the transaction */
  89.     printTransaction(transaction);
  90.  
  91.     /* Skip 4 byte checksum */
  92.     coded_input->ReadLittleEndian32(&checksum);
  93.  
  94.     if (do_checksum)
  95.     {
  96.       if (checksum != drizzled::hash::crc32(buffer, static_cast<size_t>(length)))
  97.       {
  98.         fprintf(stderr, _("Checksum failed. Wanted %" PRIu32
  99.                           " got %" PRIu32 "\n"),
  100.                 checksum,
  101.                 drizzled::hash::crc32(buffer, static_cast<size_t>(length)));
  102.       }
  103.     }
  104.  
  105.     previous_length= length;
  106.   }
  107.   if (buffer)
  108.     free(buffer);
  109.  
  110.   delete coded_input;
  111.   delete raw_input;
  112.  
  113.   return (result == true ? 0 : 1);
  114. }

Shortcomings of the Transaction Log

So far, we've generally focused on a scalable design for the transaction log and have not spent too much time on performance tuning the code — and yes, performance != scalability. There are a number of problems with the current code which we will address in future versions of the transaction log. Namely:

  • Reduce calls to malloc(). Currently, each write of a transaction message to the log file incurs a call to malloc() to allocate enough memory to store the serialized log entry. Clearly, this is not optimal. We've considered a number of alternate approached to calling malloc(), including having a scoreboard approach where a vector of memory slabs are used in a round-robin fashion. This would introduce some locking, however. Also, I've thought about using a hazard pointer list on the Session object to have previously-allocated memory on the Session object be used for something like this. But, these ideas must be hashed out further.
  • There is no index into the transaction log. This is not a problem for writing the transaction log, of course, but for readers of the transaction log. I'm in the process of creating classes and a library for building indexes for a transaction log and, in addition, creating archived snapshots to enable log shipping for Drizzle replication. I'll be pushing code for this to Launchpad later this week and will write a new article about log shipping and snapshot creation.
  • Each call to apply() calls fdatasync()/fsync() on the transaction log. Certain environments may consider this to be too strict a sync requirement, since the storage engine may already keep a transaction log file of its own that is also synced. For instance, InnoDB has a transaction log that, depending on the setting of InnoDB configuration variables, may call fdatasync() upon every transaction commit. It would be best to have the syncing behaviour be user-adjustable — for instance, a setting to allow the transaction log to be synced every X number of seconds...

Summary and Request for Comments

That's it for the discussion about the transaction log. I'll post some more code examples from the replication plugins which utilize the transaction log in a later blog entry.

What do you think of the design of the transaction log? What would you change? Comments are always welcome! Cheers. :-)

Posted by Jay Pipes in MySQL, Drizzle, C/C++ at 13:08 | Comments (4) | Trackbacks (0)

Trackbacks
Trackback specific URI for this entry

No Trackbacks

Comments
Display comments as (Linear | Threaded)

Very much detailed and explanation article. Thank you.

sorry if i am coming with stupid points.
in TransactionLog::apply line no 41
it looks like offset is moving forward and then calculating and moving backward. is it something which can be avoided?

as you said the dynamic allocation is an issue.
anyway writing need to be serial. so what is the possibility of singleton for write operation?

Thank you,
Jobin.
#1 Jobin Augustine on 2009-10-29 01:52 (Reply)
Hi Jobin!

First of all, there aren't any stupid questions. :-) All of this code is extremely new!

So, on line 41:

cur_offset-= static_cast((total_envelope_length));

You are correct that we are recalculating the original offset there. The reason we do this is to "cut out" a chunk of the log file that we will use for writing the current transaction message.

Because TransactionLog::apply() is code that is run in a threaded server, the variables of the TransactionLog such as the log_offset variable must be protected against changes from another thread's execution of the TransactionLog::apply() method.

The log_offset variable is of type drizzled::atomic. This means that it's contents follow the atomic API templates implemented in /drizzled/atomic/. This template provides methods which ensure synchronized writes of a variable's contents in a threaded environment. The atomci.fetch_and_add() method atomically increments a variable's contents and returns the contents of the variable.

On line 35:

cur_offset= log_offset.fetch_and_add(static_cast(total_envelope_length));

what we are doing is carving out a chunk of space in the log file long enough to write our complete log entry. We "carve" this chunk out by simply atomically incrementing the log_offset. Once incremented, the new log_offset value is returned from fetch_and_add(), and we must therefore subtract the length of the transaction message we will write in order to return to the exact offset into the transaction log that we will write into.

Hope the above makes sense!

Cheers!

Jay
#1.1 Jay Pipes (Link) on 2009-10-29 10:09 (Reply)
Beautiful idea. :-)
carve out enough space for the current thread.
leave the other threads to carve out what they want.
..beautiful..beautiful.
threads need not wait for other threads to complete the writing. parallel writes can happen to log. am i right?

replication is going to Rock!
#1.1.1 Jobin Augustine on 2009-10-29 11:49 (Reply)
You got it! The key is the call to pwrite() (see line 83 above).

pwrite() is a standard POSIX system call that is similar to write() but allows the caller to write a supplied buffer of a supplied size to an exact offset in a file.

Here is the doc on pwrite() and write():

http://www.opengroup.org/onlinepubs/000095399/functions/write.html

Cheers!

Jay
#1.1.1.1 Jay Pipes (Link) on 2009-10-29 11:58 (Reply)

Add Comment

Enclosing asterisks marks text as bold (*word*), underscore are made via _word_.
Standard emoticons like :-) and ;-) are converted to images.
E-Mail addresses will not be displayed and will only be used for E-Mail notifications
 
 

Calendar

Back February '10
Mon Tue Wed Thu Fri Sat Sun
1 2 3 4 5 6 7
8 9 10 11 12 13 14
15 16 17 18 19 20 21
22 23 24 25 26 27 28

Quicksearch

Archives

February 2010
January 2010
December 2009
November 2009
October 2009
September 2009
Recent...
Older...

Categories


All categories

Syndicate This Blog