struct ctdb_vacuum_handle {
struct ctdb_db_context *ctdb_db;
struct ctdb_vacuum_child_context *child_ctx;
+ uint32_t fast_path_count;
};
uint32_t total;
uint32_t vacuumed;
uint32_t copied;
+ uint32_t fast_added_to_vacuum_fetch_list;
+ uint32_t fast_added_to_delete_tree;
+ uint32_t fast_deleted;
+ uint32_t fast_skipped;
+ uint32_t fast_error;
+ uint32_t fast_total;
+ uint32_t full_added_to_vacuum_fetch_list;
+ uint32_t full_added_to_delete_tree;
+ uint32_t full_skipped;
+ uint32_t full_error;
+ uint32_t full_total;
};
/* tuning information stored for every db */
struct ctdb_marshall_buffer *records;
};
-
-static int add_record_to_delete_tree(struct vacuum_data *vdata, TDB_DATA key,
- struct ctdb_ltdb_header *hdr)
+/**
+ * Store key and header in a tree, indexed by the key hash.
+ */
+static int insert_delete_record_data_into_tree(struct ctdb_context *ctdb,
+ struct ctdb_db_context *ctdb_db,
+ trbt_tree_t *tree,
+ const struct ctdb_ltdb_header *hdr,
+ TDB_DATA key)
{
- struct ctdb_context *ctdb = vdata->ctdb;
- struct ctdb_db_context *ctdb_db = vdata->ctdb_db;
- uint32_t hash;
struct delete_record_data *dd;
+ uint32_t hash;
- hash = ctdb_hash(&key);
-
- if (trbt_lookup32(vdata->delete_tree, hash)) {
- DEBUG(DEBUG_DEBUG, (__location__ " Hash collission when vacuuming, skipping this record.\n"));
- return 0;
- }
-
- /* store key and header indexed by the key hash */
- dd = talloc_zero(vdata->delete_tree, struct delete_record_data);
+ dd = talloc_zero(tree, struct delete_record_data);
if (dd == NULL) {
DEBUG(DEBUG_ERR,(__location__ " Out of memory\n"));
return -1;
}
+
dd->ctdb = ctdb;
dd->ctdb_db = ctdb_db;
dd->key.dsize = key.dsize;
dd->hdr = *hdr;
- trbt_insert32(vdata->delete_tree, hash, dd);
+ hash = ctdb_hash(&key);
+
+ trbt_insert32(tree, hash, dd);
+
+ return 0;
+}
+
+static int add_record_to_delete_tree(struct vacuum_data *vdata, TDB_DATA key,
+ struct ctdb_ltdb_header *hdr)
+{
+ struct ctdb_context *ctdb = vdata->ctdb;
+ struct ctdb_db_context *ctdb_db = vdata->ctdb_db;
+ uint32_t hash;
+ int ret;
+
+ hash = ctdb_hash(&key);
+
+ if (trbt_lookup32(vdata->delete_tree, hash)) {
+ DEBUG(DEBUG_INFO, (__location__ " Hash collission when vacuuming, skipping this record.\n"));
+ return 0;
+ }
+
+ ret = insert_delete_record_data_into_tree(ctdb, ctdb_db,
+ vdata->delete_tree,
+ hdr, key);
+ if (ret != 0) {
+ return -1;
+ }
vdata->delete_count++;
struct ctdb_ltdb_header *hdr;
int res = 0;
+ vdata->full_total++;
+
lmaster = ctdb_lmaster(ctdb, &key);
if (lmaster >= ctdb->num_nodes) {
+ vdata->full_error++;
DEBUG(DEBUG_CRIT, (__location__
" lmaster[%u] >= ctdb->num_nodes[%u] for key"
" with hash[%u]!\n",
if (data.dsize != sizeof(struct ctdb_ltdb_header)) {
/* its not a deleted record */
+ vdata->full_skipped++;
return 0;
}
hdr = (struct ctdb_ltdb_header *)data.dptr;
if (hdr->dmaster != ctdb->pnn) {
+ vdata->full_skipped++;
return 0;
}
* So we should be able to delete it.
*/
res = add_record_to_delete_tree(vdata, key, hdr);
+ if (res != 0) {
+ vdata->full_error++;
+ } else {
+ vdata->full_added_to_delete_tree++;
+ }
} else {
/*
* We are not lmaster.
* Add the record to the blob ready to send to the nodes.
*/
res = add_record_to_vacuum_fetch_list(vdata, key);
+ if (res != 0) {
+ vdata->full_error++;
+ } else {
+ vdata->full_added_to_vacuum_fetch_list++;
+ }
}
return res;
memcpy(old_size+(uint8_t *)(recs->records), rec, rec->length);
}
+/**
+ * traverse function for the traversal of the delete_queue,
+ * the fast-path vacuuming list.
+ *
+ * - If the record has been migrated off the node
+ * or has been revived (filled with data) on the node,
+ * then skip the record.
+ *
+ * - If the current node is the record's lmaster and it is
+ * a record that has never been migrated with data, then
+ * delete the record from the local tdb.
+ *
+ * - If the current node is the record's lmaster and it has
+ * been migrated with data, then schedule it for the normal
+ * vacuuming procedure (i.e. add it to the delete_list).
+ *
+ * - If the current node is NOT the record's lmaster then
+ * add it to the list of records that are to be sent to
+ * the lmaster with the VACUUM_FETCH message.
+ */
+static void delete_queue_traverse(void *param, void *data)
+{
+ struct delete_record_data *dd =
+ talloc_get_type(data, struct delete_record_data);
+ struct vacuum_data *vdata = talloc_get_type(param, struct vacuum_data);
+ struct ctdb_db_context *ctdb_db = dd->ctdb_db;
+ struct ctdb_context *ctdb = ctdb_db->ctdb; /* or dd->ctdb ??? */
+ int res;
+ struct ctdb_ltdb_header *header;
+ TDB_DATA tdb_data;
+ uint32_t lmaster;
+
+ vdata->fast_total++;
+
+ res = tdb_chainlock(ctdb_db->ltdb->tdb, dd->key);
+ if (res != 0) {
+ DEBUG(DEBUG_ERR, (__location__ " Error getting chainlock.\n"));
+ vdata->fast_error++;
+ return;
+ }
+
+ tdb_data = tdb_fetch(ctdb_db->ltdb->tdb, dd->key);
+ if (tdb_data.dsize < sizeof(struct ctdb_ltdb_header)) {
+ /* Does not exist or not a ctdb record. Skip. */
+ goto skipped;
+ }
+
+ if (tdb_data.dsize > sizeof(struct ctdb_ltdb_header)) {
+ /* The record has been recycled (filled with data). Skip. */
+ goto skipped;
+ }
+
+ header = (struct ctdb_ltdb_header *)tdb_data.dptr;
+
+ if (header->dmaster != ctdb->pnn) {
+ /* The record has been migrated off the node. Skip. */
+ goto skipped;
+ }
+
+
+ if (header->rsn != dd->hdr.rsn) {
+ /*
+ * The record has been migrated off the node and back again.
+ * But not requeued for deletion. Skip it.
+ */
+ goto skipped;
+ }
+
+ /*
+ * We are dmaster, and the record has no data, and it has
+ * not been migrated after it has been queued for deletion.
+ *
+ * At this stage, the record could still have been revived locally
+ * and last been written with empty data. This can only be
+ * fixed with the addition of an active or delete flag. (TODO)
+ */
+
+ lmaster = ctdb_lmaster(ctdb_db->ctdb, &dd->key);
+
+ if (lmaster != ctdb->pnn) {
+ res = add_record_to_vacuum_fetch_list(vdata, dd->key);
+
+ if (res != 0) {
+ DEBUG(DEBUG_ERR,
+ (__location__ " Error adding record to list "
+ "of records to send to lmaster.\n"));
+ vdata->fast_error++;
+ } else {
+ vdata->fast_added_to_vacuum_fetch_list++;
+ }
+ goto done;
+ }
+
+ /* use header->flags or dd->hdr.flags ?? */
+ if (dd->hdr.flags & CTDB_REC_FLAG_MIGRATED_WITH_DATA) {
+ res = add_record_to_delete_tree(vdata, dd->key, &dd->hdr);
+
+ if (res != 0) {
+ DEBUG(DEBUG_ERR,
+ (__location__ " Error adding record to list "
+ "of records for deletion on lmaster.\n"));
+ vdata->fast_error++;
+ } else {
+ vdata->fast_added_to_delete_tree++;
+ }
+ } else {
+ res = tdb_delete(ctdb_db->ltdb->tdb, dd->key);
+
+ if (res != 0) {
+ DEBUG(DEBUG_ERR,
+ (__location__ " Error deleting record from local "
+ "data base.\n"));
+ vdata->fast_error++;
+ } else {
+ vdata->fast_deleted++;
+ }
+ }
+
+ goto done;
+
+skipped:
+ vdata->fast_skipped++;
+
+done:
+ if (tdb_data.dptr != NULL) {
+ free(tdb_data.dptr);
+ }
+ tdb_chainunlock(ctdb_db->ltdb->tdb, dd->key);
+
+ return;
+}
+
/*
* read-only traverse the database in order to find
* records that can be deleted and try to delete these
* records on the other nodes
* this executes in the child context
*/
-static int ctdb_vacuum_db(struct ctdb_db_context *ctdb_db, struct vacuum_data *vdata)
+static int ctdb_vacuum_db(struct ctdb_db_context *ctdb_db,
+ struct vacuum_data *vdata,
+ bool full_vacuum_run)
{
struct ctdb_context *ctdb = ctdb_db->ctdb;
const char *name = ctdb_db->db_name;
int ret, i, pnn;
+ DEBUG(DEBUG_INFO, (__location__ " Entering %s vacuum run for db "
+ "%s db_id[0x%08x]\n",
+ full_vacuum_run ? "full" : "fast",
+ ctdb_db->db_name, ctdb_db->db_id));
+
ret = ctdb_ctrl_getvnnmap(ctdb, TIMELIMIT(), CTDB_CURRENT_NODE, ctdb, &ctdb->vnn_map);
if (ret != 0) {
DEBUG(DEBUG_ERR, ("Unable to get vnnmap from local node\n"));
}
ctdb->pnn = pnn;
+
+ vdata->fast_added_to_delete_tree = 0;
+ vdata->fast_added_to_vacuum_fetch_list = 0;
+ vdata->fast_deleted = 0;
+ vdata->fast_skipped = 0;
+ vdata->fast_error = 0;
+ vdata->fast_total = 0;
+ vdata->full_added_to_delete_tree = 0;
+ vdata->full_added_to_vacuum_fetch_list = 0;
+ vdata->full_skipped = 0;
+ vdata->full_error = 0;
+ vdata->full_total = 0;
+
/* the list needs to be of length num_nodes */
vdata->list = talloc_array(vdata, struct ctdb_marshall_buffer *, ctdb->num_nodes);
if (vdata->list == NULL) {
vdata->list[i]->db_id = ctdb_db->db_id;
}
- /* read-only traverse, looking for records that might be able to be vacuumed */
- if (tdb_traverse_read(ctdb_db->ltdb->tdb, vacuum_traverse, vdata) == -1 ||
- vdata->traverse_error) {
- DEBUG(DEBUG_ERR,(__location__ " Traverse error in vacuuming '%s'\n", name));
- return -1;
+ /*
+ * Traverse the delete_queue.
+ * This builds the same lists as the db traverse.
+ */
+ trbt_traversearray32(ctdb_db->delete_queue, 1, delete_queue_traverse, vdata);
+
+ if (vdata->fast_total > 0) {
+ DEBUG(DEBUG_INFO,
+ (__location__
+ " fast vacuuming delete_queue traverse statistics: "
+ "db[%s] "
+ "total[%u] "
+ "del[%u] "
+ "skp[%u] "
+ "err[%u] "
+ "adt[%u] "
+ "avf[%u]\n",
+ ctdb_db->db_name,
+ (unsigned)vdata->fast_total,
+ (unsigned)vdata->fast_deleted,
+ (unsigned)vdata->fast_skipped,
+ (unsigned)vdata->fast_error,
+ (unsigned)vdata->fast_added_to_delete_tree,
+ (unsigned)vdata->fast_added_to_vacuum_fetch_list));
+ }
+
+ /*
+ * read-only traverse of the database, looking for records that
+ * might be able to be vacuumed.
+ *
+ * This is not done each time but only every tunable
+ * VacuumFastPathCount times.
+ */
+ if (full_vacuum_run) {
+ ret = tdb_traverse_read(ctdb_db->ltdb->tdb, vacuum_traverse, vdata);
+ if (ret == -1 || vdata->traverse_error) {
+ DEBUG(DEBUG_ERR,(__location__ " Traverse error in vacuuming '%s'\n", name));
+ return -1;
+ }
+ if (vdata->full_total > 0) {
+ DEBUG(DEBUG_INFO,
+ (__location__
+ " full vacuuming db traverse statistics: "
+ "db[%s] "
+ "total[%u] "
+ "skp[%u] "
+ "err[%u] "
+ "adt[%u] "
+ "avf[%u]\n",
+ ctdb_db->db_name,
+ (unsigned)vdata->full_total,
+ (unsigned)vdata->full_skipped,
+ (unsigned)vdata->full_error,
+ (unsigned)vdata->full_added_to_delete_tree,
+ (unsigned)vdata->full_added_to_vacuum_fetch_list));
+ }
}
/*
* called from the child context
*/
static int ctdb_vacuum_and_repack_db(struct ctdb_db_context *ctdb_db,
- TALLOC_CTX *mem_ctx)
+ TALLOC_CTX *mem_ctx,
+ bool full_vacuum_run)
{
uint32_t repack_limit = ctdb_db->ctdb->tunable.repack_limit;
uint32_t vacuum_limit = ctdb_db->ctdb->tunable.vacuum_limit;
/*
* gather all records that can be deleted in vdata
*/
- if (ctdb_vacuum_db(ctdb_db, vdata) != 0) {
+ if (ctdb_vacuum_db(ctdb_db, vdata, full_vacuum_run) != 0) {
DEBUG(DEBUG_ERR,(__location__ " Failed to vacuum '%s'\n", name));
}
if (child_ctx->child_pid != -1) {
kill(child_ctx->child_pid, SIGKILL);
+ } else {
+ /* Bump the number of successful fast-path runs. */
+ child_ctx->vacuum_handle->fast_path_count++;
}
DLIST_REMOVE(ctdb->vacuumers, child_ctx);
return;
}
+ if (vacuum_handle->fast_path_count > ctdb->tunable.vacuum_fast_path_count) {
+ vacuum_handle->fast_path_count = 0;
+ }
+
child_ctx->child_pid = ctdb_fork(ctdb);
if (child_ctx->child_pid == (pid_t)-1) {
close(child_ctx->fd[0]);
if (child_ctx->child_pid == 0) {
char cc = 0;
+ bool full_vacuum_run = false;
close(child_ctx->fd[0]);
DEBUG(DEBUG_INFO,("Vacuuming child process %d for db %s started\n", getpid(), ctdb_db->db_name));
/*
* repack the db
*/
- cc = ctdb_vacuum_and_repack_db(ctdb_db, child_ctx);
+ if ((ctdb->tunable.vacuum_fast_path_count > 0) &&
+ (vacuum_handle->fast_path_count == 0))
+ {
+ full_vacuum_run = true;
+ }
+ cc = ctdb_vacuum_and_repack_db(ctdb_db, child_ctx,
+ full_vacuum_run);
write(child_ctx->fd[1], &cc, 1);
_exit(0);
return 0;
}
+
+/**
+ * Insert a record into the ctdb_db context's delete queue,
+ * handling hash collisions.
+ */
+static int insert_record_into_delete_queue(struct ctdb_db_context *ctdb_db,
+ const struct ctdb_ltdb_header *hdr,
+ TDB_DATA key)
+{
+ struct delete_record_data *kd;
+ uint32_t hash;
+ int ret;
+
+ hash = (uint32_t)ctdb_hash(&key);
+
+ DEBUG(DEBUG_INFO, (__location__ " Schedule for deletion: db[%s] "
+ "db_id[0x%08x] "
+ "key_hash[0x%08x] "
+ "lmaster[%u] "
+ "migrated_with_data[%s]\n",
+ ctdb_db->db_name, ctdb_db->db_id,
+ hash,
+ ctdb_lmaster(ctdb_db->ctdb, &key),
+ hdr->flags & CTDB_REC_FLAG_MIGRATED_WITH_DATA ? "yes" : "no"));
+
+ kd = (struct delete_record_data *)trbt_lookup32(ctdb_db->delete_queue, hash);
+ if (kd != NULL) {
+ if ((kd->key.dsize != key.dsize) ||
+ (memcmp(kd->key.dptr, key.dptr, key.dsize) != 0))
+ {
+ DEBUG(DEBUG_INFO,
+ ("schedule for deletion: Hash collision (0x%08x)."
+ " Skipping the record.\n", hash));
+ return 0;
+ } else {
+ DEBUG(DEBUG_DEBUG,
+ ("schedule for deletetion: Overwriting entry for "
+ "key with hash 0x%08x.\n", hash));
+ }
+ }
+
+ ret = insert_delete_record_data_into_tree(ctdb_db->ctdb, ctdb_db,
+ ctdb_db->delete_queue,
+ hdr, key);
+ if (ret != 0) {
+ return -1;
+ }
+
+ return 0;
+}
+
+/**
+ * Schedule a record for deletetion.
+ * Called from the parent context.
+ */
+int32_t ctdb_control_schedule_for_deletion(struct ctdb_context *ctdb,
+ TDB_DATA indata)
+{
+ struct ctdb_control_schedule_for_deletion *dd;
+ struct ctdb_db_context *ctdb_db;
+ int ret;
+ TDB_DATA key;
+
+ dd = (struct ctdb_control_schedule_for_deletion *)indata.dptr;
+
+ ctdb_db = find_ctdb_db(ctdb, dd->db_id);
+ if (ctdb_db == NULL) {
+ DEBUG(DEBUG_ERR, (__location__ " Unknown db id 0x%08x\n",
+ dd->db_id));
+ return -1;
+ }
+
+ key.dsize = dd->keylen;
+ key.dptr = dd->key;
+
+ ret = insert_record_into_delete_queue(ctdb_db, &dd->hdr, key);
+
+ return ret;
+}
+
+int32_t ctdb_local_schedule_for_deletion(struct ctdb_db_context *ctdb_db,
+ const struct ctdb_ltdb_header *hdr,
+ TDB_DATA key)
+{
+ int ret;
+ struct ctdb_control_schedule_for_deletion *dd;
+ TDB_DATA indata;
+ int32_t status;
+
+ if (ctdb_db->ctdb->ctdbd_pid == getpid()) {
+ /* main daemon - directly queue */
+ ret = insert_delete_record_data_into_tree(ctdb_db->ctdb,
+ ctdb_db,
+ ctdb_db->delete_queue,
+ hdr, key);
+ if (ret != 0) {
+ return -1;
+ }
+ return 0;
+ }
+
+ /* child process: send the main daemon a control */
+
+ indata.dsize = offsetof(struct ctdb_control_schedule_for_deletion, key) + key.dsize;
+ indata.dptr = talloc_zero_array(ctdb_db, uint8_t, indata.dsize);
+ if (indata.dptr == NULL) {
+ DEBUG(DEBUG_ERR, (__location__ " out of memory\n"));
+ return -1;
+ }
+ dd = (struct ctdb_control_schedule_for_deletion *)(void *)indata.dptr;
+ dd->db_id = ctdb_db->db_id;
+ dd->hdr = *hdr;
+ dd->keylen = key.dsize;
+ memcpy(dd->key, key.dptr, key.dsize);
+
+ ret = ctdb_control(ctdb_db->ctdb,
+ CTDB_CURRENT_NODE,
+ ctdb_db->db_id,
+ CTDB_CONTROL_SCHEDULE_FOR_DELETION,
+ CTDB_CTRL_FLAG_NOREPLY, /* flags */
+ indata,
+ NULL, /* mem_ctx */
+ NULL, /* outdata */
+ &status,
+ NULL, /* timeout : NULL == wait forever */
+ NULL); /* error message */
+
+ talloc_free(indata.dptr);
+
+ if (ret != 0 || status != 0) {
+ DEBUG(DEBUG_ERR, (__location__ " Error sending "
+ "SCHEDULE_FOR_DELETION "
+ "control.\n"));
+ if (status != 0) {
+ ret = -1;
+ }
+ }
+
+ return ret;
+}
git.samba.org / ctdb.git / blobdiff