Merge lp:~jaypipes/drizzle/replication-ddl into lp:~drizzle-trunk/drizzle/development

[Jay Pipes]

    * Fixes drizzled's atomics:

    - fetch_and_add() was actually add_and_fetch() - fixed to have both methods correct
    - compare_and_swap() was incorrect for all traits classes. Fixed to return a bool
    true only when the supplied value is actually swapped
    - fixes increment() and decrement() methods and operator+=() in outer atomics class
    template to call proper add_and_fetch() methods on traits classes
    - Now that above are fixed, removed the hacks in Query_id and TransactionLog to
    have query ID and the new transactoin ID start properly at 1.

    * Transaction messages sent over replication stream now use
    a real transaction ID, managed by drizzled::TransactionServices. Previously,
    the Query_id was being used, resulting in SELECT statements incrementing the
    transaction ID.

    * Added a test case to ensure that DDL ops are given a transaction ID and SELECT
    ops do not increment the transaction ID.

    The transaction ID will be paired with a channel ID to become the global
    transaction identifier. ReplicationServices will manage the pairing of
    channel and transaction ID and understand how far a particular subscriber
    node has applied.

[Monty Taylor (mordred)]

It's missing an update to the sun studio atomic trait... but other than that looks good.

review: Needs Fixing

[Jay Pipes (jaypipes)]

Pushed Sun atomics fixes. Please review r1419. If we could run it through a hudson builder for Sun-only, that would be cool...

[Jay Pipes (jaypipes)]

OK, all green in build. Fixed Sun Studio atomics.

review: Needs Resubmitting

[lp:~jaypipes/drizzle/replication-ddl]

1423. By Jay Pipes <jpipes@serialcoder> on 2010-04-01

Fix cpplint header guard.

[Brian Aker (brianaker)]

This needs to be remerged with Trunk.

Thanks,
   -Brian

review: Needs Fixing

[Jay Pipes (jaypipes)]

This has already been merged with trunk.

[Brian Aker (brianaker)]

Sorry, I must have clicked on the wrong link. Though I am not seeing
this in the merge history.

On Apr 1, 2010, at 12:55 PM, Jay Pipes wrote:

> This has already been merged with trunk.
> --
> https://code.launchpad.net/~jaypipes/drizzle/replication-ddl/+merge/22558
> You are reviewing the proposed merge of lp:~jaypipes/drizzle/
> replication-ddl into lp:drizzle.

[Jay Pipes (jaypipes)]

No worries. I'm pretty sure I'm all merged up, though, as I see no revisions to pull if I bzr merge trunk locally...

[Jay Pipes (jaypipes)]

fyi, r1422 on this branch contains the most recent merge with trunk

[lp:~jaypipes/drizzle/replication-ddl]

1424. By Jay Pipes <jpipes@serialcoder> on 2010-04-02

merge trunk

1425. By Jay Pipes <jpipes@serialcoder> on 2010-04-02

merge replication-api

1426. By Jay Pipes <jpipes@serialcoder> on 2010-04-05

Merge trunk

1427. By Jay Pipes <jpipes@serialcoder> on 2010-04-06

Try another const_cast<> technique on pthread_traits.

=== modified file 'drizzled/atomic/gcc_traits.h'
--- drizzled/atomic/gcc_traits.h	2009-07-17 20:59:45 +0000
+++ drizzled/atomic/gcc_traits.h	2010-04-06 19:16:34 +0000
@@ -32,36 +32,37 @@
 
   gcc_traits() {}
 
-  /* YES. I know these are semantically backwards...
-   * so... TODO: Ensure we're doing the "right" thing here
-   */
+  inline value_type add_and_fetch(volatile value_type *value, D addend )
+  {
+    return __sync_add_and_fetch(value, addend);
+  }
+
   inline value_type fetch_and_add(volatile value_type *value, D addend )
   {
-    return __sync_add_and_fetch(value, addend);
+    return __sync_fetch_and_add(value, addend);
   }
 
   inline value_type fetch_and_increment(volatile value_type *value)
   {
-    return __sync_add_and_fetch(value, 1);
+    return __sync_fetch_and_add(value, 1);
   }
 
   inline value_type fetch_and_decrement(volatile value_type *value)
   {
-    return __sync_sub_and_fetch(value, 1);
+    return __sync_fetch_and_sub(value, 1);
   }
 
   inline value_type fetch_and_store(volatile value_type *value,
                                     value_type new_value)
   {
-    /* TODO: Is this the right one? */
     return __sync_lock_test_and_set(value, new_value);
   }
 
-  inline value_type compare_and_swap(volatile value_type *value,
+  inline bool compare_and_swap(volatile value_type *value,
                                      value_type new_value,
                                      value_type comparand )
   {
-    return __sync_val_compare_and_swap(value, comparand, new_value);
+    return __sync_bool_compare_and_swap(value, comparand, new_value);
   }
 
   inline value_type fetch(const volatile value_type *value) const volatile
@@ -76,7 +77,7 @@
      * Look at how to rewrite the below to something that ICC feels is
      * OK and yet respects memory barriers.
      */
-    return __sync_add_and_fetch(const_cast<value_type *>(value), 0);
+    return __sync_fetch_and_add(const_cast<value_type *>(value), 0);
   }
 
   inline value_type store_with_release(volatile value_type *value,

=== modified file 'drizzled/atomic/pthread_traits.h'
--- drizzled/atomic/pthread_traits.h	2009-12-14 19:51:51 +0000
+++ drizzled/atomic/pthread_traits.h	2010-04-06 19:16:34 +0000
@@ -67,11 +67,20 @@
 
   pthread_traits() {}
 
+  inline value_type add_and_fetch(volatile value_type *value, D addend )
+  {
+    my_lock.lock();
+    *value += addend;
+    value_type ret= *value;
+    my_lock.unlock();
+    return ret;
+  }
+
   inline value_type fetch_and_add(volatile value_type *value, D addend )
   {
     my_lock.lock();
+    value_type ret= *value;
     *value += addend;
-    value_type ret= *value;
     my_lock.unlock();
     return ret;
   }
@@ -79,8 +88,8 @@
   inline value_type fetch_and_increment(volatile value_type *value)
   {
     my_lock.lock();
+    value_type ret= *value;
     *value++;
-    value_type ret= *value;
     my_lock.unlock();
     return ret;
   }
@@ -88,8 +97,8 @@
   inline value_type fetch_and_decrement(volatile value_type *value)
   {
     my_lock.lock();
+    value_type ret= *value;
     *value--;
-    value_type ret= *value;
     my_lock.unlock();
     return ret;
   }
@@ -98,27 +107,30 @@
                                     value_type new_value )
   {
     my_lock.lock();
+    value_type ret= *value;
     *value= new_value;
-    value_type ret= *value;
     my_lock.unlock();
     return ret;
   }
 
-  inline value_type compare_and_swap(volatile value_type *value,
+  inline bool compare_and_swap(volatile value_type *value,
                                      value_type new_value,
                                      value_type comparand )
   {
     my_lock.lock();
-    if (*value == comparand)
+    bool ret= (*value == comparand);
+    if (ret)
       *value= new_value;
-    value_type ret= *value;
     my_lock.unlock();
     return ret;
   }
 
   inline value_type fetch(const volatile value_type *value) const volatile
   {
-    return *value;
+    const_cast<pthread_traits *>(this)->my_lock.lock();
+    value_type ret= *value;
+    const_cast<pthread_traits *>(this)->my_lock.unlock();
+    return ret;
   }
 
   inline value_type store_with_release(volatile value_type *value,

=== modified file 'drizzled/atomic/sun_studio.h'
--- drizzled/atomic/sun_studio.h	2009-08-30 00:26:17 +0000
+++ drizzled/atomic/sun_studio.h	2010-04-06 19:16:34 +0000
@@ -24,91 +24,177 @@
 #include <atomic.h>
 #undef _KERNEL
 
+inline bool __sync_fetch_and_add(volatile bool* ptr, bool val)
+{
+  bool ret= *ptr;
+  (val == true) ? atomic_inc_8((volatile uint8_t *)ptr) : atomic_add_8((volatile uint8_t *)ptr, (int8_t)val);
+  return ret;
+}
+ 
+inline int8_t __sync_fetch_and_add(volatile int8_t* ptr, int8_t val)
+{
+  int8_t ret= *ptr; 
+  (val == 1) ? atomic_inc_8((volatile uint8_t*)ptr) : atomic_add_8((volatile uint8_t*)ptr, val);
+  return ret;
+}
+
+inline int16_t __sync_fetch_and_add(volatile int16_t* ptr, int16_t val)
+{
+  int16_t ret= *ptr;
+  (val == 1) ? atomic_inc_16((volatile uint16_t*)ptr) : atomic_add_16((volatile uint16_t*)ptr, val);
+  return ret;
+}
+
+inline int32_t __sync_fetch_and_add(volatile int32_t* ptr, int32_t val)
+{
+  int32_t ret= *ptr;
+  (val == 1) ? atomic_inc_32((volatile uint32_t*)ptr) : atomic_add_32((volatile uint32_t*)ptr, val);
+  return ret;
+}
+
+inline uint8_t __sync_fetch_and_add(volatile uint8_t* ptr, uint8_t val)
+{
+  uint8_t ret= *ptr;
+  (val == 1) ? atomic_inc_8(ptr) : atomic_add_8(ptr, (int8_t)val);
+  return ret;
+}
+
+inline uint16_t __sync_fetch_and_add(volatile uint16_t* ptr, uint16_t val)
+{
+  uint16_t ret= *ptr;
+  (val == 1) ? atomic_inc_16(ptr) : atomic_add_16(ptr, (int16_t)val);
+  return ret;
+}
+
+inline uint32_t __sync_fetch_and_add(volatile uint32_t* ptr, uint32_t val)
+{
+  uint32_t ret= *ptr;
+  (val == 1) ? atomic_inc_32(ptr) : atomic_add_32(ptr, (int32_t)val);
+  return ret;
+}
+
+# if defined(_KERNEL) || defined(_INT64_TYPE)
+inline uint64_t __sync_fetch_and_add(volatile uint64_t* ptr, uint64_t val)
+{
+  uint64_t ret= *ptr;
+  (val == 1) ? atomic_inc_64(ptr) : atomic_add_64(ptr, (int64_t)val);
+  return ret;
+}
+
+inline int64_t __sync_fetch_and_add(volatile int64_t* ptr, int64_t val)
+{
+  int64_t ret= *ptr;
+  (val == 1) ? atomic_inc_64((volatile uint64_t*)ptr) : atomic_add_64((volatile uint64_t*)ptr, val);
+  return ret;
+}
+# endif /* defined(_KERNEL) || defined(_INT64_TYPE) */
+
+inline uint8_t __sync_fetch_and_sub(volatile uint8_t* ptr, uint8_t val)
+{
+  uint8_t ret= *ptr;
+  (val == 1) ? atomic_dec_8(ptr) : atomic_add_8(ptr, 0-(int8_t)val);
+  return ret;
+}
+
+inline uint16_t __sync_fetch_and_sub(volatile uint16_t* ptr, uint16_t val)
+{
+  uint16_t ret= *ptr;
+  (val == 1) ? atomic_dec_16(ptr) : atomic_add_16(ptr, 0-(int16_t)val);
+  return ret;
+}
+
+inline uint32_t __sync_fetch_and_sub(volatile uint32_t* ptr, uint32_t val)
+{
+  uint32_t ret= *ptr;
+  (val == 1) ? atomic_dec_32(ptr) : atomic_add_32(ptr, 0-(int32_t)val);
+  return ret;
+}
+
+# if defined(_KERNEL) || defined(_INT64_TYPE)
+inline uint64_t __sync_fetch_and_sub(volatile uint64_t* ptr, uint64_t val)
+{
+  uint64_t ret= *ptr;
+  (val == 1) ? atomic_dec_64(ptr) : atomic_add_64(ptr, 0-(int64_t)val);
+  return ret;
+}
+inline int64_t __sync_fetch_and_sub(volatile int64_t* ptr, uint64_t val)
+{
+  int64_t ret= *ptr;
+  (val == 1) ? atomic_dec_64((volatile uint64_t *) ptr) : atomic_add_64((volatile uint64_t *) ptr, 0-(int64_t)val);
+  return ret;
+}
+# endif /* defined(_KERNEL) || defined(_INT64_TYPE) */
+
 inline bool __sync_add_and_fetch(volatile bool* ptr, bool val)
 {
-  (val == true) ? atomic_inc_8((volatile uint8_t *)ptr) : atomic_add_8((volatile uint8_t *)ptr, (int8_t)val);
-  return *ptr;
+  return (val == true) ? atomic_inc_8_nv((volatile uint8_t *)ptr) : atomic_add_8_nv((volatile uint8_t *)ptr, (int8_t)val);
 }
  
 inline int8_t __sync_add_and_fetch(volatile int8_t* ptr, int8_t val)
 {
-  (val == 1) ? atomic_inc_8((volatile uint8_t*)ptr) : atomic_add_8((volatile uint8_t*)ptr, val);
-  return *ptr;
+  return (val == 1) ? atomic_inc_8_nv((volatile uint8_t*)ptr) : atomic_add_8_nv((volatile uint8_t*)ptr, val);
 }
 
 inline int16_t __sync_add_and_fetch(volatile int16_t* ptr, int16_t val)
 {
-  (val == 1) ? atomic_inc_16((volatile uint16_t*)ptr) : atomic_add_16((volatile uint16_t*)ptr, val);
-  return *ptr;
+  return (val == 1) ? atomic_inc_16_nv((volatile uint16_t*)ptr) : atomic_add_16_nv((volatile uint16_t*)ptr, val);
 }
 
 inline int32_t __sync_add_and_fetch(volatile int32_t* ptr, int32_t val)
 {
-  (val == 1) ? atomic_inc_32((volatile uint32_t*)ptr) : atomic_add_32((volatile uint32_t*)ptr, val);
-  return *ptr;
+  return (val == 1) ? atomic_inc_32_nv((volatile uint32_t*)ptr) : atomic_add_32_nv((volatile uint32_t*)ptr, val);
 }
 
 inline uint8_t __sync_add_and_fetch(volatile uint8_t* ptr, uint8_t val)
 {
-  (val == 1) ? atomic_inc_8(ptr) : atomic_add_8(ptr, (int8_t)val);
-  return *ptr;
+  return (val == 1) ? atomic_inc_8_nv(ptr) : atomic_add_8_nv(ptr, (int8_t)val);
 }
 
 inline uint16_t __sync_add_and_fetch(volatile uint16_t* ptr, uint16_t val)
 {
-  (val == 1) ? atomic_inc_16(ptr) : atomic_add_16(ptr, (int16_t)val);
-  return *ptr;
+  return (val == 1) ? atomic_inc_16_nv(ptr) : atomic_add_16_nv(ptr, (int16_t)val);
 }
 
 inline uint32_t __sync_add_and_fetch(volatile uint32_t* ptr, uint32_t val)
 {
-  (val == 1) ? atomic_inc_32(ptr) : atomic_add_32(ptr, (int32_t)val);
-  return *ptr;
+  return (val == 1) ? atomic_inc_32_nv(ptr) : atomic_add_32_nv(ptr, (int32_t)val);
 }
 
 # if defined(_KERNEL) || defined(_INT64_TYPE)
 inline uint64_t __sync_add_and_fetch(volatile uint64_t* ptr, uint64_t val)
 {
-  (val == 1) ? atomic_inc_64(ptr) : atomic_add_64(ptr, (int64_t)val);
-  return *ptr;
+  return (val == 1) ? atomic_inc_64_nv(ptr) : atomic_add_64_nv(ptr, (int64_t)val);
 }
 
 inline int64_t __sync_add_and_fetch(volatile int64_t* ptr, int64_t val)
 {
-  (val == 1) ? atomic_inc_64((volatile uint64_t*)ptr) : atomic_add_64((volatile uint64_t*)ptr, val);
-  return *ptr;
+  return (val == 1) ? atomic_inc_64_nv((volatile uint64_t*)ptr) : atomic_add_64_nv((volatile uint64_t*)ptr, val);
 }
 # endif /* defined(_KERNEL) || defined(_INT64_TYPE) */
 
-
 inline uint8_t __sync_sub_and_fetch(volatile uint8_t* ptr, uint8_t val)
 {
-  (val == 1) ? atomic_dec_8(ptr) : atomic_add_8(ptr, 0-(int8_t)val);
-  return *ptr;
+  return (val == 1) ? atomic_dec_8_nv(ptr) : atomic_add_8_nv(ptr, 0-(int8_t)val);
 }
 
 inline uint16_t __sync_sub_and_fetch(volatile uint16_t* ptr, uint16_t val)
 {
-  (val == 1) ? atomic_dec_16(ptr) : atomic_add_16(ptr, 0-(int16_t)val);
-  return *ptr;
+  return (val == 1) ? atomic_dec_16_nv(ptr) : atomic_add_16_nv(ptr, 0-(int16_t)val);
 }
 
 inline uint32_t __sync_sub_and_fetch(volatile uint32_t* ptr, uint32_t val)
 {
-  (val == 1) ? atomic_dec_32(ptr) : atomic_add_32(ptr, 0-(int32_t)val);
-  return *ptr;
+  return (val == 1) ? atomic_dec_32_nv(ptr) : atomic_add_32_nv(ptr, 0-(int32_t)val);
 }
 
 # if defined(_KERNEL) || defined(_INT64_TYPE)
 inline uint64_t __sync_sub_and_fetch(volatile uint64_t* ptr, uint64_t val)
 {
-  (val == 1) ? atomic_dec_64(ptr) : atomic_add_64(ptr, 0-(int64_t)val);
-  return *ptr;
+  return (val == 1) ? atomic_dec_64_nv(ptr) : atomic_add_64_nv(ptr, 0-(int64_t)val);
 }
 inline int64_t __sync_sub_and_fetch(volatile int64_t* ptr, uint64_t val)
 {
-  (val == 1) ? atomic_dec_64((volatile uint64_t *) ptr) : atomic_add_64((volatile uint64_t *) ptr, 0-(int64_t)val);
-  return *ptr;
+  return (val == 1) ? atomic_dec_64_nv((volatile uint64_t *) ptr) : atomic_add_64_nv((volatile uint64_t *) ptr, 0-(int64_t)val);
 }
 # endif /* defined(_KERNEL) || defined(_INT64_TYPE) */
 
@@ -175,4 +261,41 @@
 }
 #endif /* defined(_KERNEL) || defined(_INT64_TYPE) */
 
+inline int8_t __sync_bool_compare_and_swap(volatile int8_t* ptr,
+                                           int8_t old_val, int8_t val)
+{
+  int8_t orig= *ptr;
+  return orig == atomic_cas_8((volatile uint8_t *)ptr, old_val, val);
+}
+
+inline uint8_t __sync_bool_compare_and_swap(volatile uint8_t* ptr,
+                                           uint8_t old_val, uint8_t val)
+{
+  uint8_t orig= *ptr;
+  return orig == atomic_cas_8(ptr, old_val, val);
+}
+
+inline uint16_t __sync_bool_compare_and_swap(volatile uint16_t* ptr,
+                                            uint16_t old_val, uint16_t val)
+{
+  uint16_t orig= *ptr;
+  return orig == atomic_cas_16(ptr, old_val, val);
+}
+
+inline uint32_t __sync_bool_compare_and_swap(volatile uint32_t* ptr,
+                                            uint32_t old_val, uint32_t val)
+{
+  uint32_t orig= *ptr;
+  return orig == atomic_cas_32(ptr, old_val, val);
+}
+
+# if defined(_KERNEL) || defined(_INT64_TYPE)
+inline uint64_t __sync_bool_compare_and_swap(volatile uint64_t* ptr,
+                                            uint64_t old_val, uint64_t val)
+{
+  uint64_t orig= *ptr;
+  return orig == atomic_cas_64(ptr, old_val, val);
+}
+#endif /* defined(_KERNEL) || defined(_INT64_TYPE) */
+
 #endif /* DRIZZLED_ATOMIC_SUN_STUDIO_H */

=== modified file 'drizzled/atomics.h'
--- drizzled/atomics.h	2010-01-04 22:52:52 +0000
+++ drizzled/atomics.h	2010-04-06 19:16:34 +0000
@@ -58,6 +58,11 @@
 
   atomic_impl() : atomic_base<I>(), traits() {}
 
+  value_type add_and_fetch( D addend )
+  {
+    return traits.add_and_fetch(&this->my_value, addend);
+  }
+
   value_type fetch_and_add( D addend )
   {
     return traits.fetch_and_add(&this->my_value, addend);
@@ -78,7 +83,7 @@
     return traits.fetch_and_store(&this->my_value, value);
   }
 
-  value_type compare_and_swap( value_type value, value_type comparand )
+  bool compare_and_swap( value_type value, value_type comparand )
   {
     return traits.compare_and_swap(&this->my_value, value, comparand);
   }
@@ -102,7 +107,7 @@
 public:
   atomic_impl<I,D,T>& operator+=( D addend )
   {
-    fetch_and_add(addend)+addend;
+    increment(addend);
     return *this;
   }
 
@@ -113,15 +118,15 @@
     return operator+=(D(0)-addend);
   }
 
-  value_type increment() {
-    return fetch_and_add(1)+1;
-  }
-
-  value_type decrement() {
-    return fetch_and_add(D(-1))-1;
-  }
-
-
+  value_type increment()
+  {
+    return add_and_fetch(1);
+  }
+
+  value_type decrement()
+  {
+    return add_and_fetch(D(-1));
+  }
 };
 
 } /* namespace internal */

=== modified file 'drizzled/cursor.cc'
--- drizzled/cursor.cc	2010-04-01 06:30:46 +0000
+++ drizzled/cursor.cc	2010-04-06 19:16:34 +0000
@@ -36,7 +36,6 @@
 #include "drizzled/session.h"
 #include "drizzled/sql_base.h"
 #include "drizzled/transaction_services.h"
-#include "drizzled/replication_services.h"
 #include "drizzled/lock.h"
 #include "drizzled/item/int.h"
 #include "drizzled/item/empty_string.h"
@@ -44,7 +43,6 @@
 #include "drizzled/message/table.pb.h"
 #include "drizzled/plugin/client.h"
 #include "drizzled/internal/my_sys.h"
-#include "drizzled/transaction_services.h"
 
 using namespace std;
 
@@ -1297,10 +1295,9 @@
                                     const unsigned char *after_record)
 {
   TransactionServices &transaction_services= TransactionServices::singleton();
-  ReplicationServices &replication_services= ReplicationServices::singleton();
   Session *const session= table->in_use;
 
-  if (table->s->tmp_table || not replication_services.isActive())
+  if (table->s->tmp_table || not transaction_services.shouldConstructMessages())
     return false;
 
   bool result= false;

=== modified file 'drizzled/include.am'
--- drizzled/include.am	2010-04-02 07:45:12 +0000
+++ drizzled/include.am	2010-04-06 19:16:34 +0000
@@ -307,6 +307,7 @@
 			      drizzled/plugin/query_cache.h \
 		              drizzled/plugin/query_rewrite.h \
 			      drizzled/plugin/registry.h \
+			      drizzled/plugin/replication.h \
 			      drizzled/plugin/scheduler.h \
 			      drizzled/plugin/storage_engine.h \
 			      drizzled/plugin/table_function.h \

=== added file 'drizzled/plugin/replication.h'
--- drizzled/plugin/replication.h	1970-01-01 00:00:00 +0000
+++ drizzled/plugin/replication.h	2010-04-06 19:16:34 +0000
@@ -0,0 +1,46 @@
+/* -*- mode: c++; c-basic-offset: 2; indent-tabs-mode: nil; -*-
+ *  vim:expandtab:shiftwidth=2:tabstop=2:smarttab:
+ *
+ *  Copyright (c) 2010 Jay Pipes
+ *
+ *  Authors:
+ *
+ *    Jay Pipes <[e-mail]>
+ *
+ *  This program is free software; you can redistribute it and/or modify
+ *  it under the terms of the GNU General Public License as published by
+ *  the Free Software Foundation; version 2 of the License.
+ *
+ *  This program is distributed in the hope that it will be useful,
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *  GNU General Public License for more details.
+ *
+ *  You should have received a copy of the GNU General Public License
+ *  along with this program; if not, write to the Free Software
+ *  Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
+ */
+
+#ifndef DRIZZLED_PLUGIN_REPLICATION_H
+#define DRIZZLED_PLUGIN_REPLICATION_H
+
+/**
+ * @file Common structs and enums for the replication API
+ */
+
+namespace drizzled
+{
+
+namespace plugin
+{
+
+enum ReplicationReturnCode
+{
+  SUCCESS= 0, /* no error */
+  UNKNOWN_ERROR= 1
+};
+
+} /* namespace plugin */
+} /* namespace drizzled */
+
+#endif /* DRIZZLED_PLUGIN_REPLICATION_H */

=== modified file 'drizzled/plugin/transaction_applier.h'
--- drizzled/plugin/transaction_applier.h	2009-10-28 16:38:30 +0000
+++ drizzled/plugin/transaction_applier.h	2010-04-06 19:16:34 +0000
@@ -2,10 +2,11 @@
  *  vim:expandtab:shiftwidth=2:tabstop=2:smarttab:
  *
  *  Copyright (C) 2008-2009 Sun Microsystems
+ *  Copyright (c) 2010 Jay Pipes
  *
  *  Authors:
  *
- *    Jay Pipes <[e-mail]>
+ *    Jay Pipes <[e-mail]>
  *
  *  This program is free software; you can redistribute it and/or modify
  *  it under the terms of the GNU General Public License as published by
@@ -33,11 +34,14 @@
  */
 
 #include "drizzled/plugin/plugin.h"
+#include "drizzled/plugin/replication.h"
 #include "drizzled/atomics.h"
 
 namespace drizzled
 {
 
+class Session;
+
 namespace message { class Transaction; }
 
 namespace plugin
@@ -74,7 +78,8 @@
    *
    * @param Transaction message to be replicated
    */
-  virtual void apply(const message::Transaction &to_apply)= 0;
+  virtual ReplicationReturnCode apply(Session &in_session,
+                                      const message::Transaction &to_apply)= 0;
 
   static bool addPlugin(TransactionApplier *applier);
   static void removePlugin(TransactionApplier *applier);

=== modified file 'drizzled/plugin/transaction_replicator.h'
--- drizzled/plugin/transaction_replicator.h	2009-12-04 23:47:14 +0000
+++ drizzled/plugin/transaction_replicator.h	2010-04-06 19:16:34 +0000
@@ -2,10 +2,11 @@
  *  vim:expandtab:shiftwidth=2:tabstop=2:smarttab:
  *
  *  Copyright (C) 2008-2009 Sun Microsystems
+ *  Copyright (c) 2010 Jay Pipes
  *
  *  Authors:
  *
- *    Jay Pipes <[e-mail]>
+ *    Jay Pipes <[e-mail]>
  *
  *  This program is free software; you can redistribute it and/or modify
  *  it under the terms of the GNU General Public License as published by
@@ -25,6 +26,7 @@
 #define DRIZZLED_PLUGIN_TRANSACTION_REPLICATOR_H
 
 #include "drizzled/atomics.h"
+#include "drizzled/plugin/replication.h"
 #include "drizzled/plugin/plugin.h"
 
 /**
@@ -36,7 +38,6 @@
  * An applier is responsible for applying events, not a replicator...
  */
 
-
 namespace drizzled
 {
 namespace message
@@ -45,6 +46,8 @@
   class Statement;
 }
 
+class Session;
+
 namespace plugin
 {
 
@@ -83,8 +86,9 @@
    * @param Pointer to the applier of the command message
    * @param Transaction message to be replicated
    */
-  virtual void replicate(TransactionApplier *in_applier, 
-                         message::Transaction &to_replicate)= 0;
+  virtual ReplicationReturnCode replicate(TransactionApplier *in_applier, 
+                                          Session &session,
+                                          message::Transaction &to_replicate)= 0;
   static bool addPlugin(TransactionReplicator *replicator);
   static void removePlugin(TransactionReplicator *replicator);
 

=== modified file 'drizzled/replication_services.cc'
--- drizzled/replication_services.cc	2010-03-17 03:11:34 +0000
+++ drizzled/replication_services.cc	2010-04-06 19:16:34 +0000
@@ -39,8 +39,6 @@
 #include "drizzled/plugin/transaction_replicator.h"
 #include "drizzled/plugin/transaction_applier.h"
 #include "drizzled/message/transaction.pb.h"
-#include "drizzled/message/table.pb.h"
-#include "drizzled/message/statement_transform.h"
 #include "drizzled/gettext.h"
 #include "drizzled/session.h"
 #include "drizzled/error.h"
@@ -145,7 +143,8 @@
   return is_active;
 }
 
-void ReplicationServices::pushTransactionMessage(message::Transaction &to_push)
+plugin::ReplicationReturnCode ReplicationServices::pushTransactionMessage(Session &in_session,
+                                                                          message::Transaction &to_push)
 {
   vector<plugin::TransactionReplicator *>::iterator repl_iter= replicators.begin();
   vector<plugin::TransactionApplier *>::iterator appl_start_iter, appl_iter;
@@ -154,6 +153,8 @@
   plugin::TransactionReplicator *cur_repl;
   plugin::TransactionApplier *cur_appl;
 
+  plugin::ReplicationReturnCode result= plugin::SUCCESS;
+
   while (repl_iter != replicators.end())
   {
     cur_repl= *repl_iter;
@@ -174,19 +175,25 @@
         continue;
       }
 
-      cur_repl->replicate(cur_appl, to_push);
-      
-      /* 
-       * We update the timestamp for the last applied Transaction so that
-       * publisher plugins can ask the replication services when the
-       * last known applied Transaction was using the getLastAppliedTimestamp()
-       * method.
-       */
-      last_applied_timestamp.fetch_and_store(to_push.transaction_context().end_timestamp());
-      ++appl_iter;
+      result= cur_repl->replicate(cur_appl, in_session, to_push);
+
+      if (result == plugin::SUCCESS)
+      {
+        /* 
+         * We update the timestamp for the last applied Transaction so that
+         * publisher plugins can ask the replication services when the
+         * last known applied Transaction was using the getLastAppliedTimestamp()
+         * method.
+         */
+        last_applied_timestamp.fetch_and_store(to_push.transaction_context().end_timestamp());
+        ++appl_iter;
+      }
+      else
+        return result;
     }
     ++repl_iter;
   }
+  return result;
 }
 
 } /* namespace drizzled */

=== modified file 'drizzled/replication_services.h'
--- drizzled/replication_services.h	2010-03-16 21:30:44 +0000
+++ drizzled/replication_services.h	2010-04-06 19:16:34 +0000
@@ -26,6 +26,7 @@
 #define DRIZZLED_REPLICATION_SERVICES_H
 
 #include "drizzled/atomics.h"
+#include "drizzled/plugin/replication.h"
 
 #include <vector>
 
@@ -96,9 +97,11 @@
    * Helper method which pushes a constructed message out to the registered
    * replicator and applier plugins.
    *
+   * @param Session descriptor
    * @param Message to push out
    */
-  void pushTransactionMessage(message::Transaction &to_push);
+  plugin::ReplicationReturnCode pushTransactionMessage(Session &in_session,
+                                                       message::Transaction &to_push);
   /**
    * Constructor
    */

=== modified file 'drizzled/session.cc'
--- drizzled/session.cc	2010-03-31 22:22:16 +0000
+++ drizzled/session.cc	2010-04-06 19:16:34 +0000
@@ -369,7 +369,7 @@
 #endif
   {
     TransactionServices &transaction_services= TransactionServices::singleton();
-    transaction_services.ha_rollback_trans(this, true);
+    transaction_services.rollbackTransaction(this, true);
     xid_cache_delete(&transaction.xid_state);
   }
   hash_free(&user_vars);
@@ -772,7 +772,7 @@
        * (Which of course should never happen...)
        */
       server_status&= ~SERVER_STATUS_IN_TRANS;
-      if (transaction_services.ha_commit_trans(this, true))
+      if (transaction_services.commitTransaction(this, true))
         result= false;
       options&= ~(OPTION_BEGIN);
       break;
@@ -789,7 +789,7 @@
     case ROLLBACK_AND_CHAIN:
     {
       server_status&= ~SERVER_STATUS_IN_TRANS;
-      if (transaction_services.ha_rollback_trans(this, true))
+      if (transaction_services.rollbackTransaction(this, true))
         result= false;
       options&= ~(OPTION_BEGIN);
       if (result == true && (completion == ROLLBACK_AND_CHAIN))
@@ -822,7 +822,7 @@
   if (options & (OPTION_NOT_AUTOCOMMIT | OPTION_BEGIN))
   {
     server_status&= ~SERVER_STATUS_IN_TRANS;
-    if (transaction_services.ha_commit_trans(this, true))
+    if (transaction_services.commitTransaction(this, true))
       result= false;
   }
   options&= ~(OPTION_BEGIN);
@@ -1947,7 +1947,7 @@
   {
     TransactionServices &transaction_services= TransactionServices::singleton();
     main_da.can_overwrite_status= true;
-    transaction_services.ha_autocommit_or_rollback(this, is_error());
+    transaction_services.autocommitOrRollback(this, is_error());
     main_da.can_overwrite_status= false;
     transaction.stmt.reset();
   }

=== modified file 'drizzled/set_var.cc'
--- drizzled/set_var.cc	2010-03-30 09:29:02 +0000
+++ drizzled/set_var.cc	2010-04-06 19:16:34 +0000
@@ -1381,7 +1381,7 @@
       session->options&= ~(uint64_t) (OPTION_BEGIN);
       session->server_status|= SERVER_STATUS_AUTOCOMMIT;
       TransactionServices &transaction_services= TransactionServices::singleton();
-      if (transaction_services.ha_commit_trans(session, true))
+      if (transaction_services.commitTransaction(session, true))
         return 1;
     }
     else

=== modified file 'drizzled/sql_delete.cc'
--- drizzled/sql_delete.cc	2010-02-06 03:22:59 +0000
+++ drizzled/sql_delete.cc	2010-04-06 19:16:34 +0000
@@ -396,7 +396,7 @@
     Safety, in case the engine ignored ha_enable_transaction(false)
     above. Also clears session->transaction.*.
   */
-  error= transaction_services.ha_autocommit_or_rollback(&session, error);
+  error= transaction_services.autocommitOrRollback(&session, error);
   session.options= save_options;
 
   return error;

=== modified file 'drizzled/sql_insert.cc'
--- drizzled/sql_insert.cc	2010-03-31 21:15:40 +0000
+++ drizzled/sql_insert.cc	2010-04-06 19:16:34 +0000
@@ -1325,7 +1325,7 @@
   {
     /*
       We must invalidate the table in the query cache before binlog writing
-      and ha_autocommit_or_rollback.
+      and autocommitOrRollback.
     */
     if (session->transaction.stmt.hasModifiedNonTransData())
       session->transaction.all.markModifiedNonTransData();
@@ -1711,7 +1711,7 @@
     if (!table->s->tmp_table)
     {
       TransactionServices &transaction_services= TransactionServices::singleton();
-      transaction_services.ha_autocommit_or_rollback(session, 0);
+      transaction_services.autocommitOrRollback(session, 0);
       (void) session->endActiveTransaction();
     }
 

=== modified file 'drizzled/sql_parse.cc'
--- drizzled/sql_parse.cc	2010-03-31 21:15:40 +0000
+++ drizzled/sql_parse.cc	2010-04-06 19:16:34 +0000
@@ -254,7 +254,7 @@
   /* If commit fails, we should be able to reset the OK status. */
   session->main_da.can_overwrite_status= true;
   TransactionServices &transaction_services= TransactionServices::singleton();
-  transaction_services.ha_autocommit_or_rollback(session, session->is_error());
+  transaction_services.autocommitOrRollback(session, session->is_error());
   session->main_da.can_overwrite_status= false;
 
   session->transaction.stmt.reset();

=== modified file 'drizzled/sql_table.cc'
--- drizzled/sql_table.cc	2010-03-31 21:15:40 +0000
+++ drizzled/sql_table.cc	2010-04-06 19:16:34 +0000
@@ -1829,7 +1829,7 @@
       length= snprintf(buff, sizeof(buff), ER(ER_OPEN_AS_READONLY),
                        table_name);
       session->client->store(buff, length);
-      transaction_services.ha_autocommit_or_rollback(session, false);
+      transaction_services.autocommitOrRollback(session, false);
       session->endTransaction(COMMIT);
       session->close_thread_tables();
       lex->reset_query_tables_list(false);
@@ -1952,7 +1952,7 @@
         }
       }
     }
-    transaction_services.ha_autocommit_or_rollback(session, false);
+    transaction_services.autocommitOrRollback(session, false);
     session->endTransaction(COMMIT);
     session->close_thread_tables();
     table->table=0;				// For query cache
@@ -1964,7 +1964,7 @@
   return(false);
 
 err:
-  transaction_services.ha_autocommit_or_rollback(session, true);
+  transaction_services.autocommitOrRollback(session, true);
   session->endTransaction(ROLLBACK);
   session->close_thread_tables();			// Shouldn't be needed
   if (table)

=== modified file 'drizzled/sql_update.cc'
--- drizzled/sql_update.cc	2010-03-26 21:33:42 +0000
+++ drizzled/sql_update.cc	2010-04-06 19:16:34 +0000
@@ -523,7 +523,7 @@
     last one without error. error > 0 means an error (e.g. unique key
     violation and no IGNORE or REPLACE). error == 0 is also an error (if
     preparing the record or invoking before triggers fails). See
-    ha_autocommit_or_rollback(error>=0) and return(error>=0) below.
+    autocommitOrRollback(error>=0) and return(error>=0) below.
     Sometimes we want to binlog even if we updated no rows, in case user used
     it to be sure master and slave are in same state.
   */

=== modified file 'drizzled/statement/alter_table.cc'
--- drizzled/statement/alter_table.cc	2010-03-31 22:22:16 +0000
+++ drizzled/statement/alter_table.cc	2010-04-06 19:16:34 +0000
@@ -619,7 +619,7 @@
     goto err;
 
   /* The ALTER Table is always in its own transaction */
-  error= transaction_services.ha_autocommit_or_rollback(session, false);
+  error= transaction_services.autocommitOrRollback(session, false);
   if (! session->endActiveTransaction())
     error=1;
   if (error)
@@ -627,7 +627,7 @@
   write_bin_log(session, session->query.c_str());
 
 err:
-  (void) transaction_services.ha_autocommit_or_rollback(session, error);
+  (void) transaction_services.autocommitOrRollback(session, error);
   session->tablespace_op=false;
 
   if (error == 0)
@@ -1480,7 +1480,7 @@
     Ensure that the new table is saved properly to disk so that we
     can do a rename
   */
-  if (transaction_services.ha_autocommit_or_rollback(session, false))
+  if (transaction_services.autocommitOrRollback(session, false))
     error=1;
   if (! session->endActiveTransaction())
     error=1;

=== modified file 'drizzled/statement/release_savepoint.cc'
--- drizzled/statement/release_savepoint.cc	2010-01-28 20:48:31 +0000
+++ drizzled/statement/release_savepoint.cc	2010-04-06 19:16:34 +0000
@@ -59,7 +59,7 @@
   if (iter != savepoints.end())
   {
     NamedSavepoint &sv= *iter;
-    (void) transaction_services.ha_release_savepoint(session, sv);
+    (void) transaction_services.releaseSavepoint(session, sv);
     savepoints.erase(iter);
     session->my_ok();
   }

=== modified file 'drizzled/statement/rollback_to_savepoint.cc'
--- drizzled/statement/rollback_to_savepoint.cc	2010-02-06 03:22:59 +0000
+++ drizzled/statement/rollback_to_savepoint.cc	2010-04-06 19:16:34 +0000
@@ -73,7 +73,7 @@
                      first_savepoint_name.size()) == 0)
     {
       /* Found the named savepoint we want to rollback to */
-      (void) transaction_services.ha_rollback_to_savepoint(session, first_savepoint);
+      (void) transaction_services.rollbackToSavepoint(session, first_savepoint);
 
       if (session->transaction.all.hasModifiedNonTransData())
       {
@@ -111,7 +111,7 @@
       /* Found the named savepoint we want to rollback to */
       found= true;
 
-      (void) transaction_services.ha_rollback_to_savepoint(session, sv);
+      (void) transaction_services.rollbackToSavepoint(session, sv);
     }
     if (found)
     {

=== modified file 'drizzled/statement/savepoint.cc'
--- drizzled/statement/savepoint.cc	2010-01-28 20:48:31 +0000
+++ drizzled/statement/savepoint.cc	2010-04-06 19:16:34 +0000
@@ -65,13 +65,13 @@
     if (iter != savepoints.end())
     {
       NamedSavepoint &sv= *iter;
-      (void) transaction_services.ha_release_savepoint(session, sv);
+      (void) transaction_services.releaseSavepoint(session, sv);
       savepoints.erase(iter);
     }
     
     NamedSavepoint newsv(session->lex->ident.str, session->lex->ident.length);
 
-    if (transaction_services.ha_savepoint(session, newsv))
+    if (transaction_services.setSavepoint(session, newsv))
     {
       return true;
     }

=== modified file 'drizzled/transaction_services.cc'
--- drizzled/transaction_services.cc	2010-03-17 03:11:34 +0000
+++ drizzled/transaction_services.cc	2010-04-06 19:16:34 +0000
@@ -427,65 +427,6 @@
 }
 
 /**
-  Check if we can skip the two-phase commit.
-
-  A helper function to evaluate if two-phase commit is mandatory.
-  As a side effect, propagates the read-only/read-write flags
-  of the statement transaction to its enclosing normal transaction.
-
-  @retval true   we must run a two-phase commit. Returned
-                 if we have at least two engines with read-write changes.
-  @retval false  Don't need two-phase commit. Even if we have two
-                 transactional engines, we can run two independent
-                 commits if changes in one of the engines are read-only.
-*/
-static
-bool
-ha_check_and_coalesce_trx_read_only(Session *session,
-                                    TransactionContext::ResourceContexts &resource_contexts,
-                                    bool normal_transaction)
-{
-  /* The number of storage engines that have actual changes. */
-  unsigned num_resources_modified_data= 0;
-  ResourceContext *resource_context;
-
-  for (TransactionContext::ResourceContexts::iterator it= resource_contexts.begin();
-       it != resource_contexts.end();
-       ++it)
-  {
-    resource_context= *it;
-    if (resource_context->hasModifiedData())
-      ++num_resources_modified_data;
-
-    if (! normal_transaction)
-    {
-      ResourceContext *resource_context_normal= session->getResourceContext(resource_context->getMonitored(), true);
-      assert(resource_context != resource_context_normal);
-      /*
-        Merge read-only/read-write information about statement
-        transaction to its enclosing normal transaction. Do this
-        only if in a real transaction -- that is, if we know
-        that resource_context_all is registered in session->transaction.all.
-        Since otherwise we only clutter the normal transaction flags.
-      */
-      if (resource_context_normal->isStarted()) /* false if autocommit. */
-        resource_context_normal->coalesceWith(resource_context);
-    }
-    else if (num_resources_modified_data > 1)
-    {
-      /*
-        It is a normal transaction, so we don't need to merge read/write
-        information up, and the need for two-phase commit has been
-        already established. Break the loop prematurely.
-      */
-      break;
-    }
-  }
-  return num_resources_modified_data > 1;
-}
-
-
-/**
   @retval
     0   ok
   @retval
@@ -499,7 +440,7 @@
     stored functions or triggers. So we simply do nothing now.
     TODO: This should be fixed in later ( >= 5.1) releases.
 */
-int TransactionServices::ha_commit_trans(Session *session, bool normal_transaction)
+int TransactionServices::commitTransaction(Session *session, bool normal_transaction)
 {
   int error= 0, cookie= 0;
   /*
@@ -522,17 +463,18 @@
 
   if (resource_contexts.empty() == false)
   {
-    bool must_2pc;
-
     if (is_real_trans && wait_if_global_read_lock(session, 0, 0))
     {
-      ha_rollback_trans(session, normal_transaction);
+      rollbackTransaction(session, normal_transaction);
       return 1;
     }
 
-    must_2pc= ha_check_and_coalesce_trx_read_only(session, resource_contexts, normal_transaction);
-
-    if (! trans->no_2pc && must_2pc)
+    /*
+     * If replication is on, we do a PREPARE on the resource managers, push the
+     * Transaction message across the replication stream, and then COMMIT if the
+     * replication stream returned successfully.
+     */
+    if (shouldConstructMessages())
     {
       for (TransactionContext::ResourceContexts::iterator it= resource_contexts.begin();
            it != resource_contexts.end() && ! error;
@@ -563,14 +505,22 @@
           }
         }
       }
+      if (error == 0 && is_real_trans)
+      {
+        /*
+         * Push the constructed Transaction messages across to
+         * replicators and appliers.
+         */
+        error= commitTransactionMessage(session);
+      }
       if (error)
       {
-        ha_rollback_trans(session, normal_transaction);
+        rollbackTransaction(session, normal_transaction);
         error= 1;
         goto end;
       }
     }
-    error= ha_commit_one_phase(session, normal_transaction) ? (cookie ? 2 : 1) : 0;
+    error= commitPhaseOne(session, normal_transaction) ? (cookie ? 2 : 1) : 0;
 end:
     if (is_real_trans)
       start_waiting_global_read_lock(session);
@@ -582,7 +532,7 @@
   @note
   This function does not care about global read lock. A caller should.
 */
-int TransactionServices::ha_commit_one_phase(Session *session, bool normal_transaction)
+int TransactionServices::commitPhaseOne(Session *session, bool normal_transaction)
 {
   int error=0;
   TransactionContext *trans= normal_transaction ? &session->transaction.all : &session->transaction.stmt;
@@ -638,21 +588,10 @@
     }
   }
   trans->reset();
-  if (error == 0)
-  {
-    if (is_real_trans)
-    {
-      /* 
-       * We commit the normal transaction by finalizing the transaction message
-       * and propogating the message to all registered replicators.
-       */
-      commitTransactionMessage(session);
-    }
-  }
   return error;
 }
 
-int TransactionServices::ha_rollback_trans(Session *session, bool normal_transaction)
+int TransactionServices::rollbackTransaction(Session *session, bool normal_transaction)
 {
   int error= 0;
   TransactionContext *trans= normal_transaction ? &session->transaction.all : &session->transaction.stmt;
@@ -751,20 +690,20 @@
     the user has used LOCK TABLES then that mechanism does not know to do the
     commit.
 */
-int TransactionServices::ha_autocommit_or_rollback(Session *session, int error)
+int TransactionServices::autocommitOrRollback(Session *session, int error)
 {
   if (session->transaction.stmt.getResourceContexts().empty() == false)
   {
     if (! error)
     {
-      if (ha_commit_trans(session, false))
+      if (commitTransaction(session, false))
         error= 1;
     }
     else
     {
-      (void) ha_rollback_trans(session, false);
+      (void) rollbackTransaction(session, false);
       if (session->transaction_rollback_request)
-        (void) ha_rollback_trans(session, true);
+        (void) rollbackTransaction(session, true);
     }
 
     session->variables.tx_isolation= session->session_tx_isolation;
@@ -772,51 +711,6 @@
   return error;
 }
 
-/**
-  return the list of XID's to a client, the same way SHOW commands do.
-
-  @note
-    I didn't find in XA specs that an RM cannot return the same XID twice,
-    so mysql_xa_recover does not filter XID's to ensure uniqueness.
-    It can be easily fixed later, if necessary.
-*/
-bool TransactionServices::mysql_xa_recover(Session *session)
-{
-  List<Item> field_list;
-  int i= 0;
-  XID_STATE *xs;
-
-  field_list.push_back(new Item_int("formatID", 0, MY_INT32_NUM_DECIMAL_DIGITS));
-  field_list.push_back(new Item_int("gtrid_length", 0, MY_INT32_NUM_DECIMAL_DIGITS));
-  field_list.push_back(new Item_int("bqual_length", 0, MY_INT32_NUM_DECIMAL_DIGITS));
-  field_list.push_back(new Item_empty_string("data", DRIZZLE_XIDDATASIZE));
-
-  if (session->client->sendFields(&field_list))
-    return 1;
-
-  pthread_mutex_lock(&LOCK_xid_cache);
-  while ((xs= (XID_STATE*)hash_element(&xid_cache, i++)))
-  {
-    if (xs->xa_state==XA_PREPARED)
-    {
-      session->client->store((int64_t)xs->xid.formatID);
-      session->client->store((int64_t)xs->xid.gtrid_length);
-      session->client->store((int64_t)xs->xid.bqual_length);
-      session->client->store(xs->xid.data,
-                             xs->xid.gtrid_length+xs->xid.bqual_length);
-      if (session->client->flush())
-      {
-        pthread_mutex_unlock(&LOCK_xid_cache);
-        return 1;
-      }
-    }
-  }
-
-  pthread_mutex_unlock(&LOCK_xid_cache);
-  session->my_eof();
-  return 0;
-}
-
 struct ResourceContextCompare : public std::binary_function<ResourceContext *, ResourceContext *, bool>
 {
   result_type operator()(const ResourceContext *lhs, const ResourceContext *rhs) const
@@ -827,7 +721,7 @@
   }
 };
 
-int TransactionServices::ha_rollback_to_savepoint(Session *session, NamedSavepoint &sv)
+int TransactionServices::rollbackToSavepoint(Session *session, NamedSavepoint &sv)
 {
   int error= 0;
   TransactionContext *trans= &session->transaction.all;
@@ -923,7 +817,7 @@
   section "4.33.4 SQL-statements and transaction states",
   NamedSavepoint is *not* transaction-initiating SQL-statement
 */
-int TransactionServices::ha_savepoint(Session *session, NamedSavepoint &sv)
+int TransactionServices::setSavepoint(Session *session, NamedSavepoint &sv)
 {
   int error= 0;
   TransactionContext *trans= &session->transaction.all;
@@ -961,7 +855,7 @@
   return error;
 }
 
-int TransactionServices::ha_release_savepoint(Session *session, NamedSavepoint &sv)
+int TransactionServices::releaseSavepoint(Session *session, NamedSavepoint &sv)
 {
   int error= 0;
 
@@ -988,6 +882,12 @@
   return error;
 }
 
+bool TransactionServices::shouldConstructMessages()
+{
+  ReplicationServices &replication_services= ReplicationServices::singleton();
+  return replication_services.isActive();
+}
+
 message::Transaction *TransactionServices::getActiveTransactionMessage(Session *in_session)
 {
   message::Transaction *transaction= in_session->getTransactionMessage();
@@ -1013,7 +913,7 @@
 {
   message::TransactionContext *trx= in_transaction.mutable_transaction_context();
   trx->set_server_id(in_session->getServerId());
-  trx->set_transaction_id(in_session->getQueryId());
+  trx->set_transaction_id(getNextTransactionId());
   trx->set_start_timestamp(in_session->getCurrentTimestamp());
 }
 
@@ -1032,11 +932,11 @@
   in_session->setTransactionMessage(NULL);
 }
 
-void TransactionServices::commitTransactionMessage(Session *in_session)
+int TransactionServices::commitTransactionMessage(Session *in_session)
 {
   ReplicationServices &replication_services= ReplicationServices::singleton();
   if (! replication_services.isActive())
-    return;
+    return 0;
 
   /* If there is an active statement message, finalize it */
   message::Statement *statement= in_session->getStatementMessage();
@@ -1046,15 +946,17 @@
     finalizeStatementMessage(*statement, in_session);
   }
   else
-    return; /* No data modification occurred inside the transaction */
+    return 0; /* No data modification occurred inside the transaction */
   
   message::Transaction* transaction= getActiveTransactionMessage(in_session);
 
   finalizeTransactionMessage(*transaction, in_session);
   
-  replication_services.pushTransactionMessage(*transaction);
+  plugin::ReplicationReturnCode result= replication_services.pushTransactionMessage(*in_session, *transaction);
 
   cleanupTransactionMessage(transaction, in_session);
+
+  return static_cast<int>(result);
 }
 
 void TransactionServices::initStatementMessage(message::Statement &statement,
@@ -1114,7 +1016,7 @@
 
     finalizeTransactionMessage(*transaction, in_session);
     
-    replication_services.pushTransactionMessage(*transaction);
+    (void) replication_services.pushTransactionMessage(*in_session, *transaction);
   }
   cleanupTransactionMessage(transaction, in_session);
 }
@@ -1552,7 +1454,7 @@
 
   finalizeTransactionMessage(*transaction, in_session);
   
-  replication_services.pushTransactionMessage(*transaction);
+  (void) replication_services.pushTransactionMessage(*in_session, *transaction);
 
   cleanupTransactionMessage(transaction, in_session);
 
@@ -1582,7 +1484,7 @@
 
   finalizeTransactionMessage(*transaction, in_session);
   
-  replication_services.pushTransactionMessage(*transaction);
+  (void) replication_services.pushTransactionMessage(*in_session, *transaction);
 
   cleanupTransactionMessage(transaction, in_session);
 
@@ -1611,7 +1513,7 @@
 
   finalizeTransactionMessage(*transaction, in_session);
   
-  replication_services.pushTransactionMessage(*transaction);
+  (void) replication_services.pushTransactionMessage(*in_session, *transaction);
 
   cleanupTransactionMessage(transaction, in_session);
 }
@@ -1647,7 +1549,7 @@
 
   finalizeTransactionMessage(*transaction, in_session);
   
-  replication_services.pushTransactionMessage(*transaction);
+  (void) replication_services.pushTransactionMessage(*in_session, *transaction);
 
   cleanupTransactionMessage(transaction, in_session);
 }
@@ -1682,7 +1584,7 @@
 
   finalizeTransactionMessage(*transaction, in_session);
   
-  replication_services.pushTransactionMessage(*transaction);
+  (void) replication_services.pushTransactionMessage(*in_session, *transaction);
 
   cleanupTransactionMessage(transaction, in_session);
 }
@@ -1702,7 +1604,7 @@
 
   finalizeTransactionMessage(*transaction, in_session);
   
-  replication_services.pushTransactionMessage(*transaction);
+  (void) replication_services.pushTransactionMessage(*in_session, *transaction);
 
   cleanupTransactionMessage(transaction, in_session);
 }

=== modified file 'drizzled/transaction_services.h'
--- drizzled/transaction_services.h	2010-03-16 21:30:44 +0000
+++ drizzled/transaction_services.h	2010-04-06 19:16:34 +0000
@@ -25,6 +25,7 @@
 #ifndef DRIZZLED_TRANSACTION_SERVICES_H
 #define DRIZZLED_TRANSACTION_SERVICES_H
 
+#include "drizzled/atomics.h"
 #include "drizzled/message/transaction.pb.h"
 
 namespace drizzled
@@ -49,10 +50,17 @@
 {
 public:
   static const size_t DEFAULT_RECORD_SIZE= 100;
+  typedef uint64_t TransactionId;
   /**
    * Constructor
    */
-  TransactionServices() {}
+  TransactionServices()
+  {
+    /**
+     * @todo set transaction ID to the last one from an applier...
+     */
+    current_transaction_id= 0;
+  }
 
   /**
    * Singleton method
@@ -63,6 +71,12 @@
     static TransactionServices transaction_services;
     return transaction_services;
   }
+
+  /**
+   * Returns true if the transaction manager should construct
+   * Transaction and Statement messages, false otherwise.
+   */
+  bool shouldConstructMessages();
   /**
    * Method which returns the active Transaction message
    * for the supplied Session.  If one is not found, a new Transaction
@@ -189,7 +203,7 @@
    *
    * @param Pointer to the Session committing the transaction
    */
-  void commitTransactionMessage(Session *in_session);
+  int commitTransactionMessage(Session *in_session);
   /** 
    * Marks the current active transaction message as being rolled back and
    * pushes the transaction message out to replicators.
@@ -293,18 +307,17 @@
    */
   void rawStatement(Session *in_session, const std::string &query);
   /* transactions: interface to plugin::StorageEngine functions */
-  int ha_commit_one_phase(Session *session, bool all);
-  int ha_rollback_trans(Session *session, bool all);
+  int commitPhaseOne(Session *session, bool all);
+  int rollbackTransaction(Session *session, bool all);
 
   /* transactions: these functions never call plugin::StorageEngine functions directly */
-  int ha_commit_trans(Session *session, bool all);
-  int ha_autocommit_or_rollback(Session *session, int error);
+  int commitTransaction(Session *session, bool all);
+  int autocommitOrRollback(Session *session, int error);
 
   /* savepoints */
-  int ha_rollback_to_savepoint(Session *session, NamedSavepoint &sv);
-  int ha_savepoint(Session *session, NamedSavepoint &sv);
-  int ha_release_savepoint(Session *session, NamedSavepoint &sv);
-  bool mysql_xa_recover(Session *session);
+  int rollbackToSavepoint(Session *session, NamedSavepoint &sv);
+  int setSavepoint(Session *session, NamedSavepoint &sv);
+  int releaseSavepoint(Session *session, NamedSavepoint &sv);
 
   /**
    * Marks a storage engine as participating in a statement
@@ -383,6 +396,23 @@
                                       plugin::MonitoredInTransaction *monitored,
                                       plugin::TransactionalStorageEngine *engine,
                                       plugin::XaResourceManager *resource_manager);
+  TransactionId getNextTransactionId()
+  {
+    return current_transaction_id.increment();
+  }
+  TransactionId getCurrentTransactionId()
+  {
+    return current_transaction_id;
+  }
+  /**
+   * DEBUG ONLY.  See plugin::TransactionLog::truncate()
+   */
+  void resetTransactionId()
+  {
+    current_transaction_id= 0;
+  }
+private:
+  atomic<TransactionId> current_transaction_id;
 };
 
 } /* namespace drizzled */

=== modified file 'plugin/default_replicator/default_replicator.cc'
--- plugin/default_replicator/default_replicator.cc	2010-03-06 02:08:13 +0000
+++ plugin/default_replicator/default_replicator.cc	2010-04-06 19:16:34 +0000
@@ -62,13 +62,16 @@
   sysvar_default_replicator_enable= false;
 }
 
-void DefaultReplicator::replicate(plugin::TransactionApplier *in_applier, message::Transaction &to_replicate)
+plugin::ReplicationReturnCode
+DefaultReplicator::replicate(plugin::TransactionApplier *in_applier,
+                             Session &in_session,
+                             message::Transaction &to_replicate)
 {
   /* 
    * We do absolutely nothing but call the applier's apply() method, passing
    * along the supplied Transaction.  Yep, told you it was simple...
    */
-  in_applier->apply(to_replicate);
+  return in_applier->apply(in_session, to_replicate);
 }
 
 static DefaultReplicator *default_replicator= NULL; /* The singleton replicator */

=== modified file 'plugin/default_replicator/default_replicator.h'
--- plugin/default_replicator/default_replicator.h	2009-10-19 19:19:38 +0000
+++ plugin/default_replicator/default_replicator.h	2010-04-06 19:16:34 +0000
@@ -66,9 +66,14 @@
    * the supplied message to their own controlled memory storage
    * area.
    *
+   * @param Applier to replicate to
+   * @param Session descriptor
    * @param Transaction message to be replicated
    */
-  void replicate(drizzled::plugin::TransactionApplier *in_applier, drizzled::message::Transaction &to_replicate);
+  drizzled::plugin::ReplicationReturnCode
+  replicate(drizzled::plugin::TransactionApplier *in_applier,
+            drizzled::Session &in_session,
+            drizzled::message::Transaction &to_replicate);
   
 };
 

=== modified file 'plugin/filtered_replicator/filtered_replicator.cc'
--- plugin/filtered_replicator/filtered_replicator.cc	2010-03-06 02:08:13 +0000
+++ plugin/filtered_replicator/filtered_replicator.cc	2010-04-06 19:16:34 +0000
@@ -214,8 +214,10 @@
   sysvar_filtered_replicator_enabled= false;
 }
 
-void FilteredReplicator::replicate(plugin::TransactionApplier *in_applier, 
-                                   message::Transaction &to_replicate)
+plugin::ReplicationReturnCode
+FilteredReplicator::replicate(plugin::TransactionApplier *in_applier,
+                              Session &in_session,
+                              message::Transaction &to_replicate)
 {
   string schema_name;
   string table_name;
@@ -287,8 +289,9 @@
      */
     message::TransactionContext *tc= filtered_transaction.mutable_transaction_context();
     *tc= to_replicate.transaction_context(); /* copy construct */
-    in_applier->apply(filtered_transaction);
+    return in_applier->apply(in_session, filtered_transaction);
   }
+  return plugin::SUCCESS;
 }
 
 void FilteredReplicator::populateFilter(std::string input,

=== modified file 'plugin/filtered_replicator/filtered_replicator.h'
--- plugin/filtered_replicator/filtered_replicator.h	2009-10-19 19:19:38 +0000
+++ plugin/filtered_replicator/filtered_replicator.h	2010-04-06 19:16:34 +0000
@@ -84,10 +84,14 @@
    * the supplied message to their own controlled memory storage
    * area.
    *
+   * @param Applier to replicate to
+   * @param Session descriptor
    * @param Transaction message to be replicated
    */
-  void replicate(drizzled::plugin::TransactionApplier *in_applier, 
-                 drizzled::message::Transaction &to_replicate);
+  drizzled::plugin::ReplicationReturnCode
+  replicate(drizzled::plugin::TransactionApplier *in_applier,
+            drizzled::Session &in_session,
+            drizzled::message::Transaction &to_replicate);
   
   /**
    * Populate the vector of schemas to filter from the

=== modified file 'plugin/innobase/handler/ha_innodb.cc'
--- plugin/innobase/handler/ha_innodb.cc	2010-04-01 15:54:57 +0000
+++ plugin/innobase/handler/ha_innodb.cc	2010-04-06 19:16:34 +0000
@@ -2259,11 +2259,6 @@
 			pthread_mutex_unlock(&commit_cond_m);
 		}
 
-		if (trx->conc_state == TRX_PREPARED) {
-
-			pthread_mutex_unlock(&prepare_commit_mutex);
-		}
-
 		/* Now do a write + flush of logs. */
 		trx_commit_complete_for_mysql(trx);
 
@@ -8150,31 +8145,6 @@
 
 	srv_active_wake_master_thread();
 
-	if (all || !session_test_options(session, OPTION_NOT_AUTOCOMMIT | OPTION_BEGIN))
-	{
-
-		/* For ibbackup to work the order of transactions in binlog
-		and InnoDB must be the same. Consider the situation
-
-		  thread1> prepare; write to binlog; ...
-			  <context switch>
-		  thread2> prepare; write to binlog; commit
-		  thread1>			     ... commit
-
-		To ensure this will not happen we're taking the mutex on
-		prepare, and releasing it on commit.
-
-		Note: only do it for normal commits, done via ha_commit_trans.
-		If 2pc protocol is executed by external transaction
-		coordinator, it will be just a regular MySQL client
-		executing XA PREPARE and XA COMMIT commands.
-		In this case we cannot know how many minutes or hours
-		will be between XA PREPARE and XA COMMIT, and we don't want
-		to block for undefined period of time.
-		*/
-		pthread_mutex_lock(&prepare_commit_mutex);
-		trx->conc_state = TRX_PREPARED;
-	}
 	return(error);
 }
 

=== modified file 'plugin/transaction_log/module.cc'
--- plugin/transaction_log/module.cc	2010-03-15 22:27:00 +0000
+++ plugin/transaction_log/module.cc	2010-04-06 19:16:34 +0000
@@ -61,6 +61,10 @@
  * in truncating the log file. 
  */
 static bool sysvar_transaction_log_truncate_debug= false;
+/**
+ * The name of the main transaction log file on disk.  With no prefix,
+ * this goes into Drizzle's $datadir.
+ */
 static const char DEFAULT_LOG_FILE_PATH[]= "transaction.log"; /* In datadir... */
 /** 
  * Transaction Log plugin system variable - Should we write a CRC32 checksum for 
@@ -76,6 +80,11 @@
  * TransactionLog::SYNC_METHOD_EVERY_SECOND == 2  ... sync at most once a second
  */
 static uint32_t sysvar_transaction_log_sync_method= 0;
+/**
+ * Transaction Log plugin system variable - Number of slots to create
+ * for managing write buffers
+ */
+static uint32_t sysvar_transaction_log_num_write_buffers= 8;
 
 /** DATA_DICTIONARY views */
 static TransactionLogTool *transaction_log_tool;
@@ -99,7 +108,8 @@
   if (sysvar_transaction_log_enabled)
   {
     transaction_log= new (nothrow) TransactionLog(string(sysvar_transaction_log_file),
-                                                  sysvar_transaction_log_sync_method);
+                                                  sysvar_transaction_log_sync_method,
+                                                  sysvar_transaction_log_checksum_enabled);
 
     if (transaction_log == NULL)
     {
@@ -120,7 +130,7 @@
     /* Create the applier plugin and register it */
     transaction_log_applier= new (nothrow) TransactionLogApplier("transaction_log_applier",
                                                                  *transaction_log, 
-                                                                 sysvar_transaction_log_checksum_enabled);
+                                                                 sysvar_transaction_log_num_write_buffers);
     if (transaction_log_applier == NULL)
     {
       errmsg_printf(ERRMSG_LVL_ERROR, _("Failed to allocate the TransactionLogApplier instance.  Got error: %s\n"), 
@@ -212,7 +222,7 @@
                            set_truncate_debug, /* update func */
                            false /* default */);
 
-static DRIZZLE_SYSVAR_STR(log_file,
+static DRIZZLE_SYSVAR_STR(file,
                           sysvar_transaction_log_file,
                           PLUGIN_VAR_READONLY,
                           N_("Path to the file to use for transaction log"),
@@ -241,12 +251,24 @@
                            2,
                            0);
 
+static DRIZZLE_SYSVAR_UINT(num_write_buffers,
+                           sysvar_transaction_log_num_write_buffers,
+                           PLUGIN_VAR_OPCMDARG,
+                           N_("Number of slots for in-memory write buffers (default=8)."),
+                           NULL, /* check func */
+                           NULL, /* update func */
+                           8, /* default */
+                           4,
+                           8192,
+                           0);
+
 static drizzle_sys_var* sys_variables[]= {
   DRIZZLE_SYSVAR(enable),
   DRIZZLE_SYSVAR(truncate_debug),
-  DRIZZLE_SYSVAR(log_file),
+  DRIZZLE_SYSVAR(file),
   DRIZZLE_SYSVAR(enable_checksum),
   DRIZZLE_SYSVAR(sync_method),
+  DRIZZLE_SYSVAR(num_write_buffers),
   NULL
 };
 

=== modified file 'plugin/transaction_log/plugin.ini'
--- plugin/transaction_log/plugin.ini	2010-03-05 18:08:49 +0000
+++ plugin/transaction_log/plugin.ini	2010-04-06 19:16:34 +0000
@@ -6,8 +6,8 @@
 title=Transaction Log
 description=Log of Transaction Messages
 load_by_default=yes
-sources= background_worker.cc hexdump_transaction_message.cc module.cc print_transaction_message.cc transaction_log.cc transaction_log_applier.cc transaction_log_entry.cc transaction_log_index.cc transaction_log_reader.cc data_dictionary_schema.cc
-headers= background_worker.h hexdump_transaction_message.h print_transaction_message.h transaction_log.h transaction_log_applier.h transaction_log_entry.h transaction_log_index.h transaction_log_reader.h data_dictionary_schema.h
+sources= background_worker.cc hexdump_transaction_message.cc module.cc print_transaction_message.cc transaction_log.cc transaction_log_applier.cc transaction_log_entry.cc transaction_log_index.cc transaction_log_reader.cc data_dictionary_schema.cc write_buffer.cc
+headers= background_worker.h hexdump_transaction_message.h print_transaction_message.h transaction_log.h transaction_log_applier.h transaction_log_entry.h transaction_log_index.h transaction_log_reader.h data_dictionary_schema.h write_buffer.h
 libs=${top_builddir}/drizzled/algorithm/libhash.la 
 libadd=$(LIBZ)
 cxxflags=${PROTOSKIP_WARNINGS}

=== added file 'plugin/transaction_log/tests/r/ddl_transaction_id.result'
--- plugin/transaction_log/tests/r/ddl_transaction_id.result	1970-01-01 00:00:00 +0000
+++ plugin/transaction_log/tests/r/ddl_transaction_id.result	2010-04-06 19:16:34 +0000
@@ -0,0 +1,36 @@
+SET GLOBAL transaction_log_truncate_debug= true;
+DROP TABLE IF EXISTS t1;
+CREATE TABLE t1 (
+id INT NOT NULL PRIMARY KEY
+, padding VARCHAR(200) NOT NULL
+);
+INSERT INTO t1 VALUES (1, "I love testing.");
+INSERT INTO t1 VALUES (2, "I hate testing.");
+SELECT * FROM t1;
+id	padding
+1	I love testing.
+2	I hate testing.
+ALTER TABLE t1 CHANGE COLUMN padding less_padding VARCHAR(180) NOT NULL;
+SELECT * FROM t1 WHERE id = 3;
+id	less_padding
+DROP TABLE t1;
+SELECT * FROM DATA_DICTIONARY.TRANSACTION_LOG;
+FILE_NAME	FILE_LENGTH	NUM_LOG_ENTRIES	NUM_TRANSACTIONS	MIN_TRANSACTION_ID	MAX_TRANSACTION_ID	MIN_END_TIMESTAMP	MAX_END_TIMESTAMP	INDEX_SIZE_IN_BYTES
+transaction.log	X	6	6	1	6	X	X	X
+SELECT * FROM DATA_DICTIONARY.TRANSACTION_LOG_ENTRIES;
+ENTRY_OFFSET	ENTRY_TYPE	ENTRY_LENGTH
+X	TRANSACTION	X
+X	TRANSACTION	X
+X	TRANSACTION	X
+X	TRANSACTION	X
+X	TRANSACTION	X
+X	TRANSACTION	X
+SELECT * FROM DATA_DICTIONARY.TRANSACTION_LOG_TRANSACTIONS;
+ENTRY_OFFSET	TRANSACTION_ID	SERVER_ID	START_TIMESTAMP	END_TIMESTAMP	NUM_STATEMENTS	CHECKSUM
+X	1	1	X	X	1	0
+X	2	1	X	X	1	0
+X	3	1	X	X	1	0
+X	4	1	X	X	1	0
+X	5	1	X	X	1	0
+X	6	1	X	X	1	0
+SET GLOBAL transaction_log_truncate_debug= true;

=== modified file 'plugin/transaction_log/tests/r/filtered_replicator.result'
--- plugin/transaction_log/tests/r/filtered_replicator.result	2010-02-27 23:47:33 +0000
+++ plugin/transaction_log/tests/r/filtered_replicator.result	2010-04-06 19:16:34 +0000
@@ -6,6 +6,7 @@
 );
 INSERT INTO t1 VALUES (1, "I love testing.");
 INSERT INTO t1 VALUES (2, "I hate testing.");
+DROP TABLE t1;
 DROP TABLE IF EXISTS t1;
 CREATE TABLE t1 (
 id INT NOT NULL PRIMARY KEY
@@ -134,6 +135,9 @@
 INSERT INTO `test`.`t1` (`id`,`padding`) VALUES (2,'I hate testing.');
 COMMIT;
 START TRANSACTION;
+DROP TABLE `test`.`t1`;
+COMMIT;
+START TRANSACTION;
 DROP TABLE IF EXISTS `test`.`t1`;
 COMMIT;
 START TRANSACTION;

=== modified file 'plugin/transaction_log/tests/r/information_schema.result'
--- plugin/transaction_log/tests/r/information_schema.result	2010-03-05 04:44:58 +0000
+++ plugin/transaction_log/tests/r/information_schema.result	2010-04-06 19:16:34 +0000
@@ -6,19 +6,22 @@
 );
 INSERT INTO t1 VALUES (1, "I love testing.");
 INSERT INTO t1 VALUES (2, "I hate testing.");
+DROP TABLE t1;
 SELECT * FROM DATA_DICTIONARY.TRANSACTION_LOG;
 FILE_NAME	FILE_LENGTH	NUM_LOG_ENTRIES	NUM_TRANSACTIONS	MIN_TRANSACTION_ID	MAX_TRANSACTION_ID	MIN_END_TIMESTAMP	MAX_END_TIMESTAMP	INDEX_SIZE_IN_BYTES
-transaction.log	X	4	4	X	X	X	X	X
+transaction.log	X	5	5	1	5	X	X	X
 SELECT * FROM DATA_DICTIONARY.TRANSACTION_LOG_ENTRIES;
 ENTRY_OFFSET	ENTRY_TYPE	ENTRY_LENGTH
 X	TRANSACTION	X
 X	TRANSACTION	X
 X	TRANSACTION	X
 X	TRANSACTION	X
+X	TRANSACTION	X
 SELECT * FROM DATA_DICTIONARY.TRANSACTION_LOG_TRANSACTIONS;
 ENTRY_OFFSET	TRANSACTION_ID	SERVER_ID	START_TIMESTAMP	END_TIMESTAMP	NUM_STATEMENTS	CHECKSUM
-X	X	1	X	X	1	0
-X	X	1	X	X	1	0
-X	X	1	X	X	1	0
-X	X	1	X	X	1	0
+X	1	1	X	X	1	0
+X	2	1	X	X	1	0
+X	3	1	X	X	1	0
+X	4	1	X	X	1	0
+X	5	1	X	X	1	0
 SET GLOBAL transaction_log_truncate_debug= true;

=== modified file 'plugin/transaction_log/tests/r/insert.result'
--- plugin/transaction_log/tests/r/insert.result	2010-02-27 23:47:33 +0000
+++ plugin/transaction_log/tests/r/insert.result	2010-04-06 19:16:34 +0000
@@ -5,6 +5,7 @@
 );
 INSERT INTO t1 VALUES (1, "I love testing.");
 INSERT INTO t1 VALUES (2, "I hate testing.");
+DROP TABLE t1;
 START TRANSACTION;
 DROP TABLE IF EXISTS `test`.`t1`;
 COMMIT;
@@ -17,4 +18,7 @@
 START TRANSACTION;
 INSERT INTO `test`.`t1` (`id`,`padding`) VALUES (2,'I hate testing.');
 COMMIT;
+START TRANSACTION;
+DROP TABLE `test`.`t1`;
+COMMIT;
 SET GLOBAL transaction_log_truncate_debug= true;

=== modified file 'plugin/transaction_log/tests/r/truncate_log.result'
--- plugin/transaction_log/tests/r/truncate_log.result	2010-03-05 04:44:58 +0000
+++ plugin/transaction_log/tests/r/truncate_log.result	2010-04-06 19:16:34 +0000
@@ -6,6 +6,7 @@
 );
 INSERT INTO t1 VALUES (1, "I love testing.");
 INSERT INTO t1 VALUES (2, "I hate testing.");
-24var/master-data/transaction.log
+DROP TABLE t1;
+29var/master-data/transaction.log
 SET GLOBAL transaction_log_truncate_debug= true;
 0var/master-data/transaction.log

=== modified file 'plugin/transaction_log/tests/r/udf_print_transaction_message.result'
--- plugin/transaction_log/tests/r/udf_print_transaction_message.result	2010-02-11 04:18:03 +0000
+++ plugin/transaction_log/tests/r/udf_print_transaction_message.result	2010-04-06 19:16:34 +0000
@@ -5,6 +5,7 @@
 );
 INSERT INTO t1 VALUES (1, "I love testing.");
 INSERT INTO t1 VALUES (2, "I hate testing.");
+DROP TABLE t1;
 SELECT LENGTH(PRINT_TRANSACTION_MESSAGE('transaction.log', ENTRY_OFFSET)) > 0 as checked
 FROM DATA_DICTIONARY.TRANSACTION_LOG_ENTRIES
 LIMIT 1;

=== renamed file 'plugin/transaction_log/tests/r/sync_method_every_write.result' => 'plugin/transaction_log/tests/r/variables.result'
--- plugin/transaction_log/tests/r/sync_method_every_write.result	2010-02-05 08:11:15 +0000
+++ plugin/transaction_log/tests/r/variables.result	2010-04-06 19:16:34 +0000
@@ -3,6 +3,7 @@
 VARIABLE_NAME	VARIABLE_VALUE
 transaction_log_enable	ON
 transaction_log_enable_checksum	OFF
-transaction_log_log_file	transaction.log
+transaction_log_file	transaction.log
+transaction_log_num_write_buffers	8
 transaction_log_sync_method	1
 transaction_log_truncate_debug	OFF

=== added file 'plugin/transaction_log/tests/t/ddl_transaction_id-master.opt'
--- plugin/transaction_log/tests/t/ddl_transaction_id-master.opt	1970-01-01 00:00:00 +0000
+++ plugin/transaction_log/tests/t/ddl_transaction_id-master.opt	2010-04-06 19:16:34 +0000
@@ -0,0 +1,1 @@
+--default-replicator-enable --transaction-log-enable --scheduler=multi_thread

=== added file 'plugin/transaction_log/tests/t/ddl_transaction_id.inc'
--- plugin/transaction_log/tests/t/ddl_transaction_id.inc	1970-01-01 00:00:00 +0000
+++ plugin/transaction_log/tests/t/ddl_transaction_id.inc	2010-04-06 19:16:34 +0000
@@ -0,0 +1,29 @@
+# 
+# Tests ordering of transaction ID, that SELECT queries
+# do not increment the transaction ID, and that the proper
+# generation of transaction IDs is done for DDL operations.
+#
+
+--disable_warnings
+DROP TABLE IF EXISTS t1;
+--enable_warnings
+
+CREATE TABLE t1 (
+  id INT NOT NULL PRIMARY KEY
+, padding VARCHAR(200) NOT NULL
+);
+
+INSERT INTO t1 VALUES (1, "I love testing.");
+INSERT INTO t1 VALUES (2, "I hate testing.");
+
+SELECT * FROM t1;
+
+ALTER TABLE t1 CHANGE COLUMN padding less_padding VARCHAR(180) NOT NULL;
+
+# 
+# Should be no result here...
+#
+
+SELECT * FROM t1 WHERE id = 3;
+
+DROP TABLE t1;

=== added file 'plugin/transaction_log/tests/t/ddl_transaction_id.test'
--- plugin/transaction_log/tests/t/ddl_transaction_id.test	1970-01-01 00:00:00 +0000
+++ plugin/transaction_log/tests/t/ddl_transaction_id.test	2010-04-06 19:16:34 +0000
@@ -0,0 +1,14 @@
+# Truncate the log file to reset for the next test
+--source ../plugin/transaction_log/tests/t/truncate_log.inc
+
+--source ../plugin/transaction_log/tests/t/ddl_transaction_id.inc
+
+--replace_column 2 X 7 X 8 X 9 X
+SELECT * FROM DATA_DICTIONARY.TRANSACTION_LOG;
+--replace_column 1 X 3 X
+SELECT * FROM DATA_DICTIONARY.TRANSACTION_LOG_ENTRIES;
+--replace_column 1 X 4 X 5 X
+SELECT * FROM DATA_DICTIONARY.TRANSACTION_LOG_TRANSACTIONS;
+
+# Truncate the log file to reset for the next test
+--source ../plugin/transaction_log/tests/t/truncate_log.inc

=== modified file 'plugin/transaction_log/tests/t/information_schema.test'
--- plugin/transaction_log/tests/t/information_schema.test	2010-03-05 04:44:58 +0000
+++ plugin/transaction_log/tests/t/information_schema.test	2010-04-06 19:16:34 +0000
@@ -13,11 +13,11 @@
 # that the information contained in them matches
 # the transaction log.
 
---replace_column 2 X 5 X 6 X 7 X 8 X 9 X
+--replace_column 2 X 7 X 8 X 9 X
 SELECT * FROM DATA_DICTIONARY.TRANSACTION_LOG;
 --replace_column 1 X 3 X
 SELECT * FROM DATA_DICTIONARY.TRANSACTION_LOG_ENTRIES;
---replace_column 1 X 2 X 4 X 5 X
+--replace_column 1 X 4 X 5 X
 SELECT * FROM DATA_DICTIONARY.TRANSACTION_LOG_TRANSACTIONS;
 
 # Truncate the log file to reset for the next test

=== modified file 'plugin/transaction_log/tests/t/insert.inc'
--- plugin/transaction_log/tests/t/insert.inc	2009-12-11 19:08:29 +0000
+++ plugin/transaction_log/tests/t/insert.inc	2010-04-06 19:16:34 +0000
@@ -18,3 +18,5 @@
 
 INSERT INTO t1 VALUES (1, "I love testing.");
 INSERT INTO t1 VALUES (2, "I hate testing.");
+
+DROP TABLE t1;

=== renamed file 'plugin/transaction_log/tests/t/sync_method_every_write-master.opt' => 'plugin/transaction_log/tests/t/variables-master.opt'
=== renamed file 'plugin/transaction_log/tests/t/sync_method_every_write.test' => 'plugin/transaction_log/tests/t/variables.test'
=== modified file 'plugin/transaction_log/transaction_log.cc'
--- plugin/transaction_log/transaction_log.cc	2010-03-10 18:22:56 +0000
+++ plugin/transaction_log/transaction_log.cc	2010-04-06 19:16:34 +0000
@@ -6,7 +6,7 @@
  *
  *  Authors:
  *
- *  Jay Pipes <[e-mail]>
+ *    Jay Pipes <[e-mail]>
  *
  *  This program is free software; you can redistribute it and/or modify
  *  it under the terms of the GNU General Public License as published by
@@ -30,31 +30,37 @@
  *
  * @details
  *
- * Currently, the log file uses this implementation:
+ * Currently, the transaction log file uses a simple, single-file, append-only
+ * format.
  *
  * We have an atomic off_t called log_offset which keeps track of the 
- * offset into the log file for writing the next Transaction.
- *
- * We write Transaction message encapsulated in an 8-byte length/type header and a
- * 4-byte checksum trailer.
- *
- * When writing a Transaction to the log, we calculate the length of the 
- * Transaction to be written.  We then increment log_offset by the length
- * of the Transaction plus 2 * sizeof(uint32_t) plus sizeof(uint32_t) and store 
- * this new offset in a local off_t called cur_offset (see TransactionLog::apply().  
- * This compare and set is done in an atomic instruction.
- *
- * We then adjust the local off_t (cur_offset) back to the original
- * offset by subtracting the length and sizeof(uint32_t) and sizeof(uint32_t).
- *
- * We then first write a 64-bit length and then the serialized transaction/transaction
- * and optional checksum to our log file at our local cur_offset.
- *
- * --------------------------------------------------------------------------------
- * |<- 4 bytes ->|<- 4 bytes ->|<- # Bytes of Transaction Message ->|<- 4 bytes ->|
- * --------------------------------------------------------------------------------
- * |  Msg Type   |   Length    |   Serialized Transaction Message   |   Checksum  |
- * --------------------------------------------------------------------------------
+ * offset into the log file for writing the next log entry.  The log
+ * entries are written, one after the other, in the following way:
+ *
+ * <pre>
+ * --------------------------------------
+ * |<- 4 bytes ->|<- # Bytes of Entry ->|
+ * --------------------------------------
+ * |  Entry Type |  Serialized Entry    |
+ * --------------------------------------
+ * </pre>
+ *
+ * The Entry Type is an integer defined as an enumeration in the 
+ * /drizzled/message/transaction.proto file called TransactionLogEntry::Type.
+ *
+ * Each transaction log entry type is written to the log differently.  Here,
+ * we cover the format of each log entry type.
+ *
+ * Committed and Prepared Transaction Log Entries
+ * -----------------------------------------------
+ * 
+ * <pre>
+ * ------------------------------------------------------------------
+ * |<- 4 bytes ->|<- # Bytes of Transaction Message ->|<- 4 bytes ->|
+ * ------------------------------------------------------------------
+ * |   Length    |   Serialized Transaction Message   |   Checksum  |
+ * ------------------------------------------------------------------
+ * </pre>
  *
  * @todo
  *
@@ -77,19 +83,27 @@
 #include <drizzled/internal/my_sys.h> /* for internal::my_sync */
 #include <drizzled/errmsg_print.h>
 #include <drizzled/gettext.h>
+#include <drizzled/message/transaction.pb.h>
+#include <drizzled/transaction_services.h>
+#include <drizzled/algorithm/crc32.h>
+
+#include <google/protobuf/io/coded_stream.h>
 
 using namespace std;
 using namespace drizzled;
+using namespace google;
 
 TransactionLog *transaction_log= NULL; /* The singleton transaction log */
 
 TransactionLog::TransactionLog(const string in_log_file_path,
-                               uint32_t in_sync_method) : 
+                               uint32_t in_sync_method,
+                               bool in_do_checksum) : 
     state(OFFLINE),
     log_file_path(in_log_file_path),
     has_error(false),
     error_message(),
-    sync_method(in_sync_method)
+    sync_method(in_sync_method),
+    do_checksum(in_do_checksum)
 {
   /* Setup our log file and determine the next write offset... */
   log_file= open(log_file_path.c_str(), O_APPEND|O_CREAT|O_SYNC|O_WRONLY, S_IRWXU);
@@ -133,6 +147,43 @@
   }
 }
 
+uint8_t *TransactionLog::packTransactionIntoLogEntry(const message::Transaction &trx,
+                                                     uint8_t *buffer,
+                                                     uint32_t *checksum_out)
+{
+  uint8_t *orig_buffer= buffer;
+  size_t message_byte_length= trx.ByteSize();
+
+  /*
+   * Write the header information, which is the message type and
+   * the length of the transaction message into the buffer
+   */
+  buffer= protobuf::io::CodedOutputStream::WriteLittleEndian32ToArray(
+      static_cast<uint32_t>(ReplicationServices::TRANSACTION), buffer);
+  buffer= protobuf::io::CodedOutputStream::WriteLittleEndian32ToArray(
+      static_cast<uint32_t>(message_byte_length), buffer);
+  
+  /*
+   * Now write the serialized transaction message, followed
+   * by the optional checksum into the buffer.
+   */
+  buffer= trx.SerializeWithCachedSizesToArray(buffer);
+
+  if (do_checksum)
+  {
+    *checksum_out= drizzled::algorithm::crc32(
+        reinterpret_cast<char *>(buffer) - message_byte_length, message_byte_length);
+  }
+  else
+    *checksum_out= 0;
+
+  /* We always write in network byte order */
+  buffer= protobuf::io::CodedOutputStream::WriteLittleEndian32ToArray(*checksum_out, buffer);
+  /* Reset the pointer back to its original location... */
+  buffer= orig_buffer;
+  return orig_buffer;
+}
+
 off_t TransactionLog::writeEntry(const uint8_t *data, size_t data_length)
 {
   ssize_t written= 0;
@@ -142,12 +193,6 @@
    */
   off_t cur_offset= log_offset.fetch_and_add(static_cast<off_t>(data_length));
 
-  /*
-   * We adjust cur_offset back to the original log_offset before
-   * the increment above...
-   */
-  cur_offset-= static_cast<off_t>(data_length);
-
   /* 
    * Quick safety...if an error occurs above in another writer, the log 
    * file will be in a crashed state.
@@ -243,6 +288,7 @@
     result= ftruncate(log_file, log_offset);
   }
   while (result == -1 && errno == EINTR);
+  drizzled::TransactionServices::singleton().resetTransactionId();
 }
 
 bool TransactionLog::findLogFilenameContainingTransactionId(const ReplicationServices::GlobalTransactionId&,
@@ -272,3 +318,8 @@
 {
   return error_message;
 }
+
+size_t TransactionLog::getLogEntrySize(const message::Transaction &trx)
+{
+  return trx.ByteSize() + HEADER_TRAILER_BYTES;
+}

=== modified file 'plugin/transaction_log/transaction_log.h'
--- plugin/transaction_log/transaction_log.h	2010-03-05 18:08:49 +0000
+++ plugin/transaction_log/transaction_log.h	2010-04-06 19:16:34 +0000
@@ -49,10 +49,6 @@
 class TransactionLog
 {
 public:
-  static const uint32_t HEADER_TRAILER_BYTES= sizeof(uint32_t) + /* 4-byte msg type header */
-                                              sizeof(uint32_t) + /* 4-byte length header */
-                                              sizeof(uint32_t); /* 4 byte checksum trailer */
-
   typedef std::vector<TransactionLogEntry> Entries;
   typedef std::vector<TransactionLogTransactionEntry> TransactionEntries;
   /**
@@ -70,7 +66,8 @@
   static const uint32_t SYNC_METHOD_EVERY_SECOND= 2; ///< Sync no more than once a second
 public:
   TransactionLog(const std::string in_log_file_path,
-                 uint32_t in_sync_method);
+                 uint32_t in_sync_method,
+                 bool in_do_checksum);
 
   /** Destructor */
   ~TransactionLog();
@@ -102,6 +99,31 @@
   }
 
   /**
+   * Static helper method which returns the transaction
+   * log entry size in bytes of a given transaction
+   * message.
+   *
+   * @param[in] Transaction message
+   */
+  static size_t getLogEntrySize(const drizzled::message::Transaction &trx);
+
+  /**
+   * Method which packs into a raw byte buffer
+   * a transaction log entry.  Supplied buffer should
+   * be of adequate size.
+   *
+   * Returns a pointer to the start of the original
+   * buffer.
+   *
+   * @param[in]   Transaction message to pack
+   * @param[in]   Raw byte buffer
+   * @param[out]  Pointer to storage for checksum of message
+   */
+  uint8_t *packTransactionIntoLogEntry(const drizzled::message::Transaction &trx,
+                                       uint8_t *buffer,
+                                       uint32_t *checksum_out);
+
+  /**
    * Writes a chunk of data to the log file of a specified
    * length and returns the offset at which the chunk of
    * data was written.
@@ -153,6 +175,10 @@
    */
   const std::string &getErrorMessage() const;
 private:
+  static const uint32_t HEADER_TRAILER_BYTES= sizeof(uint32_t) + /* 4-byte msg type header */
+                                              sizeof(uint32_t) + /* 4-byte length header */
+                                              sizeof(uint32_t); /* 4 byte checksum trailer */
+
   /* Don't allows these */
   TransactionLog();
   TransactionLog(const TransactionLog &other);
@@ -181,6 +207,7 @@
   std::string error_message; ///< Current error message
   uint32_t sync_method; ///< Determines behaviour of syncing log file
   time_t last_sync_time; ///< Last time the log file was synced (only set in SYNC_METHOD_EVERY_SECOND)
+  bool do_checksum; ///< Do a CRC32 checksum when writing Transaction message to log?
 };
 
 #endif /* PLUGIN_TRANSACTION_LOG_TRANSACTION_LOG_H */

=== modified file 'plugin/transaction_log/transaction_log_applier.cc'
--- plugin/transaction_log/transaction_log_applier.cc	2010-03-29 22:01:35 +0000
+++ plugin/transaction_log/transaction_log_applier.cc	2010-04-06 19:16:34 +0000
@@ -6,7 +6,7 @@
  *
  *  Authors:
  *
- *  Jay Pipes <[e-mail]>
+ *    Jay Pipes <[e-mail]>
  *
  *  This program is free software; you can redistribute it and/or modify
  *  it under the terms of the GNU General Public License as published by
@@ -43,24 +43,19 @@
  */
 
 #include "config.h"
+#include "write_buffer.h"
 #include "transaction_log.h"
 #include "transaction_log_applier.h"
 #include "transaction_log_index.h"
 
-#include <sys/stat.h>
-#include <fcntl.h>
-#include <unistd.h>
-#include <errno.h>
+#include <vector>
 
-#include <drizzled/errmsg_print.h>
-#include <drizzled/gettext.h>
-#include <drizzled/algorithm/crc32.h>
 #include <drizzled/message/transaction.pb.h>
-#include <google/protobuf/io/coded_stream.h>
+#include <drizzled/util/functors.h>
+#include <drizzled/session.h>
 
 using namespace std;
 using namespace drizzled;
-using namespace google;
 
 TransactionLogApplier *transaction_log_applier= NULL; /* The singleton transaction log applier */
 
@@ -68,79 +63,59 @@
 
 TransactionLogApplier::TransactionLogApplier(const string name_arg,
                                              TransactionLog &in_transaction_log,
-                                             bool in_do_checksum) :
+                                             uint32_t in_num_write_buffers) :
   plugin::TransactionApplier(name_arg),
   transaction_log(in_transaction_log),  
-  do_checksum(in_do_checksum)
+  num_write_buffers(in_num_write_buffers),
+  write_buffers()
 {
+  /* 
+   * Create each of the buffers we need for undo log entries 
+   */
+  write_buffers.reserve(num_write_buffers);
+  for (size_t x= 0; x < num_write_buffers; ++x)
+  {
+    write_buffers.push_back(new WriteBuffer());
+  }
 }
 
 TransactionLogApplier::~TransactionLogApplier()
 {
-}
-
-void TransactionLogApplier::apply(const message::Transaction &to_apply)
-{
-  uint8_t *buffer; /* Buffer we will write serialized header, 
-                      message and trailing checksum to */
-  uint8_t *orig_buffer;
-
-  size_t message_byte_length= to_apply.ByteSize();
-  size_t total_envelope_length= TransactionLog::HEADER_TRAILER_BYTES + message_byte_length;
-
-  /* 
-   * Attempt allocation of raw memory buffer for the header, 
-   * message and trailing checksum bytes.
-   */
-  buffer= static_cast<uint8_t *>(malloc(total_envelope_length));
-  if (buffer == NULL)
-  {
-    errmsg_printf(ERRMSG_LVL_ERROR, 
-                  _("Failed to allocate enough memory to buffer header, "
-                    "transaction message, and trailing checksum bytes. Tried to allocate %" PRId64
-                    " bytes.  Error: %s\n"), 
-                  static_cast<int64_t>(total_envelope_length),
-                  strerror(errno));
-    return;
-  }
-  else
-    orig_buffer= buffer; /* We will free() orig_buffer, as buffer is moved during write */
-
-  /*
-   * Write the header information, which is the message type and
-   * the length of the transaction message into the buffer
-   */
-  buffer= protobuf::io::CodedOutputStream::WriteLittleEndian32ToArray(
-      static_cast<uint32_t>(ReplicationServices::TRANSACTION), buffer);
-  buffer= protobuf::io::CodedOutputStream::WriteLittleEndian32ToArray(
-      static_cast<uint32_t>(message_byte_length), buffer);
-  
-  /*
-   * Now write the serialized transaction message, followed
-   * by the optional checksum into the buffer.
-   */
-  buffer= to_apply.SerializeWithCachedSizesToArray(buffer);
-
-  uint32_t checksum= 0;
-  if (do_checksum)
-  {
-    checksum= drizzled::algorithm::crc32(
-        buffer - message_byte_length, message_byte_length);
-  }
-
-  /* We always write in network byte order */
-  buffer= protobuf::io::CodedOutputStream::WriteLittleEndian32ToArray(checksum, buffer);
-
-  /* Ask the transaction log to write the entry and return where it wrote it */
-  off_t written_to= transaction_log.writeEntry(orig_buffer, total_envelope_length);
-
-  free(orig_buffer);
+  for_each(write_buffers.begin(),
+           write_buffers.end(),
+           DeletePtr());
+  write_buffers.clear();
+}
+
+WriteBuffer *TransactionLogApplier::getWriteBuffer(const Session &session)
+{
+  return write_buffers[session.getSessionId() % num_write_buffers];
+}
+
+plugin::ReplicationReturnCode
+TransactionLogApplier::apply(Session &in_session,
+                             const message::Transaction &to_apply)
+{
+  size_t entry_size= TransactionLog::getLogEntrySize(to_apply);
+  WriteBuffer *write_buffer= getWriteBuffer(in_session);
+
+  uint32_t checksum;
+
+  write_buffer->lock();
+  write_buffer->resize(entry_size);
+  uint8_t *bytes= write_buffer->getRawBytes();
+  bytes= transaction_log.packTransactionIntoLogEntry(to_apply,
+                                                     bytes,
+                                                     &checksum);
+
+  off_t written_to= transaction_log.writeEntry(bytes, entry_size);
+  write_buffer->unlock();
 
   /* Add an entry to the index describing what was just applied */
   transaction_log_index->addEntry(TransactionLogEntry(ReplicationServices::TRANSACTION,
                                                       written_to,
-                                                      total_envelope_length),
+                                                      entry_size),
                                   to_apply,
                                   checksum);
-
+  return plugin::SUCCESS;
 }

=== modified file 'plugin/transaction_log/transaction_log_applier.h'
--- plugin/transaction_log/transaction_log_applier.h	2010-03-05 18:08:49 +0000
+++ plugin/transaction_log/transaction_log_applier.h	2010-04-06 19:16:34 +0000
@@ -43,14 +43,20 @@
 #include <vector>
 #include <string>
 
+namespace drizzled
+{
+  class Session;
+}
+
 class TransactionLog;
+class WriteBuffer;
 
 class TransactionLogApplier: public drizzled::plugin::TransactionApplier 
 {
 public:
   TransactionLogApplier(const std::string name_arg,
                         TransactionLog &in_transaction_log,
-                        bool in_do_checksum);
+                        uint32_t in_num_write_buffers);
 
   /** Destructor */
   ~TransactionLogApplier();
@@ -68,16 +74,27 @@
    * the supplied message to their own controlled memory storage
    * area.
    *
+   * @param Session descriptor
    * @param Transaction message to be replicated
    */
-  void apply(const drizzled::message::Transaction &to_apply);
+  drizzled::plugin::ReplicationReturnCode
+  apply(drizzled::Session &in_session,
+        const drizzled::message::Transaction &to_apply);
 private:
   /* Don't allows these */
   TransactionLogApplier();
   TransactionLogApplier(const TransactionLogApplier &other);
   TransactionLogApplier &operator=(const TransactionLogApplier &other);
   TransactionLog &transaction_log;
-  bool do_checksum; ///< Do a CRC32 checksum when writing Transaction message to log?
+  uint32_t num_write_buffers; ///< Number of write buffers used
+  std::vector<WriteBuffer *> write_buffers; ///< array of write buffers
+
+  /**
+   * Returns the write buffer for the supplied session
+   *
+   * @param Session descriptor
+   */
+  WriteBuffer *getWriteBuffer(const drizzled::Session &session);
 };
 
 #endif /* PLUGIN_TRANSACTION_LOG_TRANSACTION_LOG_APPLIER_H */

=== added file 'plugin/transaction_log/write_buffer.cc'
--- plugin/transaction_log/write_buffer.cc	1970-01-01 00:00:00 +0000
+++ plugin/transaction_log/write_buffer.cc	2010-04-06 19:16:34 +0000
@@ -0,0 +1,71 @@
+/* - mode: c; c-basic-offset: 2; indent-tabs-mode: nil; -*-
+ *  vim:expandtab:shiftwidth=2:tabstop=2:smarttab:
+ *
+ *  Copyright (c) 2010 Jay Pipes <[e-mail]>
+ *
+ *  Authors:
+ *
+ *  Jay Pipes <[e-mail]>
+ *
+ *  This program is free software; you can redistribute it and/or modify
+ *  it under the terms of the GNU General Public License as published by
+ *  the Free Software Foundation; either version 2 of the License, or
+ *  (at your option) any later version.
+ *
+ *  This program is distributed in the hope that it will be useful,
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *  GNU General Public License for more details.
+ *
+ *  You should have received a copy of the GNU General Public License
+ *  along with this program; if not, write to the Free Software
+ *  Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
+ */
+
+/**
+ * @file
+ *
+ * Defines the implementation of a simple locked write buffer
+ *
+ * @details
+ *
+ * The write buffer keeps a block of allocated raw bytes available for
+ * callers.  
+ */
+
+#include "config.h"
+#include "write_buffer.h"
+
+#include <drizzled/errmsg_print.h>
+#include <drizzled/gettext.h>
+
+#include <vector>
+
+using namespace std;
+using namespace drizzled;
+
+WriteBuffer::WriteBuffer() :
+  buffer()
+{
+  buffer.reserve(DEFAULT_WRITE_BUFFER_SIZE);
+  pthread_mutex_init(&latch, NULL);
+}
+
+WriteBuffer::~WriteBuffer()
+{
+  buffer.clear();
+  pthread_mutex_destroy(&latch);
+}
+
+void WriteBuffer::resize(size_t new_size)
+{
+  /* 
+   * Attempt allocation of raw memory buffer for the 
+   * requested size.  Does nothing if already allocated size
+   * if greater...
+   */
+  if (buffer.capacity() >= new_size)
+    return;
+
+  buffer.reserve(new_size);
+}

=== added file 'plugin/transaction_log/write_buffer.h'
--- plugin/transaction_log/write_buffer.h	1970-01-01 00:00:00 +0000
+++ plugin/transaction_log/write_buffer.h	2010-04-06 19:16:34 +0000
@@ -0,0 +1,90 @@
+/* - mode: c; c-basic-offset: 2; indent-tabs-mode: nil; -*-
+ *  vim:expandtab:shiftwidth=2:tabstop=2:smarttab:
+ *
+ *  Copyright (c) 2010 Jay Pipes <[e-mail]>
+ *
+ *  Authors:
+ *
+ *  Jay Pipes <[e-mail]>
+ *
+ *  This program is free software; you can redistribute it and/or modify
+ *  it under the terms of the GNU General Public License as published by
+ *  the Free Software Foundation; either version 2 of the License, or
+ *  (at your option) any later version.
+ *
+ *  This program is distributed in the hope that it will be useful,
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *  GNU General Public License for more details.
+ *
+ *  You should have received a copy of the GNU General Public License
+ *  along with this program; if not, write to the Free Software
+ *  Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
+ */
+
+/**
+ * @file
+ *
+ * Defines a simple structure for maintaining a write buffer.
+ */
+
+#ifndef PLUGIN_TRANSACTION_LOG_WRITE_BUFFER_H
+#define PLUGIN_TRANSACTION_LOG_WRITE_BUFFER_H
+
+#include <stdint.h>
+#include <vector>
+#include <pthread.h>
+
+class WriteBuffer
+{
+public:
+  static const size_t DEFAULT_WRITE_BUFFER_SIZE= 1024; /* Many GPB messages are < 1 KB... */
+  /**
+   * Constructor.
+   */
+  WriteBuffer();
+  ~WriteBuffer();
+  /**
+   * Locks the log write buffer
+   */
+  void lock()
+  {
+    pthread_mutex_lock(&latch);
+  }
+  /**
+   * Unlocks the log's write buffer
+   */
+  void unlock()
+  {
+    pthread_mutex_unlock(&latch);
+  }
+  /**
+   * Resizes the internal raw byte buffer
+   *
+   * @param[in] New size to allocate
+   */
+  void resize(size_t new_size);
+  /**
+   * Returns the pointer to the raw bytes.
+   */
+  uint8_t *getRawBytes()
+  {
+    return &buffer[0];
+  }
+  /**
+   * Returns the size of the write buffer
+   */
+  size_t getCapacity()
+  {
+    return buffer.size();
+  }
+private:
+  /* Prohibit these */
+  WriteBuffer(const WriteBuffer&);
+  WriteBuffer &operator=(const WriteBuffer&);
+
+  std::vector<uint8_t> buffer; ///< Raw memory buffer managed by the log
+  pthread_mutex_t latch; ///< Lock around the synchronized parts of the log (the write buffer)
+};
+
+#endif /* PLUGIN_TRANSACTION_LOG_WRITE_BUFFER_H */