mdbx: добавление mdbx_enumerate_subdb().

mdbx.h

@@ -4443,6 +4443,46 @@ LIBMDBX_API int mdbx_dbi_rename(MDBX_txn *txn, MDBX_dbi dbi, const char *name);
 LIBMDBX_API int mdbx_dbi_rename2(MDBX_txn *txn, MDBX_dbi dbi,
                                  const MDBX_val *name);
 
+/** \brief Функция обратного вызова для перечисления
+ *  пользовательских именованных таблиц.
+ *
+ * \ingroup c_statinfo
+ * \see mdbx_enumerate_subdb()
+ *
+ * \param [in] ctx       Указатель на контекст переданный аналогичным
+ *                       параметром в \ref mdbx_enumerate_subdb().
+ * \param [in] txn       Транзазакция.
+ * \param [in] name      Имя таблицы.
+ * \param [in] flags     Флаги \ref MDBX_db_flags_t.
+ * \param [in] stat      Базовая информация \ref MDBX_stat о таблице.
+ * \param [in] dbi       Отличное от 0 значение DBI-дескриптора,
+ *                       если таковой был открыт для этой таблицы.
+ *                       Либо 0 если такого открытого дескриптора нет.
+ *
+ * \returns Ноль при успехе и продолжении перечисления, при возвращении другого
+ *          значения оно будет немедленно возвращено вызывающему
+ *          без продолжения перечисления. */
+typedef int(MDBX_subdb_enum_func)(void *ctx, const MDBX_txn *txn,
+                                  const MDBX_val *name, MDBX_db_flags_t flags,
+                                  const struct MDBX_stat *stat,
+                                  MDBX_dbi dbi) MDBX_CXX17_NOEXCEPT;
+
+/** \brief Enumerate the entries in the reader lock table.
+ * \ingroup c_statinfo
+ * \see MDBX_subdb_enum_func
+ *
+ * \param [in] txn     Транзакция запущенная посредством
+ *                     \ref mdbx_txn_begin().
+ * \param [in] func    Указатель на пользовательскую функцию-перечислитель
+ *                     с сигнатурой \ref MDBX_subdb_enum_func,
+ *                     которая будет вызвана для каждой таблицы.
+ * \param [in] ctx     Указатель на некоторый контект, который будет передан
+ *                     в функцию-перечислитель как есть.
+ *
+ * \returns Ненулевое значение кода ошибки, либо 0 при успешном выполнении. */
+ LIBMDBX_API int mdbx_enumerate_subdb(const MDBX_txn *txn,
+                                     MDBX_subdb_enum_func *func, void *ctx);
+
 /** \defgroup value2key Value-to-Key functions
  * \brief Value-to-Key functions to
  * \ref avoid_custom_comparators "avoid using custom comparators"

src/audit.c

@@ -3,28 +3,24 @@
 
 #include "internals.h"
 
-__cold static tree_t *audit_db_dig(const MDBX_txn *txn, const size_t dbi,
+struct audit_ctx {
-                                   tree_t *fallback) {
+  size_t used;
-  const MDBX_txn *dig = txn;
+  uint8_t *const done_bitmap;
-  do {
+};
-    tASSERT(txn, txn->n_dbi == dig->n_dbi);
+
-    const uint8_t state = dbi_state(dig, dbi);
+static int audit_dbi(void *ctx, const MDBX_txn *txn, const MDBX_val *name,
-    if (state & DBI_LINDO)
+                     MDBX_db_flags_t flags, const struct MDBX_stat *stat,
-      switch (state & (DBI_VALID | DBI_STALE | DBI_OLDEN)) {
+                     MDBX_dbi dbi) {
-      case DBI_VALID:
+  struct audit_ctx *audit_ctx = ctx;
-      case DBI_OLDEN:
+  (void)name;
-        return dig->dbs + dbi;
+  (void)txn;
-      case 0:
+  (void)flags;
-        return nullptr;
+  audit_ctx->used += (size_t)stat->ms_branch_pages +
-      case DBI_VALID | DBI_STALE:
+                     (size_t)stat->ms_leaf_pages +
-      case DBI_OLDEN | DBI_STALE:
+                     (size_t)stat->ms_overflow_pages;
-        break;
+  if (dbi)
-      default:
+    audit_ctx->done_bitmap[dbi / CHAR_BIT] |= 1 << dbi % CHAR_BIT;
-        tASSERT(txn, !!"unexpected dig->dbi_state[dbi]");
+  return MDBX_SUCCESS;
-      }
-    dig = dig->parent;
-  } while (dig);
-  return fallback;
 }
 
 static size_t audit_db_used(const tree_t *db) {
@@ -71,8 +67,6 @@ __cold static int audit_ex_locked(MDBX_txn *txn, size_t retired_stored,
   tASSERT(txn, rc == MDBX_NOTFOUND);
 
   const size_t done_bitmap_size = (txn->n_dbi + CHAR_BIT - 1) / CHAR_BIT;
-  uint8_t *const done_bitmap = alloca(done_bitmap_size);
-  memset(done_bitmap, 0, done_bitmap_size);
   if (txn->parent) {
     tASSERT(txn, txn->n_dbi == txn->parent->n_dbi &&
                      txn->n_dbi == txn->env->txn->n_dbi);
@@ -82,51 +76,20 @@ __cold static int audit_ex_locked(MDBX_txn *txn, size_t retired_stored,
 #endif /* MDBX_ENABLE_DBI_SPARSE */
   }
 
-  size_t used = NUM_METAS +
+  struct audit_ctx ctx = {0, alloca(done_bitmap_size)};
-                audit_db_used(audit_db_dig(txn, FREE_DBI, nullptr)) +
+  memset(ctx.done_bitmap, 0, done_bitmap_size);
-                audit_db_used(audit_db_dig(txn, MAIN_DBI, nullptr));
+  ctx.used = NUM_METAS + audit_db_used(dbi_dig(txn, FREE_DBI, nullptr)) +
-  rc = cursor_init(&cx.outer, txn, MAIN_DBI);
+             audit_db_used(dbi_dig(txn, MAIN_DBI, nullptr));
-  if (unlikely(rc != MDBX_SUCCESS))
-    return rc;
 
-  rc = tree_search(&cx.outer, nullptr, Z_FIRST);
+  rc = mdbx_enumerate_subdb(txn, audit_dbi, &ctx);
-  while (rc == MDBX_SUCCESS) {
+  tASSERT(txn, rc == MDBX_SUCCESS);
-    page_t *mp = cx.outer.pg[cx.outer.top];
-    for (size_t k = 0; k < page_numkeys(mp); k++) {
-      node_t *node = page_node(mp, k);
-      if (node_flags(node) != N_SUBDATA)
-        continue;
-      if (unlikely(node_ds(node) != sizeof(tree_t))) {
-        ERROR("%s/%d: %s %u", "MDBX_CORRUPTED", MDBX_CORRUPTED,
-              "invalid dupsort sub-tree node size", (unsigned)node_ds(node));
-        return MDBX_CORRUPTED;
-      }
-
-      tree_t reside;
-      const tree_t *db = memcpy(&reside, node_data(node), sizeof(reside));
-      const MDBX_val name = {node_key(node), node_ks(node)};
-      for (size_t dbi = CORE_DBS; dbi < env->n_dbi; ++dbi) {
-        if (dbi >= txn->n_dbi || !(env->dbs_flags[dbi] & DB_VALID))
-          continue;
-        if (env->kvs[MAIN_DBI].clc.k.cmp(&name, &env->kvs[dbi].name))
-          continue;
-
-        done_bitmap[dbi / CHAR_BIT] |= 1 << dbi % CHAR_BIT;
-        db = audit_db_dig(txn, dbi, &reside);
-        break;
-      }
-      used += audit_db_used(db);
-    }
-    rc = cursor_sibling_right(&cx.outer);
-  }
-  tASSERT(txn, rc == MDBX_NOTFOUND);
 
   for (size_t dbi = CORE_DBS; dbi < txn->n_dbi; ++dbi) {
-    if (done_bitmap[dbi / CHAR_BIT] & (1 << dbi % CHAR_BIT))
+    if (ctx.done_bitmap[dbi / CHAR_BIT] & (1 << dbi % CHAR_BIT))
       continue;
-    const tree_t *db = audit_db_dig(txn, dbi, nullptr);
+    const tree_t *db = dbi_dig(txn, dbi, nullptr);
     if (db)
-      used += audit_db_used(db);
+      ctx.used += audit_db_used(db);
     else if (dbi_state(txn, dbi))
       WARNING("audit %s@%" PRIaTXN
               ": unable account dbi %zd / \"%*s\", state 0x%02x",
@@ -135,7 +98,7 @@ __cold static int audit_ex_locked(MDBX_txn *txn, size_t retired_stored,
               (const char *)env->kvs[dbi].name.iov_base, dbi_state(txn, dbi));
   }
 
-  if (pending + gc + used == txn->geo.first_unallocated)
+  if (pending + gc + ctx.used == txn->geo.first_unallocated)
     return MDBX_SUCCESS;
 
   if ((txn->flags & MDBX_TXN_RDONLY) == 0)
@@ -148,7 +111,7 @@ __cold static int audit_ex_locked(MDBX_txn *txn, size_t retired_stored,
   ERROR("audit @%" PRIaTXN ": %zu(pending) + %zu"
         "(gc) + %zu(count) = %zu(total) <> %zu"
         "(allocated)",
-        txn->txnid, pending, gc, used, pending + gc + used,
+        txn->txnid, pending, gc, ctx.used, pending + gc + ctx.used,
         (size_t)txn->geo.first_unallocated);
   return MDBX_PROBLEM;
 }

src/dbi.c

@@ -952,3 +952,83 @@ __cold int mdbx_dbi_stat(const MDBX_txn *txn, MDBX_dbi dbi, MDBX_stat *dest,
   stat_get(&txn->dbs[dbi], dest, bytes);
   return MDBX_SUCCESS;
 }
+
+__cold const tree_t *dbi_dig(const MDBX_txn *txn, const size_t dbi,
+                             tree_t *fallback) {
+  const MDBX_txn *dig = txn;
+  do {
+    tASSERT(txn, txn->n_dbi == dig->n_dbi);
+    const uint8_t state = dbi_state(dig, dbi);
+    if (state & DBI_LINDO)
+      switch (state & (DBI_VALID | DBI_STALE | DBI_OLDEN)) {
+      case DBI_VALID:
+      case DBI_OLDEN:
+        return dig->dbs + dbi;
+      case 0:
+        return nullptr;
+      case DBI_VALID | DBI_STALE:
+      case DBI_OLDEN | DBI_STALE:
+        break;
+      default:
+        tASSERT(txn, !!"unexpected dig->dbi_state[dbi]");
+      }
+    dig = dig->parent;
+  } while (dig);
+  return fallback;
+}
+
+__cold int mdbx_enumerate_subdb(const MDBX_txn *txn, MDBX_subdb_enum_func *func,
+                                void *ctx) {
+  if (unlikely(!func))
+    return MDBX_EINVAL;
+
+  int rc = check_txn(txn, MDBX_TXN_BLOCKED);
+  if (unlikely(rc != MDBX_SUCCESS))
+    return rc;
+
+  cursor_couple_t cx;
+  rc = cursor_init(&cx.outer, txn, MAIN_DBI);
+  if (unlikely(rc != MDBX_SUCCESS))
+    return rc;
+
+  cx.outer.next = txn->cursors[MAIN_DBI];
+  txn->cursors[MAIN_DBI] = &cx.outer;
+  for (rc = outer_first(&cx.outer, nullptr, nullptr); rc == MDBX_SUCCESS;
+       rc = outer_next(&cx.outer, nullptr, nullptr, MDBX_NEXT_NODUP)) {
+    node_t *node =
+        page_node(cx.outer.pg[cx.outer.top], cx.outer.ki[cx.outer.top]);
+    if (node_flags(node) != N_SUBDATA)
+      continue;
+    if (unlikely(node_ds(node) != sizeof(tree_t))) {
+      ERROR("%s/%d: %s %u", "MDBX_CORRUPTED", MDBX_CORRUPTED,
+            "invalid dupsort sub-tree node size", (unsigned)node_ds(node));
+      rc = MDBX_CORRUPTED;
+      break;
+    }
+
+    tree_t reside;
+    const tree_t *tree = memcpy(&reside, node_data(node), sizeof(reside));
+    const MDBX_val name = {node_key(node), node_ks(node)};
+    const MDBX_env *const env = txn->env;
+    MDBX_dbi dbi = 0;
+    for (size_t i = CORE_DBS; i < env->n_dbi; ++i) {
+      if (i >= txn->n_dbi || !(env->dbs_flags[i] & DB_VALID))
+        continue;
+      if (env->kvs[MAIN_DBI].clc.k.cmp(&name, &env->kvs[i].name))
+        continue;
+
+      tree = dbi_dig(txn, i, &reside);
+      dbi = (MDBX_dbi)i;
+      break;
+    }
+
+    MDBX_stat stat;
+    stat_get(tree, &stat, sizeof(stat));
+    rc = func(ctx, txn, &name, tree->flags, &stat, dbi);
+    if (rc != MDBX_SUCCESS)
+      break;
+  }
+  txn->cursors[MAIN_DBI] = cx.outer.next;
+
+  return (rc == MDBX_NOTFOUND) ? MDBX_SUCCESS : rc;
+}

src/dbi.h

@@ -131,3 +131,6 @@ MDBX_INTERNAL int dbi_open(MDBX_txn *txn, const MDBX_val *const name,
 
 MDBX_INTERNAL int dbi_bind(MDBX_txn *txn, const size_t dbi, unsigned user_flags,
                            MDBX_cmp_func *keycmp, MDBX_cmp_func *datacmp);
+
+MDBX_INTERNAL const tree_t *dbi_dig(const MDBX_txn *txn, const size_t dbi,
+                                    tree_t *fallback);