summaryrefslogtreecommitdiffstats
path: root/source3/lib/dbwrap/dbwrap_watch.c
diff options
context:
space:
mode:
Diffstat (limited to 'source3/lib/dbwrap/dbwrap_watch.c')
-rw-r--r--source3/lib/dbwrap/dbwrap_watch.c1285
1 files changed, 1285 insertions, 0 deletions
diff --git a/source3/lib/dbwrap/dbwrap_watch.c b/source3/lib/dbwrap/dbwrap_watch.c
new file mode 100644
index 0000000..df93119
--- /dev/null
+++ b/source3/lib/dbwrap/dbwrap_watch.c
@@ -0,0 +1,1285 @@
+/*
+ Unix SMB/CIFS implementation.
+ Watch dbwrap record changes
+ Copyright (C) Volker Lendecke 2012
+
+ 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 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 "system/filesys.h"
+#include "lib/util/server_id.h"
+#include "dbwrap/dbwrap.h"
+#include "dbwrap_watch.h"
+#include "dbwrap_open.h"
+#include "lib/util/util_tdb.h"
+#include "lib/util/tevent_ntstatus.h"
+#include "serverid.h"
+#include "server_id_watch.h"
+#include "lib/dbwrap/dbwrap_private.h"
+
+struct dbwrap_watcher {
+ /*
+ * Process watching this record
+ */
+ struct server_id pid;
+ /*
+ * Individual instance inside the waiter, incremented each
+ * time a watcher is created
+ */
+ uint64_t instance;
+};
+
+#define DBWRAP_WATCHER_BUF_LENGTH (SERVER_ID_BUF_LENGTH + sizeof(uint64_t))
+#define DBWRAP_MAX_WATCHERS (INT32_MAX/DBWRAP_WATCHER_BUF_LENGTH)
+
+/*
+ * Watched records contain a header of:
+ *
+ * [uint32] num_records
+ * 0 [DBWRAP_WATCHER_BUF_LENGTH] \
+ * 1 [DBWRAP_WATCHER_BUF_LENGTH] |
+ * .. |- Array of watchers
+ * (num_records-1)[DBWRAP_WATCHER_BUF_LENGTH] /
+ *
+ * [Remainder of record....]
+ *
+ * If this header is absent then this is a
+ * fresh record of length zero (no watchers).
+ */
+
+static bool dbwrap_watch_rec_parse(
+ TDB_DATA data,
+ uint8_t **pwatchers,
+ size_t *pnum_watchers,
+ TDB_DATA *pdata)
+{
+ size_t num_watchers;
+
+ if (data.dsize == 0) {
+ /* Fresh record */
+ if (pwatchers != NULL) {
+ *pwatchers = NULL;
+ }
+ if (pnum_watchers != NULL) {
+ *pnum_watchers = 0;
+ }
+ if (pdata != NULL) {
+ *pdata = (TDB_DATA) { .dptr = NULL };
+ }
+ return true;
+ }
+
+ if (data.dsize < sizeof(uint32_t)) {
+ /* Invalid record */
+ return false;
+ }
+
+ num_watchers = IVAL(data.dptr, 0);
+
+ data.dptr += sizeof(uint32_t);
+ data.dsize -= sizeof(uint32_t);
+
+ if (num_watchers > data.dsize/DBWRAP_WATCHER_BUF_LENGTH) {
+ /* Invalid record */
+ return false;
+ }
+
+ if (pwatchers != NULL) {
+ *pwatchers = data.dptr;
+ }
+ if (pnum_watchers != NULL) {
+ *pnum_watchers = num_watchers;
+ }
+ if (pdata != NULL) {
+ size_t watchers_len = num_watchers * DBWRAP_WATCHER_BUF_LENGTH;
+ *pdata = (TDB_DATA) {
+ .dptr = data.dptr + watchers_len,
+ .dsize = data.dsize - watchers_len
+ };
+ }
+
+ return true;
+}
+
+static void dbwrap_watcher_get(struct dbwrap_watcher *w,
+ const uint8_t buf[DBWRAP_WATCHER_BUF_LENGTH])
+{
+ server_id_get(&w->pid, buf);
+ w->instance = BVAL(buf, SERVER_ID_BUF_LENGTH);
+}
+
+static void dbwrap_watcher_put(uint8_t buf[DBWRAP_WATCHER_BUF_LENGTH],
+ const struct dbwrap_watcher *w)
+{
+ server_id_put(buf, w->pid);
+ SBVAL(buf, SERVER_ID_BUF_LENGTH, w->instance);
+}
+
+static void dbwrap_watch_log_invalid_record(
+ struct db_context *db, TDB_DATA key, TDB_DATA value)
+{
+ DBG_ERR("Found invalid record in %s\n", dbwrap_name(db));
+ dump_data(1, key.dptr, key.dsize);
+ dump_data(1, value.dptr, value.dsize);
+}
+
+struct db_watched_ctx {
+ struct db_context *backend;
+ struct messaging_context *msg;
+};
+
+struct db_watched_record {
+ struct db_record *rec;
+ struct server_id self;
+ struct {
+ struct db_record *rec;
+ TDB_DATA initial_value;
+ bool initial_valid;
+ } backend;
+ bool force_fini_store;
+ struct dbwrap_watcher added;
+ bool removed_first;
+ struct {
+ /*
+ * The is the number of watcher records
+ * parsed from backend.initial_value
+ */
+ size_t count;
+ /*
+ * This is the pointer to
+ * the optentially first watcher record
+ * parsed from backend.initial_value
+ *
+ * The pointer actually points to memory
+ * in backend.initial_value.
+ *
+ * Note it might be NULL, if count is 0.
+ */
+ uint8_t *first;
+ /*
+ * This remembers if we already
+ * notified the watchers.
+ *
+ * As we only need to do that once during:
+ * do_locked
+ * or:
+ * between rec = fetch_locked
+ * and
+ * TALLOC_FREE(rec)
+ */
+ bool alerted;
+ } watchers;
+ struct {
+ struct dbwrap_watcher watcher;
+ } wakeup;
+};
+
+static struct db_watched_record *db_record_get_watched_record(struct db_record *rec)
+{
+ /*
+ * we can't use wrec = talloc_get_type_abort() here!
+ * because wrec is likely a stack variable in
+ * dbwrap_watched_do_locked_fn()
+ *
+ * In order to have a least some protection
+ * we verify the cross reference pointers
+ * between rec and wrec
+ */
+ struct db_watched_record *wrec =
+ (struct db_watched_record *)rec->private_data;
+ SMB_ASSERT(wrec->rec == rec);
+ return wrec;
+}
+
+static NTSTATUS dbwrap_watched_record_storev(
+ struct db_watched_record *wrec,
+ const TDB_DATA *dbufs, int num_dbufs, int flags);
+static NTSTATUS dbwrap_watched_storev(struct db_record *rec,
+ const TDB_DATA *dbufs, int num_dbufs,
+ int flags);
+static NTSTATUS dbwrap_watched_delete(struct db_record *rec);
+static void dbwrap_watched_trigger_wakeup(struct messaging_context *msg_ctx,
+ struct dbwrap_watcher *watcher);
+static int db_watched_record_destructor(struct db_watched_record *wrec);
+
+static void db_watched_record_init(struct db_context *db,
+ struct messaging_context *msg_ctx,
+ struct db_record *rec,
+ struct db_watched_record *wrec,
+ struct db_record *backend_rec,
+ TDB_DATA backend_value)
+{
+ bool ok;
+
+ *rec = (struct db_record) {
+ .db = db,
+ .key = dbwrap_record_get_key(backend_rec),
+ .storev = dbwrap_watched_storev,
+ .delete_rec = dbwrap_watched_delete,
+ .private_data = wrec,
+ };
+
+ *wrec = (struct db_watched_record) {
+ .rec = rec,
+ .self = messaging_server_id(msg_ctx),
+ .backend = {
+ .rec = backend_rec,
+ .initial_value = backend_value,
+ .initial_valid = true,
+ },
+ };
+
+ ok = dbwrap_watch_rec_parse(backend_value,
+ &wrec->watchers.first,
+ &wrec->watchers.count,
+ &rec->value);
+ if (!ok) {
+ dbwrap_watch_log_invalid_record(rec->db, rec->key, backend_value);
+ /* wipe invalid data */
+ rec->value = (TDB_DATA) { .dptr = NULL, .dsize = 0 };
+ }
+}
+
+static struct db_record *dbwrap_watched_fetch_locked(
+ struct db_context *db, TALLOC_CTX *mem_ctx, TDB_DATA key)
+{
+ struct db_watched_ctx *ctx = talloc_get_type_abort(
+ db->private_data, struct db_watched_ctx);
+ struct db_record *rec = NULL;
+ struct db_watched_record *wrec = NULL;
+ struct db_record *backend_rec = NULL;
+ TDB_DATA backend_value = { .dptr = NULL, };
+
+ rec = talloc_zero(mem_ctx, struct db_record);
+ if (rec == NULL) {
+ return NULL;
+ }
+ wrec = talloc_zero(rec, struct db_watched_record);
+ if (wrec == NULL) {
+ TALLOC_FREE(rec);
+ return NULL;
+ }
+
+ backend_rec = dbwrap_fetch_locked(ctx->backend, wrec, key);
+ if (backend_rec == NULL) {
+ TALLOC_FREE(rec);
+ return NULL;
+ }
+ backend_value = dbwrap_record_get_value(backend_rec);
+
+ db_watched_record_init(db, ctx->msg,
+ rec, wrec,
+ backend_rec, backend_value);
+ rec->value_valid = true;
+ talloc_set_destructor(wrec, db_watched_record_destructor);
+
+ return rec;
+}
+
+struct db_watched_record_fini_state {
+ struct db_watched_record *wrec;
+ TALLOC_CTX *frame;
+ TDB_DATA dbufs[2];
+ int num_dbufs;
+ bool ok;
+};
+
+static void db_watched_record_fini_fetcher(TDB_DATA key,
+ TDB_DATA backend_value,
+ void *private_data)
+{
+ struct db_watched_record_fini_state *state =
+ (struct db_watched_record_fini_state *)private_data;
+ struct db_watched_record *wrec = state->wrec;
+ struct db_record *rec = wrec->rec;
+ TDB_DATA value = {};
+ bool ok;
+ size_t copy_size;
+
+ /*
+ * We're within dbwrap_parse_record()
+ * and backend_value directly points into
+ * the mmap'ed tdb, so we need to copy the
+ * parts we require.
+ */
+
+ ok = dbwrap_watch_rec_parse(backend_value, NULL, NULL, &value);
+ if (!ok) {
+ struct db_context *db = dbwrap_record_get_db(rec);
+
+ dbwrap_watch_log_invalid_record(db, key, backend_value);
+
+ /* wipe invalid data */
+ value = (TDB_DATA) { .dptr = NULL, .dsize = 0 };
+ }
+
+ copy_size = MIN(rec->value.dsize, value.dsize);
+ if (copy_size != 0) {
+ /*
+ * First reuse the buffer we already had
+ * as much as we can.
+ */
+ memcpy(rec->value.dptr, value.dptr, copy_size);
+ state->dbufs[state->num_dbufs++] = rec->value;
+ value.dsize -= copy_size;
+ value.dptr += copy_size;
+ }
+
+ if (value.dsize != 0) {
+ uint8_t *p = NULL;
+
+ /*
+ * There's still new data left
+ * allocate it on callers stackframe
+ */
+ p = talloc_memdup(state->frame, value.dptr, value.dsize);
+ if (p == NULL) {
+ DBG_WARNING("failed to allocate %zu bytes\n",
+ value.dsize);
+ return;
+ }
+
+ state->dbufs[state->num_dbufs++] = (TDB_DATA) {
+ .dptr = p, .dsize = value.dsize,
+ };
+ }
+
+ state->ok = true;
+}
+
+static void db_watched_record_fini(struct db_watched_record *wrec)
+{
+ struct db_watched_record_fini_state state = { .wrec = wrec, };
+ struct db_context *backend = dbwrap_record_get_db(wrec->backend.rec);
+ struct db_record *rec = wrec->rec;
+ TDB_DATA key = dbwrap_record_get_key(wrec->backend.rec);
+ NTSTATUS status;
+
+ if (!wrec->force_fini_store) {
+ return;
+ }
+
+ if (wrec->backend.initial_valid) {
+ if (rec->value.dsize != 0) {
+ state.dbufs[state.num_dbufs++] = rec->value;
+ }
+ } else {
+ /*
+ * We need to fetch the current
+ * value from the backend again,
+ * which may need to allocate memory
+ * on the provided stackframe.
+ */
+
+ state.frame = talloc_stackframe();
+
+ status = dbwrap_parse_record(backend, key,
+ db_watched_record_fini_fetcher, &state);
+ if (!NT_STATUS_IS_OK(status)) {
+ DBG_WARNING("dbwrap_parse_record failed: %s\n",
+ nt_errstr(status));
+ TALLOC_FREE(state.frame);
+ return;
+ }
+ if (!state.ok) {
+ TALLOC_FREE(state.frame);
+ return;
+ }
+ }
+
+ /*
+ * We don't want to wake up others just because
+ * we added ourself as new watcher. But if we
+ * removed outself from the first position
+ * we need to alert the next one.
+ */
+ if (!wrec->removed_first) {
+ dbwrap_watched_watch_skip_alerting(rec);
+ }
+
+ status = dbwrap_watched_record_storev(wrec, state.dbufs, state.num_dbufs, 0);
+ TALLOC_FREE(state.frame);
+ if (!NT_STATUS_IS_OK(status)) {
+ DBG_WARNING("dbwrap_watched_record_storev failed: %s\n",
+ nt_errstr(status));
+ return;
+ }
+
+ return;
+}
+
+static int db_watched_record_destructor(struct db_watched_record *wrec)
+{
+ struct db_record *rec = wrec->rec;
+ struct db_watched_ctx *ctx = talloc_get_type_abort(
+ rec->db->private_data, struct db_watched_ctx);
+
+ db_watched_record_fini(wrec);
+ TALLOC_FREE(wrec->backend.rec);
+ dbwrap_watched_trigger_wakeup(ctx->msg, &wrec->wakeup.watcher);
+ return 0;
+}
+
+struct dbwrap_watched_do_locked_state {
+ struct db_context *db;
+ struct messaging_context *msg_ctx;
+ struct db_watched_record *wrec;
+ struct db_record *rec;
+ void (*fn)(struct db_record *rec,
+ TDB_DATA value,
+ void *private_data);
+ void *private_data;
+};
+
+static void dbwrap_watched_do_locked_fn(
+ struct db_record *backend_rec,
+ TDB_DATA backend_value,
+ void *private_data)
+{
+ struct dbwrap_watched_do_locked_state *state =
+ (struct dbwrap_watched_do_locked_state *)private_data;
+
+ db_watched_record_init(state->db, state->msg_ctx,
+ state->rec, state->wrec,
+ backend_rec, backend_value);
+
+ state->fn(state->rec, state->rec->value, state->private_data);
+
+ db_watched_record_fini(state->wrec);
+}
+
+static NTSTATUS dbwrap_watched_do_locked(struct db_context *db, TDB_DATA key,
+ void (*fn)(struct db_record *rec,
+ TDB_DATA value,
+ void *private_data),
+ void *private_data)
+{
+ struct db_watched_ctx *ctx = talloc_get_type_abort(
+ db->private_data, struct db_watched_ctx);
+ struct db_watched_record wrec;
+ struct db_record rec;
+ struct dbwrap_watched_do_locked_state state = {
+ .db = db, .msg_ctx = ctx->msg,
+ .rec = &rec, .wrec = &wrec,
+ .fn = fn, .private_data = private_data,
+ };
+ NTSTATUS status;
+
+ status = dbwrap_do_locked(
+ ctx->backend, key, dbwrap_watched_do_locked_fn, &state);
+ if (!NT_STATUS_IS_OK(status)) {
+ DBG_DEBUG("dbwrap_do_locked returned %s\n", nt_errstr(status));
+ return status;
+ }
+
+ DBG_DEBUG("dbwrap_watched_do_locked_fn returned\n");
+
+ dbwrap_watched_trigger_wakeup(state.msg_ctx, &wrec.wakeup.watcher);
+
+ return NT_STATUS_OK;
+}
+
+static void dbwrap_watched_record_prepare_wakeup(
+ struct db_watched_record *wrec)
+{
+ /*
+ * Wakeup only needs to happen once (if at all)
+ */
+ if (wrec->watchers.alerted) {
+ /* already done */
+ return;
+ }
+ wrec->watchers.alerted = true;
+
+ if (wrec->watchers.count == 0) {
+ DBG_DEBUG("No watchers\n");
+ return;
+ }
+
+ while (wrec->watchers.count != 0) {
+ struct server_id_buf tmp;
+ bool exists;
+
+ dbwrap_watcher_get(&wrec->wakeup.watcher, wrec->watchers.first);
+ exists = serverid_exists(&wrec->wakeup.watcher.pid);
+ if (!exists) {
+ DBG_DEBUG("Discard non-existing waiter %s:%"PRIu64"\n",
+ server_id_str_buf(wrec->wakeup.watcher.pid, &tmp),
+ wrec->wakeup.watcher.instance);
+ wrec->watchers.first += DBWRAP_WATCHER_BUF_LENGTH;
+ wrec->watchers.count -= 1;
+ continue;
+ }
+
+ /*
+ * We will only wakeup the first waiter, via
+ * dbwrap_watched_trigger_wakeup(), but keep
+ * all (including the first one) in the list that
+ * will be flushed back to the backend record
+ * again. Waiters are removing their entries
+ * via dbwrap_watched_watch_remove_instance()
+ * when they no longer want to monitor the record.
+ */
+ DBG_DEBUG("Will alert first waiter %s:%"PRIu64"\n",
+ server_id_str_buf(wrec->wakeup.watcher.pid, &tmp),
+ wrec->wakeup.watcher.instance);
+ break;
+ }
+}
+
+static void dbwrap_watched_trigger_wakeup(struct messaging_context *msg_ctx,
+ struct dbwrap_watcher *watcher)
+{
+ struct server_id_buf tmp;
+ uint8_t instance_buf[8];
+ NTSTATUS status;
+
+ if (watcher->instance == 0) {
+ DBG_DEBUG("No one to wakeup\n");
+ return;
+ }
+
+ DBG_DEBUG("Alerting %s:%"PRIu64"\n",
+ server_id_str_buf(watcher->pid, &tmp),
+ watcher->instance);
+
+ SBVAL(instance_buf, 0, watcher->instance);
+
+ status = messaging_send_buf(
+ msg_ctx,
+ watcher->pid,
+ MSG_DBWRAP_MODIFIED,
+ instance_buf,
+ sizeof(instance_buf));
+ if (!NT_STATUS_IS_OK(status)) {
+ DBG_WARNING("messaging_send_buf to %s failed: %s - ignoring...\n",
+ server_id_str_buf(watcher->pid, &tmp),
+ nt_errstr(status));
+ }
+}
+
+static NTSTATUS dbwrap_watched_record_storev(
+ struct db_watched_record *wrec,
+ const TDB_DATA *dbufs, int num_dbufs, int flags)
+{
+ uint8_t num_watchers_buf[4] = { 0 };
+ uint8_t add_buf[DBWRAP_WATCHER_BUF_LENGTH];
+ size_t num_store_watchers;
+ TDB_DATA my_dbufs[num_dbufs+3];
+ int num_my_dbufs = 0;
+ NTSTATUS status;
+ size_t add_count = 0;
+
+ dbwrap_watched_record_prepare_wakeup(wrec);
+
+ wrec->backend.initial_valid = false;
+ wrec->force_fini_store = false;
+
+ if (wrec->added.pid.pid != 0) {
+ dbwrap_watcher_put(add_buf, &wrec->added);
+ add_count = 1;
+ }
+
+ num_store_watchers = wrec->watchers.count + add_count;
+ if (num_store_watchers == 0 && num_dbufs == 0) {
+ status = dbwrap_record_delete(wrec->backend.rec);
+ return status;
+ }
+ if (num_store_watchers >= DBWRAP_MAX_WATCHERS) {
+ DBG_WARNING("Can't handle %zu watchers\n",
+ num_store_watchers);
+ return NT_STATUS_INSUFFICIENT_RESOURCES;
+ }
+
+ SIVAL(num_watchers_buf, 0, num_store_watchers);
+
+ my_dbufs[num_my_dbufs++] = (TDB_DATA) {
+ .dptr = num_watchers_buf, .dsize = sizeof(num_watchers_buf),
+ };
+ if (wrec->watchers.count != 0) {
+ my_dbufs[num_my_dbufs++] = (TDB_DATA) {
+ .dptr = wrec->watchers.first, .dsize = wrec->watchers.count * DBWRAP_WATCHER_BUF_LENGTH,
+ };
+ }
+ if (add_count != 0) {
+ my_dbufs[num_my_dbufs++] = (TDB_DATA) {
+ .dptr = add_buf,
+ .dsize = sizeof(add_buf),
+ };
+ }
+ if (num_dbufs != 0) {
+ memcpy(my_dbufs+num_my_dbufs, dbufs, num_dbufs * sizeof(*dbufs));
+ num_my_dbufs += num_dbufs;
+ }
+
+ SMB_ASSERT(num_my_dbufs <= ARRAY_SIZE(my_dbufs));
+
+ status = dbwrap_record_storev(
+ wrec->backend.rec, my_dbufs, num_my_dbufs, flags);
+ return status;
+}
+
+static NTSTATUS dbwrap_watched_storev(struct db_record *rec,
+ const TDB_DATA *dbufs, int num_dbufs,
+ int flags)
+{
+ struct db_watched_record *wrec = db_record_get_watched_record(rec);
+
+ return dbwrap_watched_record_storev(wrec, dbufs, num_dbufs, flags);
+}
+
+static NTSTATUS dbwrap_watched_delete(struct db_record *rec)
+{
+ struct db_watched_record *wrec = db_record_get_watched_record(rec);
+
+ /*
+ * dbwrap_watched_record_storev() will figure out
+ * if the record should be deleted or if there are still
+ * watchers to be stored.
+ */
+ return dbwrap_watched_record_storev(wrec, NULL, 0, 0);
+}
+
+struct dbwrap_watched_traverse_state {
+ int (*fn)(struct db_record *rec, void *private_data);
+ void *private_data;
+};
+
+static int dbwrap_watched_traverse_fn(struct db_record *rec,
+ void *private_data)
+{
+ struct dbwrap_watched_traverse_state *state = private_data;
+ struct db_record prec = *rec;
+ bool ok;
+
+ ok = dbwrap_watch_rec_parse(rec->value, NULL, NULL, &prec.value);
+ if (!ok) {
+ return 0;
+ }
+ if (prec.value.dsize == 0) {
+ return 0;
+ }
+ prec.value_valid = true;
+
+ return state->fn(&prec, state->private_data);
+}
+
+static int dbwrap_watched_traverse(struct db_context *db,
+ int (*fn)(struct db_record *rec,
+ void *private_data),
+ void *private_data)
+{
+ struct db_watched_ctx *ctx = talloc_get_type_abort(
+ db->private_data, struct db_watched_ctx);
+ struct dbwrap_watched_traverse_state state = {
+ .fn = fn, .private_data = private_data };
+ NTSTATUS status;
+ int ret;
+
+ status = dbwrap_traverse(
+ ctx->backend, dbwrap_watched_traverse_fn, &state, &ret);
+ if (!NT_STATUS_IS_OK(status)) {
+ return -1;
+ }
+ return ret;
+}
+
+static int dbwrap_watched_traverse_read(struct db_context *db,
+ int (*fn)(struct db_record *rec,
+ void *private_data),
+ void *private_data)
+{
+ struct db_watched_ctx *ctx = talloc_get_type_abort(
+ db->private_data, struct db_watched_ctx);
+ struct dbwrap_watched_traverse_state state = {
+ .fn = fn, .private_data = private_data };
+ NTSTATUS status;
+ int ret;
+
+ status = dbwrap_traverse_read(
+ ctx->backend, dbwrap_watched_traverse_fn, &state, &ret);
+ if (!NT_STATUS_IS_OK(status)) {
+ return -1;
+ }
+ return ret;
+}
+
+static int dbwrap_watched_get_seqnum(struct db_context *db)
+{
+ struct db_watched_ctx *ctx = talloc_get_type_abort(
+ db->private_data, struct db_watched_ctx);
+ return dbwrap_get_seqnum(ctx->backend);
+}
+
+static int dbwrap_watched_transaction_start(struct db_context *db)
+{
+ struct db_watched_ctx *ctx = talloc_get_type_abort(
+ db->private_data, struct db_watched_ctx);
+ return dbwrap_transaction_start(ctx->backend);
+}
+
+static int dbwrap_watched_transaction_commit(struct db_context *db)
+{
+ struct db_watched_ctx *ctx = talloc_get_type_abort(
+ db->private_data, struct db_watched_ctx);
+ return dbwrap_transaction_commit(ctx->backend);
+}
+
+static int dbwrap_watched_transaction_cancel(struct db_context *db)
+{
+ struct db_watched_ctx *ctx = talloc_get_type_abort(
+ db->private_data, struct db_watched_ctx);
+ return dbwrap_transaction_cancel(ctx->backend);
+}
+
+struct dbwrap_watched_parse_record_state {
+ struct db_context *db;
+ void (*parser)(TDB_DATA key, TDB_DATA data, void *private_data);
+ void *private_data;
+ bool ok;
+};
+
+static void dbwrap_watched_parse_record_parser(TDB_DATA key, TDB_DATA data,
+ void *private_data)
+{
+ struct dbwrap_watched_parse_record_state *state = private_data;
+ TDB_DATA userdata;
+
+ state->ok = dbwrap_watch_rec_parse(data, NULL, NULL, &userdata);
+ if (!state->ok) {
+ dbwrap_watch_log_invalid_record(state->db, key, data);
+ return;
+ }
+
+ state->parser(key, userdata, state->private_data);
+}
+
+static NTSTATUS dbwrap_watched_parse_record(
+ struct db_context *db, TDB_DATA key,
+ void (*parser)(TDB_DATA key, TDB_DATA data, void *private_data),
+ void *private_data)
+{
+ struct db_watched_ctx *ctx = talloc_get_type_abort(
+ db->private_data, struct db_watched_ctx);
+ struct dbwrap_watched_parse_record_state state = {
+ .db = db,
+ .parser = parser,
+ .private_data = private_data,
+ };
+ NTSTATUS status;
+
+ status = dbwrap_parse_record(
+ ctx->backend, key, dbwrap_watched_parse_record_parser, &state);
+ if (!NT_STATUS_IS_OK(status)) {
+ return status;
+ }
+ if (!state.ok) {
+ return NT_STATUS_NOT_FOUND;
+ }
+ return NT_STATUS_OK;
+}
+
+static void dbwrap_watched_parse_record_done(struct tevent_req *subreq);
+
+static struct tevent_req *dbwrap_watched_parse_record_send(
+ TALLOC_CTX *mem_ctx,
+ struct tevent_context *ev,
+ struct db_context *db,
+ TDB_DATA key,
+ void (*parser)(TDB_DATA key, TDB_DATA data, void *private_data),
+ void *private_data,
+ enum dbwrap_req_state *req_state)
+{
+ struct db_watched_ctx *ctx = talloc_get_type_abort(
+ db->private_data, struct db_watched_ctx);
+ struct tevent_req *req = NULL;
+ struct tevent_req *subreq = NULL;
+ struct dbwrap_watched_parse_record_state *state = NULL;
+
+ req = tevent_req_create(mem_ctx, &state,
+ struct dbwrap_watched_parse_record_state);
+ if (req == NULL) {
+ *req_state = DBWRAP_REQ_ERROR;
+ return NULL;
+ }
+
+ *state = (struct dbwrap_watched_parse_record_state) {
+ .parser = parser,
+ .private_data = private_data,
+ .ok = true,
+ };
+
+ subreq = dbwrap_parse_record_send(state,
+ ev,
+ ctx->backend,
+ key,
+ dbwrap_watched_parse_record_parser,
+ state,
+ req_state);
+ if (tevent_req_nomem(subreq, req)) {
+ *req_state = DBWRAP_REQ_ERROR;
+ return tevent_req_post(req, ev);
+ }
+
+ tevent_req_set_callback(subreq, dbwrap_watched_parse_record_done, req);
+ return req;
+}
+
+static void dbwrap_watched_parse_record_done(struct tevent_req *subreq)
+{
+ struct tevent_req *req = tevent_req_callback_data(
+ subreq, struct tevent_req);
+ struct dbwrap_watched_parse_record_state *state = tevent_req_data(
+ req, struct dbwrap_watched_parse_record_state);
+ NTSTATUS status;
+
+ status = dbwrap_parse_record_recv(subreq);
+ TALLOC_FREE(subreq);
+ if (tevent_req_nterror(req, status)) {
+ return;
+ }
+
+ if (!state->ok) {
+ tevent_req_nterror(req, NT_STATUS_NOT_FOUND);
+ return;
+ }
+
+ tevent_req_done(req);
+ return;
+}
+
+static NTSTATUS dbwrap_watched_parse_record_recv(struct tevent_req *req)
+{
+ NTSTATUS status;
+
+ if (tevent_req_is_nterror(req, &status)) {
+ tevent_req_received(req);
+ return status;
+ }
+
+ tevent_req_received(req);
+ return NT_STATUS_OK;
+}
+
+static int dbwrap_watched_exists(struct db_context *db, TDB_DATA key)
+{
+ struct db_watched_ctx *ctx = talloc_get_type_abort(
+ db->private_data, struct db_watched_ctx);
+
+ return dbwrap_exists(ctx->backend, key);
+}
+
+static size_t dbwrap_watched_id(struct db_context *db, uint8_t *id,
+ size_t idlen)
+{
+ struct db_watched_ctx *ctx = talloc_get_type_abort(
+ db->private_data, struct db_watched_ctx);
+
+ return dbwrap_db_id(ctx->backend, id, idlen);
+}
+
+struct db_context *db_open_watched(TALLOC_CTX *mem_ctx,
+ struct db_context **backend,
+ struct messaging_context *msg)
+{
+ struct db_context *db;
+ struct db_watched_ctx *ctx;
+
+ db = talloc_zero(mem_ctx, struct db_context);
+ if (db == NULL) {
+ return NULL;
+ }
+ ctx = talloc_zero(db, struct db_watched_ctx);
+ if (ctx == NULL) {
+ TALLOC_FREE(db);
+ return NULL;
+ }
+ db->private_data = ctx;
+
+ ctx->msg = msg;
+
+ ctx->backend = talloc_move(ctx, backend);
+ db->lock_order = ctx->backend->lock_order;
+ ctx->backend->lock_order = DBWRAP_LOCK_ORDER_NONE;
+
+ db->fetch_locked = dbwrap_watched_fetch_locked;
+ db->do_locked = dbwrap_watched_do_locked;
+ db->traverse = dbwrap_watched_traverse;
+ db->traverse_read = dbwrap_watched_traverse_read;
+ db->get_seqnum = dbwrap_watched_get_seqnum;
+ db->transaction_start = dbwrap_watched_transaction_start;
+ db->transaction_commit = dbwrap_watched_transaction_commit;
+ db->transaction_cancel = dbwrap_watched_transaction_cancel;
+ db->parse_record = dbwrap_watched_parse_record;
+ db->parse_record_send = dbwrap_watched_parse_record_send;
+ db->parse_record_recv = dbwrap_watched_parse_record_recv;
+ db->exists = dbwrap_watched_exists;
+ db->id = dbwrap_watched_id;
+ db->name = dbwrap_name(ctx->backend);
+
+ return db;
+}
+
+uint64_t dbwrap_watched_watch_add_instance(struct db_record *rec)
+{
+ struct db_watched_record *wrec = db_record_get_watched_record(rec);
+ static uint64_t global_instance = 1;
+
+ SMB_ASSERT(wrec->added.instance == 0);
+
+ wrec->added = (struct dbwrap_watcher) {
+ .pid = wrec->self,
+ .instance = global_instance++,
+ };
+
+ wrec->force_fini_store = true;
+
+ return wrec->added.instance;
+}
+
+void dbwrap_watched_watch_remove_instance(struct db_record *rec, uint64_t instance)
+{
+ struct db_watched_record *wrec = db_record_get_watched_record(rec);
+ struct dbwrap_watcher clear_watcher = {
+ .pid = wrec->self,
+ .instance = instance,
+ };
+ size_t i;
+ struct server_id_buf buf;
+
+ if (instance == 0) {
+ return;
+ }
+
+ if (wrec->added.instance == instance) {
+ SMB_ASSERT(server_id_equal(&wrec->added.pid, &wrec->self));
+ DBG_DEBUG("Watcher %s:%"PRIu64" reverted from adding\n",
+ server_id_str_buf(clear_watcher.pid, &buf),
+ clear_watcher.instance);
+ ZERO_STRUCT(wrec->added);
+ }
+
+ for (i=0; i < wrec->watchers.count; i++) {
+ struct dbwrap_watcher watcher;
+ size_t off = i*DBWRAP_WATCHER_BUF_LENGTH;
+ size_t next_off;
+ size_t full_len;
+ size_t move_len;
+
+ dbwrap_watcher_get(&watcher, wrec->watchers.first + off);
+
+ if (clear_watcher.instance != watcher.instance) {
+ continue;
+ }
+ if (!server_id_equal(&clear_watcher.pid, &watcher.pid)) {
+ continue;
+ }
+
+ wrec->force_fini_store = true;
+
+ if (i == 0) {
+ DBG_DEBUG("Watcher %s:%"PRIu64" removed from first position of %zu\n",
+ server_id_str_buf(clear_watcher.pid, &buf),
+ clear_watcher.instance,
+ wrec->watchers.count);
+ wrec->watchers.first += DBWRAP_WATCHER_BUF_LENGTH;
+ wrec->watchers.count -= 1;
+ wrec->removed_first = true;
+ return;
+ }
+ if (i == (wrec->watchers.count-1)) {
+ DBG_DEBUG("Watcher %s:%"PRIu64" removed from last position of %zu\n",
+ server_id_str_buf(clear_watcher.pid, &buf),
+ clear_watcher.instance,
+ wrec->watchers.count);
+ wrec->watchers.count -= 1;
+ return;
+ }
+
+ DBG_DEBUG("Watcher %s:%"PRIu64" cleared at position %zu from %zu\n",
+ server_id_str_buf(clear_watcher.pid, &buf),
+ clear_watcher.instance, i+1,
+ wrec->watchers.count);
+
+ next_off = off + DBWRAP_WATCHER_BUF_LENGTH;
+ full_len = wrec->watchers.count * DBWRAP_WATCHER_BUF_LENGTH;
+ move_len = full_len - next_off;
+ memmove(wrec->watchers.first + off,
+ wrec->watchers.first + next_off,
+ move_len);
+ wrec->watchers.count -= 1;
+ return;
+ }
+
+ DBG_DEBUG("Watcher %s:%"PRIu64" not found in %zu watchers\n",
+ server_id_str_buf(clear_watcher.pid, &buf),
+ clear_watcher.instance,
+ wrec->watchers.count);
+ return;
+}
+
+void dbwrap_watched_watch_skip_alerting(struct db_record *rec)
+{
+ struct db_watched_record *wrec = db_record_get_watched_record(rec);
+
+ wrec->wakeup.watcher = (struct dbwrap_watcher) { .instance = 0, };
+ wrec->watchers.alerted = true;
+}
+
+void dbwrap_watched_watch_reset_alerting(struct db_record *rec)
+{
+ struct db_watched_record *wrec = db_record_get_watched_record(rec);
+
+ wrec->wakeup.watcher = (struct dbwrap_watcher) { .instance = 0, };
+ wrec->watchers.alerted = false;
+}
+
+void dbwrap_watched_watch_force_alerting(struct db_record *rec)
+{
+ struct db_watched_record *wrec = db_record_get_watched_record(rec);
+
+ dbwrap_watched_record_prepare_wakeup(wrec);
+}
+
+struct dbwrap_watched_watch_state {
+ struct db_context *db;
+ TDB_DATA key;
+ struct dbwrap_watcher watcher;
+ struct server_id blocker;
+ bool blockerdead;
+};
+
+static bool dbwrap_watched_msg_filter(struct messaging_rec *rec,
+ void *private_data);
+static void dbwrap_watched_watch_done(struct tevent_req *subreq);
+static void dbwrap_watched_watch_blocker_died(struct tevent_req *subreq);
+static int dbwrap_watched_watch_state_destructor(
+ struct dbwrap_watched_watch_state *state);
+
+struct tevent_req *dbwrap_watched_watch_send(TALLOC_CTX *mem_ctx,
+ struct tevent_context *ev,
+ struct db_record *rec,
+ uint64_t resumed_instance,
+ struct server_id blocker)
+{
+ struct db_context *db = dbwrap_record_get_db(rec);
+ struct db_watched_ctx *ctx = talloc_get_type_abort(
+ db->private_data, struct db_watched_ctx);
+ struct db_watched_record *wrec = db_record_get_watched_record(rec);
+ struct tevent_req *req, *subreq;
+ struct dbwrap_watched_watch_state *state;
+ uint64_t instance;
+
+ req = tevent_req_create(mem_ctx, &state,
+ struct dbwrap_watched_watch_state);
+ if (req == NULL) {
+ return NULL;
+ }
+ state->db = db;
+ state->blocker = blocker;
+
+ if (ctx->msg == NULL) {
+ tevent_req_nterror(req, NT_STATUS_NOT_SUPPORTED);
+ return tevent_req_post(req, ev);
+ }
+
+ if (resumed_instance == 0 && wrec->added.instance == 0) {
+ /*
+ * Adding a new instance
+ */
+ instance = dbwrap_watched_watch_add_instance(rec);
+ } else if (resumed_instance != 0 && wrec->added.instance == 0) {
+ /*
+ * Resuming an existing instance that was
+ * already present before do_locked started
+ */
+ instance = resumed_instance;
+ } else if (resumed_instance == wrec->added.instance) {
+ /*
+ * The caller used dbwrap_watched_watch_add_instance()
+ * already during this do_locked() invocation.
+ */
+ instance = resumed_instance;
+ } else {
+ tevent_req_nterror(req, NT_STATUS_REQUEST_NOT_ACCEPTED);
+ return tevent_req_post(req, ev);
+ }
+
+ state->watcher = (struct dbwrap_watcher) {
+ .pid = messaging_server_id(ctx->msg),
+ .instance = instance,
+ };
+
+ state->key = tdb_data_talloc_copy(state, rec->key);
+ if (tevent_req_nomem(state->key.dptr, req)) {
+ return tevent_req_post(req, ev);
+ }
+
+ subreq = messaging_filtered_read_send(
+ state, ev, ctx->msg, dbwrap_watched_msg_filter, state);
+ if (tevent_req_nomem(subreq, req)) {
+ return tevent_req_post(req, ev);
+ }
+ tevent_req_set_callback(subreq, dbwrap_watched_watch_done, req);
+
+ talloc_set_destructor(state, dbwrap_watched_watch_state_destructor);
+
+ if (blocker.pid != 0) {
+ subreq = server_id_watch_send(state, ev, blocker);
+ if (tevent_req_nomem(subreq, req)) {
+ return tevent_req_post(req, ev);
+ }
+ tevent_req_set_callback(
+ subreq, dbwrap_watched_watch_blocker_died, req);
+ }
+
+ return req;
+}
+
+static void dbwrap_watched_watch_blocker_died(struct tevent_req *subreq)
+{
+ struct tevent_req *req = tevent_req_callback_data(
+ subreq, struct tevent_req);
+ struct dbwrap_watched_watch_state *state = tevent_req_data(
+ req, struct dbwrap_watched_watch_state);
+ int ret;
+
+ ret = server_id_watch_recv(subreq, NULL);
+ TALLOC_FREE(subreq);
+ if (ret != 0) {
+ tevent_req_nterror(req, map_nt_error_from_unix(ret));
+ return;
+ }
+ state->blockerdead = true;
+ tevent_req_done(req);
+}
+
+static void dbwrap_watched_watch_state_destructor_fn(
+ struct db_record *rec,
+ TDB_DATA value,
+ void *private_data)
+{
+ struct dbwrap_watched_watch_state *state = talloc_get_type_abort(
+ private_data, struct dbwrap_watched_watch_state);
+
+ /*
+ * Here we just remove ourself from the in memory
+ * watchers array and let db_watched_record_fini()
+ * call dbwrap_watched_record_storev() to do the magic
+ * of writing back the modified in memory copy.
+ */
+ dbwrap_watched_watch_remove_instance(rec, state->watcher.instance);
+ return;
+}
+
+static int dbwrap_watched_watch_state_destructor(
+ struct dbwrap_watched_watch_state *state)
+{
+ NTSTATUS status;
+
+ status = dbwrap_do_locked(
+ state->db,
+ state->key,
+ dbwrap_watched_watch_state_destructor_fn,
+ state);
+ if (!NT_STATUS_IS_OK(status)) {
+ DBG_WARNING("dbwrap_do_locked failed: %s\n",
+ nt_errstr(status));
+ }
+ return 0;
+}
+
+static bool dbwrap_watched_msg_filter(struct messaging_rec *rec,
+ void *private_data)
+{
+ struct dbwrap_watched_watch_state *state = talloc_get_type_abort(
+ private_data, struct dbwrap_watched_watch_state);
+ uint64_t instance;
+
+ if (rec->msg_type != MSG_DBWRAP_MODIFIED) {
+ return false;
+ }
+ if (rec->num_fds != 0) {
+ return false;
+ }
+
+ if (rec->buf.length != sizeof(instance)) {
+ DBG_DEBUG("Got size %zu, expected %zu\n",
+ rec->buf.length,
+ sizeof(instance));
+ return false;
+ }
+
+ instance = BVAL(rec->buf.data, 0);
+
+ if (instance != state->watcher.instance) {
+ DBG_DEBUG("Got instance %"PRIu64", expected %"PRIu64"\n",
+ instance,
+ state->watcher.instance);
+ return false;
+ }
+
+ return true;
+}
+
+static void dbwrap_watched_watch_done(struct tevent_req *subreq)
+{
+ struct tevent_req *req = tevent_req_callback_data(
+ subreq, struct tevent_req);
+ struct dbwrap_watched_watch_state *state = tevent_req_data(
+ req, struct dbwrap_watched_watch_state);
+ struct messaging_rec *rec;
+ int ret;
+
+ ret = messaging_filtered_read_recv(subreq, state, &rec);
+ TALLOC_FREE(subreq);
+ if (ret != 0) {
+ tevent_req_nterror(req, map_nt_error_from_unix(ret));
+ return;
+ }
+ tevent_req_done(req);
+}
+
+NTSTATUS dbwrap_watched_watch_recv(struct tevent_req *req,
+ uint64_t *pkeep_instance,
+ bool *blockerdead,
+ struct server_id *blocker)
+{
+ struct dbwrap_watched_watch_state *state = tevent_req_data(
+ req, struct dbwrap_watched_watch_state);
+ NTSTATUS status;
+
+ if (tevent_req_is_nterror(req, &status)) {
+ tevent_req_received(req);
+ return status;
+ }
+ if (pkeep_instance != NULL) {
+ *pkeep_instance = state->watcher.instance;
+ /*
+ * No need to remove ourselves anymore,
+ * the caller will take care of removing itself.
+ */
+ talloc_set_destructor(state, NULL);
+ }
+ if (blockerdead != NULL) {
+ *blockerdead = state->blockerdead;
+ }
+ if (blocker != NULL) {
+ *blocker = state->blocker;
+ }
+ tevent_req_received(req);
+ return NT_STATUS_OK;
+}
+
generated by cgit v1.2.3 (git 2.39.1) at 2024-06-20 15:45:10 +0000