chop: offline event log truncation

pkg/c3/defs.h

@@ -156,5 +156,9 @@
         unlink(a);})
 #     define c3_fopen(a, b) ({                                  \
         fopen(a, b);})
+#     define c3_remove(a) ({                                    \
+        remove(a);})
+#     define c3_rename(a, b) ({                                 \
+        rename(a, b);})
 
 #endif /* ifndef C3_DEFS_H */

pkg/c3/motes.h

@@ -201,6 +201,7 @@
 #   define c3__chew   c3_s4('c','h','e','w')
 #   define c3__chis   c3_s4('c','h','i','s')
 #   define c3__chob   c3_s4('c','h','o','b')
+#   define c3__chop   c3_s4('c','h','o','p')
 #   define c3__chug   c3_s4('c','h','u','g')
 #   define c3__claf   c3_s4('c','l','a','f')
 #   define c3__clam   c3_s4('c','l','a','m')

pkg/noun/events.c

@@ -992,10 +992,10 @@ _ce_image_copy(u3e_image* fom_u, u3e_image* tou_u)
   return c3y;
 }
 
-/* _ce_backup();
+/* u3e_backup();
 */
-static void
-_ce_backup(void)
+c3_o
+u3e_backup(c3_o ovw_o)
 {
   u3e_image nop_u = { .nam_c = "north", .pgs_w = 0 };
   u3e_image sop_u = { .nam_c = "south", .pgs_w = 0 };
@@ -1004,25 +1004,25 @@ _ce_backup(void)
 
   snprintf(ful_c, 8192, "%s/.urb/bhk", u3P.dir_c);
 
-  if ( c3_mkdir(ful_c, 0700) ) {
+  if ( (c3n == ovw_o) && c3_mkdir(ful_c, 0700) ) {
     if ( EEXIST != errno ) {
       fprintf(stderr, "loom: image backup: %s\r\n", strerror(errno));
     }
-    return;
+    return c3n;
   }
 
   snprintf(ful_c, 8192, "%s/.urb/bhk/%s.bin", u3P.dir_c, nop_u.nam_c);
 
   if ( -1 == (nop_u.fid_i = c3_open(ful_c, mod_i, 0666)) ) {
     fprintf(stderr, "loom: c3_open %s: %s\r\n", ful_c, strerror(errno));
-    return;
+    return c3n;
   }
 
   snprintf(ful_c, 8192, "%s/.urb/bhk/%s.bin", u3P.dir_c, sop_u.nam_c);
 
   if ( -1 == (sop_u.fid_i = c3_open(ful_c, mod_i, 0666)) ) {
     fprintf(stderr, "loom: c3_open %s: %s\r\n", ful_c, strerror(errno));
-    return;
+    return c3n;
   }
 
   if (  (c3n == _ce_image_copy(&u3P.nor_u, &nop_u))
@@ -1034,10 +1034,14 @@ _ce_backup(void)
     c3_unlink(ful_c);
     snprintf(ful_c, 8192, "%s/.urb/bhk", u3P.dir_c);
     c3_rmdir(ful_c);
+    fprintf(stderr, "loom: image backup failed\r\n");
+    return c3n;
   }
 
   close(nop_u.fid_i);
   close(sop_u.fid_i);
+  fprintf(stderr, "loom: image backup complete\r\n");
+  return c3y;
 }
 
 /*
@@ -1103,7 +1107,7 @@ u3e_save(void)
   _ce_patch_free(pat_u);
   _ce_patch_delete();
 
-  _ce_backup();
+  u3e_backup(c3n);
 }
 
 /* u3e_live(): start the checkpointing system.

pkg/noun/events.h

@@ -64,6 +64,11 @@
 
   /** Functions.
   **/
+    /* u3e_backup(): copy the snapshot from chk to bhk. 
+    */
+      c3_o 
+      u3e_backup(c3_o ovw_o);
+
     /* u3e_fault(): handle a memory event with libsigsegv protocol.
     */
       c3_i

pkg/vere/db/lmdb.c

@@ -504,6 +504,112 @@ u3_lmdb_save_meta(MDB_env*    env_u,
   return c3y;
 }
 
+/* u3_lmdb_walk_init(): initialize db iterator.
+*/
+c3_o
+u3_lmdb_walk_init(MDB_env*      env_u,
+                  u3_lmdb_walk* itr_u,
+                  c3_d          nex_d,
+                  c3_d          las_d)
+{
+  //  XX assumes little-endian
+  //
+  MDB_val key_u = { .mv_size = sizeof(c3_d), .mv_data = &nex_d };
+  MDB_val val_u;
+  c3_w    ops_w, ret_w;
+
+  itr_u->red_o = c3n;
+  itr_u->nex_d = nex_d;
+  itr_u->las_d = las_d;
+
+  //  create a read-only transaction.
+  //
+  ops_w = MDB_RDONLY;
+  if ( (ret_w = mdb_txn_begin(env_u, 0, ops_w, &itr_u->txn_u)) ) {
+    mdb_logerror(stderr, ret_w, "lmdb: read txn_begin fail");
+    return c3n;
+  }
+  //  open the database in the transaction
+  //
+  ops_w = MDB_CREATE | MDB_INTEGERKEY;
+  if ( (ret_w = mdb_dbi_open(itr_u->txn_u, "EVENTS", ops_w, &itr_u->mdb_u)) ) {
+    mdb_logerror(stderr, ret_w, "lmdb: read: dbi_open fail");
+    //  XX confirm
+    //
+    mdb_txn_abort(itr_u->txn_u);
+    return c3n;
+  }
+
+  //  creates a cursor to iterate over keys starting at [eve_d]
+  //
+  if ( (ret_w = mdb_cursor_open(itr_u->txn_u, itr_u->mdb_u, &itr_u->cur_u)) ) {
+    mdb_logerror(stderr, ret_w, "lmdb: read: cursor_open fail");
+    //  XX confirm
+    //
+    mdb_txn_abort(itr_u->txn_u);
+    return c3n;
+  }
+
+  //  set the cursor to the position of [eve_d]
+  //
+  ops_w = MDB_SET_KEY;
+  if ( (ret_w = mdb_cursor_get(itr_u->cur_u, &key_u, &val_u, ops_w)) ) {
+    mdb_logerror(stderr, ret_w, "lmdb: read: initial cursor_get failed");
+    fprintf(stderr, "    at %" PRIu64 "\r\n", nex_d);
+    mdb_cursor_close(itr_u->cur_u);
+    //  XX confirm
+    //
+    mdb_txn_abort(itr_u->txn_u);
+    return c3n;
+  }
+
+  return c3y;
+}
+
+/* u3_lmdb_walk_next(): synchronously read next event from iterator.
+*/
+c3_o
+u3_lmdb_walk_next(u3_lmdb_walk* itr_u, size_t* len_i, void** buf_v)
+{
+  MDB_val key_u, val_u;
+  c3_w    ret_w, ops_w;
+
+  c3_assert( itr_u->nex_d <= itr_u->las_d );
+
+  ops_w = ( c3y == itr_u->red_o ) ? MDB_NEXT : MDB_GET_CURRENT;
+  if ( (ret_w = mdb_cursor_get(itr_u->cur_u, &key_u, &val_u, ops_w)) ) {
+    mdb_logerror(stderr, ret_w, "lmdb: walk error");
+    return c3n;
+  }
+
+  //  sanity check: ensure contiguous event numbers
+  //
+  if ( *(c3_d*)key_u.mv_data != itr_u->nex_d ) {
+    fprintf(stderr, "lmdb: read gap: expected %" PRIu64
+                    ", received %" PRIu64 "\r\n",
+                    itr_u->nex_d,
+                    *(c3_d*)key_u.mv_data);
+    return c3n;
+  }
+
+  *len_i = val_u.mv_size;
+  *buf_v = val_u.mv_data;
+
+  itr_u->nex_d++;
+  itr_u->red_o = c3y;
+
+  return c3y;
+}
+
+/* u3_lmdb_walk_done(): close iterator.
+*/
+void
+u3_lmdb_walk_done(u3_lmdb_walk* itr_u)
+{
+  mdb_cursor_close(itr_u->cur_u);
+  mdb_txn_abort(itr_u->txn_u);
+}
+
 /* mdb_logerror(): writes an error message and lmdb error code to f.
 */
 void mdb_logerror(FILE* f, int err, const char* fmt, ...)

pkg/vere/db/lmdb.h

@@ -8,6 +8,16 @@
 
   /* lmdb api wrapper
   */
+    /* u3_lmdb_walk: event iterator
+    */
+    typedef struct _u3_lmdb_walk {
+      MDB_txn*    txn_u;  //  transaction handle
+      MDB_dbi     mdb_u;  //  db handle
+      MDB_cursor* cur_u;  //  db cursor
+      c3_o        red_o;  //  have we read from this yet?
+      c3_d        nex_d;  //  next event number
+      c3_d        las_d;  //  final event number, inclusive
+    } u3_lmdb_walk;
 
     /* u3_lmdb_init(): open lmdb at [pax_c], mmap up to [siz_i].
     */
@@ -19,7 +29,6 @@
       void
       u3_lmdb_exit(MDB_env* env_u);
 
-
     /* u3_lmdb_stat(): print env stats.
     */
       void
@@ -64,4 +73,22 @@
                         size_t      val_i,
                         void*       val_p);
 
+    /* u3_lmdb_walk_init(): initialize db iterator.
+    */
+      c3_o
+      u3_lmdb_walk_init(MDB_env*      env_u,
+                        u3_lmdb_walk* itr_u,
+                        c3_d          nex_d,
+                        c3_d          las_d);
+
+    /* u3_lmdb_walk_next(): synchronously read next event from iterator.
+    */
+      c3_o
+      u3_lmdb_walk_next(u3_lmdb_walk* itr_u, size_t* len_i, void** buf_v);
+
+    /* u3_lmdb_walk_done(): close iterator.
+    */
+      void
+      u3_lmdb_walk_done(u3_lmdb_walk* itr_u);
+
 #endif /* ifndef U3_VERE_DB_LMDB_H */

pkg/vere/disk.c

@@ -464,14 +464,14 @@ u3_disk_read(u3_disk* log_u, c3_d eve_d, c3_d len_d)
 /* _disk_save_meta(): serialize atom, save as metadata at [key_c].
 */
 static c3_o
-_disk_save_meta(u3_disk* log_u, const c3_c* key_c, u3_atom dat)
+_disk_save_meta(MDB_env* mdb_u, const c3_c* key_c, u3_atom dat)
 {
   c3_w  len_w = u3r_met(3, dat);
   c3_y* byt_y = c3_malloc(len_w);
   u3r_bytes(0, len_w, byt_y, dat);
 
   {
-    c3_o ret_o = u3_lmdb_save_meta(log_u->mdb_u, key_c, len_w, byt_y);
+    c3_o ret_o = u3_lmdb_save_meta(mdb_u, key_c, len_w, byt_y);
     c3_free(byt_y);
     return ret_o;
   }
@@ -480,17 +480,17 @@ _disk_save_meta(u3_disk* log_u, const c3_c* key_c, u3_atom dat)
 /* u3_disk_save_meta(): save metadata.
 */
 c3_o
-u3_disk_save_meta(u3_disk* log_u,
+u3_disk_save_meta(MDB_env* mdb_u,
                   c3_d     who_d[2],
                   c3_o     fak_o,
                   c3_w     lif_w)
 {
   c3_assert( c3y == u3a_is_cat(lif_w) );
 
-  if (  (c3n == _disk_save_meta(log_u, "version", 1))
-     || (c3n == _disk_save_meta(log_u, "who", u3i_chubs(2, who_d)))
-     || (c3n == _disk_save_meta(log_u, "fake", fak_o))
-     || (c3n == _disk_save_meta(log_u, "life", lif_w)) )
+  if (  (c3n == _disk_save_meta(mdb_u, "version", 1))
+     || (c3n == _disk_save_meta(mdb_u, "who", u3i_chubs(2, who_d)))
+     || (c3n == _disk_save_meta(mdb_u, "fake", fak_o))
+     || (c3n == _disk_save_meta(mdb_u, "life", lif_w)) )
   {
     return c3n;
   }
@@ -513,36 +513,36 @@ _disk_meta_read_cb(void* ptr_v, size_t val_i, void* val_p)
 /* _disk_read_meta(): read metadata at [key_c], deserialize.
 */
 static u3_weak
-_disk_read_meta(u3_disk* log_u, const c3_c* key_c)
+_disk_read_meta(MDB_env* mdb_u, const c3_c* key_c)
 {
   u3_weak dat = u3_none;
-  u3_lmdb_read_meta(log_u->mdb_u, &dat, key_c, _disk_meta_read_cb);
+  u3_lmdb_read_meta(mdb_u, &dat, key_c, _disk_meta_read_cb);
   return dat;
 }
 
 /* u3_disk_read_meta(): read metadata.
 */
 c3_o
-u3_disk_read_meta(u3_disk* log_u,
+u3_disk_read_meta(MDB_env* mdb_u,
                   c3_d*    who_d,
                   c3_o*    fak_o,
                   c3_w*    lif_w)
 {
   u3_weak ver, who, fak, lif;
 
-  if ( u3_none == (ver = _disk_read_meta(log_u, "version")) ) {
+  if ( u3_none == (ver = _disk_read_meta(mdb_u, "version")) ) {
     fprintf(stderr, "disk: read meta: no version\r\n");
     return c3n;
   }
-  if ( u3_none == (who = _disk_read_meta(log_u, "who")) ) {
+  if ( u3_none == (who = _disk_read_meta(mdb_u, "who")) ) {
     fprintf(stderr, "disk: read meta: no indentity\r\n");
     return c3n;
   }
-  if ( u3_none == (fak = _disk_read_meta(log_u, "fake")) ) {
+  if ( u3_none == (fak = _disk_read_meta(mdb_u, "fake")) ) {
     fprintf(stderr, "disk: read meta: no fake bit\r\n");
     return c3n;
   }
-  if ( u3_none == (lif = _disk_read_meta(log_u, "life")) ) {
+  if ( u3_none == (lif = _disk_read_meta(mdb_u, "life")) ) {
     fprintf(stderr, "disk: read meta: no lifecycle length\r\n");
     return c3n;
   }