multi-key; multi-val; multi-timer aware mmap dict

mmbd.py

#//////////           .          ._________|______________________________*
#/////////                    .            .                             /       .
#////////         .                   _____|                            /
#///////              .          . \//_.|._\\/                  .______/         __.
#//////   .         .             ///_/.|.\_\\\                 /      |        /    .  
#/////                          _\|| /.:|:.\ ||/_      ._______/       |       /
#//////                   .    / //||__/|\__||\\ \                 .___|      /   .    
#///////    .      .          / //|| \\   // ||\\ \       .            |_____•   .      
#////////       .           _/ _/_|_| \*  /  ||_ \_ \_  .              |              .   
#/////////                             |                               .              
#//////////      .                     |____________.                  H


import threading
import hashlib
import struct
import fcntl
import mmap
import time
import os


#____________________________________________________________________________________


HEADER_FMT = '<8sI I I H H I Q Q Q Q'
HEADER_SIZE = 64
COVNT = b'MMBDv1\x00\x00'
WAL_SET = 1
WAL_APPEND = 2
WAL_DELETE = 3
WAL_DUP = 4
SLOT_HDR_FMT = '<B B Q I H'
SLOT_HDR_SIZE = 16

#____________________________________________________________________________________

def _round_up(n, a):
    return ((n+a-1)//a)*a
#____________________________________________________________________________________

def now_ms():
    return int(time.time()*1000)
#____________________________________________________________________________________

class MemoryMmapDict:
    def __init__(self, filepath):
        
        self.filepath = filepath
        self.fd = None
        self.m = None
        self._header = None
        self._slot_size = None
        self._capacity = None
        self._max_key_len = None
        self._max_slots_per_key = None
        self._hash_table_size = None
        self._slot_region_offset = None
        self._slot_bitmap_offset = None
        self._index_region_offset = None
        self._wal_offset = None
        self._bucket_size = None
        self._total_size = None
        self.handlers = {}
        self._writer_lock_fd = None
        self._timer_thread = None
        self._timer_stop = threading.Event()
        self._timer_interval_ms = 200
#____________________________________________________________________________________

    @classmethod
    def create(cls, filepath, slot_size, capacity, max_key_len=64, max_slots_per_key=4, initial_wal_mb=10):
        
        if os.path.exists(filepath):
            raise FileExistsError(filepath)
        if slot_size < SLOT_HDR_SIZE:
            raise ValueError(f"slot_size must be >= {SLOT_HDR_SIZE}")
        ht = 1
        while ht < capacity*2: ht<<=1
        bucket_size = 1+2+2+max_key_len+(4*max_slots_per_key)
        hdr = HEADER_SIZE
        slot_region_offset = _round_up(hdr, 4096)
        slot_region_size = capacity*slot_size
        slot_bitmap_offset = _round_up(slot_region_offset+slot_region_size, 4096)
        bitmap_size = _round_up((capacity+7)//8, 4096)
        index_region_offset = _round_up(slot_bitmap_offset + bitmap_size, 4096)
        index_region_size = ht*bucket_size
        wal_offset = _round_up(index_region_offset+index_region_size, 4096)
        total_size = wal_offset + initial_wal_mb*1024*1024
        with open(filepath, 'wb') as f: f.truncate(total_size)
        inst = cls(filepath)
        inst._init_file(slot_size, capacity, max_key_len, max_slots_per_key,
                        ht, slot_region_offset, slot_bitmap_offset, index_region_offset,
                        bucket_size, total_size, wal_offset)
        inst._write_header()
        inst._map_file()
        inst.m.flush()
        return inst
#____________________________________________________________________________________

    @classmethod
    def open(cls, filepath):
        
        if not os.path.exists(filepath):
            raise FileNotFoundError(filepath)
        inst = cls(filepath)
        inst._map_file()
        inst._read_header()
        inst._verify_header()
        inst._replay_wal()
        return inst
#____________________________________________________________________________________

    def _init_file(self, slot_size, capacity, max_key_len, max_slots_per_key,
                   ht, slot_region_offset, slot_bitmap_offset, index_region_offset,
                   bucket_size, total_size, wal_offset):
                       
        self._slot_size = int(slot_size)
        self._capacity = int(capacity)
        self._max_key_len = int(max_key_len)
        self._max_slots_per_key = int(max_slots_per_key)
        self._hash_table_size = int(ht)
        self._slot_region_offset = int(slot_region_offset)
        self._slot_bitmap_offset = int(slot_bitmap_offset)
        self._index_region_offset = int(index_region_offset)
        self._wal_offset = int(wal_offset)
        self._bucket_size = int(bucket_size)
        self._total_size = int(total_size)
#____________________________________________________________________________________

    def _write_header(self):
        
        with open(self.filepath, 'r+b') as f:
            parts = (COVNT, 1, self._slot_size, self._capacity, self._max_key_len,
                     self._max_slots_per_key, self._hash_table_size, self._slot_region_offset,
                     self._slot_bitmap_offset, self._index_region_offset, self._wal_offset)
            packed = struct.pack(HEADER_FMT, *parts)
            if len(packed) > HEADER_SIZE:
                raise RuntimeError("HEADER_FMT too large")
            packed = packed.ljust(HEADER_SIZE, b'\x00')
            f.seek(0)
            f.write(packed)
            f.flush()
            os.fsync(f.fileno())
#____________________________________________________________________________________

    def _map_file(self):
        
        self.fd = open(self.filepath, 'r+b')
        self.m = mmap.mmap(self.fd.fileno(), 0)
#____________________________________________________________________________________

    def _read_header(self):
        
        raw = self.m[0:HEADER_SIZE]
        parts = struct.unpack(HEADER_FMT, raw[:struct.calcsize(HEADER_FMT)])
        COVNT = parts[0]
        if COVNT != COVNT:
            raise RuntimeError("bad COVNT")
        self._slot_size = parts[2]
        self._capacity = parts[3]
        self._max_key_len = parts[4]
        self._max_slots_per_key = parts[5]
        self._hash_table_size = parts[6]
        self._slot_region_offset = parts[7]
        self._slot_bitmap_offset = parts[8]
        self._index_region_offset = parts[9]
        self._wal_offset = parts[10]
        self._bucket_size = 1+2+2+self._max_key_len+(4*self._max_slots_per_key)
        self._total_size = len(self.m)
        self._header = parts
#____________________________________________________________________________________

    def _verify_header(self):
        
        if self._slot_size <= 0 or self._capacity <= 0:
            raise RuntimeError("corrupt header")
#____________________________________________________________________________________

    def close(self):
        
        self.stop_timer()
        if self.m:
            self.m.flush()
            self.m.close()
            self.m = None
        if self.fd:
            self.fd.close()
            self.fd = None
#____________________________________________________________________________________

    def acquire_writer_lock(self):
        
        if self._writer_lock_fd is None:
            self._writer_lock_fd = open(self.filepath + '.lock', 'w+b')
        #fcntl.flock(self._writer_lock_fd, fcntl.LOCK_EX)
#____________________________________________________________________________________

    def release_writer_lock(self):
        
        if self._writer_lock_fd:
            pass
            #fcntl.flock(self._writer_lock_fd, fcntl.LOCK_UN)
#____________________________________________________________________________________

    def _slot_offset(self, slot_idx):
        
        if not (0 <= slot_idx < self._capacity):
            raise IndexError("slot idx out of range")
        return self._slot_region_offset+slot_idx*self._slot_size
#____________________________________________________________________________________

    def _read_slot_header(self, slot_idx):
        
        off = self._slot_offset(slot_idx)
        raw = self.m[off: off+SLOT_HDR_SIZE]
        type_id, flags, next_fire_ts, seq, _pad = struct.unpack(SLOT_HDR_FMT, raw)
        return type_id, flags, next_fire_ts, seq
#____________________________________________________________________________________

    def _write_slot_header(self, slot_idx, type_id, flags, next_fire_ts, seq):
        
        off = self._slot_offset(slot_idx)
        buf = struct.pack(SLOT_HDR_FMT, type_id, flags, int(next_fire_ts), int(seq), 0)
        self.m[off:off+SLOT_HDR_SIZE] = buf
#____________________________________________________________________________________

    def _slot_payload_offset(self, slot_idx):
        
        return self._slot_offset(slot_idx)+SLOT_HDR_SIZE
#____________________________________________________________________________________

    def _read_slot_payload(self, slot_idx):
        
        poff = self._slot_payload_offset(slot_idx)
        return bytes(self.m[poff: poff+(self._slot_size-SLOT_HDR_SIZE)])
#____________________________________________________________________________________

    def _write_slot_payload(self, slot_idx, payload_bytes):
        
        if len(payload_bytes) != (self._slot_size - SLOT_HDR_SIZE):
            raise ValueError("payload size mismatch")
        poff = self._slot_payload_offset(slot_idx)
        self.m[poff: poff+len(payload_bytes)] = payload_bytes
#____________________________________________________________________________________

    def _bitmap_byte_offset(self, bit_index):
        
        return self._slot_bitmap_offset+(bit_index//8)
#____________________________________________________________________________________

    def _is_slot_used(self, slot_idx):
        
        b = self.m[self._bitmap_byte_offset(slot_idx)]
        mask = 1<<(slot_idx%8)
        return bool(b & mask)
#____________________________________________________________________________________

    def _set_slot_used(self, slot_idx, used: bool):
        
        off = self._bitmap_byte_offset(slot_idx)
        b = self.m[off]
        mask = 1<<(slot_idx%8)
        nb = (b | mask) if used else (b & (~mask & 0xFF))
        if nb != b:
            self.m[off:off+1] = bytes([nb])
#____________________________________________________________________________________

    def _find_free_slot(self):
        
        for i in range(self._capacity):
            if not self._is_slot_used(i):
                return i
        raise MemoryError("no free slot")
#____________________________________________________________________________________

    def _bucket_offset(self, bucket_idx):
        
        return self._index_region_offset+bucket_idx*self._bucket_size
#____________________________________________________________________________________

    def _read_bucket(self, bucket_idx):
        
        off = self._bucket_offset(bucket_idx)
        raw = self.m[off:off+self._bucket_size]
        state = raw[0]
        key_len = struct.unpack_from('<H', raw, 1)[0]
        count = struct.unpack_from('<H', raw, 3)[0]
        key_bytes = raw[5:5 + self._max_key_len][:key_len]
        slots = []
        slot_area_off = 5 + self._max_key_len
        for i in range(self._max_slots_per_key):
            s = struct.unpack_from('<I', raw, slot_area_off+i*4)[0]
            if i < count:
                slots.append(s)
        return state, key_bytes.decode('utf-8'), slots
#____________________________________________________________________________________

    def _write_bucket(self, bucket_idx, state, key_bytes, slots):
        
        if isinstance(key_bytes, str):
            key_bytes = key_bytes.encode('utf-8')
        if len(key_bytes) > self._max_key_len:
            raise ValueError("key too long")
        if len(slots) > self._max_slots_per_key:
            raise ValueError("too many slots for key")
        off = self._bucket_offset(bucket_idx)
        buf = bytearray(self._bucket_size)
        buf[0] = state
        struct.pack_into('<H', buf, 1, len(key_bytes))
        struct.pack_into('<H', buf, 3, len(slots))
        buf[5:5+len(key_bytes)] = key_bytes
        slot_area_off = 5+self._max_key_len
        for i, s in enumerate(slots):
            struct.pack_into('<I', buf, slot_area_off+i*4, s)
        self.m[off:off + self._bucket_size] = bytes(buf)
#____________________________________________________________________________________

    def _hash_key(self, key):
        
        h = hashlib.blake2b(key.encode('utf-8'), digest_size=8).digest()
        return struct.unpack_from('<Q', h)[0]
#____________________________________________________________________________________

    def _find_bucket_for_key(self, key):
        
        h = self._hash_key(key)
        mask = self._hash_table_size-1
        for i in range(self._hash_table_size):
            idx = (h+i)&mask
            state, kbytes, slots = self._read_bucket(idx)
            if state == 0:
                return idx, None
            if state == 1 and kbytes == key:
                return idx, (state, kbytes, slots)
        raise RuntimeError("hash table full")
#____________________________________________________________________________________

    def _wal_append_record(self, rec_type, payload_bytes):
        
        rec_len = 1+len(payload_bytes)
        header_pos = self._wal_offset
        pos = header_pos
        end = self._total_size
        while pos+4 < end:
            cur = struct.unpack_from('<I', self.m, pos)[0]
            if cur == 0:
                break
            pos+=4+cur
        if pos+4+rec_len > end:
            raise RuntimeError("WAL full")
        self.m[pos:pos+4] = struct.pack('<I', 0)
        self.m[pos+4:pos+4+1+len(payload_bytes)] = bytes([rec_type])+payload_bytes
        self.m.flush()
        self.m[pos:pos+4] = struct.pack('<I', rec_len)
        self.m.flush()
        return pos
#____________________________________________________________________________________

    def _replay_wal(self):
        
        pos = self._wal_offset
        end = self._total_size
        while pos+4 <= end:
            rec_len = struct.unpack_from('<I', self.m, pos)[0]
            if rec_len == 0:
                break
            rec_type = self.m[pos+4]
            payload = self.m[pos+5:pos+4+rec_len]
            try:
                if rec_type == WAL_SET:
                    self._apply_set_payload(payload, durable=False)
                elif rec_type == WAL_APPEND:
                    self._apply_append_payload(payload, durable=False)
                elif rec_type == WAL_DELETE:
                    self._apply_delete_payload(payload, durable=False)
                elif rec_type == WAL_DUP:
                    self._apply_dup_payload(payload, durable=False)
            except Exception:
                pass
            pos+=4+rec_len
        self.m[self._wal_offset:self._wal_offset+4] = struct.pack('<I', 0)
        self.m.flush()
#____________________________________________________________________________________

    def _apply_set_payload(self, payload, durable=True):
        
        key_len = struct.unpack_from('<H', payload, 0)[0]
        key = payload[2:2+key_len].decode('utf-8')
        slot_bytes = payload[2+key_len:2+key_len+self._slot_size]
        hdr = slot_bytes[:SLOT_HDR_SIZE]
        payload_bytes = slot_bytes[SLOT_HDR_SIZE:]
        bidx, content = self._find_bucket_for_key(key)
        if content:
            for s in content[2]:
                self._set_slot_used(s, False)
        new_slot = self._find_free_slot()
        self.m[self._slot_offset(new_slot): self._slot_offset(new_slot)+self._slot_size] = slot_bytes
        self._set_slot_used(new_slot, True)
        self._write_bucket(bidx, 1, key, [new_slot])
        if durable:
            self.m.flush()
#____________________________________________________________________________________

    def _apply_append_payload(self, payload, durable=True):
        
        key_len = struct.unpack_from('<H', payload, 0)[0]
        key = payload[2:2+key_len].decode('utf-8')
        slot_bytes = payload[2+key_len:2+key_len+self._slot_size]
        bidx, content = self._find_bucket_for_key(key)
        if content is None:
            new_slot = self._find_free_slot()
            self.m[self._slot_offset(new_slot): self._slot_offset(new_slot)+self._slot_size] = slot_bytes
            self._set_slot_used(new_slot, True)
            self._write_bucket(bidx, 1, key, [new_slot])
        else:
            slots = content[2]
            if len(slots) >= self._max_slots_per_key:
                raise RuntimeError("max slots per key reached")
            new_slot = self._find_free_slot()
            self.m[self._slot_offset(new_slot): self._slot_offset(new_slot)+self._slot_size] = slot_bytes
            self._set_slot_used(new_slot, True)
            self._write_bucket(bidx, 1, key, slots+[new_slot])
        if durable:
            self.m.flush()
#____________________________________________________________________________________

    def _apply_delete_payload(self, payload, durable=True):
        
        key_len = struct.unpack_from('<H', payload, 0)[0]
        key = payload[2:2+key_len].decode('utf-8')
        bidx, content = self._find_bucket_for_key(key)
        if content is None:
            return
        for s in content[2]:
            self._set_slot_used(s, False)
        self._write_bucket(bidx, 2, b'', [])
        if durable:
            self.m.flush()
#____________________________________________________________________________________

    def _apply_dup_payload(self, payload, durable=True):
        
        key_len = struct.unpack_from('<H', payload, 0)[0]
        key = payload[2:2+key_len].decode('utf-8')
        bidx, content = self._find_bucket_for_key(key)
        if content is None:
            raise KeyError(key)
        slots = content[2]
        if len(slots) != 1:
            raise RuntimeError("duplicate_if_single requires exactly one slot")
        src = slots[0]
        new_slot = self._find_free_slot()
        val = self.m[self._slot_offset(src): self._slot_offset(src)+self._slot_size]
        self.m[self._slot_offset(new_slot): self._slot_offset(new_slot)+self._slot_size] = val
        self._set_slot_used(new_slot, True)
        self._write_bucket(bidx, 1, key, [src, new_slot])
        if durable:
            self.m.flush()
#____________________________________________________________________________________

    def set_single(self, key, type_id: int, payload_bytes: bytes, next_fire_ts_ms=0, flags=0):
        
        if not isinstance(key, str): raise TypeError
        if not (0 <= type_id <= 255): raise ValueError
        if len(payload_bytes) != (self._slot_size-SLOT_HDR_SIZE): raise ValueError
        kb = key.encode('utf-8')
        if len(kb) > self._max_key_len: raise ValueError
        seq = 0
        hdr = struct.pack(SLOT_HDR_FMT, type_id, flags, int(next_fire_ts_ms), seq, 0)
        slot_bytes = hdr+payload_bytes
        payload = struct.pack('<H', len(kb))+kb+slot_bytes
        self.acquire_writer_lock()
        try:
            self._wal_append_record(WAL_SET, payload)
            self._apply_set_payload(payload, durable=True)
        finally:
            self.release_writer_lock()
#____________________________________________________________________________________

    def append_slot(self, key, type_id: int, payload_bytes: bytes, next_fire_ts_ms=0, flags=0):
        
        if not isinstance(key, str): raise TypeError
        if not (0 <= type_id <= 255): raise ValueError
        if len(payload_bytes) != (self._slot_size-SLOT_HDR_SIZE): raise ValueError
        kb = key.encode('utf-8')
        if len(kb) > self._max_key_len: raise ValueError
        seq = 0
        hdr = struct.pack(SLOT_HDR_FMT, type_id, flags, int(next_fire_ts_ms), seq, 0)
        slot_bytes = hdr+payload_bytes
        payload = struct.pack('<H', len(kb))+kb+slot_bytes
        self.acquire_writer_lock()
        try:
            self._wal_append_record(WAL_APPEND, payload)
            self._apply_append_payload(payload, durable=True)
        finally:
            self.release_writer_lock()
#____________________________________________________________________________________

    def delete(self, key):
        
        kb = key.encode('utf-8')
        if len(kb) > self._max_key_len: raise ValueError
        payload = struct.pack('<H', len(kb))+kb
        self.acquire_writer_lock()
        try:
            self._wal_append_record(WAL_DELETE, payload)
            self._apply_delete_payload(payload, durable=True)
        finally:
            self.release_writer_lock()
#____________________________________________________________________________________

    def duplicate_if_single(self, key):
        
        kb = key.encode('utf-8')
        if len(kb) > self._max_key_len: raise ValueError
        payload = struct.pack('<H', len(kb))+kb
        self.acquire_writer_lock()
        try:
            self._wal_append_record(WAL_DUP, payload)
            self._apply_dup_payload(payload, durable=True)
        finally:
            self.release_writer_lock()
#____________________________________________________________________________________

    def get(self, key):
        
        bidx, content = self._find_bucket_for_key(key)
        if content is None:
            return []
        out = []
        for s in content[2]:
            off = self._slot_offset(s)
            out.append(bytes(self.m[off:off+self._slot_size]))
        return out
#____________________________________________________________________________________

    def keys(self):
        
        out = []
        for i in range(self._hash_table_size):
            state, k, slots = self._read_bucket(i)
            if state == 1:
                out.append(k)
        return out
#____________________________________________________________________________________

    def register_handler(self, type_id: int, handler):
        
        if not (0 <= type_id <= 255): raise ValueError
        self.handlers[type_id] = handler
    def _mark_bucket_active(self, bucket_idx, key, slots):
        k = ('\x01'+key).encode('utf-8')
        if len(k)-1 > self._max_key_len:
            raise ValueError("key too long for active marking")
        self._write_bucket(bucket_idx, 1, k, slots)
#____________________________________________________________________________________

    def _unmark_bucket_active(self, bucket_idx, key, slots):
        
        self._write_bucket(bucket_idx, 1, key, slots)
#____________________________________________________________________________________

    def _bucket_is_active(self, bucket_idx):
        
        state, key, slots = self._read_bucket(bucket_idx)
        if state != 1:
            return False
        return key.startswith('\x01')
#____________________________________________________________________________________

    def mark_key_active(self, key):
        
        bidx, content = self._find_bucket_for_key(key)
        if content is None:
            raise KeyError(key)
        try:
            self._mark_bucket_active(bidx, key, content[2])
            self.m.flush()
        finally:
            self.release_writer_lock()
#____________________________________________________________________________________

    def mark_key_inactive(self, key):
        
        bidx, content = self._find_bucket_for_key(key)
        if content is None:
            return
        self.acquire_writer_lock()
        try:
            self._unmark_bucket_active(bidx, key, content[2])
            self.m.flush()
        finally:
            self.release_writer_lock()
#____________________________________________________________________________________

    def start_timer(self, interval_ms=200):
        
        if self._timer_thread and self._timer_thread.is_alive():
            return
        self._timer_interval_ms = int(interval_ms)
        self._timer_stop.clear()
        t = threading.Thread(target=self._timer_loop, daemon=True)
        self._timer_thread = t
        t.start()
#____________________________________________________________________________________

    def stop_timer(self):
        
        if self._timer_thread:
            self._timer_stop.set()
            self._timer_thread.join(timeout=1.0)
            self._timer_thread = None
#____________________________________________________________________________________

    def _timer_loop(self):
        
        while not self._timer_stop.is_set():
            try:
                self._timer_tick()
            except Exception:
                pass
            time.sleep(self._timer_interval_ms/1000.0)
#____________________________________________________________________________________

    def _timer_tick(self):
        
        now = now_ms()
        for bidx in range(self._hash_table_size):
            if not self._bucket_is_active(bidx):
                continue
            state, key, slots = self._read_bucket(bidx)
            for slot_idx in slots:
                type_id, flags, next_fire_ts, seq = self._read_slot_header(slot_idx)
                if next_fire_ts == 0:
                    continue
                if next_fire_ts <= now:
                    handler = self.handlers.get(type_id)
                    if handler is None:
                        continue
                    payload = self._read_slot_payload(slot_idx)
                    try:
                        new_payload, reschedule_ts = handler.on_timer(key, (type_id, flags, next_fire_ts, seq), payload)
                    except Exception:
                        new_payload = None
                        reschedule_ts = None
                    if new_payload is None and reschedule_ts is None:
                        continue
                    self.acquire_writer_lock()
                    try:
                        cur_type, cur_flags, cur_next, cur_seq = self._read_slot_header(slot_idx)
                        if cur_seq != seq:
                            continue
                        new_seq = cur_seq + 1
                        new_next = reschedule_ts if (reschedule_ts is not None) else 0
                        final_payload = payload if new_payload is None else new_payload
                        if len(final_payload) != (self._slot_size-SLOT_HDR_SIZE):
                            raise ValueError("handler returned bad payload size")
                        hdr = struct.pack(SLOT_HDR_FMT, cur_type, cur_flags, int(new_next), int(new_seq), 0)
                        slot_bytes = hdr+final_payload
                        kb = key.encode('utf-8')
                        payload_for_wal = struct.pack('<H', len(kb))+kb+slot_bytes
                        self._wal_append_record(WAL_SET, payload_for_wal)
                        self.m[self._slot_offset(slot_idx): self._slot_offset(slot_idx)+self._slot_size] = slot_bytes
                        self.m.flush()
                    finally:
                        self.release_writer_lock()
#=================================================================================

class EchoHandler:
    
    def __init__(self, delay_ms=1000):
        
        self.delay_ms = delay_ms
#____________________________________________________________________________________

    def on_timer(self, key, header, payload_bytes):
        
        if len(payload_bytes) < 4:
            return None, None
        counter = struct.unpack_from('>I', payload_bytes, 0)[0]
        counter+=1
        new_payload = bytearray(payload_bytes)
        struct.pack_into('>I', new_payload, 0, counter)
        return bytes(new_payload), now_ms() + self.delay_ms

#=================================================================================
# NON BRIDGE-TIMER ACTIVATION TENET MMBD-OBJECT EXAMPLE:
#=================================================================================
	
fn = 'mmbd1.mmbd'

# Remove any store of @fn in memory
try:
    os.remove(fn)
except Exception:
    pass
try:
    os.remove(f'{fn}.lock')
except Exception:
    pass
    
print()
    
# Create a new file (dual-reverse-key-tenet access)
m = MemoryMmapDict.create(fn, slot_size=64, capacity=128,
                          max_key_len=64, max_slots_per_key=4)
                          
# Set a single value (replaces any existing)
mssg = b'initial tenet signature sentence(cell:2:2)'
payload = struct.pack('>I', 0) + mssg.ljust(64-SLOT_HDR_SIZE-4, b'\x00')
m.set_single('cell:2:2', type_id=1, payload_bytes=payload)
# Append extra slot for same key (key now has 2 slots, "de")
mssg = b'second tenet signature sentence(cell:2:2)'
payload = struct.pack('>I', 2) + mssg.ljust(64-SLOT_HDR_SIZE-4, b'\x00')
m.append_slot('cell:2:2', type_id=2, payload_bytes=payload)
# Append the time agent, transfer key:
payload = struct.pack('>I', 1) + mssg.ljust(64-SLOT_HDR_SIZE-4, b'\x00')
m.append_slot('user:T_AGENT1', type_id=4, payload_bytes=payload)
payload = struct.pack('>I', 2) + b't=28739'.ljust(64-SLOT_HDR_SIZE-4, b'\x00')
m.append_slot('user:T_AGENT1', type_id=3, payload_bytes=payload)
# Read the slots for a key tenets:
vals1 = m.get('cell:2:2')
vals2 = m.get('user:T_AGENT1')
print(vals1[0][:64].decode())
print(vals1[1][:64].decode())
print(vals2[0][:64].decode())
print(vals2[1][:64].decode())
# Create final key with single slot
mssg = b'A final tenet key sentence with ascii'
payload = struct.pack('>I', 0) + mssg.ljust(64-SLOT_HDR_SIZE-4, b'\x00')
m.set_single('T_session:1', type_id=3, payload_bytes=payload)
m.duplicate_if_single('T_session:1')
# T_session:1 now has two tenet slots, "ja",  is
# tricky, does this fast with mmap and memoryview
print(m.get('T_session:1'))
# Delete a key
m.delete('T_session:1')
print(m.get('T_session:1')) # Empty list []
# List the keys, "vu":
print('keys:', m.keys())
vals1 = m.get('cell:2:2')
vals2 = m.get('user:T_AGENT1')
print(vals1[0][:64].decode())
print(vals2[0][:64].decode())
m.close()

#=================================================================================
# BRIDGE-TIMER ACTIVATION TENET MMBD-OBJECT EXAMPLE:
#=================================================================================

fn = 'mmbd.mmbd'

try:
    os.remove(fn)
except Exception:
    pass
try:
    os.remove(f'{fn}.lock')
except Exception:
    pass

m = MemoryMmapDict.create(fn, slot_size=64, capacity=128, 
    	                      max_key_len=64, max_slots_per_key=2)
    
mssg = b'third tenet signature sentence(cell:3:3)'
payload = struct.pack('>I', 0) + mssg.ljust(64-SLOT_HDR_SIZE-4, b'\x00')
m.append_slot('cell:3:3', type_id=1, payload_bytes=payload)
vals1 = m.get('cell:3:3')
print(vals1[0][:64].decode())
m.register_handler(0, EchoHandler(delay_ms=500))
payload = struct.pack('>I', 1) + b'ARRIVE=93782'.ljust(64-SLOT_HDR_SIZE-4, b'\x00')
m.set_single('cell:3:3', type_id=1, payload_bytes=payload, next_fire_ts_ms=now_ms()+1000)
m.mark_key_active('cell:3:3')
# start bridge-timer(aware)
m.start_timer(interval_ms=200)
print("payload timer arrival is activated(cell:3:3)\nCorrect sauce is most usually boss")

Mathematically, lots of different scopes of reasoning can be applied with such a module. The first, reducing activation node structure from a timed property on subsets of type of information. For example, during WW2 one third of all the ships were in mathematical fact lost, sunk, destroyed. And the Pearl Harbor Attack a epic moment of that mathematical fact included. (Written in the book of Revelation before it happened, nearly 2,000 years before it happened) Secondly, activation of a node composed of two distinct activations is more factual, mathematically sound to a true neural network we are modeled to before leaving the womb. Third. A pearl is formed by a symmetrical circumstance of time and resistance in the shell. The mathematical properties of the pearl's hardness and layers already pre-formed to the reasoning effects of current before applied.