?  search:
summary | log | graphiclog1 | graphiclog2 | commit | commitdiff | tree | refs | edit | fork
blame | history | raw | HEAD
python: Port ntvfs posix bindings to Python 3 compatible form
[Samba.git] / ctdb / client / client_db.c
blobe86830e7bfb99009e9a7010c347c6b129121107c
1 /*
2 CTDB client code
3
4 Copyright (C) Amitay Isaacs 2015
5
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.
10
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.
15
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/>.
18 */
19
20 #include "replace.h"
21 #include "system/network.h"
22 #include "system/filesys.h"
23
24 #include <talloc.h>
25 #include <tevent.h>
26 #include <tdb.h>
27
28 #include "common/logging.h"
29
30 #include "lib/tdb_wrap/tdb_wrap.h"
31 #include "lib/util/tevent_unix.h"
32 #include "lib/util/dlinklist.h"
33 #include "lib/util/debug.h"
34
35 #include "protocol/protocol.h"
36 #include "protocol/protocol_api.h"
37 #include "client/client_private.h"
38 #include "client/client.h"
39
40 static struct ctdb_db_context *client_db_handle(
41 struct ctdb_client_context *client,
42 const char *db_name)
43 {
44 struct ctdb_db_context *db;
45
46 for (db = client->db; db != NULL; db = db->next) {
47 if (strcmp(db_name, db->db_name) == 0) {
48 return db;
49 }
50 }
51
52 return NULL;
53 }
54
55 static bool ctdb_db_persistent(struct ctdb_db_context *db)
56 {
57 if (db->db_flags & CTDB_DB_FLAGS_PERSISTENT) {
58 return true;
59 }
60 return false;
61 }
62
63 static bool ctdb_db_replicated(struct ctdb_db_context *db)
64 {
65 if (db->db_flags & CTDB_DB_FLAGS_REPLICATED) {
66 return true;
67 }
68 return false;
69 }
70
71 static bool ctdb_db_volatile(struct ctdb_db_context *db)
72 {
73 if (db->db_flags & CTDB_DB_FLAGS_PERSISTENT ||
74 db->db_flags & CTDB_DB_FLAGS_REPLICATED) {
75 return false;
76 }
77 return true;
78 }
79
80 struct ctdb_set_db_flags_state {
81 struct tevent_context *ev;
82 struct ctdb_client_context *client;
83 struct timeval timeout;
84 uint32_t db_id;
85 uint8_t db_flags;
86 bool readonly_done, sticky_done;
87 uint32_t *pnn_list;
88 int count;
89 };
90
91 static void ctdb_set_db_flags_nodemap_done(struct tevent_req *subreq);
92 static void ctdb_set_db_flags_readonly_done(struct tevent_req *subreq);
93 static void ctdb_set_db_flags_sticky_done(struct tevent_req *subreq);
94
95 static struct tevent_req *ctdb_set_db_flags_send(
96 TALLOC_CTX *mem_ctx,
97 struct tevent_context *ev,
98 struct ctdb_client_context *client,
99 uint32_t destnode, struct timeval timeout,
100 uint32_t db_id, uint8_t db_flags)
101 {
102 struct tevent_req *req, *subreq;
103 struct ctdb_set_db_flags_state *state;
104 struct ctdb_req_control request;
105
106 req = tevent_req_create(mem_ctx, &state,
107 struct ctdb_set_db_flags_state);
108 if (req == NULL) {
109 return NULL;
110 }
111
112 if (! (db_flags & (CTDB_DB_FLAGS_READONLY | CTDB_DB_FLAGS_STICKY))) {
113 tevent_req_done(req);
114 return tevent_req_post(req, ev);
115 }
116
117 state->ev = ev;
118 state->client = client;
119 state->timeout = timeout;
120 state->db_id = db_id;
121 state->db_flags = db_flags;
122
123 ctdb_req_control_get_nodemap(&request);
124 subreq = ctdb_client_control_send(state, ev, client, destnode, timeout,
125 &request);
126 if (tevent_req_nomem(subreq, req)) {
127 return tevent_req_post(req, ev);
128 }
129 tevent_req_set_callback(subreq, ctdb_set_db_flags_nodemap_done, req);
130
131 return req;
132 }
133
134 static void ctdb_set_db_flags_nodemap_done(struct tevent_req *subreq)
135 {
136 struct tevent_req *req = tevent_req_callback_data(
137 subreq, struct tevent_req);
138 struct ctdb_set_db_flags_state *state = tevent_req_data(
139 req, struct ctdb_set_db_flags_state);
140 struct ctdb_req_control request;
141 struct ctdb_reply_control *reply;
142 struct ctdb_node_map *nodemap;
143 int ret;
144 bool status;
145
146 status = ctdb_client_control_recv(subreq, &ret, state, &reply);
147 TALLOC_FREE(subreq);
148 if (! status) {
149 DEBUG(DEBUG_ERR,
150 ("set_db_flags: 0x%08x GET_NODEMAP failed, ret=%d\n",
151 state->db_id, ret));
152 tevent_req_error(req, ret);
153 return;
154 }
155
156 ret = ctdb_reply_control_get_nodemap(reply, state, &nodemap);
157 talloc_free(reply);
158 if (ret != 0) {
159 DEBUG(DEBUG_ERR,
160 ("set_db_flags: 0x%08x GET_NODEMAP parse failed, ret=%d\n",
161 state->db_id, ret));
162 tevent_req_error(req, ret);
163 return;
164 }
165
166 state->count = list_of_connected_nodes(nodemap, CTDB_UNKNOWN_PNN,
167 state, &state->pnn_list);
168 talloc_free(nodemap);
169 if (state->count <= 0) {
170 DEBUG(DEBUG_ERR,
171 ("set_db_flags: 0x%08x no connected nodes, count=%d\n",
172 state->db_id, state->count));
173 tevent_req_error(req, ENOMEM);
174 return;
175 }
176
177 if (state->db_flags & CTDB_DB_FLAGS_READONLY) {
178 ctdb_req_control_set_db_readonly(&request, state->db_id);
179 subreq = ctdb_client_control_multi_send(
180 state, state->ev, state->client,
181 state->pnn_list, state->count,
182 state->timeout, &request);
183 if (tevent_req_nomem(subreq, req)) {
184 return;
185 }
186 tevent_req_set_callback(subreq,
187 ctdb_set_db_flags_readonly_done, req);
188 } else {
189 state->readonly_done = true;
190 }
191
192 if (state->db_flags & CTDB_DB_FLAGS_STICKY) {
193 ctdb_req_control_set_db_sticky(&request, state->db_id);
194 subreq = ctdb_client_control_multi_send(
195 state, state->ev, state->client,
196 state->pnn_list, state->count,
197 state->timeout, &request);
198 if (tevent_req_nomem(subreq, req)) {
199 return;
200 }
201 tevent_req_set_callback(subreq, ctdb_set_db_flags_sticky_done,
202 req);
203 } else {
204 state->sticky_done = true;
205 }
206 }
207
208 static void ctdb_set_db_flags_readonly_done(struct tevent_req *subreq)
209 {
210 struct tevent_req *req = tevent_req_callback_data(
211 subreq, struct tevent_req);
212 struct ctdb_set_db_flags_state *state = tevent_req_data(
213 req, struct ctdb_set_db_flags_state);
214 int ret;
215 bool status;
216
217 status = ctdb_client_control_multi_recv(subreq, &ret, NULL, NULL,
218 NULL);
219 TALLOC_FREE(subreq);
220 if (! status) {
221 DEBUG(DEBUG_ERR,
222 ("set_db_flags: 0x%08x SET_DB_READONLY failed, ret=%d\n",
223 state->db_id, ret));
224 tevent_req_error(req, ret);
225 return;
226 }
227
228 state->readonly_done = true;
229
230 if (state->readonly_done && state->sticky_done) {
231 tevent_req_done(req);
232 }
233 }
234
235 static void ctdb_set_db_flags_sticky_done(struct tevent_req *subreq)
236 {
237 struct tevent_req *req = tevent_req_callback_data(
238 subreq, struct tevent_req);
239 struct ctdb_set_db_flags_state *state = tevent_req_data(
240 req, struct ctdb_set_db_flags_state);
241 int ret;
242 bool status;
243
244 status = ctdb_client_control_multi_recv(subreq, &ret, NULL, NULL,
245 NULL);
246 TALLOC_FREE(subreq);
247 if (! status) {
248 DEBUG(DEBUG_ERR,
249 ("set_db_flags: 0x%08x SET_DB_STICKY failed, ret=%d\n",
250 state->db_id, ret));
251 tevent_req_error(req, ret);
252 return;
253 }
254
255 state->sticky_done = true;
256
257 if (state->readonly_done && state->sticky_done) {
258 tevent_req_done(req);
259 }
260 }
261
262 static bool ctdb_set_db_flags_recv(struct tevent_req *req, int *perr)
263 {
264 int err;
265
266 if (tevent_req_is_unix_error(req, &err)) {
267 if (perr != NULL) {
268 *perr = err;
269 }
270 return false;
271 }
272 return true;
273 }
274
275 struct ctdb_attach_state {
276 struct tevent_context *ev;
277 struct ctdb_client_context *client;
278 struct timeval timeout;
279 uint32_t destnode;
280 uint8_t db_flags;
281 struct ctdb_db_context *db;
282 };
283
284 static void ctdb_attach_dbid_done(struct tevent_req *subreq);
285 static void ctdb_attach_dbpath_done(struct tevent_req *subreq);
286 static void ctdb_attach_health_done(struct tevent_req *subreq);
287 static void ctdb_attach_flags_done(struct tevent_req *subreq);
288 static void ctdb_attach_open_flags_done(struct tevent_req *subreq);
289
290 struct tevent_req *ctdb_attach_send(TALLOC_CTX *mem_ctx,
291 struct tevent_context *ev,
292 struct ctdb_client_context *client,
293 struct timeval timeout,
294 const char *db_name, uint8_t db_flags)
295 {
296 struct tevent_req *req, *subreq;
297 struct ctdb_attach_state *state;
298 struct ctdb_req_control request;
299
300 req = tevent_req_create(mem_ctx, &state, struct ctdb_attach_state);
301 if (req == NULL) {
302 return NULL;
303 }
304
305 state->db = client_db_handle(client, db_name);
306 if (state->db != NULL) {
307 tevent_req_done(req);
308 return tevent_req_post(req, ev);
309 }
310
311 state->ev = ev;
312 state->client = client;
313 state->timeout = timeout;
314 state->destnode = ctdb_client_pnn(client);
315 state->db_flags = db_flags;
316
317 state->db = talloc_zero(client, struct ctdb_db_context);
318 if (tevent_req_nomem(state->db, req)) {
319 return tevent_req_post(req, ev);
320 }
321
322 state->db->db_name = talloc_strdup(state->db, db_name);
323 if (tevent_req_nomem(state->db, req)) {
324 return tevent_req_post(req, ev);
325 }
326
327 state->db->db_flags = db_flags;
328
329 if (ctdb_db_persistent(state->db)) {
330 ctdb_req_control_db_attach_persistent(&request,
331 state->db->db_name);
332 } else if (ctdb_db_replicated(state->db)) {
333 ctdb_req_control_db_attach_replicated(&request,
334 state->db->db_name);
335 } else {
336 ctdb_req_control_db_attach(&request, state->db->db_name);
337 }
338
339 subreq = ctdb_client_control_send(state, state->ev, state->client,
340 state->destnode, state->timeout,
341 &request);
342 if (tevent_req_nomem(subreq, req)) {
343 return tevent_req_post(req, ev);
344 }
345 tevent_req_set_callback(subreq, ctdb_attach_dbid_done, req);
346
347 return req;
348 }
349
350 static void ctdb_attach_dbid_done(struct tevent_req *subreq)
351 {
352 struct tevent_req *req = tevent_req_callback_data(
353 subreq, struct tevent_req);
354 struct ctdb_attach_state *state = tevent_req_data(
355 req, struct ctdb_attach_state);
356 struct ctdb_req_control request;
357 struct ctdb_reply_control *reply;
358 bool status;
359 int ret;
360
361 status = ctdb_client_control_recv(subreq, &ret, state, &reply);
362 TALLOC_FREE(subreq);
363 if (! status) {
364 DEBUG(DEBUG_ERR, ("attach: %s %s failed, ret=%d\n",
365 state->db->db_name,
366 (ctdb_db_persistent(state->db)
367 ? "DB_ATTACH_PERSISTENT"
368 : (ctdb_db_replicated(state->db)
369 ? "DB_ATTACH_REPLICATED"
370 : "DB_ATTACH")),
371 ret));
372 tevent_req_error(req, ret);
373 return;
374 }
375
376 if (ctdb_db_persistent(state->db)) {
377 ret = ctdb_reply_control_db_attach_persistent(
378 reply, &state->db->db_id);
379 } else if (ctdb_db_replicated(state->db)) {
380 ret = ctdb_reply_control_db_attach_replicated(
381 reply, &state->db->db_id);
382 } else {
383 ret = ctdb_reply_control_db_attach(reply, &state->db->db_id);
384 }
385 talloc_free(reply);
386 if (ret != 0) {
387 DEBUG(DEBUG_ERR, ("attach: %s failed to get db_id, ret=%d\n",
388 state->db->db_name, ret));
389 tevent_req_error(req, ret);
390 return;
391 }
392
393 ctdb_req_control_getdbpath(&request, state->db->db_id);
394 subreq = ctdb_client_control_send(state, state->ev, state->client,
395 state->destnode, state->timeout,
396 &request);
397 if (tevent_req_nomem(subreq, req)) {
398 return;
399 }
400 tevent_req_set_callback(subreq, ctdb_attach_dbpath_done, req);
401 }
402
403 static void ctdb_attach_dbpath_done(struct tevent_req *subreq)
404 {
405 struct tevent_req *req = tevent_req_callback_data(
406 subreq, struct tevent_req);
407 struct ctdb_attach_state *state = tevent_req_data(
408 req, struct ctdb_attach_state);
409 struct ctdb_reply_control *reply;
410 struct ctdb_req_control request;
411 bool status;
412 int ret;
413
414 status = ctdb_client_control_recv(subreq, &ret, state, &reply);
415 TALLOC_FREE(subreq);
416 if (! status) {
417 DEBUG(DEBUG_ERR, ("attach: %s GETDBPATH failed, ret=%d\n",
418 state->db->db_name, ret));
419 tevent_req_error(req, ret);
420 return;
421 }
422
423 ret = ctdb_reply_control_getdbpath(reply, state->db,
424 &state->db->db_path);
425 talloc_free(reply);
426 if (ret != 0) {
427 DEBUG(DEBUG_ERR, ("attach: %s GETDBPATH parse failed, ret=%d\n",
428 state->db->db_name, ret));
429 tevent_req_error(req, ret);
430 return;
431 }
432
433 ctdb_req_control_db_get_health(&request, state->db->db_id);
434 subreq = ctdb_client_control_send(state, state->ev, state->client,
435 state->destnode, state->timeout,
436 &request);
437 if (tevent_req_nomem(subreq, req)) {
438 return;
439 }
440 tevent_req_set_callback(subreq, ctdb_attach_health_done, req);
441 }
442
443 static void ctdb_attach_health_done(struct tevent_req *subreq)
444 {
445 struct tevent_req *req = tevent_req_callback_data(
446 subreq, struct tevent_req);
447 struct ctdb_attach_state *state = tevent_req_data(
448 req, struct ctdb_attach_state);
449 struct ctdb_reply_control *reply;
450 const char *reason;
451 bool status;
452 int ret;
453
454 status = ctdb_client_control_recv(subreq, &ret, state, &reply);
455 TALLOC_FREE(subreq);
456 if (! status) {
457 DEBUG(DEBUG_ERR, ("attach: %s DB_GET_HEALTH failed, ret=%d\n",
458 state->db->db_name, ret));
459 tevent_req_error(req, ret);
460 return;
461 }
462
463 ret = ctdb_reply_control_db_get_health(reply, state, &reason);
464 if (ret != 0) {
465 DEBUG(DEBUG_ERR,
466 ("attach: %s DB_GET_HEALTH parse failed, ret=%d\n",
467 state->db->db_name, ret));
468 tevent_req_error(req, ret);
469 return;
470 }
471
472 if (reason != NULL) {
473 /* Database unhealthy, avoid attach */
474 DEBUG(DEBUG_ERR, ("attach: %s database unhealthy (%s)\n",
475 state->db->db_name, reason));
476 tevent_req_error(req, EIO);
477 return;
478 }
479
480 subreq = ctdb_set_db_flags_send(state, state->ev, state->client,
481 state->destnode, state->timeout,
482 state->db->db_id, state->db_flags);
483 if (tevent_req_nomem(subreq, req)) {
484 return;
485 }
486 tevent_req_set_callback(subreq, ctdb_attach_flags_done, req);
487 }
488
489 static void ctdb_attach_flags_done(struct tevent_req *subreq)
490 {
491 struct tevent_req *req = tevent_req_callback_data(
492 subreq, struct tevent_req);
493 struct ctdb_attach_state *state = tevent_req_data(
494 req, struct ctdb_attach_state);
495 struct ctdb_req_control request;
496 bool status;
497 int ret;
498
499 status = ctdb_set_db_flags_recv(subreq, &ret);
500 TALLOC_FREE(subreq);
501 if (! status) {
502 DEBUG(DEBUG_ERR, ("attach: %s set db flags 0x%08x failed\n",
503 state->db->db_name, state->db_flags));
504 tevent_req_error(req, ret);
505 return;
506 }
507
508 ctdb_req_control_db_open_flags(&request, state->db->db_id);
509 subreq = ctdb_client_control_send(state, state->ev, state->client,
510 state->destnode, state->timeout,
511 &request);
512 if (tevent_req_nomem(subreq, req)) {
513 return;
514 }
515 tevent_req_set_callback(subreq, ctdb_attach_open_flags_done, req);
516 }
517
518 static void ctdb_attach_open_flags_done(struct tevent_req *subreq)
519 {
520 struct tevent_req *req = tevent_req_callback_data(
521 subreq, struct tevent_req);
522 struct ctdb_attach_state *state = tevent_req_data(
523 req, struct ctdb_attach_state);
524 struct ctdb_reply_control *reply;
525 bool status;
526 int ret, tdb_flags;
527
528 status = ctdb_client_control_recv(subreq, &ret, state, &reply);
529 TALLOC_FREE(subreq);
530 if (! status) {
531 DEBUG(DEBUG_ERR, ("attach: %s DB_OPEN_FLAGS failed, ret=%d\n",
532 state->db->db_name, ret));
533 tevent_req_error(req, ret);
534 return;
535 }
536
537 ret = ctdb_reply_control_db_open_flags(reply, &tdb_flags);
538 talloc_free(reply);
539 if (ret != 0) {
540 DEBUG(DEBUG_ERR, ("attach: %s DB_OPEN_FLAGS parse failed,"
541 " ret=%d\n", state->db->db_name, ret));
542 tevent_req_error(req, ret);
543 return;
544 }
545
546 state->db->ltdb = tdb_wrap_open(state->db, state->db->db_path, 0,
547 tdb_flags, O_RDWR, 0);
548 if (tevent_req_nomem(state->db->ltdb, req)) {
549 DEBUG(DEBUG_ERR, ("attach: %s tdb_wrap_open failed\n",
550 state->db->db_name));
551 return;
552 }
553 DLIST_ADD(state->client->db, state->db);
554
555 tevent_req_done(req);
556 }
557
558 bool ctdb_attach_recv(struct tevent_req *req, int *perr,
559 struct ctdb_db_context **out)
560 {
561 struct ctdb_attach_state *state = tevent_req_data(
562 req, struct ctdb_attach_state);
563 int err;
564
565 if (tevent_req_is_unix_error(req, &err)) {
566 if (perr != NULL) {
567 *perr = err;
568 }
569 return false;
570 }
571
572 if (out != NULL) {
573 *out = state->db;
574 }
575 return true;
576 }
577
578 int ctdb_attach(struct tevent_context *ev,
579 struct ctdb_client_context *client,
580 struct timeval timeout,
581 const char *db_name, uint8_t db_flags,
582 struct ctdb_db_context **out)
583 {
584 TALLOC_CTX *mem_ctx;
585 struct tevent_req *req;
586 bool status;
587 int ret;
588
589 mem_ctx = talloc_new(client);
590 if (mem_ctx == NULL) {
591 return ENOMEM;
592 }
593
594 req = ctdb_attach_send(mem_ctx, ev, client, timeout,
595 db_name, db_flags);
596 if (req == NULL) {
597 talloc_free(mem_ctx);
598 return ENOMEM;
599 }
600
601 tevent_req_poll(req, ev);
602
603 status = ctdb_attach_recv(req, &ret, out);
604 if (! status) {
605 talloc_free(mem_ctx);
606 return ret;
607 }
608
609 /*
610 ctdb_set_call(db, CTDB_NULL_FUNC, ctdb_null_func);
611 ctdb_set_call(db, CTDB_FETCH_FUNC, ctdb_fetch_func);
612 ctdb_set_call(db, CTDB_FETCH_WITH_HEADER_FUNC, ctdb_fetch_with_header_func);
613 */
614
615 talloc_free(mem_ctx);
616 return 0;
617 }
618
619 struct ctdb_detach_state {
620 struct ctdb_client_context *client;
621 struct tevent_context *ev;
622 struct timeval timeout;
623 uint32_t db_id;
624 const char *db_name;
625 };
626
627 static void ctdb_detach_dbname_done(struct tevent_req *subreq);
628 static void ctdb_detach_done(struct tevent_req *subreq);
629
630 struct tevent_req *ctdb_detach_send(TALLOC_CTX *mem_ctx,
631 struct tevent_context *ev,
632 struct ctdb_client_context *client,
633 struct timeval timeout, uint32_t db_id)
634 {
635 struct tevent_req *req, *subreq;
636 struct ctdb_detach_state *state;
637 struct ctdb_req_control request;
638
639 req = tevent_req_create(mem_ctx, &state, struct ctdb_detach_state);
640 if (req == NULL) {
641 return NULL;
642 }
643
644 state->client = client;
645 state->ev = ev;
646 state->timeout = timeout;
647 state->db_id = db_id;
648
649 ctdb_req_control_get_dbname(&request, db_id);
650 subreq = ctdb_client_control_send(state, ev, client,
651 ctdb_client_pnn(client), timeout,
652 &request);
653 if (tevent_req_nomem(subreq, req)) {
654 return tevent_req_post(req, ev);
655 }
656 tevent_req_set_callback(subreq, ctdb_detach_dbname_done, req);
657
658 return req;
659 }
660
661 static void ctdb_detach_dbname_done(struct tevent_req *subreq)
662 {
663 struct tevent_req *req = tevent_req_callback_data(
664 subreq, struct tevent_req);
665 struct ctdb_detach_state *state = tevent_req_data(
666 req, struct ctdb_detach_state);
667 struct ctdb_reply_control *reply;
668 struct ctdb_req_control request;
669 int ret;
670 bool status;
671
672 status = ctdb_client_control_recv(subreq, &ret, state, &reply);
673 TALLOC_FREE(subreq);
674 if (! status) {
675 DEBUG(DEBUG_ERR, ("detach: 0x%x GET_DBNAME failed, ret=%d\n",
676 state->db_id, ret));
677 tevent_req_error(req, ret);
678 return;
679 }
680
681 ret = ctdb_reply_control_get_dbname(reply, state, &state->db_name);
682 if (ret != 0) {
683 DEBUG(DEBUG_ERR, ("detach: 0x%x GET_DBNAME failed, ret=%d\n",
684 state->db_id, ret));
685 tevent_req_error(req, ret);
686 return;
687 }
688
689 ctdb_req_control_db_detach(&request, state->db_id);
690 subreq = ctdb_client_control_send(state, state->ev, state->client,
691 ctdb_client_pnn(state->client),
692 state->timeout, &request);
693 if (tevent_req_nomem(subreq, req)) {
694 return;
695 }
696 tevent_req_set_callback(subreq, ctdb_detach_done, req);
697
698 }
699
700 static void ctdb_detach_done(struct tevent_req *subreq)
701 {
702 struct tevent_req *req = tevent_req_callback_data(
703 subreq, struct tevent_req);
704 struct ctdb_detach_state *state = tevent_req_data(
705 req, struct ctdb_detach_state);
706 struct ctdb_reply_control *reply;
707 struct ctdb_db_context *db;
708 int ret;
709 bool status;
710
711 status = ctdb_client_control_recv(subreq, &ret, state, &reply);
712 TALLOC_FREE(subreq);
713 if (! status) {
714 DEBUG(DEBUG_ERR, ("detach: %s DB_DETACH failed, ret=%d\n",
715 state->db_name, ret));
716 tevent_req_error(req, ret);
717 return;
718 }
719
720 ret = ctdb_reply_control_db_detach(reply);
721 if (ret != 0) {
722 DEBUG(DEBUG_ERR, ("detach: %s DB_DETACH failed, ret=%d\n",
723 state->db_name, ret));
724 tevent_req_error(req, ret);
725 return;
726 }
727
728 db = client_db_handle(state->client, state->db_name);
729 if (db != NULL) {
730 DLIST_REMOVE(state->client->db, db);
731 TALLOC_FREE(db);
732 }
733
734 tevent_req_done(req);
735 }
736
737 bool ctdb_detach_recv(struct tevent_req *req, int *perr)
738 {
739 int ret;
740
741 if (tevent_req_is_unix_error(req, &ret)) {
742 if (perr != NULL) {
743 *perr = ret;
744 }
745 return false;
746 }
747
748 return true;
749 }
750
751 int ctdb_detach(struct tevent_context *ev,
752 struct ctdb_client_context *client,
753 struct timeval timeout, uint32_t db_id)
754 {
755 TALLOC_CTX *mem_ctx;
756 struct tevent_req *req;
757 int ret;
758 bool status;
759
760 mem_ctx = talloc_new(client);
761 if (mem_ctx == NULL) {
762 return ENOMEM;
763 }
764
765 req = ctdb_detach_send(mem_ctx, ev, client, timeout, db_id);
766 if (req == NULL) {
767 talloc_free(mem_ctx);
768 return ENOMEM;
769 }
770
771 tevent_req_poll(req, ev);
772
773 status = ctdb_detach_recv(req, &ret);
774 if (! status) {
775 talloc_free(mem_ctx);
776 return ret;
777 }
778
779 talloc_free(mem_ctx);
780 return 0;
781 }
782
783 uint32_t ctdb_db_id(struct ctdb_db_context *db)
784 {
785 return db->db_id;
786 }
787
788 struct ctdb_db_traverse_local_state {
789 ctdb_rec_parser_func_t parser;
790 void *private_data;
791 bool extract_header;
792 int error;
793 };
794
795 static int ctdb_db_traverse_local_handler(struct tdb_context *tdb,
796 TDB_DATA key, TDB_DATA data,
797 void *private_data)
798 {
799 struct ctdb_db_traverse_local_state *state =
800 (struct ctdb_db_traverse_local_state *)private_data;
801 int ret;
802
803 if (state->extract_header) {
804 struct ctdb_ltdb_header header;
805
806 ret = ctdb_ltdb_header_extract(&data, &header);
807 if (ret != 0) {
808 state->error = ret;
809 return 1;
810 }
811
812 ret = state->parser(0, &header, key, data, state->private_data);
813 } else {
814 ret = state->parser(0, NULL, key, data, state->private_data);
815 }
816
817 if (ret != 0) {
818 state->error = ret;
819 return 1;
820 }
821
822 return 0;
823 }
824
825 int ctdb_db_traverse_local(struct ctdb_db_context *db, bool readonly,
826 bool extract_header,
827 ctdb_rec_parser_func_t parser, void *private_data)
828 {
829 struct ctdb_db_traverse_local_state state;
830 int ret;
831
832 state.parser = parser;
833 state.private_data = private_data;
834 state.extract_header = extract_header;
835 state.error = 0;
836
837 if (readonly) {
838 ret = tdb_traverse_read(db->ltdb->tdb,
839 ctdb_db_traverse_local_handler,
840 &state);
841 } else {
842 ret = tdb_traverse(db->ltdb->tdb,
843 ctdb_db_traverse_local_handler, &state);
844 }
845
846 if (ret == -1) {
847 return EIO;
848 }
849
850 return state.error;
851 }
852
853 struct ctdb_db_traverse_state {
854 struct tevent_context *ev;
855 struct ctdb_client_context *client;
856 struct ctdb_db_context *db;
857 uint32_t destnode;
858 uint64_t srvid;
859 struct timeval timeout;
860 ctdb_rec_parser_func_t parser;
861 void *private_data;
862 int result;
863 };
864
865 static void ctdb_db_traverse_handler_set(struct tevent_req *subreq);
866 static void ctdb_db_traverse_started(struct tevent_req *subreq);
867 static void ctdb_db_traverse_handler(uint64_t srvid, TDB_DATA data,
868 void *private_data);
869 static void ctdb_db_traverse_remove_handler(struct tevent_req *req);
870 static void ctdb_db_traverse_handler_removed(struct tevent_req *subreq);
871
872 struct tevent_req *ctdb_db_traverse_send(TALLOC_CTX *mem_ctx,
873 struct tevent_context *ev,
874 struct ctdb_client_context *client,
875 struct ctdb_db_context *db,
876 uint32_t destnode,
877 struct timeval timeout,
878 ctdb_rec_parser_func_t parser,
879 void *private_data)
880 {
881 struct tevent_req *req, *subreq;
882 struct ctdb_db_traverse_state *state;
883
884 req = tevent_req_create(mem_ctx, &state,
885 struct ctdb_db_traverse_state);
886 if (req == NULL) {
887 return NULL;
888 }
889
890 state->ev = ev;
891 state->client = client;
892 state->db = db;
893 state->destnode = destnode;
894 state->srvid = CTDB_SRVID_CLIENT_RANGE | getpid();
895 state->timeout = timeout;
896 state->parser = parser;
897 state->private_data = private_data;
898
899 subreq = ctdb_client_set_message_handler_send(state, ev, client,
900 state->srvid,
901 ctdb_db_traverse_handler,
902 req);
903 if (tevent_req_nomem(subreq, req)) {
904 return tevent_req_post(req, ev);
905 }
906 tevent_req_set_callback(subreq, ctdb_db_traverse_handler_set, req);
907
908 return req;
909 }
910
911 static void ctdb_db_traverse_handler_set(struct tevent_req *subreq)
912 {
913 struct tevent_req *req = tevent_req_callback_data(
914 subreq, struct tevent_req);
915 struct ctdb_db_traverse_state *state = tevent_req_data(
916 req, struct ctdb_db_traverse_state);
917 struct ctdb_traverse_start_ext traverse;
918 struct ctdb_req_control request;
919 int ret = 0;
920 bool status;
921
922 status = ctdb_client_set_message_handler_recv(subreq, &ret);
923 TALLOC_FREE(subreq);
924 if (! status) {
925 tevent_req_error(req, ret);
926 return;
927 }
928
929 traverse = (struct ctdb_traverse_start_ext) {
930 .db_id = ctdb_db_id(state->db),
931 .reqid = 0,
932 .srvid = state->srvid,
933 .withemptyrecords = false,
934 };
935
936 ctdb_req_control_traverse_start_ext(&request, &traverse);
937 subreq = ctdb_client_control_send(state, state->ev, state->client,
938 state->destnode, state->timeout,
939 &request);
940 if (subreq == NULL) {
941 state->result = ENOMEM;
942 ctdb_db_traverse_remove_handler(req);
943 return;
944 }
945 tevent_req_set_callback(subreq, ctdb_db_traverse_started, req);
946 }
947
948 static void ctdb_db_traverse_started(struct tevent_req *subreq)
949 {
950 struct tevent_req *req = tevent_req_callback_data(
951 subreq, struct tevent_req);
952 struct ctdb_db_traverse_state *state = tevent_req_data(
953 req, struct ctdb_db_traverse_state);
954 struct ctdb_reply_control *reply;
955 int ret = 0;
956 bool status;
957
958 status = ctdb_client_control_recv(subreq, &ret, state, &reply);
959 TALLOC_FREE(subreq);
960 if (! status) {
961 DEBUG(DEBUG_ERR, ("traverse: control failed, ret=%d\n", ret));
962 state->result = ret;
963 ctdb_db_traverse_remove_handler(req);
964 return;
965 }
966
967 ret = ctdb_reply_control_traverse_start_ext(reply);
968 talloc_free(reply);
969 if (ret != 0) {
970 DEBUG(DEBUG_ERR, ("traverse: control reply failed, ret=%d\n",
971 ret));
972 state->result = ret;
973 ctdb_db_traverse_remove_handler(req);
974 return;
975 }
976 }
977
978 static void ctdb_db_traverse_handler(uint64_t srvid, TDB_DATA data,
979 void *private_data)
980 {
981 struct tevent_req *req = talloc_get_type_abort(
982 private_data, struct tevent_req);
983 struct ctdb_db_traverse_state *state = tevent_req_data(
984 req, struct ctdb_db_traverse_state);
985 struct ctdb_rec_data *rec;
986 struct ctdb_ltdb_header header;
987 size_t np;
988 int ret;
989
990 ret = ctdb_rec_data_pull(data.dptr, data.dsize, state, &rec, &np);
991 if (ret != 0) {
992 return;
993 }
994
995 if (rec->key.dsize == 0 && rec->data.dsize == 0) {
996 talloc_free(rec);
997 ctdb_db_traverse_remove_handler(req);
998 return;
999 }
1000
1001 ret = ctdb_ltdb_header_extract(&rec->data, &header);
1002 if (ret != 0) {
1003 talloc_free(rec);
1004 return;
1005 }
1006
1007 if (rec->data.dsize == 0) {
1008 talloc_free(rec);
1009 return;
1010 }
1011
1012 ret = state->parser(rec->reqid, &header, rec->key, rec->data,
1013 state->private_data);
1014 talloc_free(rec);
1015 if (ret != 0) {
1016 state->result = ret;
1017 ctdb_db_traverse_remove_handler(req);
1018 }
1019 }
1020
1021 static void ctdb_db_traverse_remove_handler(struct tevent_req *req)
1022 {
1023 struct ctdb_db_traverse_state *state = tevent_req_data(
1024 req, struct ctdb_db_traverse_state);
1025 struct tevent_req *subreq;
1026
1027 subreq = ctdb_client_remove_message_handler_send(state, state->ev,
1028 state->client,
1029 state->srvid, req);
1030 if (tevent_req_nomem(subreq, req)) {
1031 return;
1032 }
1033 tevent_req_set_callback(subreq, ctdb_db_traverse_handler_removed, req);
1034 }
1035
1036 static void ctdb_db_traverse_handler_removed(struct tevent_req *subreq)
1037 {
1038 struct tevent_req *req = tevent_req_callback_data(
1039 subreq, struct tevent_req);
1040 struct ctdb_db_traverse_state *state = tevent_req_data(
1041 req, struct ctdb_db_traverse_state);
1042 int ret;
1043 bool status;
1044
1045 status = ctdb_client_remove_message_handler_recv(subreq, &ret);
1046 TALLOC_FREE(subreq);
1047 if (! status) {
1048 tevent_req_error(req, ret);
1049 return;
1050 }
1051
1052 if (state->result != 0) {
1053 tevent_req_error(req, state->result);
1054 return;
1055 }
1056
1057 tevent_req_done(req);
1058 }
1059
1060 bool ctdb_db_traverse_recv(struct tevent_req *req, int *perr)
1061 {
1062 int ret;
1063
1064 if (tevent_req_is_unix_error(req, &ret)) {
1065 if (perr != NULL) {
1066 *perr = ret;
1067 }
1068 return false;
1069 }
1070
1071 return true;
1072 }
1073
1074 int ctdb_db_traverse(TALLOC_CTX *mem_ctx, struct tevent_context *ev,
1075 struct ctdb_client_context *client,
1076 struct ctdb_db_context *db,
1077 uint32_t destnode, struct timeval timeout,
1078 ctdb_rec_parser_func_t parser, void *private_data)
1079 {
1080 struct tevent_req *req;
1081 int ret = 0;
1082 bool status;
1083
1084 req = ctdb_db_traverse_send(mem_ctx, ev, client, db, destnode,
1085 timeout, parser, private_data);
1086 if (req == NULL) {
1087 return ENOMEM;
1088 }
1089
1090 tevent_req_poll(req, ev);
1091
1092 status = ctdb_db_traverse_recv(req, &ret);
1093 if (! status) {
1094 return ret;
1095 }
1096
1097 return 0;
1098 }
1099
1100 int ctdb_ltdb_fetch(struct ctdb_db_context *db, TDB_DATA key,
1101 struct ctdb_ltdb_header *header,
1102 TALLOC_CTX *mem_ctx, TDB_DATA *data)
1103 {
1104 TDB_DATA rec;
1105 size_t np;
1106 int ret;
1107
1108 rec = tdb_fetch(db->ltdb->tdb, key);
1109 if (rec.dsize < sizeof(struct ctdb_ltdb_header)) {
1110 /* No record present */
1111 if (rec.dptr != NULL) {
1112 free(rec.dptr);
1113 }
1114
1115 if (tdb_error(db->ltdb->tdb) != TDB_ERR_NOEXIST) {
1116 return EIO;
1117 }
1118
1119 *header = (struct ctdb_ltdb_header) {
1120 .dmaster = CTDB_UNKNOWN_PNN,
1121 };
1122
1123 if (data != NULL) {
1124 *data = tdb_null;
1125 }
1126 return 0;
1127 }
1128
1129 ret = ctdb_ltdb_header_pull(rec.dptr, rec.dsize, header, &np);
1130 if (ret != 0) {
1131 return ret;
1132 }
1133
1134 ret = 0;
1135 if (data != NULL) {
1136 data->dsize = rec.dsize - np;
1137 data->dptr = talloc_memdup(mem_ctx, rec.dptr + np,
1138 data->dsize);
1139 if (data->dptr == NULL) {
1140 ret = ENOMEM;
1141 }
1142 }
1143
1144 free(rec.dptr);
1145 return ret;
1146 }
1147
1148 /*
1149 * Fetch a record from volatile database
1150 *
1151 * Steps:
1152 * 1. Get a lock on the hash chain
1153 * 2. If the record does not exist, migrate the record
1154 * 3. If readonly=true and delegations do not exist, migrate the record.
1155 * 4. If readonly=false and delegations exist, migrate the record.
1156 * 5. If the local node is not dmaster, migrate the record.
1157 * 6. Return record
1158 */
1159
1160 struct ctdb_fetch_lock_state {
1161 struct tevent_context *ev;
1162 struct ctdb_client_context *client;
1163 struct ctdb_record_handle *h;
1164 bool readonly;
1165 uint32_t pnn;
1166 };
1167
1168 static int ctdb_fetch_lock_check(struct tevent_req *req);
1169 static void ctdb_fetch_lock_migrate(struct tevent_req *req);
1170 static void ctdb_fetch_lock_migrate_done(struct tevent_req *subreq);
1171
1172 struct tevent_req *ctdb_fetch_lock_send(TALLOC_CTX *mem_ctx,
1173 struct tevent_context *ev,
1174 struct ctdb_client_context *client,
1175 struct ctdb_db_context *db,
1176 TDB_DATA key, bool readonly)
1177 {
1178 struct ctdb_fetch_lock_state *state;
1179 struct tevent_req *req;
1180 int ret;
1181
1182 req = tevent_req_create(mem_ctx, &state, struct ctdb_fetch_lock_state);
1183 if (req == NULL) {
1184 return NULL;
1185 }
1186
1187 state->ev = ev;
1188 state->client = client;
1189
1190 state->h = talloc_zero(db, struct ctdb_record_handle);
1191 if (tevent_req_nomem(state->h, req)) {
1192 return tevent_req_post(req, ev);
1193 }
1194 state->h->client = client;
1195 state->h->db = db;
1196 state->h->key.dptr = talloc_memdup(state->h, key.dptr, key.dsize);
1197 if (tevent_req_nomem(state->h->key.dptr, req)) {
1198 return tevent_req_post(req, ev);
1199 }
1200 state->h->key.dsize = key.dsize;
1201 state->h->readonly = false;
1202
1203 state->readonly = readonly;
1204 state->pnn = ctdb_client_pnn(client);
1205
1206 /* Check that database is not persistent */
1207 if (! ctdb_db_volatile(db)) {
1208 DEBUG(DEBUG_ERR, ("fetch_lock: %s database not volatile\n",
1209 db->db_name));
1210 tevent_req_error(req, EINVAL);
1211 return tevent_req_post(req, ev);
1212 }
1213
1214 ret = ctdb_fetch_lock_check(req);
1215 if (ret == 0) {
1216 tevent_req_done(req);
1217 return tevent_req_post(req, ev);
1218 }
1219 if (ret != EAGAIN) {
1220 tevent_req_error(req, ret);
1221 return tevent_req_post(req, ev);
1222 }
1223 return req;
1224 }
1225
1226 static int ctdb_fetch_lock_check(struct tevent_req *req)
1227 {
1228 struct ctdb_fetch_lock_state *state = tevent_req_data(
1229 req, struct ctdb_fetch_lock_state);
1230 struct ctdb_record_handle *h = state->h;
1231 struct ctdb_ltdb_header header;
1232 TDB_DATA data = tdb_null;
1233 size_t np;
1234 int ret, err = 0;
1235 bool do_migrate = false;
1236
1237 ret = tdb_chainlock(h->db->ltdb->tdb, h->key);
1238 if (ret != 0) {
1239 DEBUG(DEBUG_ERR,
1240 ("fetch_lock: %s tdb_chainlock failed, %s\n",
1241 h->db->db_name, tdb_errorstr(h->db->ltdb->tdb)));
1242 err = EIO;
1243 goto failed;
1244 }
1245
1246 data = tdb_fetch(h->db->ltdb->tdb, h->key);
1247 if (data.dptr == NULL) {
1248 if (tdb_error(h->db->ltdb->tdb) == TDB_ERR_NOEXIST) {
1249 goto migrate;
1250 } else {
1251 err = EIO;
1252 goto failed;
1253 }
1254 }
1255
1256 /* Got the record */
1257 ret = ctdb_ltdb_header_pull(data.dptr, data.dsize, &header, &np);
1258 if (ret != 0) {
1259 err = ret;
1260 goto failed;
1261 }
1262
1263 if (! state->readonly) {
1264 /* Read/write access */
1265 if (header.dmaster == state->pnn &&
1266 header.flags & CTDB_REC_RO_HAVE_DELEGATIONS) {
1267 goto migrate;
1268 }
1269
1270 if (header.dmaster != state->pnn) {
1271 goto migrate;
1272 }
1273 } else {
1274 /* Readonly access */
1275 if (header.dmaster != state->pnn &&
1276 ! (header.flags & (CTDB_REC_RO_HAVE_READONLY |
1277 CTDB_REC_RO_HAVE_DELEGATIONS))) {
1278 goto migrate;
1279 }
1280 }
1281
1282 /* We are the dmaster or readonly delegation */
1283 h->header = header;
1284 h->data = data;
1285 if (header.flags & (CTDB_REC_RO_HAVE_READONLY |
1286 CTDB_REC_RO_HAVE_DELEGATIONS)) {
1287 h->readonly = true;
1288 }
1289 return 0;
1290
1291 migrate:
1292 do_migrate = true;
1293 err = EAGAIN;
1294
1295 failed:
1296 if (data.dptr != NULL) {
1297 free(data.dptr);
1298 }
1299 ret = tdb_chainunlock(h->db->ltdb->tdb, h->key);
1300 if (ret != 0) {
1301 DEBUG(DEBUG_ERR,
1302 ("fetch_lock: %s tdb_chainunlock failed, %s\n",
1303 h->db->db_name, tdb_errorstr(h->db->ltdb->tdb)));
1304 return EIO;
1305 }
1306
1307 if (do_migrate) {
1308 ctdb_fetch_lock_migrate(req);
1309 }
1310 return err;
1311 }
1312
1313 static void ctdb_fetch_lock_migrate(struct tevent_req *req)
1314 {
1315 struct ctdb_fetch_lock_state *state = tevent_req_data(
1316 req, struct ctdb_fetch_lock_state);
1317 struct ctdb_req_call request;
1318 struct tevent_req *subreq;
1319
1320 ZERO_STRUCT(request);
1321 request.flags = CTDB_IMMEDIATE_MIGRATION;
1322 if (state->readonly) {
1323 request.flags |= CTDB_WANT_READONLY;
1324 }
1325 request.db_id = state->h->db->db_id;
1326 request.callid = CTDB_NULL_FUNC;
1327 request.key = state->h->key;
1328 request.calldata = tdb_null;
1329
1330 subreq = ctdb_client_call_send(state, state->ev, state->client,
1331 &request);
1332 if (tevent_req_nomem(subreq, req)) {
1333 return;
1334 }
1335
1336 tevent_req_set_callback(subreq, ctdb_fetch_lock_migrate_done, req);
1337 }
1338
1339 static void ctdb_fetch_lock_migrate_done(struct tevent_req *subreq)
1340 {
1341 struct tevent_req *req = tevent_req_callback_data(
1342 subreq, struct tevent_req);
1343 struct ctdb_fetch_lock_state *state = tevent_req_data(
1344 req, struct ctdb_fetch_lock_state);
1345 struct ctdb_reply_call *reply;
1346 int ret;
1347 bool status;
1348
1349 status = ctdb_client_call_recv(subreq, state, &reply, &ret);
1350 TALLOC_FREE(subreq);
1351 if (! status) {
1352 DEBUG(DEBUG_ERR, ("fetch_lock: %s CALL failed, ret=%d\n",
1353 state->h->db->db_name, ret));
1354 tevent_req_error(req, ret);
1355 return;
1356 }
1357
1358 if (reply->status != 0) {
1359 tevent_req_error(req, EIO);
1360 return;
1361 }
1362 talloc_free(reply);
1363
1364 ret = ctdb_fetch_lock_check(req);
1365 if (ret != 0) {
1366 if (ret != EAGAIN) {
1367 tevent_req_error(req, ret);
1368 }
1369 return;
1370 }
1371
1372 tevent_req_done(req);
1373 }
1374
1375 static int ctdb_record_handle_destructor(struct ctdb_record_handle *h)
1376 {
1377 int ret;
1378
1379 ret = tdb_chainunlock(h->db->ltdb->tdb, h->key);
1380 if (ret != 0) {
1381 DEBUG(DEBUG_ERR,
1382 ("fetch_lock: %s tdb_chainunlock failed, %s\n",
1383 h->db->db_name, tdb_errorstr(h->db->ltdb->tdb)));
1384 }
1385 free(h->data.dptr);
1386 return 0;
1387 }
1388
1389 struct ctdb_record_handle *ctdb_fetch_lock_recv(struct tevent_req *req,
1390 struct ctdb_ltdb_header *header,
1391 TALLOC_CTX *mem_ctx,
1392 TDB_DATA *data, int *perr)
1393 {
1394 struct ctdb_fetch_lock_state *state = tevent_req_data(
1395 req, struct ctdb_fetch_lock_state);
1396 struct ctdb_record_handle *h = state->h;
1397 int err;
1398
1399 if (tevent_req_is_unix_error(req, &err)) {
1400 if (perr != NULL) {
1401 TALLOC_FREE(state->h);
1402 *perr = err;
1403 }
1404 return NULL;
1405 }
1406
1407 if (header != NULL) {
1408 *header = h->header;
1409 }
1410 if (data != NULL) {
1411 size_t offset;
1412
1413 offset = ctdb_ltdb_header_len(&h->header);
1414
1415 data->dsize = h->data.dsize - offset;
1416 data->dptr = talloc_memdup(mem_ctx, h->data.dptr + offset,
1417 data->dsize);
1418 if (data->dptr == NULL) {
1419 TALLOC_FREE(state->h);
1420 if (perr != NULL) {
1421 *perr = ENOMEM;
1422 }
1423 return NULL;
1424 }
1425 }
1426
1427 talloc_set_destructor(h, ctdb_record_handle_destructor);
1428 return h;
1429 }
1430
1431 int ctdb_fetch_lock(TALLOC_CTX *mem_ctx, struct tevent_context *ev,
1432 struct ctdb_client_context *client,
1433 struct ctdb_db_context *db, TDB_DATA key, bool readonly,
1434 struct ctdb_record_handle **out,
1435 struct ctdb_ltdb_header *header, TDB_DATA *data)
1436 {
1437 struct tevent_req *req;
1438 struct ctdb_record_handle *h;
1439 int ret;
1440
1441 req = ctdb_fetch_lock_send(mem_ctx, ev, client, db, key, readonly);
1442 if (req == NULL) {
1443 return ENOMEM;
1444 }
1445
1446 tevent_req_poll(req, ev);
1447
1448 h = ctdb_fetch_lock_recv(req, header, mem_ctx, data, &ret);
1449 if (h == NULL) {
1450 return ret;
1451 }
1452
1453 *out = h;
1454 return 0;
1455 }
1456
1457 int ctdb_store_record(struct ctdb_record_handle *h, TDB_DATA data)
1458 {
1459 uint8_t header[sizeof(struct ctdb_ltdb_header)];
1460 TDB_DATA rec[2];
1461 size_t np;
1462 int ret;
1463
1464 /* Cannot modify the record if it was obtained as a readonly copy */
1465 if (h->readonly) {
1466 return EINVAL;
1467 }
1468
1469 /* Check if the new data is same */
1470 if (h->data.dsize == data.dsize &&
1471 memcmp(h->data.dptr, data.dptr, data.dsize) == 0) {
1472 /* No need to do anything */
1473 return 0;
1474 }
1475
1476 ctdb_ltdb_header_push(&h->header, header, &np);
1477
1478 rec[0].dsize = np;
1479 rec[0].dptr = header;
1480
1481 rec[1].dsize = data.dsize;
1482 rec[1].dptr = data.dptr;
1483
1484 ret = tdb_storev(h->db->ltdb->tdb, h->key, rec, 2, TDB_REPLACE);
1485 if (ret != 0) {
1486 DEBUG(DEBUG_ERR,
1487 ("store_record: %s tdb_storev failed, %s\n",
1488 h->db->db_name, tdb_errorstr(h->db->ltdb->tdb)));
1489 return EIO;
1490 }
1491
1492 return 0;
1493 }
1494
1495 struct ctdb_delete_record_state {
1496 struct ctdb_record_handle *h;
1497 };
1498
1499 static void ctdb_delete_record_done(struct tevent_req *subreq);
1500
1501 struct tevent_req *ctdb_delete_record_send(TALLOC_CTX *mem_ctx,
1502 struct tevent_context *ev,
1503 struct ctdb_record_handle *h)
1504 {
1505 struct tevent_req *req, *subreq;
1506 struct ctdb_delete_record_state *state;
1507 struct ctdb_key_data key;
1508 struct ctdb_req_control request;
1509 uint8_t header[sizeof(struct ctdb_ltdb_header)];
1510 TDB_DATA rec;
1511 size_t np;
1512 int ret;
1513
1514 req = tevent_req_create(mem_ctx, &state,
1515 struct ctdb_delete_record_state);
1516 if (req == NULL) {
1517 return NULL;
1518 }
1519
1520 state->h = h;
1521
1522 /* Cannot delete the record if it was obtained as a readonly copy */
1523 if (h->readonly) {
1524 DEBUG(DEBUG_ERR, ("fetch_lock delete: %s readonly record\n",
1525 h->db->db_name));
1526 tevent_req_error(req, EINVAL);
1527 return tevent_req_post(req, ev);
1528 }
1529
1530 ctdb_ltdb_header_push(&h->header, header, &np);
1531
1532 rec.dsize = np;
1533 rec.dptr = header;
1534
1535 ret = tdb_store(h->db->ltdb->tdb, h->key, rec, TDB_REPLACE);
1536 if (ret != 0) {
1537 DEBUG(DEBUG_ERR,
1538 ("fetch_lock delete: %s tdb_sore failed, %s\n",
1539 h->db->db_name, tdb_errorstr(h->db->ltdb->tdb)));
1540 tevent_req_error(req, EIO);
1541 return tevent_req_post(req, ev);
1542 }
1543
1544 key.db_id = h->db->db_id;
1545 key.header = h->header;
1546 key.key = h->key;
1547
1548 ctdb_req_control_schedule_for_deletion(&request, &key);
1549 subreq = ctdb_client_control_send(state, ev, h->client,
1550 ctdb_client_pnn(h->client),
1551 tevent_timeval_zero(),
1552 &request);
1553 if (tevent_req_nomem(subreq, req)) {
1554 return tevent_req_post(req, ev);
1555 }
1556 tevent_req_set_callback(subreq, ctdb_delete_record_done, req);
1557
1558 return req;
1559 }
1560
1561 static void ctdb_delete_record_done(struct tevent_req *subreq)
1562 {
1563 struct tevent_req *req = tevent_req_callback_data(
1564 subreq, struct tevent_req);
1565 struct ctdb_delete_record_state *state = tevent_req_data(
1566 req, struct ctdb_delete_record_state);
1567 int ret;
1568 bool status;
1569
1570 status = ctdb_client_control_recv(subreq, &ret, NULL, NULL);
1571 TALLOC_FREE(subreq);
1572 if (! status) {
1573 DEBUG(DEBUG_ERR,
1574 ("delete_record: %s SCHDULE_FOR_DELETION failed, "
1575 "ret=%d\n", state->h->db->db_name, ret));
1576 tevent_req_error(req, ret);
1577 return;
1578 }
1579
1580 tevent_req_done(req);
1581 }
1582
1583 bool ctdb_delete_record_recv(struct tevent_req *req, int *perr)
1584 {
1585 int err;
1586
1587 if (tevent_req_is_unix_error(req, &err)) {
1588 if (perr != NULL) {
1589 *perr = err;
1590 }
1591 return false;
1592 }
1593
1594 return true;
1595 }
1596
1597
1598 int ctdb_delete_record(struct ctdb_record_handle *h)
1599 {
1600 struct tevent_context *ev = h->ev;
1601 TALLOC_CTX *mem_ctx;
1602 struct tevent_req *req;
1603 int ret;
1604 bool status;
1605
1606 mem_ctx = talloc_new(NULL);
1607 if (mem_ctx == NULL) {
1608 return ENOMEM;
1609 }
1610
1611 req = ctdb_delete_record_send(mem_ctx, ev, h);
1612 if (req == NULL) {
1613 talloc_free(mem_ctx);
1614 return ENOMEM;
1615 }
1616
1617 tevent_req_poll(req, ev);
1618
1619 status = ctdb_delete_record_recv(req, &ret);
1620 talloc_free(mem_ctx);
1621 if (! status) {
1622 return ret;
1623 }
1624
1625 return 0;
1626 }
1627
1628 /*
1629 * Global lock functions
1630 */
1631
1632 struct ctdb_g_lock_lock_state {
1633 struct tevent_context *ev;
1634 struct ctdb_client_context *client;
1635 struct ctdb_db_context *db;
1636 TDB_DATA key;
1637 struct ctdb_server_id my_sid;
1638 enum ctdb_g_lock_type lock_type;
1639 struct ctdb_record_handle *h;
1640 /* state for verification of active locks */
1641 struct ctdb_g_lock_list *lock_list;
1642 unsigned int current;
1643 };
1644
1645 static void ctdb_g_lock_lock_fetched(struct tevent_req *subreq);
1646 static void ctdb_g_lock_lock_process_locks(struct tevent_req *req);
1647 static void ctdb_g_lock_lock_checked(struct tevent_req *subreq);
1648 static int ctdb_g_lock_lock_update(struct tevent_req *req);
1649 static void ctdb_g_lock_lock_retry(struct tevent_req *subreq);
1650
1651 static bool ctdb_g_lock_conflicts(enum ctdb_g_lock_type l1,
1652 enum ctdb_g_lock_type l2)
1653 {
1654 if ((l1 == CTDB_G_LOCK_READ) && (l2 == CTDB_G_LOCK_READ)) {
1655 return false;
1656 }
1657 return true;
1658 }
1659
1660 struct tevent_req *ctdb_g_lock_lock_send(TALLOC_CTX *mem_ctx,
1661 struct tevent_context *ev,
1662 struct ctdb_client_context *client,
1663 struct ctdb_db_context *db,
1664 const char *keyname,
1665 struct ctdb_server_id *sid,
1666 bool readonly)
1667 {
1668 struct tevent_req *req, *subreq;
1669 struct ctdb_g_lock_lock_state *state;
1670
1671 req = tevent_req_create(mem_ctx, &state,
1672 struct ctdb_g_lock_lock_state);
1673 if (req == NULL) {
1674 return NULL;
1675 }
1676
1677 state->ev = ev;
1678 state->client = client;
1679 state->db = db;
1680 state->key.dptr = discard_const(keyname);
1681 state->key.dsize = strlen(keyname) + 1;
1682 state->my_sid = *sid;
1683 state->lock_type = (readonly ? CTDB_G_LOCK_READ : CTDB_G_LOCK_WRITE);
1684
1685 subreq = ctdb_fetch_lock_send(state, ev, client, db, state->key,
1686 false);
1687 if (tevent_req_nomem(subreq, req)) {
1688 return tevent_req_post(req, ev);
1689 }
1690 tevent_req_set_callback(subreq, ctdb_g_lock_lock_fetched, req);
1691
1692 return req;
1693 }
1694
1695 static void ctdb_g_lock_lock_fetched(struct tevent_req *subreq)
1696 {
1697 struct tevent_req *req = tevent_req_callback_data(
1698 subreq, struct tevent_req);
1699 struct ctdb_g_lock_lock_state *state = tevent_req_data(
1700 req, struct ctdb_g_lock_lock_state);
1701 TDB_DATA data;
1702 size_t np;
1703 int ret = 0;
1704
1705 state->h = ctdb_fetch_lock_recv(subreq, NULL, state, &data, &ret);
1706 TALLOC_FREE(subreq);
1707 if (state->h == NULL) {
1708 DEBUG(DEBUG_ERR, ("g_lock_lock: %s fetch lock failed\n",
1709 (char *)state->key.dptr));
1710 tevent_req_error(req, ret);
1711 return;
1712 }
1713
1714 if (state->lock_list != NULL) {
1715 TALLOC_FREE(state->lock_list);
1716 state->current = 0;
1717 }
1718
1719 ret = ctdb_g_lock_list_pull(data.dptr, data.dsize, state,
1720 &state->lock_list, &np);
1721 talloc_free(data.dptr);
1722 if (ret != 0) {
1723 DEBUG(DEBUG_ERR, ("g_lock_lock: %s invalid lock data\n",
1724 (char *)state->key.dptr));
1725 tevent_req_error(req, ret);
1726 return;
1727 }
1728
1729 ctdb_g_lock_lock_process_locks(req);
1730 }
1731
1732 static void ctdb_g_lock_lock_process_locks(struct tevent_req *req)
1733 {
1734 struct ctdb_g_lock_lock_state *state = tevent_req_data(
1735 req, struct ctdb_g_lock_lock_state);
1736 struct tevent_req *subreq;
1737 struct ctdb_g_lock *lock;
1738 bool check_server = false;
1739 int ret;
1740
1741 while (state->current < state->lock_list->num) {
1742 lock = &state->lock_list->lock[state->current];
1743
1744 /* We should not ask for the same lock more than once */
1745 if (ctdb_server_id_equal(&lock->sid, &state->my_sid)) {
1746 DEBUG(DEBUG_ERR, ("g_lock_lock: %s deadlock\n",
1747 (char *)state->key.dptr));
1748 tevent_req_error(req, EDEADLK);
1749 return;
1750 }
1751
1752 if (ctdb_g_lock_conflicts(lock->type, state->lock_type)) {
1753 check_server = true;
1754 break;
1755 }
1756
1757 state->current += 1;
1758 }
1759
1760 if (check_server) {
1761 struct ctdb_req_control request;
1762
1763 ctdb_req_control_process_exists(&request, lock->sid.pid);
1764 subreq = ctdb_client_control_send(state, state->ev,
1765 state->client,
1766 lock->sid.vnn,
1767 tevent_timeval_zero(),
1768 &request);
1769 if (tevent_req_nomem(subreq, req)) {
1770 return;
1771 }
1772 tevent_req_set_callback(subreq, ctdb_g_lock_lock_checked, req);
1773 return;
1774 }
1775
1776 /* There is no conflict, add ourself to the lock_list */
1777 state->lock_list->lock = talloc_realloc(state->lock_list,
1778 state->lock_list->lock,
1779 struct ctdb_g_lock,
1780 state->lock_list->num + 1);
1781 if (state->lock_list->lock == NULL) {
1782 tevent_req_error(req, ENOMEM);
1783 return;
1784 }
1785
1786 lock = &state->lock_list->lock[state->lock_list->num];
1787 lock->type = state->lock_type;
1788 lock->sid = state->my_sid;
1789 state->lock_list->num += 1;
1790
1791 ret = ctdb_g_lock_lock_update(req);
1792 if (ret != 0) {
1793 tevent_req_error(req, ret);
1794 return;
1795 }
1796
1797 TALLOC_FREE(state->h);
1798 tevent_req_done(req);
1799 }
1800
1801 static void ctdb_g_lock_lock_checked(struct tevent_req *subreq)
1802 {
1803 struct tevent_req *req = tevent_req_callback_data(
1804 subreq, struct tevent_req);
1805 struct ctdb_g_lock_lock_state *state = tevent_req_data(
1806 req, struct ctdb_g_lock_lock_state);
1807 struct ctdb_reply_control *reply;
1808 int ret, value;
1809 bool status;
1810
1811 status = ctdb_client_control_recv(subreq, &ret, state, &reply);
1812 TALLOC_FREE(subreq);
1813 if (! status) {
1814 DEBUG(DEBUG_ERR,
1815 ("g_lock_lock: %s PROCESS_EXISTS failed, ret=%d\n",
1816 (char *)state->key.dptr, ret));
1817 tevent_req_error(req, ret);
1818 return;
1819 }
1820
1821 ret = ctdb_reply_control_process_exists(reply, &value);
1822 if (ret != 0) {
1823 tevent_req_error(req, ret);
1824 return;
1825 }
1826 talloc_free(reply);
1827
1828 if (value == 0) {
1829 /* server process exists, need to retry */
1830 TALLOC_FREE(state->h);
1831 subreq = tevent_wakeup_send(state, state->ev,
1832 tevent_timeval_current_ofs(0,1000));
1833 if (tevent_req_nomem(subreq, req)) {
1834 return;
1835 }
1836 tevent_req_set_callback(subreq, ctdb_g_lock_lock_retry, req);
1837 return;
1838 }
1839
1840 /* server process does not exist, remove conflicting entry */
1841 state->lock_list->lock[state->current] =
1842 state->lock_list->lock[state->lock_list->num-1];
1843 state->lock_list->num -= 1;
1844
1845 ret = ctdb_g_lock_lock_update(req);
1846 if (ret != 0) {
1847 tevent_req_error(req, ret);
1848 return;
1849 }
1850
1851 ctdb_g_lock_lock_process_locks(req);
1852 }
1853
1854 static int ctdb_g_lock_lock_update(struct tevent_req *req)
1855 {
1856 struct ctdb_g_lock_lock_state *state = tevent_req_data(
1857 req, struct ctdb_g_lock_lock_state);
1858 TDB_DATA data;
1859 size_t np;
1860 int ret;
1861
1862 data.dsize = ctdb_g_lock_list_len(state->lock_list);
1863 data.dptr = talloc_size(state, data.dsize);
1864 if (data.dptr == NULL) {
1865 return ENOMEM;
1866 }
1867
1868 ctdb_g_lock_list_push(state->lock_list, data.dptr, &np);
1869 ret = ctdb_store_record(state->h, data);
1870 talloc_free(data.dptr);
1871 return ret;
1872 }
1873
1874 static void ctdb_g_lock_lock_retry(struct tevent_req *subreq)
1875 {
1876 struct tevent_req *req = tevent_req_callback_data(
1877 subreq, struct tevent_req);
1878 struct ctdb_g_lock_lock_state *state = tevent_req_data(
1879 req, struct ctdb_g_lock_lock_state);
1880 bool success;
1881
1882 success = tevent_wakeup_recv(subreq);
1883 TALLOC_FREE(subreq);
1884 if (! success) {
1885 tevent_req_error(req, ENOMEM);
1886 return;
1887 }
1888
1889 subreq = ctdb_fetch_lock_send(state, state->ev, state->client,
1890 state->db, state->key, false);
1891 if (tevent_req_nomem(subreq, req)) {
1892 return;
1893 }
1894 tevent_req_set_callback(subreq, ctdb_g_lock_lock_fetched, req);
1895 }
1896
1897 bool ctdb_g_lock_lock_recv(struct tevent_req *req, int *perr)
1898 {
1899 struct ctdb_g_lock_lock_state *state = tevent_req_data(
1900 req, struct ctdb_g_lock_lock_state);
1901 int err;
1902
1903 TALLOC_FREE(state->h);
1904
1905 if (tevent_req_is_unix_error(req, &err)) {
1906 if (perr != NULL) {
1907 *perr = err;
1908 }
1909 return false;
1910 }
1911
1912 return true;
1913 }
1914
1915 struct ctdb_g_lock_unlock_state {
1916 struct tevent_context *ev;
1917 struct ctdb_client_context *client;
1918 struct ctdb_db_context *db;
1919 TDB_DATA key;
1920 struct ctdb_server_id my_sid;
1921 struct ctdb_record_handle *h;
1922 struct ctdb_g_lock_list *lock_list;
1923 };
1924
1925 static void ctdb_g_lock_unlock_fetched(struct tevent_req *subreq);
1926 static int ctdb_g_lock_unlock_update(struct tevent_req *req);
1927 static void ctdb_g_lock_unlock_deleted(struct tevent_req *subreq);
1928
1929 struct tevent_req *ctdb_g_lock_unlock_send(TALLOC_CTX *mem_ctx,
1930 struct tevent_context *ev,
1931 struct ctdb_client_context *client,
1932 struct ctdb_db_context *db,
1933 const char *keyname,
1934 struct ctdb_server_id sid)
1935 {
1936 struct tevent_req *req, *subreq;
1937 struct ctdb_g_lock_unlock_state *state;
1938
1939 req = tevent_req_create(mem_ctx, &state,
1940 struct ctdb_g_lock_unlock_state);
1941 if (req == NULL) {
1942 return NULL;
1943 }
1944
1945 state->ev = ev;
1946 state->client = client;
1947 state->db = db;
1948 state->key.dptr = discard_const(keyname);
1949 state->key.dsize = strlen(keyname) + 1;
1950 state->my_sid = sid;
1951
1952 subreq = ctdb_fetch_lock_send(state, ev, client, db, state->key,
1953 false);
1954 if (tevent_req_nomem(subreq, req)) {
1955 return tevent_req_post(req, ev);
1956 }
1957 tevent_req_set_callback(subreq, ctdb_g_lock_unlock_fetched, req);
1958
1959 return req;
1960 }
1961
1962 static void ctdb_g_lock_unlock_fetched(struct tevent_req *subreq)
1963 {
1964 struct tevent_req *req = tevent_req_callback_data(
1965 subreq, struct tevent_req);
1966 struct ctdb_g_lock_unlock_state *state = tevent_req_data(
1967 req, struct ctdb_g_lock_unlock_state);
1968 TDB_DATA data;
1969 size_t np;
1970 int ret = 0;
1971
1972 state->h = ctdb_fetch_lock_recv(subreq, NULL, state, &data, &ret);
1973 TALLOC_FREE(subreq);
1974 if (state->h == NULL) {
1975 DEBUG(DEBUG_ERR, ("g_lock_unlock: %s fetch lock failed\n",
1976 (char *)state->key.dptr));
1977 tevent_req_error(req, ret);
1978 return;
1979 }
1980
1981 ret = ctdb_g_lock_list_pull(data.dptr, data.dsize, state,
1982 &state->lock_list, &np);
1983 if (ret != 0) {
1984 DEBUG(DEBUG_ERR, ("g_lock_unlock: %s invalid lock data\n",
1985 (char *)state->key.dptr));
1986 tevent_req_error(req, ret);
1987 return;
1988 }
1989
1990 ret = ctdb_g_lock_unlock_update(req);
1991 if (ret != 0) {
1992 tevent_req_error(req, ret);
1993 return;
1994 }
1995
1996 if (state->lock_list->num == 0) {
1997 subreq = ctdb_delete_record_send(state, state->ev, state->h);
1998 if (tevent_req_nomem(subreq, req)) {
1999 return;
2000 }
2001 tevent_req_set_callback(subreq, ctdb_g_lock_unlock_deleted,
2002 req);
2003 return;
2004 }
2005
2006 TALLOC_FREE(state->h);
2007 tevent_req_done(req);
2008 }
2009
2010 static int ctdb_g_lock_unlock_update(struct tevent_req *req)
2011 {
2012 struct ctdb_g_lock_unlock_state *state = tevent_req_data(
2013 req, struct ctdb_g_lock_unlock_state);
2014 struct ctdb_g_lock *lock;
2015 int ret, i;
2016
2017 for (i=0; i<state->lock_list->num; i++) {
2018 lock = &state->lock_list->lock[i];
2019
2020 if (ctdb_server_id_equal(&lock->sid, &state->my_sid)) {
2021 break;
2022 }
2023 }
2024
2025 if (i < state->lock_list->num) {
2026 state->lock_list->lock[i] =
2027 state->lock_list->lock[state->lock_list->num-1];
2028 state->lock_list->num -= 1;
2029 }
2030
2031 if (state->lock_list->num != 0) {
2032 TDB_DATA data;
2033 size_t np;
2034
2035 data.dsize = ctdb_g_lock_list_len(state->lock_list);
2036 data.dptr = talloc_size(state, data.dsize);
2037 if (data.dptr == NULL) {
2038 return ENOMEM;
2039 }
2040
2041 ctdb_g_lock_list_push(state->lock_list, data.dptr, &np);
2042 ret = ctdb_store_record(state->h, data);
2043 talloc_free(data.dptr);
2044 if (ret != 0) {
2045 return ret;
2046 }
2047 }
2048
2049 return 0;
2050 }
2051
2052 static void ctdb_g_lock_unlock_deleted(struct tevent_req *subreq)
2053 {
2054 struct tevent_req *req = tevent_req_callback_data(
2055 subreq, struct tevent_req);
2056 struct ctdb_g_lock_unlock_state *state = tevent_req_data(
2057 req, struct ctdb_g_lock_unlock_state);
2058 int ret;
2059 bool status;
2060
2061 status = ctdb_delete_record_recv(subreq, &ret);
2062 if (! status) {
2063 DEBUG(DEBUG_ERR,
2064 ("g_lock_unlock %s delete record failed, ret=%d\n",
2065 (char *)state->key.dptr, ret));
2066 tevent_req_error(req, ret);
2067 return;
2068 }
2069
2070 TALLOC_FREE(state->h);
2071 tevent_req_done(req);
2072 }
2073
2074 bool ctdb_g_lock_unlock_recv(struct tevent_req *req, int *perr)
2075 {
2076 struct ctdb_g_lock_unlock_state *state = tevent_req_data(
2077 req, struct ctdb_g_lock_unlock_state);
2078 int err;
2079
2080 TALLOC_FREE(state->h);
2081
2082 if (tevent_req_is_unix_error(req, &err)) {
2083 if (perr != NULL) {
2084 *perr = err;
2085 }
2086 return false;
2087 }
2088
2089 return true;
2090 }
2091
2092 /*
2093 * Persistent database functions
2094 */
2095 struct ctdb_transaction_start_state {
2096 struct tevent_context *ev;
2097 struct ctdb_client_context *client;
2098 struct timeval timeout;
2099 struct ctdb_transaction_handle *h;
2100 uint32_t destnode;
2101 };
2102
2103 static void ctdb_transaction_g_lock_attached(struct tevent_req *subreq);
2104 static void ctdb_transaction_g_lock_done(struct tevent_req *subreq);
2105
2106 struct tevent_req *ctdb_transaction_start_send(TALLOC_CTX *mem_ctx,
2107 struct tevent_context *ev,
2108 struct ctdb_client_context *client,
2109 struct timeval timeout,
2110 struct ctdb_db_context *db,
2111 bool readonly)
2112 {
2113 struct ctdb_transaction_start_state *state;
2114 struct tevent_req *req, *subreq;
2115 struct ctdb_transaction_handle *h;
2116
2117 req = tevent_req_create(mem_ctx, &state,
2118 struct ctdb_transaction_start_state);
2119 if (req == NULL) {
2120 return NULL;
2121 }
2122
2123 if (ctdb_db_volatile(db)) {
2124 tevent_req_error(req, EINVAL);
2125 return tevent_req_post(req, ev);
2126 }
2127
2128 state->ev = ev;
2129 state->client = client;
2130 state->destnode = ctdb_client_pnn(client);
2131
2132 h = talloc_zero(db, struct ctdb_transaction_handle);
2133 if (tevent_req_nomem(h, req)) {
2134 return tevent_req_post(req, ev);
2135 }
2136
2137 h->ev = ev;
2138 h->client = client;
2139 h->db = db;
2140 h->readonly = readonly;
2141 h->updated = false;
2142
2143 /* SRVID is unique for databases, so client can have transactions
2144 * active for multiple databases */
2145 h->sid = ctdb_client_get_server_id(client, db->db_id);
2146
2147 h->recbuf = ctdb_rec_buffer_init(h, db->db_id);
2148 if (tevent_req_nomem(h->recbuf, req)) {
2149 return tevent_req_post(req, ev);
2150 }
2151
2152 h->lock_name = talloc_asprintf(h, "transaction_db_0x%08x", db->db_id);
2153 if (tevent_req_nomem(h->lock_name, req)) {
2154 return tevent_req_post(req, ev);
2155 }
2156
2157 state->h = h;
2158
2159 subreq = ctdb_attach_send(state, ev, client, timeout, "g_lock.tdb", 0);
2160 if (tevent_req_nomem(subreq, req)) {
2161 return tevent_req_post(req, ev);
2162 }
2163 tevent_req_set_callback(subreq, ctdb_transaction_g_lock_attached, req);
2164
2165 return req;
2166 }
2167
2168 static void ctdb_transaction_g_lock_attached(struct tevent_req *subreq)
2169 {
2170 struct tevent_req *req = tevent_req_callback_data(
2171 subreq, struct tevent_req);
2172 struct ctdb_transaction_start_state *state = tevent_req_data(
2173 req, struct ctdb_transaction_start_state);
2174 bool status;
2175 int ret;
2176
2177 status = ctdb_attach_recv(subreq, &ret, &state->h->db_g_lock);
2178 TALLOC_FREE(subreq);
2179 if (! status) {
2180 DEBUG(DEBUG_ERR,
2181 ("transaction_start: %s attach g_lock.tdb failed\n",
2182 state->h->db->db_name));
2183 tevent_req_error(req, ret);
2184 return;
2185 }
2186
2187 subreq = ctdb_g_lock_lock_send(state, state->ev, state->client,
2188 state->h->db_g_lock,
2189 state->h->lock_name,
2190 &state->h->sid, state->h->readonly);
2191 if (tevent_req_nomem(subreq, req)) {
2192 return;
2193 }
2194 tevent_req_set_callback(subreq, ctdb_transaction_g_lock_done, req);
2195 }
2196
2197 static void ctdb_transaction_g_lock_done(struct tevent_req *subreq)
2198 {
2199 struct tevent_req *req = tevent_req_callback_data(
2200 subreq, struct tevent_req);
2201 struct ctdb_transaction_start_state *state = tevent_req_data(
2202 req, struct ctdb_transaction_start_state);
2203 int ret;
2204 bool status;
2205
2206 status = ctdb_g_lock_lock_recv(subreq, &ret);
2207 TALLOC_FREE(subreq);
2208 if (! status) {
2209 DEBUG(DEBUG_ERR,
2210 ("transaction_start: %s g_lock lock failed, ret=%d\n",
2211 state->h->db->db_name, ret));
2212 tevent_req_error(req, ret);
2213 return;
2214 }
2215
2216 tevent_req_done(req);
2217 }
2218
2219 struct ctdb_transaction_handle *ctdb_transaction_start_recv(
2220 struct tevent_req *req,
2221 int *perr)
2222 {
2223 struct ctdb_transaction_start_state *state = tevent_req_data(
2224 req, struct ctdb_transaction_start_state);
2225 int err;
2226
2227 if (tevent_req_is_unix_error(req, &err)) {
2228 if (perr != NULL) {
2229 *perr = err;
2230 }
2231 return NULL;
2232 }
2233
2234 return state->h;
2235 }
2236
2237 int ctdb_transaction_start(TALLOC_CTX *mem_ctx, struct tevent_context *ev,
2238 struct ctdb_client_context *client,
2239 struct timeval timeout,
2240 struct ctdb_db_context *db, bool readonly,
2241 struct ctdb_transaction_handle **out)
2242 {
2243 struct tevent_req *req;
2244 struct ctdb_transaction_handle *h;
2245 int ret;
2246
2247 req = ctdb_transaction_start_send(mem_ctx, ev, client, timeout, db,
2248 readonly);
2249 if (req == NULL) {
2250 return ENOMEM;
2251 }
2252
2253 tevent_req_poll(req, ev);
2254
2255 h = ctdb_transaction_start_recv(req, &ret);
2256 if (h == NULL) {
2257 return ret;
2258 }
2259
2260 *out = h;
2261 return 0;
2262 }
2263
2264 struct ctdb_transaction_record_fetch_state {
2265 TDB_DATA key, data;
2266 struct ctdb_ltdb_header header;
2267 bool found;
2268 };
2269
2270 static int ctdb_transaction_record_fetch_traverse(
2271 uint32_t reqid,
2272 struct ctdb_ltdb_header *nullheader,
2273 TDB_DATA key, TDB_DATA data,
2274 void *private_data)
2275 {
2276 struct ctdb_transaction_record_fetch_state *state =
2277 (struct ctdb_transaction_record_fetch_state *)private_data;
2278
2279 if (state->key.dsize == key.dsize &&
2280 memcmp(state->key.dptr, key.dptr, key.dsize) == 0) {
2281 int ret;
2282
2283 ret = ctdb_ltdb_header_extract(&data, &state->header);
2284 if (ret != 0) {
2285 DEBUG(DEBUG_ERR,
2286 ("record_fetch: Failed to extract header, "
2287 "ret=%d\n", ret));
2288 return 1;
2289 }
2290
2291 state->data = data;
2292 state->found = true;
2293 }
2294
2295 return 0;
2296 }
2297
2298 static int ctdb_transaction_record_fetch(struct ctdb_transaction_handle *h,
2299 TDB_DATA key,
2300 struct ctdb_ltdb_header *header,
2301 TDB_DATA *data)
2302 {
2303 struct ctdb_transaction_record_fetch_state state;
2304 int ret;
2305
2306 state.key = key;
2307 state.found = false;
2308
2309 ret = ctdb_rec_buffer_traverse(h->recbuf,
2310 ctdb_transaction_record_fetch_traverse,
2311 &state);
2312 if (ret != 0) {
2313 return ret;
2314 }
2315
2316 if (state.found) {
2317 if (header != NULL) {
2318 *header = state.header;
2319 }
2320 if (data != NULL) {
2321 *data = state.data;
2322 }
2323 return 0;
2324 }
2325
2326 return ENOENT;
2327 }
2328
2329 int ctdb_transaction_fetch_record(struct ctdb_transaction_handle *h,
2330 TDB_DATA key,
2331 TALLOC_CTX *mem_ctx, TDB_DATA *data)
2332 {
2333 TDB_DATA tmp_data;
2334 struct ctdb_ltdb_header header;
2335 int ret;
2336
2337 ret = ctdb_transaction_record_fetch(h, key, NULL, &tmp_data);
2338 if (ret == 0) {
2339 data->dptr = talloc_memdup(mem_ctx, tmp_data.dptr,
2340 tmp_data.dsize);
2341 if (data->dptr == NULL) {
2342 return ENOMEM;
2343 }
2344 data->dsize = tmp_data.dsize;
2345 return 0;
2346 }
2347
2348 ret = ctdb_ltdb_fetch(h->db, key, &header, mem_ctx, data);
2349 if (ret != 0) {
2350 return ret;
2351 }
2352
2353 ret = ctdb_rec_buffer_add(h, h->recbuf, 0, &header, key, *data);
2354 if (ret != 0) {
2355 return ret;
2356 }
2357
2358 return 0;
2359 }
2360
2361 int ctdb_transaction_store_record(struct ctdb_transaction_handle *h,
2362 TDB_DATA key, TDB_DATA data)
2363 {
2364 TALLOC_CTX *tmp_ctx;
2365 struct ctdb_ltdb_header header;
2366 TDB_DATA old_data;
2367 int ret;
2368
2369 if (h->readonly) {
2370 return EINVAL;
2371 }
2372
2373 tmp_ctx = talloc_new(h);
2374 if (tmp_ctx == NULL) {
2375 return ENOMEM;
2376 }
2377
2378 ret = ctdb_transaction_record_fetch(h, key, &header, &old_data);
2379 if (ret != 0) {
2380 ret = ctdb_ltdb_fetch(h->db, key, &header, tmp_ctx, &old_data);
2381 if (ret != 0) {
2382 return ret;
2383 }
2384 }
2385
2386 if (old_data.dsize == data.dsize &&
2387 memcmp(old_data.dptr, data.dptr, data.dsize) == 0) {
2388 talloc_free(tmp_ctx);
2389 return 0;
2390 }
2391
2392 header.dmaster = ctdb_client_pnn(h->client);
2393 header.rsn += 1;
2394
2395 ret = ctdb_rec_buffer_add(h, h->recbuf, 0, &header, key, data);
2396 talloc_free(tmp_ctx);
2397 if (ret != 0) {
2398 return ret;
2399 }
2400 h->updated = true;
2401
2402 return 0;
2403 }
2404
2405 int ctdb_transaction_delete_record(struct ctdb_transaction_handle *h,
2406 TDB_DATA key)
2407 {
2408 return ctdb_transaction_store_record(h, key, tdb_null);
2409 }
2410
2411 static int ctdb_transaction_fetch_db_seqnum(struct ctdb_transaction_handle *h,
2412 uint64_t *seqnum)
2413 {
2414 const char *keyname = CTDB_DB_SEQNUM_KEY;
2415 TDB_DATA key, data;
2416 struct ctdb_ltdb_header header;
2417 int ret;
2418
2419 key.dptr = discard_const(keyname);
2420 key.dsize = strlen(keyname) + 1;
2421
2422 ret = ctdb_ltdb_fetch(h->db, key, &header, h, &data);
2423 if (ret != 0) {
2424 DEBUG(DEBUG_ERR,
2425 ("transaction_commit: %s seqnum fetch failed, ret=%d\n",
2426 h->db->db_name, ret));
2427 return ret;
2428 }
2429
2430 if (data.dsize == 0) {
2431 /* initial data */
2432 *seqnum = 0;
2433 return 0;
2434 }
2435
2436 if (data.dsize != sizeof(uint64_t)) {
2437 talloc_free(data.dptr);
2438 return EINVAL;
2439 }
2440
2441 *seqnum = *(uint64_t *)data.dptr;
2442
2443 talloc_free(data.dptr);
2444 return 0;
2445 }
2446
2447 static int ctdb_transaction_store_db_seqnum(struct ctdb_transaction_handle *h,
2448 uint64_t seqnum)
2449 {
2450 const char *keyname = CTDB_DB_SEQNUM_KEY;
2451 TDB_DATA key, data;
2452
2453 key.dptr = discard_const(keyname);
2454 key.dsize = strlen(keyname) + 1;
2455
2456 data.dptr = (uint8_t *)&seqnum;
2457 data.dsize = sizeof(seqnum);
2458
2459 return ctdb_transaction_store_record(h, key, data);
2460 }
2461
2462 struct ctdb_transaction_commit_state {
2463 struct tevent_context *ev;
2464 struct timeval timeout;
2465 struct ctdb_transaction_handle *h;
2466 uint64_t seqnum;
2467 };
2468
2469 static void ctdb_transaction_commit_done(struct tevent_req *subreq);
2470 static void ctdb_transaction_commit_g_lock_done(struct tevent_req *subreq);
2471
2472 struct tevent_req *ctdb_transaction_commit_send(
2473 TALLOC_CTX *mem_ctx,
2474 struct tevent_context *ev,
2475 struct timeval timeout,
2476 struct ctdb_transaction_handle *h)
2477 {
2478 struct tevent_req *req, *subreq;
2479 struct ctdb_transaction_commit_state *state;
2480 struct ctdb_req_control request;
2481 int ret;
2482
2483 req = tevent_req_create(mem_ctx, &state,
2484 struct ctdb_transaction_commit_state);
2485 if (req == NULL) {
2486 return NULL;
2487 }
2488
2489 state->ev = ev;
2490 state->timeout = timeout;
2491 state->h = h;
2492
2493 ret = ctdb_transaction_fetch_db_seqnum(h, &state->seqnum);
2494 if (ret != 0) {
2495 tevent_req_error(req, ret);
2496 return tevent_req_post(req, ev);
2497 }
2498
2499 ret = ctdb_transaction_store_db_seqnum(h, state->seqnum+1);
2500 if (ret != 0) {
2501 tevent_req_error(req, ret);
2502 return tevent_req_post(req, ev);
2503 }
2504
2505 ctdb_req_control_trans3_commit(&request, h->recbuf);
2506 subreq = ctdb_client_control_send(state, ev, h->client,
2507 ctdb_client_pnn(h->client),
2508 timeout, &request);
2509 if (tevent_req_nomem(subreq, req)) {
2510 return tevent_req_post(req, ev);
2511 }
2512 tevent_req_set_callback(subreq, ctdb_transaction_commit_done, req);
2513
2514 return req;
2515 }
2516
2517 static void ctdb_transaction_commit_done(struct tevent_req *subreq)
2518 {
2519 struct tevent_req *req = tevent_req_callback_data(
2520 subreq, struct tevent_req);
2521 struct ctdb_transaction_commit_state *state = tevent_req_data(
2522 req, struct ctdb_transaction_commit_state);
2523 struct ctdb_transaction_handle *h = state->h;
2524 struct ctdb_reply_control *reply;
2525 uint64_t seqnum;
2526 int ret;
2527 bool status;
2528
2529 status = ctdb_client_control_recv(subreq, &ret, state, &reply);
2530 TALLOC_FREE(subreq);
2531 if (! status) {
2532 DEBUG(DEBUG_ERR,
2533 ("transaction_commit: %s TRANS3_COMMIT failed, ret=%d\n",
2534 h->db->db_name, ret));
2535 tevent_req_error(req, ret);
2536 return;
2537 }
2538
2539 ret = ctdb_reply_control_trans3_commit(reply);
2540 talloc_free(reply);
2541
2542 if (ret != 0) {
2543 /* Control failed due to recovery */
2544
2545 ret = ctdb_transaction_fetch_db_seqnum(h, &seqnum);
2546 if (ret != 0) {
2547 tevent_req_error(req, ret);
2548 return;
2549 }
2550
2551 if (seqnum == state->seqnum) {
2552 struct ctdb_req_control request;
2553
2554 /* try again */
2555 ctdb_req_control_trans3_commit(&request,
2556 state->h->recbuf);
2557 subreq = ctdb_client_control_send(
2558 state, state->ev, state->h->client,
2559 ctdb_client_pnn(state->h->client),
2560 state->timeout, &request);
2561 if (tevent_req_nomem(subreq, req)) {
2562 return;
2563 }
2564 tevent_req_set_callback(subreq,
2565 ctdb_transaction_commit_done,
2566 req);
2567 return;
2568 }
2569
2570 if (seqnum != state->seqnum + 1) {
2571 DEBUG(DEBUG_ERR,
2572 ("transaction_commit: %s seqnum mismatch "
2573 "0x%"PRIx64" != 0x%"PRIx64" + 1\n",
2574 state->h->db->db_name, seqnum, state->seqnum));
2575 tevent_req_error(req, EIO);
2576 return;
2577 }
2578 }
2579
2580 /* trans3_commit successful */
2581 subreq = ctdb_g_lock_unlock_send(state, state->ev, h->client,
2582 h->db_g_lock, h->lock_name, h->sid);
2583 if (tevent_req_nomem(subreq, req)) {
2584 return;
2585 }
2586 tevent_req_set_callback(subreq, ctdb_transaction_commit_g_lock_done,
2587 req);
2588 }
2589
2590 static void ctdb_transaction_commit_g_lock_done(struct tevent_req *subreq)
2591 {
2592 struct tevent_req *req = tevent_req_callback_data(
2593 subreq, struct tevent_req);
2594 struct ctdb_transaction_commit_state *state = tevent_req_data(
2595 req, struct ctdb_transaction_commit_state);
2596 int ret;
2597 bool status;
2598
2599 status = ctdb_g_lock_unlock_recv(subreq, &ret);
2600 TALLOC_FREE(subreq);
2601 if (! status) {
2602 DEBUG(DEBUG_ERR,
2603 ("transaction_commit: %s g_lock unlock failed, ret=%d\n",
2604 state->h->db->db_name, ret));
2605 tevent_req_error(req, ret);
2606 return;
2607 }
2608
2609 talloc_free(state->h);
2610 tevent_req_done(req);
2611 }
2612
2613 bool ctdb_transaction_commit_recv(struct tevent_req *req, int *perr)
2614 {
2615 int err;
2616
2617 if (tevent_req_is_unix_error(req, &err)) {
2618 if (perr != NULL) {
2619 *perr = err;
2620 }
2621 return false;
2622 }
2623
2624 return true;
2625 }
2626
2627 int ctdb_transaction_commit(struct ctdb_transaction_handle *h)
2628 {
2629 struct tevent_context *ev = h->ev;
2630 TALLOC_CTX *mem_ctx;
2631 struct tevent_req *req;
2632 int ret;
2633 bool status;
2634
2635 if (h->readonly || ! h->updated) {
2636 return ctdb_transaction_cancel(h);
2637 }
2638
2639 mem_ctx = talloc_new(NULL);
2640 if (mem_ctx == NULL) {
2641 return ENOMEM;
2642 }
2643
2644 req = ctdb_transaction_commit_send(mem_ctx, ev,
2645 tevent_timeval_zero(), h);
2646 if (req == NULL) {
2647 talloc_free(mem_ctx);
2648 return ENOMEM;
2649 }
2650
2651 tevent_req_poll(req, ev);
2652
2653 status = ctdb_transaction_commit_recv(req, &ret);
2654 if (! status) {
2655 talloc_free(mem_ctx);
2656 return ret;
2657 }
2658
2659 talloc_free(mem_ctx);
2660 return 0;
2661 }
2662
2663 struct ctdb_transaction_cancel_state {
2664 struct tevent_context *ev;
2665 struct ctdb_transaction_handle *h;
2666 struct timeval timeout;
2667 };
2668
2669 static void ctdb_transaction_cancel_done(struct tevent_req *subreq);
2670
2671 struct tevent_req *ctdb_transaction_cancel_send(
2672 TALLOC_CTX *mem_ctx,
2673 struct tevent_context *ev,
2674 struct timeval timeout,
2675 struct ctdb_transaction_handle *h)
2676 {
2677 struct tevent_req *req, *subreq;
2678 struct ctdb_transaction_cancel_state *state;
2679
2680 req = tevent_req_create(mem_ctx, &state,
2681 struct ctdb_transaction_cancel_state);
2682 if (req == NULL) {
2683 return NULL;
2684 }
2685
2686 state->ev = ev;
2687 state->h = h;
2688 state->timeout = timeout;
2689
2690 subreq = ctdb_g_lock_unlock_send(state, state->ev, state->h->client,
2691 state->h->db_g_lock,
2692 state->h->lock_name, state->h->sid);
2693 if (tevent_req_nomem(subreq, req)) {
2694 return tevent_req_post(req, ev);
2695 }
2696 tevent_req_set_callback(subreq, ctdb_transaction_cancel_done,
2697 req);
2698
2699 return req;
2700 }
2701
2702 static void ctdb_transaction_cancel_done(struct tevent_req *subreq)
2703 {
2704 struct tevent_req *req = tevent_req_callback_data(
2705 subreq, struct tevent_req);
2706 struct ctdb_transaction_cancel_state *state = tevent_req_data(
2707 req, struct ctdb_transaction_cancel_state);
2708 int ret;
2709 bool status;
2710
2711 status = ctdb_g_lock_unlock_recv(subreq, &ret);
2712 TALLOC_FREE(subreq);
2713 if (! status) {
2714 DEBUG(DEBUG_ERR,
2715 ("transaction_cancel: %s g_lock unlock failed, ret=%d\n",
2716 state->h->db->db_name, ret));
2717 talloc_free(state->h);
2718 tevent_req_error(req, ret);
2719 return;
2720 }
2721
2722 talloc_free(state->h);
2723 tevent_req_done(req);
2724 }
2725
2726 bool ctdb_transaction_cancel_recv(struct tevent_req *req, int *perr)
2727 {
2728 int err;
2729
2730 if (tevent_req_is_unix_error(req, &err)) {
2731 if (perr != NULL) {
2732 *perr = err;
2733 }
2734 return false;
2735 }
2736
2737 return true;
2738 }
2739
2740 int ctdb_transaction_cancel(struct ctdb_transaction_handle *h)
2741 {
2742 struct tevent_context *ev = h->ev;
2743 struct tevent_req *req;
2744 TALLOC_CTX *mem_ctx;
2745 int ret;
2746 bool status;
2747
2748 mem_ctx = talloc_new(NULL);
2749 if (mem_ctx == NULL) {
2750 talloc_free(h);
2751 return ENOMEM;
2752 }
2753
2754 req = ctdb_transaction_cancel_send(mem_ctx, ev,
2755 tevent_timeval_zero(), h);
2756 if (req == NULL) {
2757 talloc_free(mem_ctx);
2758 talloc_free(h);
2759 return ENOMEM;
2760 }
2761
2762 tevent_req_poll(req, ev);
2763
2764 status = ctdb_transaction_cancel_recv(req, &ret);
2765 if (! status) {
2766 talloc_free(mem_ctx);
2767 return ret;
2768 }
2769
2770 talloc_free(mem_ctx);
2771 return 0;
2772 }
2773
2774 /*
2775 * TODO:
2776 *
2777 * In future Samba should register SERVER_ID.
2778 * Make that structure same as struct srvid {}.
2779 */
Samba GIT mirror