← Back to branch summary

~drizzle-trunk/drizzle/development

« back to all changes in this revision

Viewing changes to drizzled/replication_services.cc

This patch adds the following functionality:

* Removes the need to manually enable replicators in order
  for an applier to work.
* Removes the enabled/disabled setting of both transaction
  applier plugins and transaction replicator plugins
* Pairs a replicator with an applier into a "ReplicationStream"
  and removes all checks for "enabled" replicators and appliers
* Allows modules that implement a TransactionApplier (such as
  the transaction_log module) to specify which replicator to
  use via a configuration variable.  For instance, the transaction
  log module now has --transaction-log-use-replicator=[default|filtered..]
  instead of the user having to do --default-replicator-enable and such
* Adds a new data dictionary table for REPLICATION_STREAMS, which
  allows querying of activated replication-to-applier streams
  managed by drizzled::ReplicationServices

Show diffs side-by-side

added added

removed removed

Lines of Context:
drizzled/replication_services.cc
43
43
#include "drizzled/session.h"
44
44
#include "drizzled/error.h"
45
45
 
 
46
#include <string>
46
47
#include <vector>
 
48
#include <algorithm>
47
49
 
48
50
using namespace std;
49
51
 
50
52
namespace drizzled
51
53
{
52
54
 
53
 
ReplicationServices::ReplicationServices()
 
55
ReplicationServices::ReplicationServices() :
 
56
  is_active(false)
54
57
{
55
 
  is_active= false;
56
58
}
57
59
 
58
 
void ReplicationServices::evaluateActivePlugins()
 
60
void ReplicationServices::normalizeReplicatorName(string &name)
59
61
{
60
 
  /* 
61
 
   * We loop through replicators and appliers, evaluating
62
 
   * whether or not there is at least one active replicator
63
 
   * and one active applier.  If not, we set is_active
64
 
   * to false.
65
 
   */
66
 
  bool tmp_is_active= false;
67
 
 
68
 
  if (replicators.empty() || appliers.empty())
69
 
  {
70
 
    is_active= false;
71
 
    return;
72
 
  }
73
 
 
74
 
  /* 
75
 
   * Determine if any remaining replicators and if those
76
 
   * replicators are active...if not, set is_active
77
 
   * to false
78
 
   */
79
 
  for (Replicators::iterator repl_iter= replicators.begin();
80
 
       repl_iter != replicators.end();
81
 
       ++repl_iter)
82
 
  {
83
 
    if ((*repl_iter)->isEnabled())
 
62
  transform(name.begin(),
 
63
            name.end(),
 
64
            name.begin(),
 
65
            ::tolower);
 
66
  if (name.find("replicator") == string::npos)
 
67
    name.append("replicator", 10);
 
68
  {
 
69
    size_t found_underscore= name.find('_');
 
70
    while (found_underscore != string::npos)
84
71
    {
85
 
      tmp_is_active= true;
86
 
      break;
 
72
      name.erase(found_underscore, 1);
 
73
      found_underscore= name.find('_');
87
74
    }
88
75
  }
89
 
  if (! tmp_is_active)
90
 
  {
91
 
    /* No active replicators. Set is_active to false and exit. */
92
 
    is_active= false;
93
 
    return;
94
 
  }
 
76
}
95
77
 
 
78
bool ReplicationServices::evaluateRegisteredPlugins()
 
79
{
96
80
  /* 
97
 
   * OK, we know there's at least one active replicator.
98
 
   *
99
 
   * Now determine if any remaining replicators and if those
100
 
   * replicators are active...if not, set is_active
101
 
   * to false
 
81
   * We loop through appliers that have registered with us
 
82
   * and attempts to pair the applier with its requested
 
83
   * replicator.  If an applier has requested a replicator
 
84
   * that has either not been built or has not registered
 
85
   * with the replication services, we print an error and
 
86
   * return false
102
87
   */
 
88
  if (appliers.empty())
 
89
    return true;
 
90
 
 
91
  if (replicators.empty() && not appliers.empty())
 
92
  {
 
93
    errmsg_printf(ERRMSG_LVL_ERROR,
 
94
                  N_("You registered a TransactionApplier plugin but no "
 
95
                     "TransactionReplicator plugins were registered.\n"));
 
96
    return false;
 
97
  }
 
98
 
103
99
  for (Appliers::iterator appl_iter= appliers.begin();
104
100
       appl_iter != appliers.end();
105
101
       ++appl_iter)
106
102
  {
107
 
    if ((*appl_iter)->isEnabled())
108
 
    {
109
 
      is_active= true;
110
 
      return;
 
103
    plugin::TransactionApplier *applier= (*appl_iter).second;
 
104
    string requested_replicator_name= (*appl_iter).first;
 
105
    normalizeReplicatorName(requested_replicator_name);
 
106
 
 
107
    bool found= false;
 
108
    Replicators::iterator repl_iter;
 
109
    for (repl_iter= replicators.begin();
 
110
         repl_iter != replicators.end();
 
111
         ++repl_iter)
 
112
    {
 
113
      string replicator_name= (*repl_iter)->getName();
 
114
      normalizeReplicatorName(replicator_name);
 
115
 
 
116
      if (requested_replicator_name.compare(replicator_name) == 0)
 
117
      {
 
118
        found= true;
 
119
        break;
 
120
      }
 
121
    }
 
122
    if (not found)
 
123
    {
 
124
      errmsg_printf(ERRMSG_LVL_ERROR,
 
125
                    N_("You registered a TransactionApplier plugin but no "
 
126
                       "TransactionReplicator plugins were registered that match the "
 
127
                       "requested replicator name of '%s'.\n"
 
128
                       "We have deactivated the TransactionApplier '%s'.\n"),
 
129
                       requested_replicator_name.c_str(),
 
130
                       applier->getName().c_str());
 
131
      applier->deactivate();
 
132
      return false;
 
133
    }
 
134
    else
 
135
    {
 
136
      replication_streams.push_back(make_pair(*repl_iter, applier));
111
137
    }
112
138
  }
113
 
  /* If we get here, there are no active appliers */
114
 
  is_active= false;
 
139
  is_active= true;
 
140
  return true;
115
141
}
116
142
 
117
143
void ReplicationServices::attachReplicator(plugin::TransactionReplicator *in_replicator)
118
144
{
119
145
  replicators.push_back(in_replicator);
120
 
  evaluateActivePlugins();
121
146
}
122
147
 
123
148
void ReplicationServices::detachReplicator(plugin::TransactionReplicator *in_replicator)
124
149
{
125
150
  replicators.erase(std::find(replicators.begin(), replicators.end(), in_replicator));
126
 
  evaluateActivePlugins();
127
 
}
128
 
 
129
 
void ReplicationServices::attachApplier(plugin::TransactionApplier *in_applier)
130
 
{
131
 
  appliers.push_back(in_applier);
132
 
  evaluateActivePlugins();
133
 
}
134
 
 
135
 
void ReplicationServices::detachApplier(plugin::TransactionApplier *in_applier)
136
 
{
137
 
  appliers.erase(std::find(appliers.begin(), appliers.end(), in_applier));
138
 
  evaluateActivePlugins();
 
151
}
 
152
 
 
153
void ReplicationServices::attachApplier(plugin::TransactionApplier *in_applier, const string &requested_replicator_name)
 
154
{
 
155
  appliers.push_back(make_pair(requested_replicator_name, in_applier));
 
156
}
 
157
 
 
158
void ReplicationServices::detachApplier(plugin::TransactionApplier *)
 
159
{
139
160
}
140
161
 
141
162
bool ReplicationServices::isActive() const
146
167
plugin::ReplicationReturnCode ReplicationServices::pushTransactionMessage(Session &in_session,
147
168
                                                                          message::Transaction &to_push)
148
169
{
149
 
  vector<plugin::TransactionReplicator *>::iterator repl_iter= replicators.begin();
150
 
  vector<plugin::TransactionApplier *>::iterator appl_start_iter, appl_iter;
151
 
  appl_start_iter= appliers.begin();
152
 
 
153
 
  plugin::TransactionReplicator *cur_repl;
154
 
  plugin::TransactionApplier *cur_appl;
155
 
 
156
170
  plugin::ReplicationReturnCode result= plugin::SUCCESS;
157
171
 
158
 
  while (repl_iter != replicators.end())
 
172
  for (ReplicationStreams::iterator iter= replication_streams.begin();
 
173
       iter != replication_streams.end();
 
174
       ++iter)
159
175
  {
160
 
    cur_repl= *repl_iter;
161
 
    if (! cur_repl->isEnabled())
162
 
    {
163
 
      ++repl_iter;
164
 
      continue;
165
 
    }
166
 
    
167
 
    appl_iter= appl_start_iter;
168
 
    while (appl_iter != appliers.end())
169
 
    {
170
 
      cur_appl= *appl_iter;
171
 
 
172
 
      if (! cur_appl->isEnabled())
173
 
      {
174
 
        ++appl_iter;
175
 
        continue;
176
 
      }
177
 
 
178
 
      result= cur_repl->replicate(cur_appl, in_session, to_push);
179
 
 
180
 
      if (result == plugin::SUCCESS)
181
 
      {
182
 
        /* 
183
 
         * We update the timestamp for the last applied Transaction so that
184
 
         * publisher plugins can ask the replication services when the
185
 
         * last known applied Transaction was using the getLastAppliedTimestamp()
186
 
         * method.
187
 
         */
188
 
        last_applied_timestamp.fetch_and_store(to_push.transaction_context().end_timestamp());
189
 
        ++appl_iter;
190
 
      }
191
 
      else
192
 
        return result;
193
 
    }
194
 
    ++repl_iter;
 
176
    plugin::TransactionReplicator *cur_repl= (*iter).first;
 
177
    plugin::TransactionApplier *cur_appl= (*iter).second;
 
178
 
 
179
    result= cur_repl->replicate(cur_appl, in_session, to_push);
 
180
 
 
181
    if (result == plugin::SUCCESS)
 
182
    {
 
183
      /* 
 
184
       * We update the timestamp for the last applied Transaction so that
 
185
       * publisher plugins can ask the replication services when the
 
186
       * last known applied Transaction was using the getLastAppliedTimestamp()
 
187
       * method.
 
188
       */
 
189
      last_applied_timestamp.fetch_and_store(to_push.transaction_context().end_timestamp());
 
190
    }
 
191
    else
 
192
      return result;
195
193
  }
196
194
  return result;
197
195
}
198
196
 
 
197
ReplicationServices::ReplicationStreams &ReplicationServices::getReplicationStreams()
 
198
{
 
199
  return replication_streams;
 
200
}
 
201
 
199
202
} /* namespace drizzled */