2 Unix SMB/CIFS implementation.
3 Database interface wrapper around ctdbd
4 Copyright (C) Volker Lendecke 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 #ifdef CLUSTER_SUPPORT
23 #include "ctdb_private.h"
24 #include "ctdbd_conn.h"
27 struct tdb_wrap *wtdb;
32 struct db_ctdb_ctx *ctdb_ctx;
33 struct ctdb_ltdb_header header;
36 static NTSTATUS db_ctdb_store(struct db_record *rec, TDB_DATA data, int flag)
38 struct db_ctdb_rec *crec = talloc_get_type_abort(
39 rec->private_data, struct db_ctdb_rec);
43 cdata.dsize = sizeof(crec->header) + data.dsize;
45 if (!(cdata.dptr = SMB_MALLOC_ARRAY(uint8, cdata.dsize))) {
46 return NT_STATUS_NO_MEMORY;
49 memcpy(cdata.dptr, &crec->header, sizeof(crec->header));
50 memcpy(cdata.dptr + sizeof(crec->header), data.dptr, data.dsize);
52 ret = tdb_store(crec->ctdb_ctx->wtdb->tdb, rec->key, cdata, TDB_REPLACE);
54 SAFE_FREE(cdata.dptr);
56 return (ret == 0) ? NT_STATUS_OK : NT_STATUS_INTERNAL_DB_CORRUPTION;
60 /* for persistent databases the store is a bit different. We have to
61 ask the ctdb daemon to push the record to all nodes after the
63 static NTSTATUS db_ctdb_store_persistent(struct db_record *rec, TDB_DATA data, int flag)
65 struct db_ctdb_rec *crec = talloc_get_type_abort(
66 rec->private_data, struct db_ctdb_rec);
71 cdata.dsize = sizeof(crec->header) + data.dsize;
73 if (!(cdata.dptr = SMB_MALLOC_ARRAY(uint8, cdata.dsize))) {
74 return NT_STATUS_NO_MEMORY;
79 memcpy(cdata.dptr, &crec->header, sizeof(crec->header));
80 memcpy(cdata.dptr + sizeof(crec->header), data.dptr, data.dsize);
82 status = ctdbd_start_persistent_update(messaging_ctdbd_connection(), crec->ctdb_ctx->db_id, rec->key, cdata);
84 /* now tell ctdbd to update this record on all other nodes */
85 if (NT_STATUS_IS_OK(status)) {
86 ret = tdb_store(crec->ctdb_ctx->wtdb->tdb, rec->key, cdata, TDB_REPLACE);
87 status = (ret == 0) ? NT_STATUS_OK : NT_STATUS_INTERNAL_DB_CORRUPTION;
90 /* now tell ctdbd to update this record on all other nodes */
91 if (NT_STATUS_IS_OK(status)) {
92 status = ctdbd_persistent_store(messaging_ctdbd_connection(), crec->ctdb_ctx->db_id, rec->key, cdata);
94 ctdbd_cancel_persistent_update(messaging_ctdbd_connection(), crec->ctdb_ctx->db_id, rec->key, cdata);
97 SAFE_FREE(cdata.dptr);
102 static NTSTATUS db_ctdb_delete(struct db_record *rec)
104 struct db_ctdb_rec *crec = talloc_get_type_abort(
105 rec->private_data, struct db_ctdb_rec);
110 * We have to store the header with empty data. TODO: Fix the
114 data.dptr = (uint8 *)&crec->header;
115 data.dsize = sizeof(crec->header);
117 ret = tdb_store(crec->ctdb_ctx->wtdb->tdb, rec->key, data, TDB_REPLACE);
119 return (ret == 0) ? NT_STATUS_OK : NT_STATUS_INTERNAL_DB_CORRUPTION;
122 static int db_ctdb_record_destr(struct db_record* data)
124 struct db_ctdb_rec *crec = talloc_get_type_abort(
125 data->private_data, struct db_ctdb_rec);
127 DEBUG(10, (DEBUGLEVEL > 10
128 ? "Unlocking db %u key %s\n"
129 : "Unlocking db %u key %.20s\n",
130 (int)crec->ctdb_ctx->db_id,
131 hex_encode(data, (unsigned char *)data->key.dptr,
134 if (tdb_chainunlock(crec->ctdb_ctx->wtdb->tdb, data->key) != 0) {
135 DEBUG(0, ("tdb_chainunlock failed\n"));
142 static struct db_record *db_ctdb_fetch_locked(struct db_context *db,
146 struct db_ctdb_ctx *ctx = talloc_get_type_abort(db->private_data,
148 struct db_record *result;
149 struct db_ctdb_rec *crec;
152 int migrate_attempts = 0;
154 if (!(result = talloc(mem_ctx, struct db_record))) {
155 DEBUG(0, ("talloc failed\n"));
159 if (!(crec = TALLOC_ZERO_P(result, struct db_ctdb_rec))) {
160 DEBUG(0, ("talloc failed\n"));
165 result->private_data = (void *)crec;
166 crec->ctdb_ctx = ctx;
168 result->key.dsize = key.dsize;
169 result->key.dptr = (uint8 *)talloc_memdup(result, key.dptr, key.dsize);
170 if (result->key.dptr == NULL) {
171 DEBUG(0, ("talloc failed\n"));
177 * Do a blocking lock on the record
181 if (DEBUGLEVEL >= 10) {
182 char *keystr = hex_encode(result, key.dptr, key.dsize);
183 DEBUG(10, (DEBUGLEVEL > 10
184 ? "Locking db %u key %s\n"
185 : "Locking db %u key %.20s\n",
186 (int)crec->ctdb_ctx->db_id, keystr));
190 if (tdb_chainlock(ctx->wtdb->tdb, key) != 0) {
191 DEBUG(3, ("tdb_chainlock failed\n"));
196 if (db->persistent) {
197 result->store = db_ctdb_store_persistent;
199 result->store = db_ctdb_store;
201 result->delete_rec = db_ctdb_delete;
202 talloc_set_destructor(result, db_ctdb_record_destr);
204 ctdb_data = tdb_fetch(ctx->wtdb->tdb, key);
207 * See if we have a valid record and we are the dmaster. If so, we can
208 * take the shortcut and just return it.
211 if ((ctdb_data.dptr == NULL) ||
212 (ctdb_data.dsize < sizeof(struct ctdb_ltdb_header)) ||
213 ((struct ctdb_ltdb_header *)ctdb_data.dptr)->dmaster != get_my_vnn()
215 || (random() % 2 != 0)
218 SAFE_FREE(ctdb_data.dptr);
219 tdb_chainunlock(ctx->wtdb->tdb, key);
220 talloc_set_destructor(result, NULL);
222 migrate_attempts += 1;
224 DEBUG(10, ("ctdb_data.dptr = %p, dmaster = %u (%u)\n",
225 ctdb_data.dptr, ctdb_data.dptr ?
226 ((struct ctdb_ltdb_header *)ctdb_data.dptr)->dmaster : -1,
229 status = ctdbd_migrate(messaging_ctdbd_connection(),ctx->db_id, key);
230 if (!NT_STATUS_IS_OK(status)) {
231 DEBUG(5, ("ctdb_migrate failed: %s\n",
236 /* now its migrated, try again */
240 if (migrate_attempts > 10) {
241 DEBUG(0, ("db_ctdb_fetch_locked needed %d attempts\n",
245 memcpy(&crec->header, ctdb_data.dptr, sizeof(crec->header));
247 result->value.dsize = ctdb_data.dsize - sizeof(crec->header);
248 result->value.dptr = NULL;
250 if ((result->value.dsize != 0)
251 && !(result->value.dptr = (uint8 *)talloc_memdup(
252 result, ctdb_data.dptr + sizeof(crec->header),
253 result->value.dsize))) {
254 DEBUG(0, ("talloc failed\n"));
258 SAFE_FREE(ctdb_data.dptr);
264 fetch (unlocked, no migration) operation on ctdb
266 static int db_ctdb_fetch(struct db_context *db, TALLOC_CTX *mem_ctx,
267 TDB_DATA key, TDB_DATA *data)
269 struct db_ctdb_ctx *ctx = talloc_get_type_abort(db->private_data,
274 /* try a direct fetch */
275 ctdb_data = tdb_fetch(ctx->wtdb->tdb, key);
278 * See if we have a valid record and we are the dmaster. If so, we can
279 * take the shortcut and just return it.
280 * we bypass the dmaster check for persistent databases
282 if ((ctdb_data.dptr != NULL) &&
283 (ctdb_data.dsize >= sizeof(struct ctdb_ltdb_header)) &&
285 ((struct ctdb_ltdb_header *)ctdb_data.dptr)->dmaster == get_my_vnn())) {
286 /* we are the dmaster - avoid the ctdb protocol op */
288 data->dsize = ctdb_data.dsize - sizeof(struct ctdb_ltdb_header);
289 if (data->dsize == 0) {
290 SAFE_FREE(ctdb_data.dptr);
295 data->dptr = (uint8 *)talloc_memdup(
296 mem_ctx, ctdb_data.dptr+sizeof(struct ctdb_ltdb_header),
299 SAFE_FREE(ctdb_data.dptr);
301 if (data->dptr == NULL) {
307 SAFE_FREE(ctdb_data.dptr);
309 /* we weren't able to get it locally - ask ctdb to fetch it for us */
310 status = ctdbd_fetch(messaging_ctdbd_connection(),ctx->db_id, key, mem_ctx, data);
311 if (!NT_STATUS_IS_OK(status)) {
312 DEBUG(5, ("ctdbd_fetch failed: %s\n", nt_errstr(status)));
319 struct traverse_state {
320 struct db_context *db;
321 int (*fn)(struct db_record *rec, void *private_data);
325 static void traverse_callback(TDB_DATA key, TDB_DATA data, void *private_data)
327 struct traverse_state *state = (struct traverse_state *)private_data;
328 struct db_record *rec;
329 TALLOC_CTX *tmp_ctx = talloc_new(state->db);
330 /* we have to give them a locked record to prevent races */
331 rec = db_ctdb_fetch_locked(state->db, tmp_ctx, key);
332 if (rec && rec->value.dsize > 0) {
333 state->fn(rec, state->private_data);
335 talloc_free(tmp_ctx);
338 static int traverse_persistent_callback(TDB_CONTEXT *tdb, TDB_DATA kbuf, TDB_DATA dbuf,
341 struct traverse_state *state = (struct traverse_state *)private_data;
342 struct db_record *rec;
343 TALLOC_CTX *tmp_ctx = talloc_new(state->db);
345 /* we have to give them a locked record to prevent races */
346 rec = db_ctdb_fetch_locked(state->db, tmp_ctx, kbuf);
347 if (rec && rec->value.dsize > 0) {
348 ret = state->fn(rec, state->private_data);
350 talloc_free(tmp_ctx);
354 static int db_ctdb_traverse(struct db_context *db,
355 int (*fn)(struct db_record *rec,
359 struct db_ctdb_ctx *ctx = talloc_get_type_abort(db->private_data,
361 struct traverse_state state;
365 state.private_data = private_data;
367 if (db->persistent) {
368 /* for persistent databases we don't need to do a ctdb traverse,
369 we can do a faster local traverse */
370 return tdb_traverse(ctx->wtdb->tdb, traverse_persistent_callback, &state);
374 ctdbd_traverse(ctx->db_id, traverse_callback, &state);
378 static NTSTATUS db_ctdb_store_deny(struct db_record *rec, TDB_DATA data, int flag)
380 return NT_STATUS_MEDIA_WRITE_PROTECTED;
383 static NTSTATUS db_ctdb_delete_deny(struct db_record *rec)
385 return NT_STATUS_MEDIA_WRITE_PROTECTED;
388 static void traverse_read_callback(TDB_DATA key, TDB_DATA data, void *private_data)
390 struct traverse_state *state = (struct traverse_state *)private_data;
391 struct db_record rec;
394 rec.store = db_ctdb_store_deny;
395 rec.delete_rec = db_ctdb_delete_deny;
396 rec.private_data = state->db;
397 state->fn(&rec, state->private_data);
400 static int traverse_persistent_callback_read(TDB_CONTEXT *tdb, TDB_DATA kbuf, TDB_DATA dbuf,
403 struct traverse_state *state = (struct traverse_state *)private_data;
404 struct db_record rec;
407 rec.store = db_ctdb_store_deny;
408 rec.delete_rec = db_ctdb_delete_deny;
409 rec.private_data = state->db;
411 if (rec.value.dsize <= sizeof(struct ctdb_ltdb_header)) {
412 /* a deleted record */
415 rec.value.dsize -= sizeof(struct ctdb_ltdb_header);
416 rec.value.dptr += sizeof(struct ctdb_ltdb_header);
418 return state->fn(&rec, state->private_data);
421 static int db_ctdb_traverse_read(struct db_context *db,
422 int (*fn)(struct db_record *rec,
426 struct db_ctdb_ctx *ctx = talloc_get_type_abort(db->private_data,
428 struct traverse_state state;
432 state.private_data = private_data;
434 if (db->persistent) {
435 /* for persistent databases we don't need to do a ctdb traverse,
436 we can do a faster local traverse */
437 return tdb_traverse_read(ctx->wtdb->tdb, traverse_persistent_callback_read, &state);
440 ctdbd_traverse(ctx->db_id, traverse_read_callback, &state);
444 static int db_ctdb_get_seqnum(struct db_context *db)
446 struct db_ctdb_ctx *ctx = talloc_get_type_abort(db->private_data,
448 return tdb_get_seqnum(ctx->wtdb->tdb);
451 static int db_ctdb_trans_dummy(struct db_context *db)
454 * Not implemented yet, just return ok
459 struct db_context *db_open_ctdb(TALLOC_CTX *mem_ctx,
461 int hash_size, int tdb_flags,
462 int open_flags, mode_t mode)
464 struct db_context *result;
465 struct db_ctdb_ctx *db_ctdb;
468 if (!lp_clustering()) {
469 DEBUG(10, ("Clustering disabled -- no ctdb\n"));
473 if (!(result = TALLOC_ZERO_P(mem_ctx, struct db_context))) {
474 DEBUG(0, ("talloc failed\n"));
479 if (!(db_ctdb = TALLOC_P(result, struct db_ctdb_ctx))) {
480 DEBUG(0, ("talloc failed\n"));
485 if (!NT_STATUS_IS_OK(ctdbd_db_attach(messaging_ctdbd_connection(),name, &db_ctdb->db_id, tdb_flags))) {
486 DEBUG(0, ("ctdbd_db_attach failed for %s\n", name));
491 db_path = ctdbd_dbpath(messaging_ctdbd_connection(), db_ctdb, db_ctdb->db_id);
493 result->persistent = ((tdb_flags & TDB_CLEAR_IF_FIRST) == 0);
495 /* only pass through specific flags */
496 tdb_flags &= TDB_SEQNUM;
498 /* honor permissions if user has specified O_CREAT */
499 if (open_flags & O_CREAT) {
500 chmod(db_path, mode);
503 db_ctdb->wtdb = tdb_wrap_open(db_ctdb, db_path, hash_size, tdb_flags, O_RDWR, 0);
504 if (db_ctdb->wtdb == NULL) {
505 DEBUG(0, ("Could not open tdb %s: %s\n", db_path, strerror(errno)));
509 talloc_free(db_path);
511 result->private_data = (void *)db_ctdb;
512 result->fetch_locked = db_ctdb_fetch_locked;
513 result->fetch = db_ctdb_fetch;
514 result->traverse = db_ctdb_traverse;
515 result->traverse_read = db_ctdb_traverse_read;
516 result->get_seqnum = db_ctdb_get_seqnum;
517 result->transaction_start = db_ctdb_trans_dummy;
518 result->transaction_commit = db_ctdb_trans_dummy;
519 result->transaction_cancel = db_ctdb_trans_dummy;
521 DEBUG(3,("db_open_ctdb: opened database '%s' with dbid 0x%x\n",
522 name, db_ctdb->db_id));