//-< DBSCLS.CXX >----------------------------------------------------*--------*
// GOODS                     Version 1.0         (c) 1997  GARRET    *     ?  *
// (Generic Object Oriented Database System)                         *   /\|  *
//                                                                   *  /  \  *
//                          Created:     28-Mar-97    K.A. Knizhnik  * / [] \ *
//                          Last update: 18-Jun-97    K.A. Knizhnik  * GARRET *
//-------------------------------------------------------------------*--------*
// Some common database classes
//-------------------------------------------------------------------*--------*

#include "dbscls.h"

BEGIN_GOODS_NAMESPACE

field_descriptor& String::describe_components()
{
    return NO_FIELDS;
}

//
// Registration of array classes
//

REGISTER_TEMPLATE(ArrayOfByte, object, pessimistic_repeatable_read_scheme);
REGISTER_TEMPLATE(ArrayOfInt, object, pessimistic_repeatable_read_scheme);
REGISTER_TEMPLATE(ArrayOfDouble, object, pessimistic_repeatable_read_scheme);
REGISTER_TEMPLATE(ArrayOfObject, object, pessimistic_repeatable_read_scheme);
REGISTER(String, ArrayOfByte, pessimistic_repeatable_read_scheme);

//
// Set methods
//

ref<set_member> set_member::first() const
{
    return owner->first; 
}

ref<set_member> set_member::last() const
{
    return owner->last; 
}

int set_member::compare(const char* key) const
{
    return compare(key, strlen(key)+1);
}

int set_member::compare(ref<set_member> mbr) const
{
    return -mbr->compare(key, size - offsetof(set_member, key)); 
}

int set_member::compare(const char* key, size_t size) const
{
    size_t this_size = this->size - offsetof(set_member, key);
    size_t len = size < this_size ? size : this_size; 
    int diff = memcmp(this->key, key, len);
    return diff == 0 ? this_size - size : diff; 
}

int set_member::compareIgnoreCase(const char* key) const
{
    int n = size - offsetof(set_member, key);
    char const* p = this->key;
    while (--n >= 0) { 
        if (*key == '\0') 
            { 
            return 1;
        }
        if (toupper(*key & 0xFF) != toupper(*p & 0xFF)) { 
            return toupper(*key & 0xFF) - toupper(*p & 0xFF);
        }
        p += 1;
        key += 1;
    }
    return -(*key & 0xFF);
}

size_t set_member::copyKeyTo(char* buf, size_t bufSize) const
{
    size_t len = size - offsetof(set_member, key);
    if (len < bufSize) { 
        bufSize = len;
        buf[len] = '\0';
    }
    memcpy(buf, key, bufSize);
    return len;
}

skey_t set_member::get_key() const
{
    return str2key(key, size - offsetof(set_member, key)); 
}

skey_t set_member::str2key(const char* key_str, size_t key_len)
{
#ifdef USE_LOCALE_SETTINGS
    char buf[sizeof(skey_t)];
    strxfrm(buf, (char*)key_str, sizeof buf);
    key_str = buf;
#endif
    skey_t key = 0;
    nat1* s = (nat1*)key_str;
    for (int n = sizeof(skey_t); key_len != 0 && --n >= 0; key_len -= 1) { 
        key |= skey_t(*s++) << (n << 3);
    } 
    return key;
} 

field_descriptor& set_member::describe_components()
{
    return FIELD(next), FIELD(prev), FIELD(owner), FIELD(obj), VARYING(key); 
}

ref<set_member> set_owner::find(const char* key, size_t key_len) const
{
    for (ref<set_member> mbr = first; mbr != NULL; mbr = mbr->next) { 
        if (mbr->compare(key, key_len) == 0) { 
            return mbr; 
        }
    }
    return NULL;
} 

void set_owner::put_first(ref<set_member> mbr)
{
    assert(mbr->owner == NULL && mbr->next == NULL && mbr->prev == NULL);
    modify(mbr)->owner = this;
    modify(mbr)->next = first;
    modify(mbr)->prev = NULL; 
    if (!first.is_nil()) { 
        modify(first)->prev = mbr;
    } else { 
        last = mbr; 
    }
    first = mbr; 
    n_members += 1;
}

void set_owner::put_last(ref<set_member> mbr)
{
    assert(mbr->owner == NULL && mbr->next == NULL && mbr->prev == NULL);
    modify(mbr)->owner = this;
    modify(mbr)->next = NULL;
    modify(mbr)->prev = last; 
    if (!last.is_nil()) { 
        modify(last)->next = mbr;
    } else { 
        first = mbr; 
    }
    last = mbr; 
    n_members += 1;
}

void set_owner::put_before(ref<set_member> before, ref<set_member> mbr)
{
    assert(mbr->owner == NULL && mbr->next == NULL && mbr->prev == NULL);
    if (first == before) { 
        modify(this)->put_first(mbr);
    } else { 
        modify(mbr)->owner = this;
        modify(mbr)->next = before;
        modify(mbr)->prev = before->prev; 
        modify(before)->prev = mbr;
        modify(mbr->prev)->next = mbr;
        n_members += 1;
    }
}


void set_owner::put_after(ref<set_member> after, ref<set_member> mbr)
{
    assert(mbr->owner == NULL && mbr->next == NULL && mbr->prev == NULL);
    if (last == after) { 
        modify(this)->put_last(mbr);
    } else {  
        modify(mbr)->owner = this;
        modify(mbr)->next = after->next;
        modify(mbr)->prev = after; 
        modify(after)->next = mbr;
        modify(mbr->next)->prev = mbr;
        n_members += 1;
    }
}

void set_owner::remove(ref<set_member> mbr)
{
    if (mbr == first) { 
        (void)get_first();
    } else if (mbr == last) { 
        (void)get_last();
    } else { 
        assert(mbr->owner == this);
        modify(mbr->prev)->next = mbr->next; 
        modify(mbr->next)->prev = mbr->prev; 
        modify(mbr)->next = NULL; 
        modify(mbr)->prev = NULL; 
        modify(mbr)->owner = NULL; 
        n_members -= 1;
    }    
}
   
ref<set_member> set_owner::get_first()
{
    ref<set_member> mbr = first; 
    if (!mbr.is_nil()) { 
        first = first->next; 
        if (first.is_nil()) { 
            last = NULL; 
        } else { 
            modify(first)->prev = NULL; 
        }
        modify(mbr)->next = NULL; 
        modify(mbr)->prev = NULL; 
        modify(mbr)->owner = NULL; 
        n_members -= 1;
    }
    return mbr; 
}

ref<set_member> set_owner::get_last()
{
    ref<set_member> mbr = last; 
    if (!mbr.is_nil()) { 
        last = last->prev; 
        if (last.is_nil()) { 
            first = NULL; 
        } else { 
            modify(last)->next = NULL; 
        }
        modify(mbr)->next = NULL; 
        modify(mbr)->prev = NULL; 
        modify(mbr)->owner = NULL; 
        n_members -= 1;
    }
    return mbr; 
}

size_t set_owner::iterate(member_function f, void const* arg) const
{
    size_t count = 0;
    ref<set_member> next, mbr;
    for (mbr = first; !mbr.is_nil(); mbr = next) {
        next = mbr->next;
        f(mbr, arg);
        count += 1;
    }
    return count;
}


field_descriptor& set_owner::describe_components()
{
    return FIELD(first), FIELD(last), FIELD(obj), FIELD(n_members); 
}

REGISTER(set_member, object, optimistic_scheme);
REGISTER(set_owner, object, pessimistic_repeatable_read_scheme);

//
// B tree
//

ref<set_member> B_tree::find(skey_t key) const
{
    if (root.is_nil()) { 
        return NULL; 
    } else { 
        return root->find(height, key, n); 
    }
}

ref<set_member> B_tree::findGE(skey_t key) const
{
    if (root.is_nil()) { 
        return NULL; 
    } else { 
        return root->findGE(height, key, n); 
    }
}

ref<set_member> B_tree::find(const char* str, size_t len, skey_t key) const
{
    if (root != NULL) { 
        ref<set_member> mbr = find(key);
        while (mbr != NULL) { 
            int diff = mbr->compare(str, len);
            if (diff == 0) { 
                return mbr;
            } else if (diff > 0) { 
                return NULL; 
            }
            mbr = mbr->next;
        }
    }
    return NULL;
}

ref<set_member> B_tree::findGE(const char* str, size_t len, skey_t key) const
{
    if (root != NULL) { 
        ref<set_member> mbr = findGE(key);
        while (mbr != NULL) { 
            int diff = mbr->compare(str, len);
            if (diff >= 0) { 
                return mbr;
            }
            mbr = mbr->next;
        }
    }
    return NULL;
}

ref<set_member> B_tree::find(const char* str) const
{
    size_t len = strlen(str); 
    return find(str, len + 1, set_member::str2key(str, len)); 
}

ref<set_member> B_tree::findGE(const char* str) const
{
    size_t len = strlen(str); 
    return findGE(str, len, set_member::str2key(str, len)); 
}

#define MAX_KEY ((skey_t)-1)

void B_tree::insert(ref<set_member> mbr) 
{       
    B_page::item ins;
    ins.p = mbr;
    ins.key = mbr->get_key();

    if (root == NULL) { 
        put_last(mbr); 
        root = B_page::create(NULL, ins, n); 
        height = 1;
    } else if (root->find_insert_position(height, this, ins, n) 
               == B_page::overflow) 
    { 
        root = B_page::create(root, ins, n);
        height += 1;
    } 
}

void B_tree::remove(ref<set_member> mbr)
{
    B_page::item rem; 
    rem.key = mbr->get_key(); 
    rem.p = mbr; 
    if (modify(root)->find_remove_position(height, rem, n) == B_page::underflow
        && root->m == n*2-1) 
    { 
        root = root->e[n*2-1].p;
        height -= 1;
    }
    set_owner::remove(mbr);
}

void B_tree::ok() const 
{
    if (root != NULL) { 
        root->ok(0, MAX_KEY, 0, height, n);
    }
}

void B_tree::dump() const
{ 
    console::output("B_tree[%08lx]: height = %d, records = %d\n",
                    hnd->opid, height, n_members); 
    if (root != NULL) { 
        root->dump(0, height, n); 
    }
}

field_descriptor& B_tree::describe_components()
{
    return FIELD(root), FIELD(height), FIELD(n); 
}


B_page::B_page(ref<B_page> const& root, item& ins, int4 n) :
    object(self_class, 2*n)
{
    if (ins.key == MAX_KEY) { 
        m = 2*n-1;
        e[2*n-1] = ins;
    } else {
        m = 2*n-2;
        e[2*n-2] = ins;
        e[2*n-1].key = MAX_KEY;
        e[2*n-1].p = root;
    }
}

ref<set_member> B_page::find(int height, skey_t key, int4 n) const
{
    int l = m, r = 2*n-1;
    while (l < r)  {
        int i = (l+r) >> 1;
        if (key > e[i].key) l = i+1; else r = i;
    }
    internal_assert(r == l && e[r].key >= key); 
    if (--height != 0) { 
        ref<B_page> child = e[r].p;
        return child->find(height, key, n); 
    } else { 
        if (key == e[r].key) { 
            return e[r].p; 
        } else { 
            return NULL;
        }
    }
}

ref<set_member> B_page::findGE(int height, skey_t key, int4 n) const
{
    int l = m, r = 2*n-1;
    while (l < r)  {
        int i = (l+r) >> 1;
        if (key > e[i].key) l = i+1; else r = i;
    }
    internal_assert(r == l && e[r].key >= key); 
    if (--height != 0) { 
        ref<B_page> child = e[r].p;
        return child->findGE(height, key, n); 
    } else { 
        return e[r].p; 
    }
}

B_page::operation_effect B_page::find_insert_position(int level, 
                                                      ref<B_tree> const& tree, 
                                                      item& ins,
                                                      int4 n) const
{
    int  i, l = m, r = 2*n-1;
    skey_t key = ins.key;
    while (l < r)  {
        i = (l+r) >> 1;
        if (key > e[i].key) l = i+1; else r = i;
    }
    internal_assert(r == l && e[r].key >= key); 
    if (--level == 0)  {
        if (e[r].key == key) { 
            ref<set_member> mbr;

            for (mbr = e[r].p; 
                 mbr != NULL && mbr->compare(ins.p) < 0; 
                 mbr = mbr->next);
            if (mbr == NULL) { 
                modify(tree)->put_last(ins.p);
            } else { 
                modify(tree)->put_before(mbr, ins.p);
                if (mbr == e[r].p) { 
                    modify(this)->e[r].p = ins.p;
                }
            }
            return done;
        } else { 
            if (e[r].p == NULL) { 
                modify(tree)->put_last(ins.p);
            } else { 
                modify(tree)->put_before(e[r].p, ins.p);
            }
        }               
    } else { 
        ref<B_page> child = e[r].p;
        if (child->find_insert_position(level, tree, ins, n) == done) { 
            return done;
        }
    }
    // insert before e[r]
    return modify(this)->insert(r, ins, n); 
}

B_page::operation_effect B_page::insert(int r, item& ins, int4 n)
{
    // insert before e[r]
    if (m > 0) {
        memmove(&e[m-1], &e[m], (r - m)*sizeof(item));
        memset(&e[r-1], 0, sizeof(item));
        m -= 1;
        e[r-1] = ins;
        return done;
    } else { // page is full then divide page
        ref<B_page> b = B_page::create(n);
        if (r < n) {
            memcpy(&modify(b)->e[n], e, r*sizeof(item));
            modify(b)->e[n+r] = ins;
            memcpy(&modify(b)->e[n+r+1], &e[r], (n-r-1)*sizeof(item));
        } else {
            memcpy(&modify(b)->e[n], e, n*sizeof(item));
            memmove(&e[n-1], &e[n], (r-n)*sizeof(item));
            memset(&e[r-1], 0, sizeof(item));
            e[r-1] = ins;
        }
        ins.key = b->e[2*n-1].key;
        ins.p = b;
        m = n-1;
        modify(b)->m = n;
        memset(e, 0, (n-1)*sizeof(item));
        return overflow;
    }
}

B_page::operation_effect B_page::find_remove_position(int level, 
                                                      B_page::item& rem,
                                                      int4 n) const
{
    int i, l = m, r = 2*n-1;
    skey_t key = rem.key;
    while (l < r) {
        i = (l+r) >> 1;
        if (key > e[i].key) l = i+1; else r = i;
    }
    internal_assert(r == l && e[r].key >= key); 
    if (--level == 0) {
        assert(e[r].key == key);  
        ref<set_member> mbr = rem.p;
        if (mbr == e[r].p) { 
            if (mbr->next != NULL && mbr->next->get_key() == key) { 
                modify(this)->e[r].p = mbr->next;
                return done; 
            } else {
                return modify(this)->remove(r, rem, n); 
            }
        }
    } else { 
        ref<B_page> child = e[r].p; 
        switch (child->find_remove_position(level, rem, n)) { 
          case underflow: 
            return modify(this)->handle_underflow(r, key, rem, n);
          case propagation: 
            if (key == e[r].key) { 
                modify(this)->e[r].key = rem.key; 
                return propagation;
            }
          default:;
        } 
    }
    return done; 
}

B_page::operation_effect B_page::remove(int r, item& rem, int4 n)
{
    e[r].p = NULL; 
    if (e[r].key == MAX_KEY) { 
        return done; 
    } else { 
        memmove(&e[m+1], &e[m], (r - m)*sizeof(item));
        memset(&e[m], 0, sizeof(item));
        rem.key = e[2*n-1].key; 
        return (++m > n) ? underflow : propagation; 
    }
}

B_page::operation_effect B_page::handle_underflow(int r, skey_t key, item& rem,
                                                  int4 n)
{
    ref<B_page> a = e[r].p;
    internal_assert(a->m == n+1);
    if (r < 2*n-1) { // exists greater page
        ref<B_page> b = e[r+1].p;
        int bm = b->m; 
        if (bm < n) {  // reallocation of nodes between pages a and b
            int i = (n - bm + 1) >> 1;
            memmove(&modify(a)->e[n+1-i], &a->e[n+1], (n-1)*sizeof(item));
            memcpy(&modify(a)->e[2*n-i], &b->e[bm], i*sizeof(item));
            memset(&modify(b)->e[bm], 0, i*sizeof(item) );
            modify(b)->m += i;
            modify(a)->m -= i;
            e[r].key = a->e[2*n-1].key;
            return done;
        } else { // merge page a to b  
            internal_assert(bm == n); 
            memcpy(&modify(b)->e[1], &a->e[n+1], (n-1)*sizeof(item));
            memset(&modify(a)->e[n+1], 0, (n-1)*sizeof(item));
            e[r].p = NULL; 
            memmove(&e[m+1], &e[m], (r-m)*sizeof(item));
            memset(&e[m], 0, sizeof(item));
            modify(b)->m = 1;
            // dismiss page 'a'
            return ++m > n ? underflow : done;
        }
    } else { // page b is before a
        ref<B_page> b = e[r-1].p;
        int bm = b->m; 
        if (key == e[r].key) { 
            e[r].key = rem.key;
        }
        if (bm < n) { // reallocation
            int i = (n - bm + 1) >> 1;
            memcpy(&modify(a)->e[n+1-i], &b->e[2*n-i], i*sizeof(item));
            memmove(&modify(b)->e[bm+i], &b->e[bm], (2*n-bm-i)*sizeof(item));
            memset(&modify(b)->e[bm], 0, i*sizeof(item));
            e[r-1].key = b->e[2*n-1].key;
            modify(b)->m += i;
            modify(a)->m -= i;
            return propagation;
        } else { // merge page b to a
            internal_assert(bm == n); 
            memcpy(&modify(a)->e[1], &b->e[n], n*sizeof(item));
            memset(&modify(b)->e[n], 0, n*sizeof(item));
            e[r-1].p = NULL; 
            memmove(&e[m+1], &e[m], (r-1-m)*sizeof(item));
            memset(&e[m], 0, sizeof(item));
            modify(a)->m = 1;
            // dismiss page 'b'
            return ++m > n ? underflow : propagation;
        }
    }
}

void B_page::dump(int level, int height, int4 n) const
{
    static const char tabs[] = "\t\t\t\t\t\t\t\t\t\t\t\t\t\t\t\t";
    char buf[8];
    union { 
        char   str[8];
        skey_t val;
    } key;
    int i;
    
    console::output(&tabs[(sizeof tabs) - 1 - level]);
    for (i = m; i < 2*n; i++) { 
        key.val = e[i].key;
        if (key.val != MAX_KEY) {
            if (sizeof(skey_t) == 4) { 
                unpack4(buf, key.str);
            } else { 
                unpack8(buf, key.str);
            }
            console::output("%.*s ", sizeof(skey_t), buf);
        } 
    }
    console::output("\n");
    if (++level < height) { 
        for (i = m; i < 2*n; i++) { 
            ref<B_page> page = e[i].p;
            page->dump(level, height, n);
        }
    }
}

void B_page::ok(skey_t min_key_val, skey_t max_key_val, 
                int level, int height, int4 n) const
{
    int i;
    assert(level == 0 || m <= n);

    if (++level == height) { // leaf page
        assert(e[m].key >= min_key_val);
        assert(e[2*n-1].key == max_key_val);

        for (i = m; i < 2*n-1; i++) { 
            ref<set_member> mbr = e[i].p;
            if (mbr.is_nil()) { 
                assert(i == 2*n-1 && e[2*n-1].key == MAX_KEY);
                break;
            }
            while (mbr->get_key() == e[i].key) {
                mbr = mbr->next; 
                if (mbr.is_nil()) { 
                    assert(i >= 2*n-2 && e[2*n-1].key == MAX_KEY);
                    break;
                }
                assert(mbr->prev->compare(mbr) <= 0);
            }
            assert((i == 2*n-1 && mbr.is_nil()) 
                   || (i < 2*n-1 && mbr == e[i+1].p));
            if (!mbr.is_nil()) { 
                assert(mbr->get_key() == e[i+1].key);
                assert(e[i].key < e[i+1].key);
            }
        }
    } else { 
        for (i = m; i < 2*n; i++) { 
            ref<B_page> p = e[i].p;
            assert(e[i].key > min_key_val 
                   || (i == 0 && e[i].key == min_key_val));
            p->ok(min_key_val, e[i].key, level, height, n);
            min_key_val = e[i].key;
        }
        assert(min_key_val == max_key_val);
    }
}

field_descriptor& B_page::describe_components()
{
    return FIELD(m), VARYING(e); 
}


REGISTER(B_tree, set_owner, pessimistic_repeatable_read_scheme);
#if 0
REGISTER_EX(B_page, object, hierarchical_access_scheme,
            class_descriptor::cls_aligned);
#else
REGISTER(B_page, object, hierarchical_access_scheme);
#endif

//
// Hash table
//


field_descriptor& hash_item::describe_components()
{
    return FIELD(next), FIELD(obj), VARYING(name); 
}

void hash_table::put(const char* name, ref<object> obj)
{
    unsigned h = string_hash_function(name) % size;
    table[h] = hash_item::create(obj, table[h], name);
    n_elems += 1;
}

ref<object> hash_table::get(const char* name) const
{
    ref<hash_item> ip = table[string_hash_function(name) % size];
    while (ip != NULL) { 
        if (ip->compare(name) == 0) { 
            return ip->obj;
        }
        ip = ip->next;
    }
    return NULL;
}

boolean hash_table::del(const char* name)
{
    unsigned h = string_hash_function(name) % size;
    ref<hash_item> curr = table[h], prev = NULL;
    while (curr != NULL) { 
        if (curr->compare(name) == 0) { 
            if (prev == NULL) { 
                table[h] = curr->next;
            } else { 
                modify(prev)->next = curr->next;
            }
            n_elems -= 1;
            return True;
        }
        prev = curr;
        curr = curr->next;
    }
    return False;    
}

boolean hash_table::del(const char* name, ref<object> obj)
{
    unsigned h = string_hash_function(name) % size;
    ref<hash_item> curr = table[h], prev = NULL;
    while (curr != NULL) { 
        if (curr->obj == obj && curr->compare(name) == 0) { 
            if (prev == NULL) { 
                table[h] = curr->next;
            } else { 
                modify(prev)->next = curr->next;
            }
            n_elems -= 1;
            return True;
        }
        prev = curr;
        curr = curr->next;
    }
    return False;    
}

ref<object> hash_table::apply(hash_item::item_function f) const
{
    for (int i = size; --i >= 0;) { 
        for (ref<hash_item> ip = table[i]; ip != NULL; ip = ip->next) { 
            if (ip->apply(f)) { 
                return ip->obj;
            }
        }
    }
    return NULL;
}

field_descriptor& hash_table::describe_components()
{
    return FIELD(size), FIELD(n_elems), VARYING(table); 
}

REGISTER(hash_table, object, pessimistic_repeatable_read_scheme);
REGISTER(hash_item, object, hierarchical_access_scheme);


//
// Hash tree implementation
//
ref<object> H_page::apply(hash_item::item_function f, int height) const 
{ 
    if (--height == 0){ 
        for (int i = 0; i < (1 << size_log); i++) {
            for (ref<hash_item> ip = p[i]; ip != NULL; ip = ip->next) { 
                if (ip->apply(f)) { 
                    return ip->obj;
                }
            }
        }
    } else { 
        for (int i = 0; i < (1 << size_log); i++) {
            ref<H_page> pg = p[i];
            if (pg != NULL) { 
                ref<object> obj = pg->apply(f, height);
                if (!obj.is_nil()) { 
                    return obj;
                }
            }
        }
    }
    return NULL;
}

void H_page::reset(int height)
{
    if (--height != 0) { 
        for (int i = 0; i < (1 << size_log); i++) {
            ref<H_page> pg = p[i];
            if (pg != NULL) { 
                modify(pg)->reset(height);
            }
        }
    }
    for (int i = 0; i < (1 << size_log); i++) {
        p[i] = NULL;
    }
}

field_descriptor& H_page::describe_components()
{
    return ARRAY(p); 
}

REGISTER(H_page, object, hierarchical_access_scheme);

void H_tree::put(const char* name, ref<object> obj)
{
    unsigned h = string_hash_function(name) % size;
    int i, j;
    ref<H_page> pg = root;
    if (pg.is_nil()) { 
        pg = NEW H_page;
        root = pg;
    }
    for (i = height; --i > 0; ) { 
        j = (h >> (i*H_page::size_log)) & ((1 << H_page::size_log) - 1);
        ref<H_page> child = pg->p[j];
        if (child == NULL) {
            child = NEW H_page;
            modify(pg)->p[j] = child;
        }
        pg = child;
    }
    i = h & ((1 << H_page::size_log) - 1);
    modify(pg)->p[i] = hash_item::create(obj, pg->p[i], name);
    n_elems += 1;
}

ref<object> H_tree::get(const char* name) const
{
    unsigned h = string_hash_function(name) % size;
    int i, j;
    ref<H_page> pg = root;
    for (i = height; --i > 0; ) { 
        if (pg.is_nil()) return NULL;
        j = (h >> (i*H_page::size_log)) & ((1 << H_page::size_log) - 1);
        pg = pg->p[j];
    }
    if (pg.is_nil()) return NULL;
    i = h & ((1 << H_page::size_log) - 1);
    ref<hash_item> ip = pg->p[i];
    while (ip != NULL) { 
        if (ip->compare(name) == 0) { 
            return ip->obj;
        }
        ip = ip->next;
    }
    return NULL;
}

boolean H_tree::del(const char* name)
{
    unsigned h = string_hash_function(name) % size;
    int i, j;
    ref<H_page> pg = root;
    for (i = height; --i > 0; ) { 
        if (pg.is_nil()) return False;
        j = (h >> (i*H_page::size_log)) & ((1 << H_page::size_log) - 1);
        pg = pg->p[j];
    }
    if (pg.is_nil()) return False;
    i = h & ((1 << H_page::size_log) - 1);
    ref<hash_item> curr = pg->p[i], prev = NULL;
    while (curr != NULL) { 
        if (curr->compare(name) == 0) { 
            if (prev == NULL) { 
                modify(pg)->p[i] = curr->next;
            } else { 
                modify(prev)->next = curr->next;
            }
            n_elems -= 1;
            return True;
        }
        prev = curr;
        curr = curr->next;
    }
    return False;    
}

boolean H_tree::del(const char* name, ref<object> obj)
{
    unsigned h = string_hash_function(name) % size;
    int i, j;
    ref<H_page> pg = root;
    for (i = height; --i > 0; ) { 
        if (pg.is_nil()) return False;
        j = (h >> (i*H_page::size_log)) & ((1 << H_page::size_log) - 1);
        pg = pg->p[j];
    }
    if (pg.is_nil()) return False;
    i = h & ((1 << H_page::size_log) - 1);
    ref<hash_item> curr = pg->p[i], prev = NULL;
    while (curr != NULL) { 
        if (curr->obj == obj && curr->compare(name) == 0) { 
            if (prev == NULL) { 
                modify(pg)->p[i] = curr->next;
            } else { 
                modify(prev)->next = curr->next;
            }
            n_elems -= 1;
            return True;
        }
        prev = curr;
        curr = curr->next;
    }
    return False;    
}

field_descriptor& H_tree::describe_components()
{
    return FIELD(size), FIELD(n_elems), FIELD(height), FIELD(root); 
}

H_tree::H_tree(size_t hash_size, class_descriptor& desc) : object(desc) 
{ 
    unsigned h = 1, pow2 = 1 << H_page::size_log;
    while (pow2 <= hash_size) { 
        pow2 <<= H_page::size_log;
        h += 1;
    }
    height = h;
    size = hash_size;
    n_elems = 0;
}

REGISTER(H_tree, object, pessimistic_repeatable_read_scheme);


//
//  Binary large object
//

int Blob::readahead = 1;

struct blob_sync { 
    boolean     cancel;
    Blob const* chain;
    semaphore   credit;
};
    
void task_proc Blob::load_next_blob_part(void* arg) 
{
    blob_sync& bs = *(blob_sync*)arg;
    ref<Blob> head = bs.chain; 
    ref<Blob> curr = bs.chain->next; 

    do { 
        curr = curr->next;
        head->notify();
        if (bs.cancel) { 
            head->notify();
            break;
        }    
        bs.credit.wait();
    } while (curr != NULL);
}

void Blob::play() const
{ 
    if (next.is_nil()) { 
        handle(); // that is all
    } else { 
        blob_sync sync;
        sync.chain = this;
        sync.cancel = False;
        for (int credit = readahead; --credit >= 0; sync.credit.signal());

        task::create(load_next_blob_part, &sync, 
                     task::pri_high, task::min_stack); 
        ref<Blob> curr = this;
        while (curr->handle()) { 
            if (!curr->next.is_nil()) { 
                wait();
                sync.credit.signal();
                curr = curr->next;
            } else { 
                return;
            }
        }
        if (!curr->next.is_nil()) { 
            sync.cancel = True;
            wait();
            wait(); // wait preloading task termination
        }
    }
}

boolean Blob::handle() const 
{
    return False;
} 


field_descriptor& Blob::describe_components()
{
    return FIELD(next), FIELD(last), VARYING(data); 
}

REGISTER(Blob, object, optimistic_scheme);

END_GOODS_NAMESPACE