volatile.h

/* Jazz (c) 2018-2026 kaalam.ai (The Authors of Jazz), using (under the same license):
 
    1. Biomodelling - The AATBlockQueue class (c) Jacques Basaldúa, 2009-2012 licensed
      exclusively for the use in the Jazz server software.
 
      Copyright 2009-2012 Jacques Basaldúa
 
    2. BBVA - Jazz: A lightweight analytical web server for data-driven applications.
 
      Copyright 2016-2017 Banco Bilbao Vizcaya Argentaria, S.A.
 
      This product includes software developed at
 
      BBVA (https://www.bbva.com/)
 
    3. LMDB, Copyright 2011-2017 Howard Chu, Symas Corp. All rights reserved.
 
      Licensed under http://www.OpenLDAP.org/license.html
 
 
      Licensed under the Apache License, Version 2.0 (the "License");
    you may not use this file except in compliance with the License.
    You may obtain a copy of the License at
 
      http://www.apache.org/licenses/LICENSE-2.0
 
      Unless required by applicable law or agreed to in writing, software
    distributed under the License is distributed on an "AS IS" BASIS,
    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    See the License for the specific language governing permissions and
    limitations under the License.
*/
 
 
#include "src/jazz_elements/channel.h"
 
#ifdef CATCH_TEST
#ifndef INCLUDED_JAZZ_CATCH2
#define INCLUDED_JAZZ_CATCH2
 
#include "src/catch2/catch.hpp"
 
#endif
#endif
 
 
#ifndef INCLUDED_JAZZ_ELEMENTS_VOLATILE
#define INCLUDED_JAZZ_ELEMENTS_VOLATILE
 
namespace jazz_elements
{
 
#define BASE_DEQUE_10BIT        0x0a4       //< First 10 bits of base "deque"
#define BASE_INDEX_10BIT        0x1c9       //< First 10 bits of base "index"
#define BASE_QUEUE_10BIT        0x2b1       //< First 10 bits of base "queue"
#define BASE_TREE_10BIT         0x254       //< First 10 bits of base "tree"
 
#define COMMAND_JUST_THE_KEY    0x000       //< No command in the key
#define COMMAND_CHILD_10BIT     0x103       //< First 10 bits of command "ch{ild}"
#define COMMAND_FIRST_10BIT     0x126       //< First 10 bits of command "fi{rst}"
#define COMMAND_GET_10BIT       0x0a7       //< First 10 bits of command "ge{t}"
#define COMMAND_HIGH_10BIT      0x128       //< First 10 bits of command "hi{ghest}"
#define COMMAND_LAST_10BIT      0x02c       //< First 10 bits of command "la{st}"
#define COMMAND_LOW_10BIT       0x1ec       //< First 10 bits of command "lo{west}"
#define COMMAND_NEXT_10BIT      0x0ae       //< First 10 bits of command "ne{xt}"
#define COMMAND_PARENT_10BIT    0x030       //< First 10 bits of command "pa{rent}"
#define COMMAND_PFIRST_10BIT    0x0d0       //< First 10 bits of command "pf{irst}"
#define COMMAND_PLAST_10BIT     0x190       //< First 10 bits of command "pl{ast}"
#define COMMAND_PREV_10BIT      0x250       //< First 10 bits of command "pr{ev}"
#define COMMAND_PUT_10BIT       0x2b0       //< First 10 bits of command "pu{t}"
#define COMMAND_XHIGH_10BIT     0x118       //< First 10 bits of command "xh{ighest}"
#define COMMAND_XLOW_10BIT      0x198       //< First 10 bits of command "xl{owest}"
#define COMMAND_SECOND_ARG      0x3ff       //< In a put call with a key, it is either a parent key or a priority.
#define COMMAND_SIZE            0x400       //< For numbers, defining a queue size, this is added to avoid overlap.
 
 
typedef struct VolatileTransaction *pVolatileTransaction;
 
 
struct VolatileTransaction: Transaction {
    union {
        pVolatileTransaction p_prev;                    
        pVolatileTransaction p_parent;                  
    };
    pVolatileTransaction p_next;                        
    union {
        pVolatileTransaction p_child;                   
        double priority;                                
    };
    union {
        int level;                                      
        int num_wins;                                   
    };
    union {
        int times_used;                                 
        int num_visits;                                 
    };
    uint64_t key_hash;                                  
};
 
 
struct EntityKeyHash {
    uint64_t ent_hash;                      
    uint64_t key_hash;                      
 
    bool operator==(const EntityKeyHash &o) const {
        return ent_hash == o.ent_hash && key_hash == o.key_hash;
    }
 
    bool operator<(const EntityKeyHash &o) const {
        return ent_hash < o.ent_hash || (ent_hash == o.ent_hash && key_hash < o.key_hash);
    }
};
 
 
struct QueueEnt {
    int                  queue_size;    
    int                  queue_use;     
    pVolatileTransaction p_root;        
};
 
 
typedef std::map<uint64_t, pVolatileTransaction> HashVolXctMap;
 
 
typedef std::map<uint64_t, QueueEnt> HashQueueEntMap;
 
 
typedef std::map<EntityKeyHash, pVolatileTransaction> EntKeyVolXctMap;
 
 
struct NameUse {
    int  use;       
    Name name;      
};
 
 
typedef std::map<uint64_t, NameUse> HashNameUseMap;
 
 
typedef String* pString;
 
 
class Volatile : public Container {
 
    public:
 
        Volatile(pLogger a_logger, pConfigFile a_config);
       ~Volatile();
 
        virtual pChar const id();
 
        StatusCode start    ();
        StatusCode shut_down();
 
        // The easy interface (Requires explicit pulling because of the native interface using the same names.)
 
        using Container::get;
        using Container::header;
        using Container::put;
        using Container::locate;
        using Container::new_entity;
        using Container::remove;
        using Container::copy;
 
        virtual StatusCode new_transaction(pTransaction &p_txn);
        virtual void destroy_transaction  (pTransaction &p_txn);
 
        // The "native" interface
 
        virtual StatusCode get       (pTransaction      &p_txn,
                                      Locator           &what);
        virtual StatusCode get       (pTransaction      &p_txn,
                                      Locator           &what,
                                      pBlock             p_row_filter);
        virtual StatusCode get       (pTransaction      &p_txn,
                                      Locator           &what,
                                      pChar              name);
        virtual StatusCode locate    (Locator           &location,
                                      Locator           &what);
        virtual StatusCode header    (StaticBlockHeader &hea,
                                      Locator           &what);
        virtual StatusCode header    (pTransaction      &p_txn,
                                      Locator           &what);
        virtual StatusCode put       (Locator           &where,
                                      pBlock             p_block,
                                      int                mode = WRITE_AS_BASE_DEFAULT);
        virtual StatusCode new_entity(Locator           &where);
        virtual StatusCode remove    (Locator           &where);
        virtual StatusCode copy      (Locator           &where,
                                      Locator           &what);
 
        // Support for container names in the BaseAPI .base_names()
 
        void base_names(BaseNames &base_names);
 
#ifndef CATCH_TEST
    private:
#endif
 
        StatusCode new_volatile();
        StatusCode destroy_volatile();
 
        inline void new_key(Name &key) {
            sprintf(key, "k%014lx", ++key_seed);
        }
 
 
        inline StatusCode populate_index(Index &index, pBlock p_block) {
 
            if (p_block->cell_type != CELL_TYPE_TUPLE || p_block->size != 2) return SERVICE_ERROR_BAD_BLOCK;
 
            pBlock p_key = pTuple(p_block)->get_block(0);
            pBlock p_val = pTuple(p_block)->get_block(1);
 
            if (   p_key->cell_type != CELL_TYPE_STRING || p_val->cell_type != CELL_TYPE_STRING
                || p_key->rank != 1 || p_val->rank != 1 || p_key->size != p_val->size) return SERVICE_ERROR_BAD_BLOCK;
 
            for (int i = 0; i < p_key->size; i++)
                index[p_key->get_string(i)] = p_val->get_string(i);
 
            return SERVICE_NO_ERROR;
        }
 
 
        inline StatusCode put_queue_insert(uint64_t ent_hash, Name &key, double priority, pBlock p_block, int mode) {
            HashQueueEntMap::iterator it_queue = queue_ent.find(ent_hash);
 
            if (it_queue == queue_ent.end()) return SERVICE_ERROR_ENTITY_NOT_FOUND;
 
            EntityKeyHash ek = {ent_hash, hash(key)};
            EntKeyVolXctMap::iterator it_item = queue_key.find(ek);
 
            if (it_item != queue_key.end()) {
                if (mode & WRITE_ONLY_IF_NOT_EXISTS) return SERVICE_ERROR_WRITE_FORBIDDEN;
 
                pVolatileTransaction p_item = it_item->second;
 
                pBlock p_new = block_malloc(p_block->total_bytes);
                if (p_new == nullptr) return SERVICE_ERROR_NO_MEM;
 
                alloc_bytes -= p_item->p_block->total_bytes;
                free(p_item->p_block);
 
                memcpy(p_new, p_block, p_block->total_bytes);
 
                p_item->status = BLOCK_STATUS_EMPTY;
 
                it_queue->second.p_root = aat_remove(p_item, it_queue->second.p_root);
 
                p_item->priority = priority;
                p_item->times_used++;
                p_item->p_block = p_new;
 
                p_item->p_block->hash64 = 0;
 
                p_item->status = BLOCK_STATUS_READY;
 
                it_queue->second.p_root = aat_insert(p_item, it_queue->second.p_root);
 
                queue_key[ek] = p_item;
 
                return SERVICE_NO_ERROR;
            }
            if (mode & WRITE_ONLY_IF_EXISTS) return SERVICE_ERROR_WRITE_FORBIDDEN;
 
            if (it_queue->second.queue_use == it_queue->second.queue_size) {
                pVolatileTransaction p_lowest = aat_lowest_priority(it_queue->second.p_root);
 
                if (p_lowest->priority >= priority) return SERVICE_ERROR_LOW_PRIORITY;
 
                destroy_item(BASE_QUEUE_10BIT, ent_hash, p_lowest);
            }
 
            pTransaction p_txn;
 
            int ret = new_transaction(p_txn);
 
            if (ret != SERVICE_NO_ERROR) return ret;
 
            pBlock p_new = block_malloc(p_block->total_bytes);
            if (p_new == nullptr) {
                destroy_transaction(p_txn);
 
                return SERVICE_ERROR_NO_MEM;
            }
            memcpy(p_new, p_block, p_block->total_bytes);
 
            p_new->hash64 = 0;
 
            pVolatileTransaction(p_txn)->priority   = priority;
            pVolatileTransaction(p_txn)->times_used = 0;
            pVolatileTransaction(p_txn)->key_hash   = add_name(ek.key_hash, key);
            pVolatileTransaction(p_txn)->p_block    = p_new;
            pVolatileTransaction(p_txn)->status     = BLOCK_STATUS_READY;
 
            it_queue->second.p_root = aat_insert((pVolatileTransaction) p_txn, it_queue->second.p_root);
            it_queue->second.queue_use++;
 
            queue_key[ek] = (pVolatileTransaction) p_txn;
 
            return SERVICE_NO_ERROR;
        }
 
 
        inline StatusCode put_in_deque(HashVolXctMap::iterator it_ent, EntityKeyHash &ek, Name &key, pBlock p_block, bool first = false) {
 
            pTransaction p_txn;
            int          ret;
 
            if ((ret = new_transaction(p_txn)) != SERVICE_NO_ERROR) return ret;
 
            pBlock p_new = block_malloc(p_block->total_bytes);
            if (p_new == nullptr) {
                destroy_transaction(p_txn);
 
                return SERVICE_ERROR_NO_MEM;
            }
            memcpy(p_new, p_block, p_block->total_bytes);
            p_txn->p_block = p_new;
            p_txn->p_block->hash64 = 0;
            p_txn->status  = BLOCK_STATUS_READY;
 
            deque_key[ek] = (pVolatileTransaction) p_txn;
            pVolatileTransaction(p_txn)->key_hash = add_name(ek.key_hash, key);
 
            if (it_ent->second == nullptr) {
                it_ent->second = (pVolatileTransaction) p_txn;
                pVolatileTransaction(p_txn)->p_next = (pVolatileTransaction) p_txn;
                pVolatileTransaction(p_txn)->p_prev = (pVolatileTransaction) p_txn;
 
                return SERVICE_NO_ERROR;
            }
 
            if (first) {
                pVolatileTransaction p_2nd  = it_ent->second;
                pVolatileTransaction p_last = p_2nd->p_prev;
 
                pVolatileTransaction(p_txn)->p_next = p_2nd;
                p_2nd->p_prev                       = (pVolatileTransaction) p_txn;
                pVolatileTransaction(p_txn)->p_prev = p_last;
                p_last->p_next                      = (pVolatileTransaction) p_txn;
 
                it_ent->second = (pVolatileTransaction) p_txn;
 
                return SERVICE_NO_ERROR;
            }
            pVolatileTransaction p_last = it_ent->second->p_prev;
 
            pVolatileTransaction(p_txn)->p_prev = p_last;
            it_ent->second->p_prev              = (pVolatileTransaction) p_txn;
            pVolatileTransaction(p_txn)->p_next = it_ent->second;
            p_last->p_next                      = (pVolatileTransaction) p_txn;
 
            return SERVICE_NO_ERROR;
        }
 
 
        inline StatusCode put_in_tree(HashVolXctMap::iterator it_ent, EntityKeyHash &ek, Name &key, Name &parent, pBlock p_block) {
 
            pVolatileTransaction p_root, p_parent = nullptr, p_next = nullptr;;
 
            if ((p_root = it_ent->second) != nullptr) {
                if (tree_key.find(ek) != tree_key.end()) return SERVICE_ERROR_WRITE_FORBIDDEN;
 
                EntityKeyHash parent_ek = {ek.ent_hash, hash(parent)};
 
                EntKeyVolXctMap::iterator it;
 
                if ((it = tree_key.find(parent_ek)) == tree_key.end()) return SERVICE_ERROR_PARENT_NOT_FOUND;
 
                p_parent = it->second;
                p_next   = p_parent->p_child;
            }
 
            pTransaction p_txn;
            int          ret;
 
            if ((ret = new_transaction(p_txn)) != SERVICE_NO_ERROR) return ret;
 
            pBlock p_new = block_malloc(p_block->total_bytes);
            if (p_new == nullptr) {
                destroy_transaction(p_txn);
 
                return SERVICE_ERROR_NO_MEM;
            }
            memcpy(p_new, p_block, p_block->total_bytes);
            p_txn->p_block = p_new;
            p_txn->p_block->hash64 = 0;
            p_txn->status  = BLOCK_STATUS_READY;
 
            tree_key[ek] = (pVolatileTransaction) p_txn;
 
            pVolatileTransaction(p_txn)->p_parent   = p_parent;
            pVolatileTransaction(p_txn)->p_child    = nullptr;
            pVolatileTransaction(p_txn)->p_next     = p_next;
            pVolatileTransaction(p_txn)->num_wins   = 0;
            pVolatileTransaction(p_txn)->num_visits = 0;
            pVolatileTransaction(p_txn)->key_hash   = add_name(ek.key_hash, key);
 
            if (p_root == nullptr)
                it_ent->second = (pVolatileTransaction) p_txn;
            else
                p_parent->p_child = (pVolatileTransaction) p_txn;
 
            return SERVICE_NO_ERROR;
        }
 
 
        inline StatusCode put_replace(pVolatileTransaction p_replace, pBlock p_block) {
 
            pBlock p_new = block_malloc(p_block->total_bytes);
 
            if (p_new == nullptr) return SERVICE_ERROR_NO_MEM;
 
            alloc_bytes -= p_replace->p_block->total_bytes;
            free(p_replace->p_block);
 
            memcpy(p_new, p_block, p_block->total_bytes);
 
            p_replace->p_block = p_new;
 
            return SERVICE_NO_ERROR;
        }
 
 
        inline StatusCode put_index(Index &index, pChar key, pBlock p_block, int mode) {
 
            if (p_block->cell_type != CELL_TYPE_STRING || p_block->size != 1) return SERVICE_ERROR_BAD_BLOCK;
 
            Index::iterator it = index.find(key);
 
            if (it == index.end()) {
                if (mode & WRITE_ONLY_IF_EXISTS) return SERVICE_ERROR_WRITE_FORBIDDEN;
 
                index[key] = p_block->get_string(0);
 
                return SERVICE_NO_ERROR;
            }
            if (mode & WRITE_ONLY_IF_NOT_EXISTS) return SERVICE_ERROR_WRITE_FORBIDDEN;
 
            index[key] = p_block->get_string(0);
 
            return SERVICE_NO_ERROR;
        }
 
 
        inline void destroy_item(int base, uint64_t ent_hash, pVolatileTransaction p_item) {
 
            EntityKeyHash ek = {ent_hash, p_item->key_hash};
            pTransaction  p_txn = p_item;
 
            switch (base) {
            case BASE_DEQUE_10BIT: {
                HashVolXctMap::iterator it_ent = deque_ent.find(ent_hash);
                if (p_item == it_ent->second) {
                    if (p_item->p_next == p_item)
                        it_ent->second = nullptr;
                    else {
                        p_item->p_next->p_prev = p_item->p_prev;
                        p_item->p_prev->p_next = p_item->p_next;
                        it_ent->second = p_item->p_next;
                    }
                } else {
                    p_item->p_next->p_prev = p_item->p_prev;
                    p_item->p_prev->p_next = p_item->p_next;
                }
                erase_name(ek.key_hash);
                deque_key.erase(ek);
                destroy_transaction(p_txn); }
 
                return;
 
            case BASE_QUEUE_10BIT: {
                HashQueueEntMap::iterator it_ent = queue_ent.find(ent_hash);
 
                it_ent->second.p_root = aat_remove(p_item, it_ent->second.p_root);
                it_ent->second.queue_use--;
 
                erase_name(ek.key_hash);
                queue_key.erase(ek);
                destroy_transaction(p_txn); }
 
                return;
 
            case BASE_TREE_10BIT:
                erase_name(ek.key_hash);
                destroy_transaction(p_txn);
                tree_key.erase(ek);
            }
        }
 
 
        inline uint64_t add_name(uint64_t hash, Name &key) {
 
            HashNameUseMap::iterator it = name.find(hash);
 
            if (it != name.end())
                name[hash].use++;
            else {
                NameUse nu;
 
                nu.use = 1;
                memcpy(&nu.name, &key, sizeof(Name));
 
                name[hash] = nu;
            }
            return hash;
        }
 
 
        inline void erase_name(uint64_t hash) {
 
            HashNameUseMap::iterator it = name.find(hash);
 
            if (it == name.end()) return;
 
            if (--name[hash].use == 0)
                name.erase(it);
        }
 
 
        inline StatusCode internal_get(pTransaction &p_txn, pString &p_str, uint64_t &pop_ent, Locator &what) {
 
            int base;
            pVolatileTransaction p_root;
            EntityKeyHash ek;
            ek.ent_hash = hash(what.entity);
 
            pop_ent = 0;
 
            switch (base = TenBitsAtAddress(what.base)) {
            case BASE_DEQUE_10BIT: {
                HashVolXctMap::iterator it_ent = deque_ent.find(ek.ent_hash);
 
                if (it_ent == deque_ent.end()) return SERVICE_ERROR_ENTITY_NOT_FOUND;
 
                p_root = it_ent->second; }
 
                break;
 
            case BASE_INDEX_10BIT: {
                HashVolXctMap::iterator it_ent = index_ent.find(ek.ent_hash);
 
                if (it_ent == index_ent.end()) return SERVICE_ERROR_ENTITY_NOT_FOUND;
 
                p_root = it_ent->second; }
 
                break;
 
            case BASE_QUEUE_10BIT: {
                HashQueueEntMap::iterator it_ent = queue_ent.find(ek.ent_hash);
 
                if (it_ent == queue_ent.end()) return SERVICE_ERROR_ENTITY_NOT_FOUND;
 
                p_root = it_ent->second.p_root; }
 
                break;
 
            case BASE_TREE_10BIT: {
                HashVolXctMap::iterator it_ent = tree_ent.find(ek.ent_hash);
 
                if (it_ent == tree_ent.end()) return SERVICE_ERROR_ENTITY_NOT_FOUND;
 
                p_root = it_ent->second; }
 
                break;
 
            default:
                return SERVICE_ERROR_WRONG_BASE;
            }
 
            if (p_root == nullptr) return SERVICE_ERROR_EMPTY_ENTITY;
 
            Name key, second;
            int  command;
 
            if (!parse_command(key, command, second, what.key, false))
                return SERVICE_ERROR_PARSING_COMMAND;
 
            switch (command) {
            case COMMAND_JUST_THE_KEY:
                switch (base) {
                case BASE_DEQUE_10BIT: {
                    ek.key_hash = hash(key);
                    EntKeyVolXctMap::iterator it;
 
                    if ((it = deque_key.find(ek)) == deque_key.end()) return SERVICE_ERROR_BLOCK_NOT_FOUND;
 
                    p_txn = it->second;
                    p_str = nullptr; }
 
                    return SERVICE_NO_ERROR;
 
                case BASE_QUEUE_10BIT: {
                    ek.key_hash = hash(key);
                    EntKeyVolXctMap::iterator it;
 
                    if ((it = queue_key.find(ek)) == queue_key.end()) return SERVICE_ERROR_BLOCK_NOT_FOUND;
 
                    p_txn = it->second;
                    p_str = nullptr; }
 
                    return SERVICE_NO_ERROR;
 
                case BASE_TREE_10BIT: {
                    ek.key_hash = hash(key);
                    EntKeyVolXctMap::iterator it;
 
                    if ((it = tree_key.find(ek)) == tree_key.end()) return SERVICE_ERROR_BLOCK_NOT_FOUND;
 
                    p_txn = it->second;
                    p_str = nullptr; }
 
                    return SERVICE_NO_ERROR;
 
                default: {
                    Index::iterator it;
 
                    if ((it = p_root->p_hea->index.find(key)) == p_root->p_hea->index.end()) return SERVICE_ERROR_BLOCK_NOT_FOUND;
 
                    p_txn = nullptr;
                    p_str = &it->second; }
 
                    return SERVICE_NO_ERROR;
                }
 
            case COMMAND_CHILD_10BIT: {
                if (base != BASE_TREE_10BIT) return SERVICE_ERROR_PARSING_COMMAND;
 
                ek.key_hash = hash(key);
                EntKeyVolXctMap::iterator it;
 
                if ((it = tree_key.find(ek)) == tree_key.end()) return SERVICE_ERROR_BLOCK_NOT_FOUND;
 
                p_txn = it->second->p_child;
 
                if (p_txn == nullptr) return SERVICE_ERROR_BLOCK_NOT_FOUND;
 
                p_str = nullptr; }
 
                return SERVICE_NO_ERROR;
 
            case COMMAND_PARENT_10BIT: {
                if (base != BASE_TREE_10BIT) return SERVICE_ERROR_PARSING_COMMAND;
 
                ek.key_hash = hash(key);
                EntKeyVolXctMap::iterator it;
 
                if ((it = tree_key.find(ek)) == tree_key.end()) return SERVICE_ERROR_BLOCK_NOT_FOUND;
 
                p_txn = it->second->p_parent;
 
                if (p_txn == nullptr) return SERVICE_ERROR_BLOCK_NOT_FOUND;
 
                p_str = nullptr; }
 
                return SERVICE_NO_ERROR;
 
            case COMMAND_NEXT_10BIT: {
                ek.key_hash = hash(key);
                EntKeyVolXctMap::iterator it;
 
                switch (base) {
                case BASE_TREE_10BIT: {
                    if ((it = tree_key.find(ek)) == tree_key.end()) return SERVICE_ERROR_BLOCK_NOT_FOUND;
 
                    p_txn = it->second->p_next;
 
                    if (p_txn == nullptr) return SERVICE_ERROR_BLOCK_NOT_FOUND;
 
                    p_str = nullptr; }
 
                    return SERVICE_NO_ERROR;
 
                case BASE_DEQUE_10BIT: {
                    if ((it = deque_key.find(ek)) == deque_key.end()) return SERVICE_ERROR_BLOCK_NOT_FOUND;
 
                    p_txn = it->second->p_next;
                    p_str = nullptr; }
 
                    return SERVICE_NO_ERROR;
                }
                return SERVICE_ERROR_PARSING_COMMAND; }
 
            case COMMAND_PREV_10BIT: {
                if (base != BASE_DEQUE_10BIT) return SERVICE_ERROR_PARSING_COMMAND;
 
                ek.key_hash = hash(key);
                EntKeyVolXctMap::iterator it;
 
                if ((it = deque_key.find(ek)) == deque_key.end()) return SERVICE_ERROR_BLOCK_NOT_FOUND;
 
                p_txn = it->second->p_parent;
                p_str = nullptr; }
 
                return SERVICE_NO_ERROR;
 
            case COMMAND_HIGH_10BIT:
            case COMMAND_XHIGH_10BIT: {
                if (base != BASE_QUEUE_10BIT) return SERVICE_ERROR_PARSING_COMMAND;
 
                if (p_root == nullptr) return SERVICE_ERROR_EMPTY_ENTITY;
 
                p_txn = aat_highest_priority(p_root);
                p_str = nullptr;
 
                if (command == COMMAND_XHIGH_10BIT)
                    pop_ent = ek.ent_hash; }
 
                return SERVICE_NO_ERROR;
 
            case COMMAND_LOW_10BIT:
            case COMMAND_XLOW_10BIT: {
                if (base != BASE_QUEUE_10BIT) return SERVICE_ERROR_PARSING_COMMAND;
 
                if (p_root == nullptr) return SERVICE_ERROR_EMPTY_ENTITY;
 
                p_txn = aat_lowest_priority(p_root);
                p_str = nullptr;
 
                if (command == COMMAND_XLOW_10BIT)
                    pop_ent = ek.ent_hash; }
 
                return SERVICE_NO_ERROR;
 
            case COMMAND_FIRST_10BIT:
            case COMMAND_PFIRST_10BIT: {
                if (   (base != BASE_DEQUE_10BIT)
                    && (base != BASE_TREE_10BIT || command == COMMAND_PFIRST_10BIT)) return SERVICE_ERROR_PARSING_COMMAND;
 
                if (p_root == nullptr) return SERVICE_ERROR_EMPTY_ENTITY;
 
                p_txn = p_root;
                p_str = nullptr;
 
                if (command == COMMAND_PFIRST_10BIT)
                    pop_ent = ek.ent_hash; }
 
                return SERVICE_NO_ERROR;
 
            case COMMAND_LAST_10BIT:
            case COMMAND_PLAST_10BIT: {
                if (base != BASE_DEQUE_10BIT) return SERVICE_ERROR_PARSING_COMMAND;
 
                if (p_root == nullptr) return SERVICE_ERROR_EMPTY_ENTITY;
 
                p_txn = p_root->p_prev;
                p_str = nullptr;
 
                if (command == COMMAND_PLAST_10BIT)
                    pop_ent = ek.ent_hash; }
 
                return SERVICE_NO_ERROR;
            }
 
//          case command == COMMAND_GET_10BIT:  This condition cannot fail or parse_command() would have.
 
            if (base != BASE_INDEX_10BIT) return SERVICE_ERROR_PARSING_COMMAND;
 
            p_txn = nullptr;
            p_str = nullptr;
            pop_ent = ek.ent_hash;
 
            return SERVICE_NO_ERROR;
        }
 
 
        inline bool parse_command(Name &key_out, int &command, Name &second, Name key_in, bool is_put) {
            if (key_in[0] == '~') {
                key_out[0] = 0;
                second [0] = 0;
 
                switch (command = TenBitsAtAddress(&key_in[1])) {
                case COMMAND_FIRST_10BIT:
                case COMMAND_LAST_10BIT:
                    return true;
 
                case COMMAND_PUT_10BIT:
                    return is_put;
 
                case COMMAND_PFIRST_10BIT:
                case COMMAND_PLAST_10BIT:
                case COMMAND_HIGH_10BIT:
                case COMMAND_LOW_10BIT:
                case COMMAND_GET_10BIT:
                case COMMAND_XHIGH_10BIT:
                case COMMAND_XLOW_10BIT:
                    return !is_put;
 
                default:
                    if (!is_put || key_in[1] < '0' || key_in[1] > '9') return false;
 
                    int size, r_len;
 
                    if (sscanf(&key_in[1], "%d%n", &size, &r_len) != 1 || size <= 0 || key_in[r_len + 1] != 0) return false;
 
                    command = COMMAND_SIZE + size;
 
                    return true;
                }
            }
            strcpy(key_out, key_in);
            pChar pc = strchr(key_out, '~');
 
            if (pc == nullptr) {
                command   = COMMAND_JUST_THE_KEY;
                second[0] = 0;
 
                return true;
            }
            (*pc++) = 0;
 
            if (*pc == 0)
                return false;
 
            if (is_put) {
                strcpy(second, pc);
                command = COMMAND_SECOND_ARG;
 
                return true;
            }
            second[0] = 0;
 
            switch (command = TenBitsAtAddress(pc)) {
            case COMMAND_CHILD_10BIT:
            case COMMAND_NEXT_10BIT:
            case COMMAND_PARENT_10BIT:
            case COMMAND_PREV_10BIT:
                return true;
            }
 
            return false;
        }
 
 
        inline uint64_t hash(Name &name) {
 
            int len = strnlen(name, sizeof(Name));
            int siz = sizeof(Name) - len;
 
            if (siz > 1)
                memset(&name[len], 0, siz);
 
            return MurmurHash64A(&name, NAME_SIZE);
        }
 
 
        inline void destroy_queue(uint64_t ent_hash, pVolatileTransaction p_txn) {
 
            if (p_txn != nullptr) {
                destroy_queue(ent_hash, p_txn->p_prev);
                destroy_queue(ent_hash, p_txn->p_next);
 
                destroy_item(BASE_QUEUE_10BIT, ent_hash, p_txn);
            }
        }
 
 
        inline void destroy_tree(uint64_t ent_hash, pVolatileTransaction p_txn) {
 
            if (p_txn != nullptr) {
                destroy_tree(ent_hash, p_txn->p_next);
                destroy_tree(ent_hash, p_txn->p_child);
 
                destroy_item(BASE_TREE_10BIT, ent_hash, p_txn);
            }
        }
 
 
        inline bool aat_to_left(pVolatileTransaction p_item, pVolatileTransaction p_tree) {
 
            return p_item->priority == p_tree->priority ? (uintptr_t) p_item < (uintptr_t) p_tree : p_item->priority < p_tree->priority;
        }
 
 
        inline pVolatileTransaction aat_highest_priority(pVolatileTransaction p_item) {
 
            if (p_item != nullptr) {
                while (p_item->p_next != nullptr)
                    p_item = p_item->p_next;
            };
 
            return p_item;
        };
 
 
        inline pVolatileTransaction aat_lowest_priority(pVolatileTransaction p_item) {
 
            if (p_item != nullptr) {
                while (p_item->p_prev != nullptr)
                    p_item = p_item->p_prev;
            };
 
            return p_item;
        };
 
 
        inline pVolatileTransaction aat_rebalance(pVolatileTransaction p_tree) {
 
            aat_decrease_level(p_tree);
 
            p_tree = aat_skew(p_tree);
            if (p_tree->p_next != nullptr) {
                p_tree->p_next = aat_skew(p_tree->p_next);
 
                if (p_tree->p_next->p_next != nullptr)
                    p_tree->p_next->p_next = aat_skew(p_tree->p_next->p_next);
            };
            p_tree = aat_split(p_tree);
 
            if (p_tree->p_next != nullptr)
                p_tree->p_next = aat_split(p_tree->p_next);
 
            return p_tree;
        }
 
 
        inline bool aat_is_in_tree(pVolatileTransaction p_item, pVolatileTransaction p_tree) {
 
            if (p_tree == nullptr || p_item == nullptr) return false;
 
            if (p_item == p_tree) return true;
 
            if (aat_to_left(p_item, p_tree)) return aat_is_in_tree(p_item, p_tree->p_prev);
 
            return aat_is_in_tree(p_item, p_tree->p_next);
        };
 
 
        inline pVolatileTransaction aat_remove_deep(pVolatileTransaction  p_kill,
                                                    pVolatileTransaction  p_parent,
                                                    pVolatileTransaction  p_tree,
                                                    pVolatileTransaction &p_deep) {
            if (p_kill->p_next != nullptr)
                p_kill->p_next = aat_remove_deep(p_kill->p_next, p_kill, p_tree, p_deep);
 
            else {
                if (p_parent == p_tree) {
                    p_tree->p_prev = p_kill->p_prev;    // Disconnect p_kill
                    p_kill->level  = p_tree->level;
                    p_kill->p_next = p_tree->p_next;
                    p_kill->p_prev = p_tree->p_prev;    // p_kill is the new p_tree
 
                    return aat_rebalance(p_kill);
 
                } else {
                    p_deep = p_kill;                    // Save p_kill for the end
 
                    return p_kill->p_prev;              // Disconnect p_kill
                }
            }
            if (p_parent != p_tree) return aat_rebalance(p_kill);
 
            else {
                p_deep->level  = p_tree->level;
                p_deep->p_next = p_tree->p_next;
                p_deep->p_prev = aat_rebalance(p_kill);
 
                return aat_rebalance(p_deep);
            }
        }
 
 
        inline pVolatileTransaction aat_remove(pVolatileTransaction p_item, pVolatileTransaction p_tree) {
 
            if (p_tree == nullptr || p_item == nullptr) return p_tree;
 
            if (p_item == p_tree) {
                if (p_tree->p_prev == nullptr) return p_tree->p_next;
 
                else {
                    if (p_tree->p_next == nullptr) return p_tree->p_prev;
 
                    else {
                        pVolatileTransaction p_deep = nullptr;
                        return aat_remove_deep(p_tree->p_prev, p_tree, p_tree, p_deep);
                    }
                }
            } else {
                if (aat_to_left(p_item, p_tree))
                    p_tree->p_prev = aat_remove(p_item, p_tree->p_prev);
 
                else
                    p_tree->p_next = aat_remove(p_item, p_tree->p_next);
            }
 
            return aat_rebalance(p_tree);
        };
 
 
        inline void aat_decrease_level(pVolatileTransaction p_item) {
 
            if (p_item->p_prev == nullptr) {
                if (p_item->p_next == nullptr)
                    p_item->level = 1;
 
                else {
                    int should_be = p_item->p_next->level + 1;
 
                    if (should_be < p_item->level)
                        p_item->level = should_be;
                }
            } else {
                if (p_item->p_next == nullptr) {
                    int should_be = p_item->p_prev->level + 1;
 
                    if (should_be < p_item->level)
                        p_item->level = should_be;
 
                } else {
                    int should_be = std::min(p_item->p_prev->level, p_item->p_next->level) + 1;
 
                    if (should_be < p_item->level) {
                        p_item->level = should_be;
 
                        if (should_be < p_item->p_next->level)
                            p_item->p_next->level = should_be;
                    }
                }
            }
        };
 
 
        inline pVolatileTransaction aat_skew(pVolatileTransaction p_item) {
            // rotate p_next if p_prev child has same level
 
            if (p_item->p_prev != nullptr && p_item->level == p_item->p_prev->level) {
                pVolatileTransaction p_left = p_item->p_prev;
 
                p_item->p_prev = p_left->p_next;
                p_left->p_next = p_item;
 
                return p_left;
            }
 
            return p_item;
        };
 
 
        inline pVolatileTransaction aat_split(pVolatileTransaction p_item) {
            // rotate p_prev if there are two p_next children on same level
 
            if (   p_item->p_next != nullptr
                && p_item->p_next->p_next != nullptr
                && p_item->level == p_item->p_next->p_next->level) {
 
                pVolatileTransaction p_right = p_item->p_next;
 
                p_item->p_next  = p_right->p_prev;
                p_right->p_prev = p_item;
                if (p_item->p_prev == nullptr && p_item->p_next == nullptr && p_item->level > 1) {
                    p_item->level = 1;
/* The following condition looks like it could be problematic (since the previous one is already an undocumented deviation from the
canonical method description). It has never been observed (in intense unit testing) and should only be considered if problems happen.
It may very well be impossible, who knows. Just keep it as a remark, unless someone smarter proves it unnecessary. */
                    // if (   p_right->level != 2
                    //  || (   p_right->p_next->p_next != nullptr
                    //      && p_right->p_next->p_next->level >= 2))
                    //      p_right->level++;   // This is actually a breakpoint possibility, not a solution!!
 
                } else
                    p_right->level++;
 
                return p_right;
            }
 
            return p_item;
        };
 
 
        inline pVolatileTransaction aat_insert(pVolatileTransaction p_new, pVolatileTransaction p_tree) {
            // Do the normal binary tree insertion procedure. Set the result of the
            // recursive call to the correct child in case a new node was created or the
            // root of the subtree changes.
 
            if (p_tree == nullptr) {
                p_new->level = 1;
                p_new->p_prev = nullptr;
                p_new->p_next = nullptr;
 
                return p_new;
 
            } else {
 
                if (aat_to_left(p_new, p_tree))
                    p_tree->p_prev = aat_insert(p_new, p_tree->p_prev);
 
                else
                    p_tree->p_next = aat_insert(p_new, p_tree->p_next);
            }
 
            // Perform aat_skew and then aat_split. The conditionals that determine whether or
            // not a rotation will occur or not are inside of the procedures, as given above.
 
            p_tree = aat_skew(p_tree);
            p_tree = aat_split(p_tree);
 
            return p_tree;
        };
 
        uint64_t        key_seed;           
        HashNameUseMap  name {};            
        HashQueueEntMap queue_ent {};       
        HashVolXctMap   deque_ent {};       
        HashVolXctMap   tree_ent {};        
        HashVolXctMap   index_ent {};       
        EntKeyVolXctMap deque_key {};       
        EntKeyVolXctMap queue_key {};       
        EntKeyVolXctMap tree_key {};        
};
typedef Volatile *pVolatile;                
 
#ifdef CATCH_TEST
 
// Instancing Volatile
// -------------------
 
extern Volatile VOL;
 
#endif
 
} // namespace jazz_elements
 
#endif // ifndef INCLUDED_JAZZ_ELEMENTS_VOLATILE