← Back to branch summary

~drizzle-trunk/drizzle/development

« back to all changes in this revision

Viewing changes to plugin/rabbitmq/rabbitmq_log.cc

[patch 109/129] Merge patch for revision 1921 from InnoDB SVN:
revno: 1921
revision-id: svn-v4:16c675df-0fcb-4bc9-8058-dcc011a37293:branches/zip:6160
parent: svn-v4:16c675df-0fcb-4bc9-8058-dcc011a37293:branches/zip:6159
committer: vasil
timestamp: Wed 2009-11-11 13:33:49 +0000
message:
  branches/zip: Merge r6152:6159 from branches/5.1:
  
  (r6158 was skipped as an equivallent change has already been merged from MySQL)
  
    ------------------------------------------------------------------------
    r6154 | calvin | 2009-11-11 02:51:17 +0200 (Wed, 11 Nov 2009) | 17 lines
    Changed paths:
       M /branches/5.1/include/os0file.h
       M /branches/5.1/os/os0file.c
    
    branches/5.1: fix bug#3139: Mysql crashes: 'windows error 995'
    after several selects on a large DB
    
    During stress environment, Windows AIO may fail with error code
    ERROR_OPERATION_ABORTED. InnoDB does not handle the error, rather
    crashes. The cause of the error is unknown, but likely due to
    faulty hardware or driver.
    
    This patch introduces a new error code OS_FILE_OPERATION_ABORTED,
    which maps to Windows ERROR_OPERATION_ABORTED (995). When the error
    is detected during AIO, the InnoDB will issue a synchronous retry
    (read/write).
    
    This patch has been extensively tested by MySQL support.
    
    Approved by: Marko
    rb://196
    ------------------------------------------------------------------------
    r6158 | vasil | 2009-11-11 14:52:14 +0200 (Wed, 11 Nov 2009) | 37 lines
    Changed paths:
       M /branches/5.1/handler/ha_innodb.cc
       M /branches/5.1/handler/ha_innodb.h
    
    branches/5.1:
    
    Merge a change from MySQL:
    (this has been reviewed by Calvin and Marko, and Calvin says Luis has
    incorporated Marko's suggestions)
    
      ------------------------------------------------------------
      revno: 3092.5.1
      committer: Luis Soares <luis.soares@sun.com>
      branch nick: mysql-5.1-bugteam
      timestamp: Thu 2009-09-24 15:52:52 +0100
      message:
        BUG#42829: binlogging enabled for all schemas regardless of
        binlog-db-db / binlog-ignore-db
              
        InnoDB will return an error if statement based replication is used
        along with transaction isolation level READ-COMMITTED (or weaker),
        even if the statement in question is filtered out according to the
        binlog-do-db rules set. In this case, an error should not be printed.
              
        This patch addresses this issue by extending the existing check in
        external_lock to take into account the filter rules before deciding to
        print an error. Furthermore, it also changes decide_logging_format to
        take into consideration whether the statement is filtered out from 
        binlog before decision is made.
      added:
        mysql-test/suite/binlog/r/binlog_stm_do_db.result
        mysql-test/suite/binlog/t/binlog_stm_do_db-master.opt
        mysql-test/suite/binlog/t/binlog_stm_do_db.test
      modified:
        sql/sql_base.cc
        sql/sql_class.cc
        storage/innobase/handler/ha_innodb.cc
        storage/innobase/handler/ha_innodb.h
        storage/innodb_plugin/handler/ha_innodb.cc
        storage/innodb_plugin/handler/ha_innodb.h
    
    ------------------------------------------------------------------------
modified:
  include/os0file.h              2@16c675df-0fcb-4bc9-8058-dcc011a37293:trunk%2Finclude%2Fos0file.h
  os/os0file.c                   2@16c675df-0fcb-4bc9-8058-dcc011a37293:trunk%2Fos%2Fos0file.c
diff:
=== modified file 'include/os0file.h'

Show diffs side-by-side

added added

removed removed

Lines of Context:
plugin/rabbitmq/rabbitmq_log.cc
40
40
using namespace drizzled;
41
41
using namespace google;
42
42
 
43
 
namespace drizzle_plugin
44
 
{
 
43
/**
 
44
 * The hostname to connect to
 
45
 */
 
46
static char* sysvar_rabbitmq_host= NULL;
45
47
 
46
48
/**
47
49
 * rabbitmq port
48
50
 */
49
 
static port_constraint sysvar_rabbitmq_port;
50
 
 
51
 
 
52
 
RabbitMQLog::RabbitMQLog(const string &name, 
53
 
                         const std::string &exchange,
54
 
                         const std::string &routingkey,
55
 
                         RabbitMQHandler* mqHandler) :
56
 
  plugin::TransactionApplier(name),
57
 
  _rabbitMQHandler(mqHandler),
58
 
  _exchange(exchange),
59
 
  _routingkey(routingkey)
60
 
{ }
 
51
static int sysvar_rabbitmq_port= 0;
 
52
 
 
53
/**
 
54
 * rabbitmq username
 
55
 */
 
56
static char* sysvar_rabbitmq_username= NULL;
 
57
 
 
58
/**
 
59
 * rabbitmq password
 
60
 */
 
61
static char* sysvar_rabbitmq_password= NULL;
 
62
 
 
63
/**
 
64
 * rabbitmq virtualhost
 
65
 */
 
66
static char* sysvar_rabbitmq_virtualhost= NULL;
 
67
 
 
68
/**
 
69
 * rabbitmq exchangename
 
70
 */
 
71
static char* sysvar_rabbitmq_exchange= NULL;
 
72
 
 
73
/**
 
74
 * rabbitmq routing key
 
75
 */
 
76
static char* sysvar_rabbitmq_routingkey= NULL;
 
77
 
 
78
/**
 
79
 * Is the rabbitmq log enabled?
 
80
 */
 
81
static bool sysvar_rabbitmq_log_enabled= false;
 
82
 
 
83
/**
 
84
 * The name of the replicator plugin
 
85
 * to pair the rabbitmq log's applier with.
 
86
 * Defaults to "default"
 
87
 */
 
88
static char *sysvar_rabbitmq_use_replicator= NULL;
 
89
static const char DEFAULT_USE_REPLICATOR[]= "default";
 
90
 
 
91
 
 
92
RabbitMQLog::RabbitMQLog(const string name_arg, 
 
93
                         RabbitMQHandler* mqHandler)
 
94
  :plugin::TransactionApplier(name_arg)
 
95
{
 
96
  rabbitMQHandler= mqHandler;
 
97
}
61
98
 
62
99
RabbitMQLog::~RabbitMQLog() 
63
 
{ }
 
100
{
 
101
}
64
102
 
65
103
plugin::ReplicationReturnCode
66
104
RabbitMQLog::apply(Session &, const message::Transaction &to_apply)
67
105
{
68
106
  size_t message_byte_length= to_apply.ByteSize();
69
 
  uint8_t* buffer= new uint8_t[message_byte_length];
 
107
  uint8_t* buffer= static_cast<uint8_t *>(malloc(message_byte_length));
70
108
  if(buffer == NULL)
71
109
  {
72
110
    errmsg_printf(ERRMSG_LVL_ERROR, _("Failed to allocate enough memory to transaction message\n"));
77
115
  to_apply.SerializeWithCachedSizesToArray(buffer);
78
116
  try
79
117
  {
80
 
    _rabbitMQHandler->publish(buffer, 
81
 
                             int(message_byte_length), 
82
 
                             _exchange,
83
 
                             _routingkey);
 
118
    rabbitMQHandler->publish(buffer, 
 
119
                             int(message_byte_length), 
 
120
                             sysvar_rabbitmq_exchange, 
 
121
                             sysvar_rabbitmq_routingkey);
84
122
  }
85
123
  catch(exception& e)
86
124
  {
88
126
    deactivate();
89
127
    return plugin::UNKNOWN_ERROR;
90
128
  }
91
 
  delete[] buffer;
 
129
  free(buffer);
92
130
  return plugin::SUCCESS;
93
131
}
94
132
 
104
142
{
105
143
  const module::option_map &vm= context.getOptions();
106
144
  
107
 
  try 
108
 
  {
109
 
    rabbitmqHandler= new RabbitMQHandler(vm["host"].as<string>(),
110
 
                                         sysvar_rabbitmq_port, 
111
 
                                         vm["username"].as<string>(), 
112
 
                                         vm["password"].as<string>(), 
113
 
                                         vm["virtualhost"].as<string>());
114
 
  } 
115
 
  catch (exception& e) 
116
 
  {
117
 
    errmsg_printf(ERRMSG_LVL_ERROR, _("Failed to allocate the RabbitMQHandler.  Got error: %s\n"),
118
 
                  e.what());
119
 
    return 1;
120
 
  }
121
 
  try 
122
 
  {
123
 
    rabbitmqLogger= new RabbitMQLog("rabbit_log_applier",
124
 
                                    vm["exchange"].as<string>(),
125
 
                                    vm["routingkey"].as<string>(),
126
 
                                    rabbitmqHandler);
127
 
  } 
128
 
  catch (exception& e) 
129
 
  {
130
 
    errmsg_printf(ERRMSG_LVL_ERROR, _("Failed to allocate the RabbitMQLog instance.  Got error: %s\n"), 
131
 
                  e.what());
132
 
    return 1;
133
 
  }
134
 
 
135
 
  context.add(rabbitmqLogger);
136
 
  ReplicationServices &replication_services= ReplicationServices::singleton();
137
 
  replication_services.attachApplier(rabbitmqLogger, vm["use-replicator"].as<string>());
138
 
 
139
 
  context.registerVariable(new sys_var_const_string_val("host", vm["host"].as<string>()));
140
 
  context.registerVariable(new sys_var_constrained_value_readonly<in_port_t>("port", sysvar_rabbitmq_port));
141
 
  context.registerVariable(new sys_var_const_string_val("username", vm["username"].as<string>()));
142
 
  context.registerVariable(new sys_var_const_string_val("password", vm["password"].as<string>()));
143
 
  context.registerVariable(new sys_var_const_string_val("virtualhost", vm["virtualhost"].as<string>()));
144
 
  context.registerVariable(new sys_var_const_string_val("exchange", vm["exchange"].as<string>()));
145
 
  context.registerVariable(new sys_var_const_string_val("routingkey", vm["routingkey"].as<string>()));
146
 
 
 
145
  if (vm.count("username"))
 
146
  {
 
147
    sysvar_rabbitmq_username= const_cast<char *>(vm["username"].as<string>().c_str());
 
148
  }
 
149
 
 
150
  if (vm.count("password"))
 
151
  {
 
152
    sysvar_rabbitmq_password= const_cast<char *>(vm["password"].as<string>().c_str());
 
153
  }
 
154
 
 
155
  if (vm.count("host"))
 
156
  {
 
157
    sysvar_rabbitmq_host= const_cast<char *>(vm["host"].as<string>().c_str());
 
158
  }
 
159
 
 
160
  if(sysvar_rabbitmq_log_enabled)
 
161
  {
 
162
    try 
 
163
    {
 
164
      rabbitmqHandler= new RabbitMQHandler(sysvar_rabbitmq_host, 
 
165
                                           sysvar_rabbitmq_port, 
 
166
                                           sysvar_rabbitmq_username, 
 
167
                                           sysvar_rabbitmq_password, 
 
168
                                           sysvar_rabbitmq_virtualhost);
 
169
    } 
 
170
    catch (exception& e) 
 
171
    {
 
172
      errmsg_printf(ERRMSG_LVL_ERROR, _("Failed to allocate the RabbitMQHandler.  Got error: %s\n"),
 
173
                    e.what());
 
174
      return 1;
 
175
    }
 
176
    try 
 
177
    {
 
178
      rabbitmqLogger= new RabbitMQLog("rabbit_log_applier", rabbitmqHandler);
 
179
    } 
 
180
    catch (exception& e) 
 
181
    {
 
182
      errmsg_printf(ERRMSG_LVL_ERROR, _("Failed to allocate the RabbitMQLog instance.  Got error: %s\n"), 
 
183
                    e.what());
 
184
      return 1;
 
185
    }
 
186
 
 
187
    context.add(rabbitmqLogger);
 
188
    ReplicationServices &replication_services= ReplicationServices::singleton();
 
189
    string replicator_name(sysvar_rabbitmq_use_replicator);
 
190
    replication_services.attachApplier(rabbitmqLogger, replicator_name);
 
191
    return 0;
 
192
  }
147
193
  return 0;
148
194
}
149
195
 
150
196
 
 
197
static DRIZZLE_SYSVAR_BOOL(enable,
 
198
                           sysvar_rabbitmq_log_enabled,
 
199
                           PLUGIN_VAR_NOCMDARG,
 
200
                           N_("Enable rabbitmq log"),
 
201
                           NULL, /* check func */
 
202
                           NULL, /* update func */
 
203
                           false /* default */);
 
204
 
 
205
 
 
206
static DRIZZLE_SYSVAR_STR(hostname,
 
207
                          sysvar_rabbitmq_host,
 
208
                          PLUGIN_VAR_READONLY,
 
209
                          N_("Host name to connect to"),
 
210
                          NULL, /* check func */
 
211
                          NULL, /* update func*/
 
212
                          "localhost" /* default */);
 
213
 
 
214
 
 
215
static DRIZZLE_SYSVAR_INT(port,
 
216
                          sysvar_rabbitmq_port,
 
217
                          PLUGIN_VAR_READONLY,
 
218
                          N_("RabbitMQ Port"),
 
219
                          NULL, /* check func */
 
220
                          NULL, /* update func */
 
221
                          5672, /* default */
 
222
                          0,
 
223
                          65535,
 
224
                          0);
 
225
 
 
226
static DRIZZLE_SYSVAR_STR(username,
 
227
                          sysvar_rabbitmq_username,
 
228
                          PLUGIN_VAR_READONLY,
 
229
                          N_("RabbitMQ username"),
 
230
                          NULL, /* check func */
 
231
                          NULL, /* update func*/
 
232
                          "guest" /* default */);
 
233
 
 
234
static DRIZZLE_SYSVAR_STR(password,
 
235
                          sysvar_rabbitmq_password,
 
236
                          PLUGIN_VAR_READONLY,
 
237
                          N_("RabbitMQ password"),
 
238
                          NULL, /* check func */
 
239
                          NULL, /* update func*/
 
240
                          "guest" /* default */);
 
241
 
 
242
static DRIZZLE_SYSVAR_STR(virtualhost,
 
243
                          sysvar_rabbitmq_virtualhost,
 
244
                          PLUGIN_VAR_READONLY,
 
245
                          N_("RabbitMQ virtualhost"),
 
246
                          NULL, /* check func */
 
247
                          NULL, /* update func*/
 
248
                          "/" /* default */);
 
249
 
 
250
static DRIZZLE_SYSVAR_STR(exchange,
 
251
                          sysvar_rabbitmq_exchange,
 
252
                          PLUGIN_VAR_READONLY,
 
253
                          N_("Name of RabbitMQ exchange to publish to"),
 
254
                          NULL, /* check func */
 
255
                          NULL, /* update func*/
 
256
                          "ReplicationExchange" /* default */);
 
257
 
 
258
static DRIZZLE_SYSVAR_STR(routingkey,
 
259
                          sysvar_rabbitmq_routingkey,
 
260
                          PLUGIN_VAR_READONLY,
 
261
                          N_("Name of RabbitMQ routing key to use"),
 
262
                          NULL, /* check func */
 
263
                          NULL, /* update func*/
 
264
                          "ReplicationRoutingKey" /* default */);
 
265
 
 
266
static DRIZZLE_SYSVAR_STR(use_replicator,
 
267
                          sysvar_rabbitmq_use_replicator,
 
268
                          PLUGIN_VAR_READONLY,
 
269
                          N_("Name of the replicator plugin to use (default='default_replicator')"),
 
270
                          NULL, /* check func */
 
271
                          NULL, /* update func*/
 
272
                          DEFAULT_USE_REPLICATOR /* default */);
 
273
 
151
274
static void init_options(drizzled::module::option_context &context)
152
275
{
 
276
  context ("enable",
 
277
           po::value<bool>(&sysvar_rabbitmq_log_enabled)->default_value(false)->zero_tokens(),
 
278
           N_("Enable rabbitmq log"));
153
279
  context("host", 
154
280
          po::value<string>()->default_value("localhost"),
155
281
          N_("Host name to connect to"));
156
 
  context("port",
157
 
          po::value<port_constraint>(&sysvar_rabbitmq_port)->default_value(5672),
158
 
          N_("Port to connect to"));
159
 
  context("virtualhost",
160
 
          po::value<string>()->default_value("/"),
161
 
          N_("RabbitMQ virtualhost"));
162
282
  context("username",
163
283
          po::value<string>()->default_value("guest"),
164
284
          N_("RabbitMQ username"));
165
285
  context("password",
166
286
          po::value<string>()->default_value("guest"),
167
287
          N_("RabbitMQ password"));
168
 
  context("use-replicator",
169
 
          po::value<string>()->default_value("default_replicator"),
170
 
          N_("Name of the replicator plugin to use (default='default_replicator')"));
171
 
  context("exchange",
172
 
          po::value<string>()->default_value("ReplicationExchange"),
173
 
          N_("Name of RabbitMQ exchange to publish to"));
174
 
  context("routingkey",
175
 
          po::value<string>()->default_value("ReplicationRoutingKey"),
176
 
          N_("Name of RabbitMQ routing key to use"));
177
288
}
178
289
 
179
 
} /* namespace drizzle_plugin */
 
290
static drizzle_sys_var* system_variables[]= {
 
291
  DRIZZLE_SYSVAR(enable),
 
292
  DRIZZLE_SYSVAR(hostname),
 
293
  DRIZZLE_SYSVAR(port),
 
294
  DRIZZLE_SYSVAR(username),
 
295
  DRIZZLE_SYSVAR(password),
 
296
  DRIZZLE_SYSVAR(virtualhost),
 
297
  DRIZZLE_SYSVAR(exchange),
 
298
  DRIZZLE_SYSVAR(routingkey),
 
299
  DRIZZLE_SYSVAR(use_replicator),
 
300
  NULL
 
301
};
180
302
 
181
 
DRIZZLE_PLUGIN(drizzle_plugin::init, NULL, drizzle_plugin::init_options);
 
303
DRIZZLE_PLUGIN(init, system_variables, init_options);
182
304