container.h

/* Jazz (c) 2018-2025 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 <map>
#include <atomic>
#include <thread>
#include <regex>
 
 
#include "src/jazz_elements/tuple.h"
 
#ifdef CATCH_TEST
#ifndef INCLUDED_JAZZ_CATCH2
#define INCLUDED_JAZZ_CATCH2
 
#include "src/catch2/catch.hpp"
 
#endif
#endif
 
 
#ifndef INCLUDED_JAZZ_ELEMENTS_CONTAINER
#define INCLUDED_JAZZ_ELEMENTS_CONTAINER
 
 
namespace jazz_elements
{
 
#define MAX_SIZE_OF_CELL_AS_TEXT        48                  
 
#define NA_AS_TEXT                      {"NA\0"}            
#define LENGTH_NA_AS_TEXT               2                   
 
#define PARSE_ERROR_UNEXPECTED_EOF      101     
#define PARSE_ERROR_UNEXPECTED_CHAR     102     
#define PARSE_ERROR_ITEM_NAME           103     
#define PARSE_ERROR_ITEM_NAME_MISMATCH  104     
#define PARSE_ERROR_TENSOR_EXPLORATION  105     
 
#define PARSE_ERROR_KIND_EXPLORATION    106     
#define PARSE_ERROR_TOO_MANY_ITEMS      107     
#define PARSE_ERROR_EXPECTED_EOF        108     
#define PARSE_ERROR_TENSOR_FILLING      109     
#define PARSE_ERROR_TEXT_FILLING        110     
 
#define FILL_NEW_DONT_FILL                0     
#define FILL_NEW_WITH_ZERO                1     
#define FILL_NEW_WITH_NA                  2     
#define FILL_WITH_TEXTFILE                3     
 
#define BUILD_TUPLE                       1     
#define BUILD_KIND                        2     
 
#define BLOCK_STATUS_READY                0     
#define BLOCK_STATUS_EMPTY                1     
#define BLOCK_STATUS_DESTROYED            2     
 
#define LOCK_NUM_RETRIES_BEFORE_YIELD   100     
 
#define LOCK_WEIGHT_OF_WRITE            46341
 
// State based parser types:
#define EIGHT_BIT_LONG                   256    
#define MAX_TRANSITION_REGEX_LEN          32    
#define PSTATE_INVALID_CHAR              255    
 
// Writing modes for put(1): What to do is block exists:
#define WRITE_ONLY_IF_EXISTS            0x01    
#define WRITE_ONLY_IF_NOT_EXISTS        0x02    
#define WRITE_ANY_RESTRICTION           0x03    
 
// Writing modes for put(2): What to write
#define WRITE_AS_BASE_DEFAULT           0x00    
#define WRITE_AS_STRING                 0x04    
#define WRITE_AS_CONTENT                0x08    
#define WRITE_AS_FULL_BLOCK             0x10    
#define WRITE_AS_ANY_WRITE              0x1C    
 
struct ParseNextStateLUT {
    unsigned char next[EIGHT_BIT_LONG];         
};
 
 
struct ParseStateTransition {
    int  from;                                  
    int  to;                                    
    char rex[MAX_TRANSITION_REGEX_LEN];         
};
 
 
void compile_next_state_LUT(ParseNextStateLUT lut[], int num_states, ParseStateTransition trans[]);
 
 
typedef std::atomic<int32_t> Lock32;
 
 
typedef class Container *pContainer;
 
 
typedef std::map<std::string, pContainer> BaseNames;
 
 
struct Transaction {
    union {
        pBlock          p_block;            
        pBlockHeader    p_hea;              
    };
    Lock32              _lock_;             
    int                 status;             
    pContainer          p_owner;            
};
typedef Transaction *pTransaction;          
 
 
typedef struct ExtraLocator *pExtraLocator; 
 
struct Locator {
    char base   [SHORT_NAME_SIZE];  
    Name entity;                    
    Name key;                       
 
    union {
        int           attribute;    
        pExtraLocator p_extra;      
    };
};
typedef Locator *pLocator;          
 
 
struct StoredTransaction: Transaction {
    StoredTransaction *p_next;                      
};
typedef StoredTransaction *pStoredTransaction;      
 
 
typedef std::map<std::string, int>  MapSI;
 
 
class Container : public Service {
 
    public:
 
        Container(pLogger a_logger, pConfigFile a_config);
       ~Container();
 
       // Service API
 
        StatusCode start    ();
        StatusCode shut_down();
 
        // .enter_read() .enter_write() .leave_read() .leave_write() .lock_container() .unlock_container()
 
        void enter_read (pTransaction p_txn);
        void enter_write(pTransaction p_txn);
        void leave_read (pTransaction p_txn);
        void leave_write(pTransaction p_txn);
 
        // - Allocation: .new_block(), .destroy_transaction()
 
        // 1. new_block(): Create a Tensor from raw data specifying everything from scratch.
        StatusCode new_block   (pTransaction       &p_txn,
                                int                 cell_type,
                                int                *dim,
                                int                 fill_tensor     = FILL_NEW_DONT_FILL,
                                int                 stringbuff_size = 0,
                                const char         *p_text          = nullptr,
                                char                eol             = '\n',
                                AttributeMap       *att             = nullptr);
 
        // 2. new_block(): Create a Kind or Tuple from arrays of StaticBlockHeader, names, and, in the case of a tuple, Tensors.
        StatusCode new_block   (pTransaction       &p_txn,
                                int                 num_items,
                                StaticBlockHeader   p_hea[],
                                Name                p_names[],
                                pBlock              p_block[],
                                AttributeMap       *dims            = nullptr,
                                AttributeMap       *att             = nullptr);
 
        // 3. new_block(): Create a Tensor by selecting rows (filtering) from another Tensor.
        StatusCode new_block   (pTransaction       &p_txn,
                                pBlock              p_from,
                                pBlock              p_row_filter,
                                AttributeMap       *att             = nullptr);
 
        // 4. new_block(): Create a Tensor by selecting an item from a Tuple.
        StatusCode new_block   (pTransaction       &p_txn,
                                pTuple              p_from,
                                pChar               name,
                                AttributeMap       *att             = nullptr);
 
        // 5. new_block(): Create a Tensor, Kind or Tuple from a Text block kept as a Tensor of CELL_TYPE_BYTE of rank == 1.
        StatusCode new_block   (pTransaction       &p_txn,
                                pBlock              p_from_text,
                                int                 cell_type,
                                pKind               p_as_kind       = nullptr,
                                AttributeMap       *att             = nullptr);
 
        // 6. new_block(): Create a Tensor of CELL_TYPE_BYTE of rank == 1 with a text serialization of a Tensor, Kind or Tuple.
        StatusCode new_block   (pTransaction       &p_txn,
                                pBlock              p_from_raw,
                                pChar               p_fmt           = nullptr,
                                bool                ret_as_string   = false,
                                AttributeMap       *att             = nullptr);
 
        // 7. new_block(): Create an empty Index block.
        StatusCode new_block   (pTransaction       &p_txn,
                                int                 cell_type);
 
        // 8. new_block(): Create a Tuple of (key:STRING[length],value:STRING[length]) with the content of an Index.
        StatusCode new_block   (pTransaction       &p_txn,
                                Index              &index);
 
        // Support for transactions creation/destruction
 
        virtual StatusCode new_transaction(pTransaction &p_txn);
        virtual void destroy_transaction  (pTransaction &p_txn);
 
        // Crud: .get(), .header(), .put(), .new_entity(), .remove(), .copy()
 
        // The "easy" interface: Uses strings instead of locators. Is translated to the native interface by an as_locator() call.
        virtual StatusCode get         (pTransaction       &p_txn,
                                        pChar               p_what);
        virtual StatusCode get         (pTransaction       &p_txn,
                                        pChar               p_what,
                                        pBlock              p_row_filter);
        virtual StatusCode get         (pTransaction       &p_txn,
                                        pChar               p_what,
                                        pChar               name);
        virtual StatusCode locate      (Locator            &location,
                                        pChar               p_what);
        virtual StatusCode header      (StaticBlockHeader  &hea,
                                        pChar               p_what);
        virtual StatusCode header      (pTransaction       &p_txn,
                                        pChar               p_what);
        virtual StatusCode put         (pChar               p_where,
                                        pBlock              p_block,
                                        int                 mode = WRITE_AS_BASE_DEFAULT);
        virtual StatusCode new_entity  (pChar               p_where);
        virtual StatusCode remove      (pChar               p_where);
        virtual StatusCode copy        (pChar               p_where,
                                        pChar               p_what);
 
        // The function call interface: exec()/modify().
        virtual StatusCode exec        (pTransaction       &p_txn,
                                        Locator            &function,
                                        pTuple              p_args);
        virtual StatusCode modify      (Locator            &function,
                                        pTuple              p_args);
 
        // The parser: This simple regex-based parser only needs override for Channels.
        virtual StatusCode as_locator  (Locator            &result,
                                        pChar               p_what);
 
        // The "native" interface: This is what really does the job and all Container descendants implement.
        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);
 
        inline StatusCode unwrap_received(pTransaction &p_txn) {
 
            pBlock p_blk = (pBlock) &p_txn->p_block->tensor.cell_byte[0];
            int    size  = p_txn->p_block->size;
 
            if (p_blk->total_bytes == size && p_blk->check_hash()) {
                pBlock p_new = block_malloc(size);
                if (p_new == nullptr) {
                    destroy_transaction(p_txn);
 
                    return SERVICE_ERROR_NO_MEM;
                }
                memcpy(p_new, p_blk, size);
 
                alloc_bytes -= p_txn->p_block->total_bytes;
                free(p_txn->p_block);
 
                p_txn->p_block = p_new;
 
                return SERVICE_NO_ERROR;
            }
            if (strnlen((pChar) p_blk, size) != size)
                return SERVICE_NO_ERROR;
 
            pTransaction p_aux;
 
            reinterpret_cast <pChar>(p_blk)[size] = 0;
 
            if (new_block(p_aux, CELL_TYPE_STRING, nullptr, FILL_WITH_TEXTFILE, 0, (pChar) p_blk, 0) != SERVICE_NO_ERROR) {
                destroy_transaction(p_txn);
 
                return SERVICE_ERROR_NO_MEM;
            }
            std::swap(p_txn->p_block, p_aux->p_block);
 
            destroy_transaction(p_aux);
 
            return SERVICE_NO_ERROR;
        }
 
        inline StatusCode unwrap_received(pTransaction &p_txn, pBlock p_maybe_block, int rec_size) {
 
            if (rec_size > sizeof(BlockHeader) && p_maybe_block->total_bytes == rec_size && p_maybe_block->check_hash()) {
                int ret = new_transaction(p_txn);
 
                if (ret != SERVICE_NO_ERROR)
                    return ret;
 
                pBlock p_new = block_malloc(rec_size);
                if (p_new == nullptr) {
                    destroy_transaction(p_txn);
 
                    return SERVICE_ERROR_NO_MEM;
                }
                memcpy(p_new, p_maybe_block, rec_size);
 
                p_txn->p_block = p_new;
                p_txn->status  = BLOCK_STATUS_READY;
 
                return SERVICE_NO_ERROR;
            }
            if (strnlen((pChar) p_maybe_block, rec_size) != rec_size) {
                int dim[MAX_TENSOR_RANK] = {rec_size, 0};
                int ret                  = new_block(p_txn, CELL_TYPE_BYTE, dim, FILL_NEW_DONT_FILL);
 
                if (ret == SERVICE_NO_ERROR)
                    memcpy(&p_txn->p_block->tensor.cell_byte[0], p_maybe_block, rec_size);
 
                return ret;
            }
 
            return new_block(p_txn, CELL_TYPE_STRING, nullptr, FILL_WITH_TEXTFILE, 0, (pChar) p_maybe_block, 0);
        }
 
        // Support for container names in the BaseAPI .base_names()
 
        void base_names(BaseNames &base_names);
 
#ifndef CATCH_TEST
    protected:
#endif
 
        inline void* malloc(size_t size) {
            if (alloc_bytes + size >= fail_alloc_bytes)
                return nullptr;
 
            void * ret = std::malloc(size);
 
            if (ret != nullptr)
                alloc_bytes += size;
 
#ifdef CATCH_TEST
            if ((uint64_t) ret & 0x7)
                log(LOG_ERROR, "Misaligned block !!");
#endif
 
            return ret;
        }
 
        inline pBlock block_malloc(size_t size) {
            pBlock p_blk = (pBlock) malloc(size);
 
            if (p_blk != nullptr) {
                p_blk->cell_type   = 0;
                p_blk->total_bytes = 0;
            }
 
            return p_blk;
        }
 
        inline void lock_container() {
            int     retry = 0;
            int32_t lock = 0;
            int32_t next = 1;
 
            while (true) {
                if (_lock_.compare_exchange_weak(lock, next))
                    return;
 
                if (++retry > LOCK_NUM_RETRIES_BEFORE_YIELD) {
                    std::this_thread::yield();
                    retry = 0;
                }
            }
        }
 
        void unlock_container() {
            _lock_ = 0;
        }
 
        StatusCode destroy_container();
 
        inline int from_hex(char c) {
            return (c < 65) ? c - 48 : (c > 96) ? c - 87 : c - 55;
        }
 
        int max_transactions;               
        uint64_t warn_alloc_bytes;          
        uint64_t fail_alloc_bytes;          
        uint64_t alloc_bytes;               
        pTransaction p_buffer;              
        pTransaction p_free;                
        bool alloc_warning_issued;          
        Lock32 _lock_;                      
 
#ifndef CATCH_TEST
    private:
#endif
 
        char HEX[16] = {'0', '1', '2', '3', '4', '5', '6', '7', '8', '9', 'a', 'b', 'c', 'd', 'e', 'f'};    
 
        StatusCode new_container();
 
        inline int skip_space(pChar &p_in, int &num_bytes) {
            while (num_bytes > 0) {
                if (*p_in == 0)
                    return 0;
 
                if (*p_in == ' ' || *p_in == '\t' || *p_in == '\n') {
                    p_in++;
                    num_bytes--;
                } else
                    break;
            }
 
            return num_bytes;
        }
 
        inline char get_char(pChar &p_in, int &num_bytes) {
            if (num_bytes < 1)
                return 0;
 
            num_bytes--;
 
            return *(p_in++);
        }
 
        inline bool get_item_name(pChar &p_in, int &num_bytes, pChar p_out, bool check_quotes = true, bool check_colon = true) {
            if (skip_space(p_in, num_bytes) <= 1)
                return false;
 
            if (check_quotes && (get_char(p_in, num_bytes) != '"'))
                return false;
 
            char ch = get_char(p_in, num_bytes);
 
            if (ch < 'A' || ch > 'z' || (ch > 'Z' && ch < 'a'))
                return false;
 
            *(p_out++) = ch;
 
            for (int i = 1; i < NAME_LENGTH; i++) {
                if (!check_quotes && (*p_in == ',' || *p_in == ']'))
                    break;
 
                ch = get_char(p_in, num_bytes);
 
                if (check_quotes && ch == '"')
                    break;
 
                if (ch < '0' || ch > 'z' || (ch > '9' && ch < 'A') || (ch > 'Z' && ch < '_') || ch == 0x60)
                    return false;
 
                *(p_out++) = ch;
            }
 
            *p_out = 0;
 
            if (check_quotes && (ch != '"') && (get_char(p_in, num_bytes) != '"'))
                return false;
 
            if (check_colon) {
                if (skip_space(p_in, num_bytes) <= 1)
                    return false;
 
                if (get_char(p_in, num_bytes) != ':')
                    return false;
            }
 
            return true;
        }
 
        inline bool sscanf_int32(pChar &p_in, int &num_bytes, int &result) {
            int r_len;
 
            if (sscanf(p_in, "%i%n", &result, &r_len) > 0) {
                p_in      += r_len;
                num_bytes -= r_len;
 
                return num_bytes >= 0;
            } else
                return false;
        }
 
        inline bool push_int_cell(pChar cell, pChar &p_st, int * &p_out) {
 
            if (p_st == cell) {
                *(p_out++) = INTEGER_NA;
 
                return true;
            }
            *p_st = 0;
            p_st  = cell;
 
            if (sscanf(p_st, "%i", p_out) != 1)
                return false;
 
            p_out++;
 
            return true;
        }
 
        inline bool push_int_cell(pChar cell, pChar &p_st, long long * &p_out) {
 
            if (p_st == cell) {
                *(p_out++) = LONG_INTEGER_NA;
 
                return true;
            }
            *p_st = 0;
            p_st  = cell;
 
            if (sscanf(p_st, "%lli", p_out) != 1)
                return false;
 
            p_out++;
 
            return true;
        }
 
        inline bool push_bool_cell(pChar cell, pChar &p_st, bool * &p_out) {
 
            if (p_st == cell) {
                *reinterpret_cast<uint8_t *>(p_out++) = BYTE_BOOLEAN_NA;
 
                return true;
            }
            *p_st = 0;
            p_st  = cell;
 
            *(p_out++) = cell[0] == '1';
 
            return true;
        }
 
        inline bool push_bool_cell(pChar cell, pChar &p_st, uint32_t * &p_out) {
 
            if (p_st == cell) {
                *(p_out++) = BOOLEAN_NA;
 
                return true;
            }
            *p_st = 0;
            p_st  = cell;
 
            *(p_out++) = cell[0] == '1';
 
            return true;
        }
 
        inline bool push_real_cell(pChar cell, pChar &p_st, float * &p_out) {
 
            if (p_st == cell) {
                *(p_out++) = SINGLE_NA;
 
                return true;
            }
            *p_st = 0;
            p_st  = cell;
 
            if (sscanf(p_st, "%f", p_out) != 1)
                return false;
 
            p_out++;
 
            return true;
        }
 
        inline bool push_real_cell(pChar cell, pChar &p_st, double * &p_out) {
 
            if (p_st == cell) {
                *(p_out++) = DOUBLE_NA;
 
                return true;
            }
            *p_st = 0;
            p_st  = cell;
 
            if (sscanf(p_st, "%lf", p_out) != 1)
                return false;
 
            p_out++;
 
            return true;
        }
 
        inline bool push_time_cell(pChar cell, pChar &p_st, time_t * &p_out, pChar fmt) {
 
            if (p_st == cell) {
                *(p_out++) = TIME_POINT_NA;
 
                return true;
            }
            *p_st = 0;
            p_st  = cell;
 
            struct tm timeinfo = {0};
 
            if (strptime(p_st, fmt, &timeinfo) == nullptr)
                return false;
 
            time_t xx = timegm(&timeinfo);
            if (xx < 0)
                return false;
 
            *(p_out++) = xx;
 
            return true;
        }
 
        bool get_type_and_shape  (pChar &p_in, int &num_bytes, ItemHeader *item_hea, MapSI &dims);
        bool get_shape_and_size  (pChar &p_in, int &num_bytes, int cell_type, ItemHeader *item_hea);
        bool fill_text_buffer    (pChar &p_in, int &num_bytes, pChar p_out, int num_cells, int is_NA[], int hasLN[]);
        bool fill_tensor         (pChar &p_in, int &num_bytes, pBlock p_block);
 
        int new_text_block       (pTransaction &p_txn, ItemHeader &item_hea, pChar &p_in, int &num_bytes, AttributeMap *att = nullptr);
 
        int tensor_int_as_text   (pBlock p_block, pChar p_dest, pChar p_fmt);
        int tensor_bool_as_text  (pBlock p_block, pChar p_dest);
        int tensor_float_as_text (pBlock p_block, pChar p_dest, pChar p_fmt);
        int tensor_string_as_text(pBlock p_block, pChar p_dest);
        int tensor_time_as_text  (pBlock p_block, pChar p_dest, pChar p_fmt);
        int tensor_tuple_as_text (pTuple p_tuple, pChar p_dest, pChar p_fmt, int item_len[]);
        int tensor_kind_as_text  (pKind  p_kind,  pChar p_dest);
 
        inline void opening_brackets(int rank, pChar &p_ret) {
            for (int i = 0; i < rank; i++)
                *(p_ret++) = '[';
 
            *p_ret = 0;
        }
 
        inline pChar as_shape(int rank, int dim[], pChar p_ret, pKind p_kind) {
            *(p_ret++) = '[';
 
            for (int i = 0; i < rank; i++) {
                int k = dim[i];
                if (k < 0) {
                    char *p_dim_name = &p_kind->p_string_buffer()->buffer[-k];
                    strcpy(p_ret, p_dim_name);
                    p_ret += strlen(p_dim_name);
 
                } else
                    p_ret += sprintf(p_ret, "%i", k);
 
                if (i < rank - 1) {
                    *(p_ret++) = ',';
                    *(p_ret++) = ' ';
                }
            }
 
            *(p_ret++) = ']';
 
            *p_ret = 0;
 
            return p_ret;
        }
 
        inline void as_hex(pChar &p_dest, uint8_t bl) {
            *(p_dest++) = '\\';
            *(p_dest++) = 'x';
            *(p_dest++) = HEX[bl >> 4];
            *(p_dest++) = HEX[bl & 0x0f];
        }
 
        inline void separator(int rank_1, int dim[], int idx[], pChar &p_ret) {
            for (int i = rank_1; i >= 0; i--) {
                idx[i]++;
 
                if (idx[i] == dim[i]) {
                    idx[i] = 0;
                    *(p_ret++) = ']';
                } else {
                    *(p_ret++) = ',';
                    *(p_ret++) = ' ';
 
                    for (int j = 0; j < rank_1 - i; j++)
                        *(p_ret++) = '[';
 
                    break;
                }
            }
            *p_ret = 0;
        }
 
        inline int separator_len(int rank_1, int dim[], int idx[]) {
            int ret = 0;
 
            for (int i = rank_1; i >= 0; i--) {
                idx[i]++;
 
                if (idx[i] == dim[i]) {
                    idx[i] = 0;
                    ret++;
                } else {
                    ret += 2;
 
                    if (i == rank_1)
                        return ret;
 
                    ret += rank_1 - i;
 
                    return ret;
                }
            }
            return ret;
        }
};
 
// Instancing container, logger and config
// ---------------------------------------
 
extern ConfigFile CONFIG;   
extern Logger     LOGGER;   
 
#ifdef CATCH_TEST
 
void compare_full_blocks(pBlock p_bl1, pBlock p_bl2, bool skip_value_check = false);
 
extern Container  CNT;      
 
#endif
 
} // namespace jazz_elements
 
#endif // ifndef INCLUDED_JAZZ_ELEMENTS_CONTAINER