summaryrefslogtreecommitdiff
path: root/plugin/feedback/sender_thread.cc
blob: bcd10f6713f25384ef2199ce59dfa977ea7c1c59 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
/* Copyright (C) 2010 Sergei Golubchik and Monty Program Ab

   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 Street, Fifth Floor, Boston, MA 02111-1301 USA */

#include "feedback.h"
#include <sql_acl.h>
#include <sql_parse.h>
#include <sql_show.h>
#include <time.h>

namespace feedback {

static THD *thd= 0;                ///< background thread thd
static my_thread_id thd_thread_id; ///< its thread_id

static size_t needed_size= 20480;

ulong startup_interval= 60*5;     ///< in seconds (5 minutes)
ulong first_interval= 60*60*24;   ///< in seconds (one day)
ulong interval= 60*60*24*7;       ///< in seconds (one week)

/**
  reads the rows from a table and puts them, concatenated, in a String

  @note
  1. only supports two column tables - no less, no more.
  2. it emulates mysql -e "select * from..." and thus it separates
     columns with \t and starts the output with column names.
*/
static int table_to_string(TABLE *table, String *result)
{
  bool res;
  char buff1[MAX_FIELD_WIDTH], buff2[MAX_FIELD_WIDTH];
  String str1(buff1, sizeof(buff1), system_charset_info);
  String str2(buff2, sizeof(buff2), system_charset_info);

  res= table->file->ha_rnd_init(1);

  dbug_tmp_use_all_columns(table, table->read_set);

  while(!res && !table->file->ha_rnd_next(table->record[0]))
  {
    table->field[0]->val_str(&str1);
    table->field[1]->val_str(&str2);
    if (result->reserve(str1.length() + str2.length() + 3))
      res= 1;
    else
    {
      result->qs_append(str1.ptr(), str1.length());
      result->qs_append('\t');
      result->qs_append(str2.ptr(), str2.length());
      result->qs_append('\n');
    }
  }

  res = res || result->append('\n');

  /*
    Note, "|=" and not "||" - because we want to call ha_rnd_end()
    even if res is already 1.
  */
  res |= table->file->ha_rnd_end();

  return res;
}

/**
  Initialize the THD and TABLE_LIST

  The structures must be sufficiently initialized for create_tmp_table()
  and fill_feedback() to work.
*/
static int prepare_for_fill(TABLE_LIST *tables)
{
  /*
    Add our thd to the list, for it to be visible in SHOW PROCESSLIST.
    But don't generate thread_id every time - use the saved value
    (every increment of global thread_id counts as a new connection
    in SHOW STATUS and we want to avoid skewing the statistics)
  */
  thd->variables.pseudo_thread_id= thd->thread_id;
  mysql_mutex_lock(&LOCK_thread_count);
  threads.append(thd);
  mysql_mutex_unlock(&LOCK_thread_count);
  thd->thread_stack= (char*) &tables;
  if (thd->store_globals())
    return 1;

  thd->mysys_var->current_cond= &sleep_condition;
  thd->mysys_var->current_mutex= &sleep_mutex;
  thd->proc_info="feedback";
  thd->set_command(COM_SLEEP);
  thd->system_thread= SYSTEM_THREAD_EVENT_WORKER; // whatever
  thd->set_time();
  thd->init_for_queries();
  thd->real_id= pthread_self();
  thd->db= NULL;
  thd->db_length= 0;
  thd->security_ctx->host_or_ip= "";
  thd->security_ctx->db_access= DB_ACLS;
  thd->security_ctx->master_access= ~NO_ACCESS;
  bzero((char*) &thd->net, sizeof(thd->net));
  lex_start(thd);
  mysql_init_select(thd->lex);

  tables->init_one_table(INFORMATION_SCHEMA_NAME.str,
                         INFORMATION_SCHEMA_NAME.length,
                         i_s_feedback->table_name,
                         strlen(i_s_feedback->table_name),
                         0, TL_READ);
  tables->schema_table= i_s_feedback;
  tables->table= create_schema_table(thd, tables);
  if (!tables->table)
    return 1;

  tables->select_lex= thd->lex->current_select;
  tables->table->pos_in_table_list= tables;

  return 0;
}

/**
  Try to detect if this thread is going down

  which can happen for different reasons:
  * plugin is being unloaded
  * mysqld server is being shut down
  * the thread is being killed

*/
static bool going_down()
{
  return shutdown_plugin || shutdown_in_progress || (thd && thd->killed);
}

/**
  just like sleep, but waits on a condition and checks "plugin shutdown" status
*/
static int slept_ok(time_t sec)
{
  struct timespec abstime;
  int ret= 0;

  set_timespec(abstime, sec);

  mysql_mutex_lock(&sleep_mutex);
  while (!going_down() && ret != ETIMEDOUT)
    ret= mysql_cond_timedwait(&sleep_condition, &sleep_mutex, &abstime);
  mysql_mutex_unlock(&sleep_mutex);

  return !going_down();
}

/**
  create a feedback report and send it to all specified urls

  If "when" argument is not null, only it and the server uid are sent.
  Otherwise a full report is generated.
*/
static void send_report(const char *when)
{
  TABLE_LIST tables;
  String str;
  int i, last_todo;
  Url **todo= (Url**)alloca(url_count*sizeof(Url*));

  str.alloc(needed_size); // preallocate it to avoid many small mallocs

  /*
    on startup and shutdown the server may not be completely
    initialized, and full report won't work.
    We send a short status notice only.
  */
  if (when)
  {
    str.length(0);
    str.append(STRING_WITH_LEN("FEEDBACK_SERVER_UID"));
    str.append('\t');
    str.append(server_uid_buf);
    str.append('\n');
    str.append(STRING_WITH_LEN("FEEDBACK_WHEN"));
    str.append('\t');
    str.append(when);
    str.append('\n');
    str.append(STRING_WITH_LEN("FEEDBACK_USER_INFO"));
    str.append('\t');
    str.append(user_info);
    str.append('\n');
    str.append('\n');
  }
  else
  {
    /*
      otherwise, prepare the THD and TABLE_LIST,
      create and fill the temporary table with data just like
      SELECT * FROM INFORMATION_SCHEMA.FEEDBACK is doing,
      read and concatenate table data into a String.
    */
    if (!(thd= new THD(thd_thread_id)))
      return;

    if (prepare_for_fill(&tables))
      goto ret;

    if (fill_feedback(thd, &tables, NULL))
      goto ret;

    if (table_to_string(tables.table, &str))
      goto ret;

    needed_size= (size_t)(str.length() * 1.1);

    free_tmp_table(thd, tables.table);
    tables.table= 0;
  }

  /*
    Try to send the report on every url from the list, remove url on success,
    keep failed in the list. Repeat until the list is empty.
  */
  memcpy(todo, urls, url_count*sizeof(Url*));
  last_todo= url_count - 1;
  do
  {
    for (i= 0; i <= last_todo;)
    {
      Url *url= todo[i];

      if (thd) // for nicer SHOW PROCESSLIST
        thd->set_query(const_cast<char*>(url->url()), (uint) url->url_length());

      if (url->send(str.ptr(), str.length()))
        i++;
      else
        todo[i]= todo[last_todo--];
    }
    if (last_todo < 0)
      break;
  } while (slept_ok(send_retry_wait)); // wait a little bit before retrying

ret:
  if (thd)
  {
    if (tables.table)
      free_tmp_table(thd, tables.table);
    thd->cleanup_after_query();
    /*
      clean up, free the thd.
      reset all thread local status variables to minimize
      the effect of the background thread on SHOW STATUS.
    */
    mysql_mutex_lock(&LOCK_thread_count);
    thd->set_status_var_init();
    thd->killed= KILL_CONNECTION;
    thd->unlink();
    mysql_cond_broadcast(&COND_thread_count);
    mysql_mutex_unlock(&LOCK_thread_count);
    delete thd;
    thd= 0;
  }
}

/**
  background sending thread
*/
pthread_handler_t background_thread(void *arg __attribute__((unused)))
{
  if (my_thread_init())
    return 0;

  thd_thread_id= next_thread_id();

  if (slept_ok(startup_interval))
  {
    send_report("startup");

    if (slept_ok(first_interval))
    {
      send_report(NULL);

      while(slept_ok(interval))
        send_report(NULL);
    }

    send_report("shutdown");
  }

  my_thread_end();
  pthread_exit(0);
  return 0;
}

} // namespace feedback