ctdb ltdb code
Copyright (C) Andrew Tridgell 2006
-
- This library is free software; you can redistribute it and/or
- modify it under the terms of the GNU Lesser General Public
- License as published by the Free Software Foundation; either
- version 2 of the License, or (at your option) any later version.
-
- This library is distributed in the hope that it will be useful,
+ Copyright (C) Ronnie sahlberg 2011
+
+ This program is free software; you can redistribute it and/or modify
+ it under the terms of the GNU General Public License as published by
+ the Free Software Foundation; either version 3 of the License, or
+ (at your option) any later version.
+
+ This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
- MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
- Lesser General Public License for more details.
-
- You should have received a copy of the GNU Lesser General Public
- License along with this library; if not, write to the Free Software
- Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
+ MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ GNU General Public License for more details.
+
+ You should have received a copy of the GNU General Public License
+ along with this program; if not, see <http://www.gnu.org/licenses/>.
*/
-#include "includes.h"
-#include "lib/events/events.h"
-#include "lib/tdb/include/tdb.h"
+#include "replace.h"
#include "system/network.h"
#include "system/filesys.h"
-#include "../include/ctdb_private.h"
-#include "db_wrap.h"
+
+#include <tdb.h>
+
+#include "lib/tdb_wrap/tdb_wrap.h"
#include "lib/util/dlinklist.h"
+#include "lib/util/debug.h"
+
+#include "ctdb_private.h"
+
+#include "common/common.h"
+#include "common/logging.h"
+
+
+/*
+ * Calculate tdb flags based on databse type
+ */
+int ctdb_db_tdb_flags(uint8_t db_flags, bool with_valgrind, bool with_mutex)
+{
+ int tdb_flags = 0;
+
+ if (db_flags & CTDB_DB_FLAGS_PERSISTENT) {
+ tdb_flags = TDB_DEFAULT;
+
+ } else if (db_flags & CTDB_DB_FLAGS_REPLICATED) {
+ tdb_flags = TDB_NOSYNC |
+ TDB_CLEAR_IF_FIRST |
+ TDB_INCOMPATIBLE_HASH;
+
+ } else {
+ tdb_flags = TDB_NOSYNC |
+ TDB_CLEAR_IF_FIRST |
+ TDB_INCOMPATIBLE_HASH;
+
+#ifdef TDB_MUTEX_LOCKING
+ if (with_mutex && tdb_runtime_check_for_robust_mutexes()) {
+ tdb_flags |= TDB_MUTEX_LOCKING;
+ }
+#endif
+
+ }
+
+ tdb_flags |= TDB_DISALLOW_NESTING;
+ if (with_valgrind) {
+ tdb_flags |= TDB_NOMMAP;
+ }
+
+ return tdb_flags;
+}
/*
find an attached ctdb_db handle given a name
return NULL;
}
+bool ctdb_db_persistent(struct ctdb_db_context *ctdb_db)
+{
+ if (ctdb_db->db_flags & CTDB_DB_FLAGS_PERSISTENT) {
+ return true;
+ }
+ return false;
+}
+
+bool ctdb_db_replicated(struct ctdb_db_context *ctdb_db)
+{
+ if (ctdb_db->db_flags & CTDB_DB_FLAGS_REPLICATED) {
+ return true;
+ }
+ return false;
+}
-/*
- this is the dummy null procedure that all databases support
-*/
-static int ctdb_null_func(struct ctdb_call_info *call)
+bool ctdb_db_volatile(struct ctdb_db_context *ctdb_db)
{
- return 0;
+ if ((ctdb_db->db_flags & CTDB_DB_FLAGS_PERSISTENT) ||
+ (ctdb_db->db_flags & CTDB_DB_FLAGS_REPLICATED)) {
+ return false;
+ }
+ return true;
}
-/*
- this is a plain fetch procedure that all databases support
-*/
-static int ctdb_fetch_func(struct ctdb_call_info *call)
+bool ctdb_db_readonly(struct ctdb_db_context *ctdb_db)
{
- call->reply_data = &call->record_data;
- return 0;
+ if (ctdb_db->db_flags & CTDB_DB_FLAGS_READONLY) {
+ return true;
+ }
+ return false;
+}
+
+void ctdb_db_set_readonly(struct ctdb_db_context *ctdb_db)
+{
+ ctdb_db->db_flags |= CTDB_DB_FLAGS_READONLY;
+}
+
+void ctdb_db_reset_readonly(struct ctdb_db_context *ctdb_db)
+{
+ ctdb_db->db_flags &= ~CTDB_DB_FLAGS_READONLY;
+}
+
+bool ctdb_db_sticky(struct ctdb_db_context *ctdb_db)
+{
+ if (ctdb_db->db_flags & CTDB_DB_FLAGS_STICKY) {
+ return true;
+ }
+ return false;
}
+void ctdb_db_set_sticky(struct ctdb_db_context *ctdb_db)
+{
+ ctdb_db->db_flags |= CTDB_DB_FLAGS_STICKY;
+}
/*
return the lmaster given a key
TDB_DATA key,
struct ctdb_ltdb_header *header)
{
- header->rsn = 0;
+ ZERO_STRUCTP(header);
/* initial dmaster is the lmaster */
header->dmaster = ctdb_lmaster(ctdb_db->ctdb, &key);
- header->laccessor = header->dmaster;
- header->lacount = 0;
+ header->flags = CTDB_REC_FLAG_AUTOMATIC;
}
rec = tdb_fetch(ctdb_db->ltdb->tdb, key);
if (rec.dsize < sizeof(*header)) {
- TDB_DATA d2;
/* return an initial header */
if (rec.dptr) free(rec.dptr);
+ if (ctdb->vnn_map == NULL) {
+ /* called from the client */
+ ZERO_STRUCTP(data);
+ header->dmaster = (uint32_t)-1;
+ return -1;
+ }
ltdb_initial_header(ctdb_db, key, header);
- ZERO_STRUCT(d2);
if (data) {
- *data = d2;
+ *data = tdb_null;
+ }
+ if (ctdb_db_persistent(ctdb_db) ||
+ header->dmaster == ctdb_db->ctdb->pnn) {
+ if (ctdb_ltdb_store(ctdb_db, key, header, tdb_null) != 0) {
+ DEBUG(DEBUG_NOTICE,
+ (__location__ "failed to store initial header\n"));
+ }
}
- ctdb_ltdb_store(ctdb_db, key, header, d2);
return 0;
}
return 0;
}
-
/*
fetch a record from the ltdb, separating out the header information
- and returning the body of the record. A valid (initial) header is
- returned if the record is not present
+ and returning the body of the record.
+ if the record does not exist, *header will be NULL
+ and data = {0, NULL}
+*/
+int ctdb_ltdb_fetch_with_header(struct ctdb_db_context *ctdb_db,
+ TDB_DATA key, struct ctdb_ltdb_header *header,
+ TALLOC_CTX *mem_ctx, TDB_DATA *data)
+{
+ TDB_DATA rec;
+
+ rec = tdb_fetch(ctdb_db->ltdb->tdb, key);
+ if (rec.dsize < sizeof(*header)) {
+ free(rec.dptr);
+
+ data->dsize = 0;
+ data->dptr = NULL;
+ return -1;
+ }
+
+ *header = *(struct ctdb_ltdb_header *)rec.dptr;
+ if (data) {
+ data->dsize = rec.dsize - sizeof(struct ctdb_ltdb_header);
+ data->dptr = talloc_memdup(mem_ctx,
+ sizeof(struct ctdb_ltdb_header)+rec.dptr,
+ data->dsize);
+ }
+
+ free(rec.dptr);
+
+ return 0;
+}
+
+
+/*
+ write a record to a normal database
*/
int ctdb_ltdb_store(struct ctdb_db_context *ctdb_db, TDB_DATA key,
struct ctdb_ltdb_header *header, TDB_DATA data)
{
struct ctdb_context *ctdb = ctdb_db->ctdb;
- TDB_DATA rec;
+ TDB_DATA rec[2];
+ uint32_t hsize = sizeof(struct ctdb_ltdb_header);
int ret;
+ bool seqnum_suppressed = false;
+
+ if (ctdb_db->ctdb_ltdb_store_fn) {
+ return ctdb_db->ctdb_ltdb_store_fn(ctdb_db, key, header, data);
+ }
if (ctdb->flags & CTDB_FLAG_TORTURE) {
+ TDB_DATA old;
struct ctdb_ltdb_header *h2;
- rec = tdb_fetch(ctdb_db->ltdb->tdb, key);
- h2 = (struct ctdb_ltdb_header *)rec.dptr;
- if (rec.dptr && rec.dsize >= sizeof(h2) && h2->rsn > header->rsn) {
- DEBUG(0,("RSN regression! %llu %llu\n",
- (unsigned long long)h2->rsn, (unsigned long long)header->rsn));
+
+ old = tdb_fetch(ctdb_db->ltdb->tdb, key);
+ h2 = (struct ctdb_ltdb_header *)old.dptr;
+ if (old.dptr != NULL && old.dsize >= hsize &&
+ h2->rsn > header->rsn) {
+ DEBUG(DEBUG_ERR,
+ ("RSN regression! %"PRIu64" %"PRIu64"\n",
+ h2->rsn, header->rsn));
+ }
+ if (old.dptr != NULL) {
+ free(old.dptr);
}
- if (rec.dptr) free(rec.dptr);
}
- rec.dsize = sizeof(*header) + data.dsize;
- rec.dptr = talloc_size(ctdb, rec.dsize);
- CTDB_NO_MEMORY(ctdb, rec.dptr);
+ rec[0].dsize = hsize;
+ rec[0].dptr = (uint8_t *)header;
+
+ rec[1].dsize = data.dsize;
+ rec[1].dptr = data.dptr;
- memcpy(rec.dptr, header, sizeof(*header));
- memcpy(rec.dptr + sizeof(*header), data.dptr, data.dsize);
+ /* Databases with seqnum updates enabled only get their seqnum
+ changes when/if we modify the data */
+ if (ctdb_db->seqnum_update != NULL) {
+ TDB_DATA old;
+ old = tdb_fetch(ctdb_db->ltdb->tdb, key);
- ret = tdb_store(ctdb_db->ltdb->tdb, key, rec, TDB_REPLACE);
- talloc_free(rec.dptr);
+ if ((old.dsize == hsize + data.dsize) &&
+ memcmp(old.dptr+hsize, data.dptr, data.dsize) == 0) {
+ tdb_remove_flags(ctdb_db->ltdb->tdb, TDB_SEQNUM);
+ seqnum_suppressed = true;
+ }
+ if (old.dptr != NULL) {
+ free(old.dptr);
+ }
+ }
+ ret = tdb_storev(ctdb_db->ltdb->tdb, key, rec, 2, TDB_REPLACE);
+ if (ret != 0) {
+ DEBUG(DEBUG_ERR, (__location__ " Failed to store dynamic data\n"));
+ }
+ if (seqnum_suppressed) {
+ tdb_add_flags(ctdb_db->ltdb->tdb, TDB_SEQNUM);
+ }
return ret;
}
-
/*
lock a record in the ltdb, given a key
*/
{
int ret = tdb_chainunlock(ctdb_db->ltdb->tdb, key);
if (ret != 0) {
- DEBUG(0,("tdb_chainunlock failed\n"));
+ DEBUG(DEBUG_ERR,("tdb_chainunlock failed on db %s [%s]\n", ctdb_db->db_name, tdb_errorstr(ctdb_db->ltdb->tdb)));
}
return ret;
}
-struct lock_fetch_state {
- struct ctdb_context *ctdb;
- void (*recv_pkt)(void *, struct ctdb_req_header *);
- void *recv_context;
- struct ctdb_req_header *hdr;
- uint32_t generation;
- bool ignore_generation;
-};
/*
- called when we should retry the operation
- */
-static void lock_fetch_callback(void *p)
-{
- struct lock_fetch_state *state = talloc_get_type(p, struct lock_fetch_state);
- if (!state->ignore_generation &&
- state->generation != state->ctdb->vnn_map->generation) {
- DEBUG(0,("Discarding previous generation lockwait packet\n"));
- talloc_free(state->hdr);
- return;
- }
- state->recv_pkt(state->recv_context, state->hdr);
- DEBUG(2,(__location__ " PACKET REQUEUED\n"));
-}
-
-
-/*
- do a non-blocking ltdb_lock, deferring this ctdb request until we
- have the chainlock
-
- It does the following:
-
- 1) tries to get the chainlock. If it succeeds, then it returns 0
-
- 2) if it fails to get a chainlock immediately then it sets up a
- non-blocking chainlock via ctdb_lockwait, and when it gets the
- chainlock it re-submits this ctdb request to the main packet
- receive function
-
- This effectively queues all ctdb requests that cannot be
- immediately satisfied until it can get the lock. This means that
- the main ctdb daemon will not block waiting for a chainlock held by
- a client
-
- There are 3 possible return values:
-
- 0: means that it got the lock immediately.
- -1: means that it failed to get the lock, and won't retry
- -2: means that it failed to get the lock immediately, but will retry
- */
-int ctdb_ltdb_lock_requeue(struct ctdb_db_context *ctdb_db,
- TDB_DATA key, struct ctdb_req_header *hdr,
- void (*recv_pkt)(void *, struct ctdb_req_header *),
- void *recv_context, bool ignore_generation)
+ delete a record from a normal database
+*/
+int ctdb_ltdb_delete(struct ctdb_db_context *ctdb_db, TDB_DATA key)
{
- int ret;
- struct tdb_context *tdb = ctdb_db->ltdb->tdb;
- struct lockwait_handle *h;
- struct lock_fetch_state *state;
-
- ret = tdb_chainlock_nonblock(tdb, key);
-
- if (ret != 0 &&
- !(errno == EACCES || errno == EAGAIN || errno == EDEADLK)) {
- /* a hard failure - don't try again */
- return -1;
- }
-
- /* when torturing, ensure we test the contended path */
- if ((ctdb_db->ctdb->flags & CTDB_FLAG_TORTURE) &&
- random() % 5 == 0) {
- ret = -1;
- tdb_chainunlock(tdb, key);
- }
-
- /* first the non-contended path */
- if (ret == 0) {
+ if (! ctdb_db_volatile(ctdb_db)) {
+ DEBUG(DEBUG_WARNING,
+ ("Ignored deletion of empty record from "
+ "non-volatile database\n"));
return 0;
}
-
- state = talloc(hdr, struct lock_fetch_state);
- state->ctdb = ctdb_db->ctdb;
- state->hdr = hdr;
- state->recv_pkt = recv_pkt;
- state->recv_context = recv_context;
- state->generation = ctdb_db->ctdb->vnn_map->generation;
- state->ignore_generation = ignore_generation;
-
- /* now the contended path */
- h = ctdb_lockwait(ctdb_db, key, lock_fetch_callback, state);
- if (h == NULL) {
- tdb_chainunlock(tdb, key);
+ if (tdb_delete(ctdb_db->ltdb->tdb, key) != 0) {
+ DEBUG(DEBUG_ERR,("Failed to delete empty record."));
return -1;
}
-
- /* we need to move the packet off the temporary context in ctdb_input_pkt(),
- so it won't be freed yet */
- talloc_steal(state, hdr);
- talloc_steal(state, h);
-
- /* now tell the caller than we will retry asynchronously */
- return -2;
+ return 0;
}
-/*
- a varient of ctdb_ltdb_lock_requeue that also fetches the record
- */
-int ctdb_ltdb_lock_fetch_requeue(struct ctdb_db_context *ctdb_db,
- TDB_DATA key, struct ctdb_ltdb_header *header,
- struct ctdb_req_header *hdr, TDB_DATA *data,
- void (*recv_pkt)(void *, struct ctdb_req_header *),
- void *recv_context, bool ignore_generation)
+int ctdb_trackingdb_add_pnn(struct ctdb_context *ctdb, TDB_DATA *data, uint32_t pnn)
{
- int ret;
+ int byte_pos = pnn / 8;
+ int bit_mask = 1 << (pnn % 8);
+
+ if (byte_pos + 1 > data->dsize) {
+ char *buf;
- ret = ctdb_ltdb_lock_requeue(ctdb_db, key, hdr, recv_pkt,
- recv_context, ignore_generation);
- if (ret == 0) {
- ret = ctdb_ltdb_fetch(ctdb_db, key, header, hdr, data);
- if (ret != 0) {
- ctdb_ltdb_unlock(ctdb_db, key);
+ buf = malloc(byte_pos + 1);
+ memset(buf, 0, byte_pos + 1);
+ if (buf == NULL) {
+ DEBUG(DEBUG_ERR, ("Out of memory when allocating buffer of %d bytes for trackingdb\n", byte_pos + 1));
+ return -1;
+ }
+ if (data->dptr != NULL) {
+ memcpy(buf, data->dptr, data->dsize);
+ free(data->dptr);
}
+ data->dptr = (uint8_t *)buf;
+ data->dsize = byte_pos + 1;
}
- return ret;
-}
-
-/*
- paraoid check to see if the db is empty
- */
-static void ctdb_check_db_empty(struct ctdb_db_context *ctdb_db)
-{
- struct tdb_context *tdb = ctdb_db->ltdb->tdb;
- int count = tdb_traverse_read(tdb, NULL, NULL);
- if (count != 0) {
- DEBUG(0,(__location__ " tdb '%s' not empty on attach! aborting\n",
- ctdb_db->db_path));
- ctdb_fatal(ctdb_db->ctdb, "database not empty on attach");
- }
+ data->dptr[byte_pos] |= bit_mask;
+ return 0;
}
-/*
- a client has asked to attach a new database
- */
-int32_t ctdb_control_db_attach(struct ctdb_context *ctdb, TDB_DATA indata,
- TDB_DATA *outdata)
+void ctdb_trackingdb_traverse(struct ctdb_context *ctdb, TDB_DATA data, ctdb_trackingdb_cb cb, void *private_data)
{
- const char *db_name = (const char *)indata.dptr;
- struct ctdb_db_context *ctdb_db, *tmp_db;
- int ret;
-
- /* see if we already have this name */
- for (tmp_db=ctdb->db_list;tmp_db;tmp_db=tmp_db->next) {
- if (strcmp(db_name, tmp_db->db_name) == 0) {
- /* this is not an error */
- outdata->dptr = (uint8_t *)&tmp_db->db_id;
- outdata->dsize = sizeof(tmp_db->db_id);
- return 0;
- }
- }
-
- ctdb_db = talloc_zero(ctdb, struct ctdb_db_context);
- CTDB_NO_MEMORY(ctdb, ctdb_db);
+ int i;
- ctdb_db->ctdb = ctdb;
- ctdb_db->db_name = talloc_strdup(ctdb_db, db_name);
- CTDB_NO_MEMORY(ctdb, ctdb_db->db_name);
+ for(i = 0; i < data.dsize; i++) {
+ int j;
- ctdb_db->db_id = ctdb_hash(&indata);
+ for (j=0; j<8; j++) {
+ int mask = 1<<j;
- outdata->dptr = (uint8_t *)&ctdb_db->db_id;
- outdata->dsize = sizeof(ctdb_db->db_id);
-
- /* check for hash collisions */
- for (tmp_db=ctdb->db_list;tmp_db;tmp_db=tmp_db->next) {
- if (tmp_db->db_id == ctdb_db->db_id) {
- DEBUG(0,("db_id 0x%x hash collision. name1='%s' name2='%s'\n",
- db_name, tmp_db->db_name));
- talloc_free(ctdb_db);
- return -1;
+ if (data.dptr[i] & mask) {
+ cb(ctdb, i * 8 + j, private_data);
+ }
}
}
-
- if (ctdb->db_directory == NULL) {
- ctdb->db_directory = VARDIR "/ctdb";
- }
-
- /* make sure the db directory exists */
- if (mkdir(ctdb->db_directory, 0700) == -1 && errno != EEXIST) {
- DEBUG(0,(__location__ " Unable to create ctdb directory '%s'\n",
- ctdb->db_directory));
- talloc_free(ctdb_db);
- return -1;
- }
-
- /* open the database */
- ctdb_db->db_path = talloc_asprintf(ctdb_db, "%s/%s.%u",
- ctdb->db_directory,
- db_name, ctdb->vnn);
-
- ctdb_db->ltdb = tdb_wrap_open(ctdb, ctdb_db->db_path, 0,
- TDB_CLEAR_IF_FIRST, O_CREAT|O_RDWR, 0666);
- if (ctdb_db->ltdb == NULL) {
- DEBUG(0,("Failed to open tdb '%s'\n", ctdb_db->db_path));
- talloc_free(ctdb_db);
- return -1;
- }
-
- ctdb_check_db_empty(ctdb_db);
-
- DLIST_ADD(ctdb->db_list, ctdb_db);
-
- /*
- all databases support the "null" function. we need this in
- order to do forced migration of records
- */
- ret = ctdb_daemon_set_call(ctdb, ctdb_db->db_id, ctdb_null_func, CTDB_NULL_FUNC);
- if (ret != 0) {
- DEBUG(0,("Failed to setup null function for '%s'\n", ctdb_db->db_name));
- talloc_free(ctdb_db);
- return -1;
- }
-
- /*
- all databases support the "fetch" function. we need this
- for efficient Samba3 ctdb fetch
- */
- ret = ctdb_daemon_set_call(ctdb, ctdb_db->db_id, ctdb_fetch_func, CTDB_FETCH_FUNC);
- if (ret != 0) {
- DEBUG(0,("Failed to setup fetch function for '%s'\n", ctdb_db->db_name));
- talloc_free(ctdb_db);
- return -1;
- }
-
- /* tell all the other nodes about this database */
- ctdb_daemon_send_control(ctdb, CTDB_BROADCAST_VNNMAP, 0,
- CTDB_CONTROL_DB_ATTACH, 0, CTDB_CTRL_FLAG_NOREPLY,
- indata, NULL, NULL);
-
- DEBUG(1,("Attached to database '%s'\n", ctdb_db->db_path));
-
- /* success */
- return 0;
}
/*
- called when a broadcast seqnum update comes in
- */
-int32_t ctdb_ltdb_update_seqnum(struct ctdb_context *ctdb, uint32_t db_id, uint32_t srcnode)
+ this is the dummy null procedure that all databases support
+*/
+int ctdb_null_func(struct ctdb_call_info *call)
{
- struct ctdb_db_context *ctdb_db;
- if (srcnode == ctdb->vnn) {
- /* don't update ourselves! */
- return 0;
- }
-
- ctdb_db = find_ctdb_db(ctdb, db_id);
- if (!ctdb_db) {
- DEBUG(0,("Unknown db_id 0x%x in ctdb_ltdb_update_seqnum\n"));
- return -1;
- }
-
- tdb_increment_seqnum_nonblock(ctdb_db->ltdb->tdb);
- ctdb_db->seqnum = tdb_get_seqnum(ctdb_db->ltdb->tdb);
return 0;
}
/*
- timer to check for seqnum changes in a ltdb and propogate them
- */
-static void ctdb_ltdb_seqnum_check(struct event_context *ev, struct timed_event *te,
- struct timeval t, void *p)
+ this is a plain fetch procedure that all databases support
+*/
+int ctdb_fetch_func(struct ctdb_call_info *call)
{
- struct ctdb_db_context *ctdb_db = talloc_get_type(p, struct ctdb_db_context);
- struct ctdb_context *ctdb = ctdb_db->ctdb;
- uint32_t new_seqnum = tdb_get_seqnum(ctdb_db->ltdb->tdb);
- if (new_seqnum != ctdb_db->seqnum) {
- /* something has changed - propogate it */
- TDB_DATA data;
- data.dptr = (uint8_t *)&ctdb_db->db_id;
- data.dsize = sizeof(uint32_t);
- ctdb_daemon_send_control(ctdb, CTDB_BROADCAST_VNNMAP, 0,
- CTDB_CONTROL_UPDATE_SEQNUM, 0, CTDB_CTRL_FLAG_NOREPLY,
- data, NULL, NULL);
- }
- ctdb_db->seqnum = new_seqnum;
-
- /* setup a new timer */
- ctdb_db->te = event_add_timed(ctdb->ev, ctdb_db,
- timeval_current_ofs(ctdb->seqnum_frequency, 0),
- ctdb_ltdb_seqnum_check, ctdb_db);
+ call->reply_data = &call->record_data;
+ return 0;
}
/*
- enable seqnum handling on this db
- */
-int32_t ctdb_ltdb_enable_seqnum(struct ctdb_context *ctdb, uint32_t db_id)
+ this is a plain fetch procedure that all databases support
+ this returns the full record including the ltdb header
+*/
+int ctdb_fetch_with_header_func(struct ctdb_call_info *call)
{
- struct ctdb_db_context *ctdb_db;
- ctdb_db = find_ctdb_db(ctdb, db_id);
- if (!ctdb_db) {
- DEBUG(0,("Unknown db_id 0x%x in ctdb_ltdb_enable_seqnum\n"));
+ call->reply_data = talloc(call, TDB_DATA);
+ if (call->reply_data == NULL) {
return -1;
}
-
- if (ctdb_db->te == NULL) {
- ctdb_db->te = event_add_timed(ctdb->ev, ctdb_db,
- timeval_current_ofs(ctdb->seqnum_frequency, 0),
- ctdb_ltdb_seqnum_check, ctdb_db);
+ call->reply_data->dsize = sizeof(struct ctdb_ltdb_header) + call->record_data.dsize;
+ call->reply_data->dptr = talloc_size(call->reply_data, call->reply_data->dsize);
+ if (call->reply_data->dptr == NULL) {
+ return -1;
}
+ memcpy(call->reply_data->dptr, call->header, sizeof(struct ctdb_ltdb_header));
+ memcpy(&call->reply_data->dptr[sizeof(struct ctdb_ltdb_header)], call->record_data.dptr, call->record_data.dsize);
- tdb_enable_seqnum(ctdb_db->ltdb->tdb);
- ctdb_db->seqnum = tdb_get_seqnum(ctdb_db->ltdb->tdb);
return 0;
}
-/*
- enable seqnum handling on this db
- */
-int32_t ctdb_ltdb_set_seqnum_frequency(struct ctdb_context *ctdb, uint32_t frequency)
-{
- ctdb->seqnum_frequency = frequency;
- return 0;
-}
git.samba.org / vlendec / samba-autobuild / .git / blobdiff