2 efficient async ctdb traverse
4 Copyright (C) Andrew Tridgell 2007
6 This program is free software; you can redistribute it and/or modify
7 it under the terms of the GNU General Public License as published by
8 the Free Software Foundation; either version 3 of the License, or
9 (at your option) any later version.
11 This program is distributed in the hope that it will be useful,
12 but WITHOUT ANY WARRANTY; without even the implied warranty of
13 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14 GNU General Public License for more details.
16 You should have received a copy of the GNU General Public License
17 along with this program; if not, see <http://www.gnu.org/licenses/>.
21 #include "lib/events/events.h"
22 #include "system/filesys.h"
23 #include "system/wait.h"
25 #include "lib/tdb/include/tdb.h"
26 #include "../include/ctdb_private.h"
28 typedef void (*ctdb_traverse_fn_t)(void *private_data, TDB_DATA key, TDB_DATA data);
31 handle returned to caller - freeing this handler will kill the child and
32 terminate the traverse
34 struct ctdb_traverse_local_handle {
35 struct ctdb_db_context *ctdb_db;
39 ctdb_traverse_fn_t callback;
40 struct timeval start_time;
41 struct ctdb_queue *queue;
45 called when data is available from the child
47 static void ctdb_traverse_local_handler(uint8_t *rawdata, size_t length, void *private_data)
49 struct ctdb_traverse_local_handle *h = talloc_get_type(private_data,
50 struct ctdb_traverse_local_handle);
52 ctdb_traverse_fn_t callback = h->callback;
53 void *p = h->private_data;
54 struct ctdb_rec_data *tdata = (struct ctdb_rec_data *)rawdata;
56 if (rawdata == NULL || length < 4 || length != tdata->length) {
59 callback(p, tdb_null, tdb_null);
63 key.dsize = tdata->keylen;
64 key.dptr = &tdata->data[0];
65 data.dsize = tdata->datalen;
66 data.dptr = &tdata->data[tdata->keylen];
68 callback(p, key, data);
72 destroy a in-flight traverse operation
74 static int traverse_local_destructor(struct ctdb_traverse_local_handle *h)
76 kill(h->child, SIGKILL);
77 waitpid(h->child, NULL, 0);
82 callback from tdb_traverse_read()
84 static int ctdb_traverse_local_fn(struct tdb_context *tdb, TDB_DATA key, TDB_DATA data, void *p)
86 struct ctdb_traverse_local_handle *h = talloc_get_type(p,
87 struct ctdb_traverse_local_handle);
88 struct ctdb_rec_data *d;
89 struct ctdb_ltdb_header *hdr;
91 /* filter out non-authoritative and zero-length records */
92 hdr = (struct ctdb_ltdb_header *)data.dptr;
93 if (data.dsize <= sizeof(struct ctdb_ltdb_header) ||
94 hdr->dmaster != h->ctdb_db->ctdb->pnn) {
98 d = ctdb_marshall_record(h, 0, key, NULL, data);
100 /* error handling is tricky in this child code .... */
104 if (write(h->fd[1], (uint8_t *)d, d->length) != d->length) {
112 setup a non-blocking traverse of a local ltdb. The callback function
113 will be called on every record in the local ltdb. To stop the
114 travserse, talloc_free() the travserse_handle.
116 The traverse is finished when the callback is called with tdb_null for key and data
118 static struct ctdb_traverse_local_handle *ctdb_traverse_local(struct ctdb_db_context *ctdb_db,
119 ctdb_traverse_fn_t callback,
122 struct ctdb_traverse_local_handle *h;
125 h = talloc_zero(ctdb_db, struct ctdb_traverse_local_handle);
139 if (h->child == (pid_t)-1) {
146 h->callback = callback;
147 h->private_data = private_data;
148 h->ctdb_db = ctdb_db;
151 /* start the traverse in the child */
153 tdb_traverse_read(ctdb_db->ltdb->tdb, ctdb_traverse_local_fn, h);
158 talloc_set_destructor(h, traverse_local_destructor);
161 setup a packet queue between the child and the parent. This
162 copes with all the async and packet boundary issues
164 h->queue = ctdb_queue_setup(ctdb_db->ctdb, h, h->fd[0], 0, ctdb_traverse_local_handler, h);
165 if (h->queue == NULL) {
170 h->start_time = timeval_current();
176 struct ctdb_traverse_all_handle {
177 struct ctdb_context *ctdb;
179 ctdb_traverse_fn_t callback;
185 destroy a traverse_all op
187 static int ctdb_traverse_all_destructor(struct ctdb_traverse_all_handle *state)
189 ctdb_reqid_remove(state->ctdb, state->reqid);
193 struct ctdb_traverse_all {
199 /* called when a traverse times out */
200 static void ctdb_traverse_all_timeout(struct event_context *ev, struct timed_event *te,
201 struct timeval t, void *private_data)
203 struct ctdb_traverse_all_handle *state = talloc_get_type(private_data, struct ctdb_traverse_all_handle);
205 state->ctdb->statistics.timeouts.traverse++;
207 state->callback(state->private_data, tdb_null, tdb_null);
212 setup a cluster-wide non-blocking traverse of a ctdb. The
213 callback function will be called on every record in the local
214 ltdb. To stop the travserse, talloc_free() the traverse_handle.
216 The traverse is finished when the callback is called with tdb_null
219 static struct ctdb_traverse_all_handle *ctdb_daemon_traverse_all(struct ctdb_db_context *ctdb_db,
220 ctdb_traverse_fn_t callback,
223 struct ctdb_traverse_all_handle *state;
224 struct ctdb_context *ctdb = ctdb_db->ctdb;
227 struct ctdb_traverse_all r;
229 state = talloc(ctdb_db, struct ctdb_traverse_all_handle);
235 state->reqid = ctdb_reqid_new(ctdb_db->ctdb, state);
236 state->callback = callback;
237 state->private_data = private_data;
238 state->null_count = 0;
240 talloc_set_destructor(state, ctdb_traverse_all_destructor);
242 r.db_id = ctdb_db->db_id;
243 r.reqid = state->reqid;
246 data.dptr = (uint8_t *)&r;
247 data.dsize = sizeof(r);
249 /* tell all the nodes in the cluster to start sending records to this node */
250 ret = ctdb_daemon_send_control(ctdb, CTDB_BROADCAST_VNNMAP, 0,
251 CTDB_CONTROL_TRAVERSE_ALL,
252 0, CTDB_CTRL_FLAG_NOREPLY, data, NULL, NULL);
258 /* timeout the traverse */
259 event_add_timed(ctdb->ev, state,
260 timeval_current_ofs(ctdb->tunable.traverse_timeout, 0),
261 ctdb_traverse_all_timeout, state);
266 struct traverse_all_state {
267 struct ctdb_context *ctdb;
268 struct ctdb_traverse_local_handle *h;
274 called for each record during a traverse all
276 static void traverse_all_callback(void *p, TDB_DATA key, TDB_DATA data)
278 struct traverse_all_state *state = talloc_get_type(p, struct traverse_all_state);
280 struct ctdb_rec_data *d;
283 d = ctdb_marshall_record(state, state->reqid, key, NULL, data);
286 DEBUG(DEBUG_ERR,("Out of memory in traverse_all_callback\n"));
290 cdata.dptr = (uint8_t *)d;
291 cdata.dsize = d->length;
293 ret = ctdb_daemon_send_control(state->ctdb, state->srcnode, 0, CTDB_CONTROL_TRAVERSE_DATA,
294 0, CTDB_CTRL_FLAG_NOREPLY, cdata, NULL, NULL);
296 DEBUG(DEBUG_ERR,("Failed to send traverse data\n"));
299 if (key.dsize == 0 && data.dsize == 0) {
306 called when a CTDB_CONTROL_TRAVERSE_ALL control comes in. We then
307 setup a traverse of our local ltdb, sending the records as
308 CTDB_CONTROL_TRAVERSE_DATA records back to the originator
310 int32_t ctdb_control_traverse_all(struct ctdb_context *ctdb, TDB_DATA data, TDB_DATA *outdata)
312 struct ctdb_traverse_all *c = (struct ctdb_traverse_all *)data.dptr;
313 struct traverse_all_state *state;
314 struct ctdb_db_context *ctdb_db;
316 if (data.dsize != sizeof(struct ctdb_traverse_all)) {
317 DEBUG(DEBUG_ERR,("Invalid size in ctdb_control_traverse_all\n"));
321 ctdb_db = find_ctdb_db(ctdb, c->db_id);
322 if (ctdb_db == NULL) {
326 state = talloc(ctdb_db, struct traverse_all_state);
331 state->reqid = c->reqid;
332 state->srcnode = c->pnn;
335 state->h = ctdb_traverse_local(ctdb_db, traverse_all_callback, state);
336 if (state->h == NULL) {
346 called when a CTDB_CONTROL_TRAVERSE_DATA control comes in. We then
347 call the traverse_all callback with the record
349 int32_t ctdb_control_traverse_data(struct ctdb_context *ctdb, TDB_DATA data, TDB_DATA *outdata)
351 struct ctdb_rec_data *d = (struct ctdb_rec_data *)data.dptr;
352 struct ctdb_traverse_all_handle *state;
354 ctdb_traverse_fn_t callback;
357 if (data.dsize < sizeof(uint32_t) || data.dsize != d->length) {
358 DEBUG(DEBUG_ERR,("Bad record size in ctdb_control_traverse_data\n"));
362 state = ctdb_reqid_find(ctdb, d->reqid, struct ctdb_traverse_all_handle);
363 if (state == NULL || d->reqid != state->reqid) {
364 /* traverse might have been terminated already */
368 key.dsize = d->keylen;
369 key.dptr = &d->data[0];
370 data.dsize = d->datalen;
371 data.dptr = &d->data[d->keylen];
373 if (key.dsize == 0 && data.dsize == 0) {
375 if (state->null_count != ctdb_get_num_active_nodes(ctdb)) {
380 callback = state->callback;
381 private_data = state->private_data;
383 callback(private_data, key, data);
384 if (key.dsize == 0 && data.dsize == 0) {
385 /* we've received all of the null replies, so all
386 nodes are finished */
392 struct traverse_start_state {
393 struct ctdb_context *ctdb;
394 struct ctdb_traverse_all_handle *h;
401 callback which sends records as messages to the client
403 static void traverse_start_callback(void *p, TDB_DATA key, TDB_DATA data)
405 struct traverse_start_state *state;
406 struct ctdb_rec_data *d;
409 state = talloc_get_type(p, struct traverse_start_state);
411 d = ctdb_marshall_record(state, state->reqid, key, NULL, data);
416 cdata.dptr = (uint8_t *)d;
417 cdata.dsize = d->length;
419 ctdb_dispatch_message(state->ctdb, state->srvid, cdata);
420 if (key.dsize == 0 && data.dsize == 0) {
421 /* end of traverse */
427 start a traverse_all - called as a control from a client
429 int32_t ctdb_control_traverse_start(struct ctdb_context *ctdb, TDB_DATA data,
430 TDB_DATA *outdata, uint32_t srcnode)
432 struct ctdb_traverse_start *d = (struct ctdb_traverse_start *)data.dptr;
433 struct traverse_start_state *state;
434 struct ctdb_db_context *ctdb_db;
436 if (data.dsize != sizeof(*d)) {
437 DEBUG(DEBUG_ERR,("Bad record size in ctdb_control_traverse_start\n"));
441 ctdb_db = find_ctdb_db(ctdb, d->db_id);
442 if (ctdb_db == NULL) {
446 state = talloc(ctdb_db, struct traverse_start_state);
451 state->srcnode = srcnode;
452 state->reqid = d->reqid;
453 state->srvid = d->srvid;
456 state->h = ctdb_daemon_traverse_all(ctdb_db, traverse_start_callback, state);
457 if (state->h == NULL) {