//-< MOP.CXX >-------------------------------------------------------*--------*
// GOODS                     Version 1.0         (c) 1997  GARRET    *     ?  *
// (Generic Object Oriented Database System)                         *   /\|  *
//                                                                   *  /  \  *
//                          Created:     20-Feb-97    K.A. Knizhnik  * / [] \ *
//                          Last update: 17-Nov-97    K.A. Knizhnik  * GARRET *
//-------------------------------------------------------------------*--------*
// Metaobjects are used to specify synchronization policy for GOODS objects
//-------------------------------------------------------------------*--------*


#include "goods.h"

BEGIN_GOODS_NAMESPACE

//
// Object monitor implementation used by basic_metaobject
//

mutex* object_monitor::global_cs = NEW mutex; 

#ifdef GOODS_SUPPORT_EXCEPTIONS
int dbException::stackUnwindCount;
#endif 

inline void object_monitor::attach(hnd_t new_owner) 
{
    internal_assert(!busy && !signaled && owner == NULL && n_nested_locks==0);
    if (hnd != 0 && IS_VALID_OBJECT(hnd->obj)) { 
        hnd->obj->monitor = 0;
    }
    hnd = new_owner;
}

inline void object_monitor::detach()
{
    internal_assert(busy == 0 && owner == NULL && n_nested_locks==0);
    hnd = 0; 
    while (signaled) { // Return semaphore to initial state
        signaled -= 1;
        sem->wait(); // should not be blocked
    }
}

//
// 'lock' and 'unlock' are executed with global mutex locked.
// To avoid deadlock first release this lock when object is locked
// by other user.
//
inline void object_monitor::lock()
{
    task* self = task::current();
    if (busy++ == 0) { 
        cs->enter(); // should not wait
    } else if (owner != self) { 
        unlock_global(); 
        cs->enter();
        lock_global(); 
        assert(owner == NULL);
    }
    owner = self;
    n_nested_locks += 1;
}

inline void object_monitor::unlock() 
{ 
    busy -= 1; 
    if (--n_nested_locks == 0) { 
        owner = NULL;
        cs->leave();
    }
}


inline void object_monitor::wait()
{
    if (sem == NULL) { 
        sem = NEW semaphorex(*cs);
    }
    task* self = owner;
    int n_locks = n_nested_locks;
    n_nested_locks = 0;
    owner = NULL;
    unlock_global();
    sem->wait();
    lock_global();
    internal_assert(signaled > 0);
    owner = self;
    n_nested_locks = n_locks;
    signaled -= 1;
}

inline boolean object_monitor::wait(time_t timeout)
{
    if (sem == NULL) { 
        sem = NEW semaphorex(*cs);
    }
    task* self = owner;
    int n_locks = n_nested_locks;
    n_nested_locks = 0;
    owner = NULL;
    unlock_global();
    if (sem->wait_with_timeout(timeout)) { 
        lock_global();
        internal_assert(signaled > 0);
        owner = self;
        n_nested_locks = n_locks;
        signaled -= 1;
        return True;
    }
    lock_global();
    owner = self;
    n_nested_locks = n_locks;
    return False;
}

inline void object_monitor::notify()
{
    signaled += 1;
    if (sem == NULL) { 
        sem = NEW semaphorex(*cs);
    }
    sem->signal();
}

inline void object_monitor::init() 
{ 
    cs = NEW mutex;
    sem = NULL; // lazy semaphore creation
    busy = 0;
    signaled = 0;
    hnd = 0; 
    n_nested_locks = 0;
    owner = NULL;
}

inline void object_monitor::destroy()
{
    delete sem;
    delete cs;
}

//
// Turnstile of object monitors used by basic_metaobject
//

int monitor_turnstile::attach_monitor(hnd_t hnd)
{
    register object_monitor* mp = &turnstile[pos];
    register int n = size - pos; 
   
    while (--n > 0) {
        mp += 1;
        if (!mp->busy && !mp->signaled) { 
            mp->attach(hnd);
            return pos = size - n; 
        } 
    }
    n = pos;  
    mp = turnstile; 
    while (--n >= 0) { 
        mp += 1;
        if (!mp->busy && !mp->signaled) {
            mp->attach(hnd);
            return pos -= n; 
        } 
    }
    object_monitor* new_turnstile = NEW object_monitor[size * 2];
    memcpy(new_turnstile, turnstile, size*sizeof(object_monitor));
    pos = size;
    size *= 2;
    for (int i = pos; i < size; i ++) { 
        new_turnstile[i].init();
    } 
    delete[] turnstile; 
    new_turnstile[pos].attach(hnd);
    turnstile = new_turnstile;
    return pos;
}

boolean monitor_turnstile::is_locked(int monitor) 
{
    return turnstile[monitor].busy != 0; 
}

void monitor_turnstile::lock(int monitor) 
{
    turnstile[monitor].lock(); 
}

void monitor_turnstile::unlock(int monitor) 
{
    turnstile[monitor].unlock(); 
}

void monitor_turnstile::wait(int monitor) 
{
    turnstile[monitor].wait(); 
}

boolean monitor_turnstile::wait(int monitor, time_t timeout) 
{
    return turnstile[monitor].wait(timeout); 
}

void monitor_turnstile::notify(int monitor) 
{
    turnstile[monitor].notify(); 
}

void monitor_turnstile::detach_monitor(int monitor) 
{
    turnstile[monitor].detach(); 
} 
        
monitor_turnstile::monitor_turnstile(int init_size) 
{
    size = init_size;
    pos = 0;
    turnstile = NEW object_monitor[init_size];
    for (int i = 1; i < init_size; i++) { 
        turnstile[i].init();
    }
    turnstile[0].busy = 0;
}

monitor_turnstile::~monitor_turnstile() { 
    for (int i = 1; i < size; i++) { 
        turnstile[i].destroy();
    }
    delete[] turnstile; 
}

//
// Cache manager 
//

#define DEFAULT_CACHE_LIMIT     (1024*1024) 

cache_manager cache_manager::default_manager;
boolean cache_manager::cooperative_mode = True;

static void default_abort_hook(metaobject*)
{
    console::error("Transaction is aborted by server\n");
}

void cache_manager::cleanup_cache(obj_storage* storage) 
{
    for (int i = 0; i < 2; i++) { 
        hnd_t hnd, next;
        for (hnd = lru[i].head; hnd != NULL; hnd = next) { 
            object_header* obj = metaobject::get_header(hnd); 
            next = obj->next;
            if (hnd->storage == storage) { 
                hnd->access = 0;
                obj->remove_object();
            }           
        }
    }
} 
    
cache_manager::cache_manager() 
{ 
    cache_size_limit[0] = cache_size_limit[1] = DEFAULT_CACHE_LIMIT; 
    trans_commit_wait_flag = False;
    trans_commit_in_progress = False;
    used_cache_size[0] = used_cache_size[1] = 0;
    abort_hook = default_abort_hook; 
    n_nested_transactions = 0;
    isolation_level = PER_PROCESS_TRANSACTION;
}


//
// Basic metaobject methods
//

#define NOTIFICATION_HASH_SIZE  113

monitor_turnstile basic_metaobject::turnstile;
basic_metaobject::notify_item* basic_metaobject::notification_hash[NOTIFICATION_HASH_SIZE];


void basic_metaobject::destroy(hnd_t hnd) 
{
    object_header* obj = get_header(hnd);
    if ((obj->state & (object_header::persistent|object_header::fixed)) ==
        object_header::persistent) 
    { 
        get_from_cache(hnd);
    }
    if (obj->monitor != 0) { 
        turnstile.detach_monitor(obj->monitor);
        obj->monitor = 0;
    }
}

void basic_metaobject::lock(hnd_t hnd) 
{
    object_header* obj = get_header(hnd);
    if (obj->monitor == 0) { 
        obj->monitor = turnstile.attach_monitor(hnd); 
    }
    turnstile.lock(obj->monitor); 
}
 
void basic_metaobject::unlock(hnd_t hnd) 
{
    object_header* obj = get_header(hnd);
    internal_assert(obj->monitor != 0);
    turnstile.unlock(obj->monitor); 
}

void basic_metaobject::wait(hnd_t hnd) 
{
    object_header* obj = get_header(hnd);
    internal_assert(obj->monitor != 0);
    turnstile.wait(obj->monitor);     
}

boolean basic_metaobject::wait(hnd_t hnd, time_t timeout) 
{
    object_header* obj = get_header(hnd);
    internal_assert(obj->monitor != 0);
    return turnstile.wait(obj->monitor, timeout);     
}

void basic_metaobject::notify(hnd_t hnd)
{
    object_header* obj = get_header(hnd);
    if (obj->monitor == 0) { 
        obj->monitor = turnstile.attach_monitor(hnd); 
    }
    turnstile.notify(obj->monitor);         
}


void basic_metaobject::make_persistent(hnd_t hnd, hnd_t parent_hnd)
{
    obj_storage* storage = parent_hnd->storage;
    object_header* obj = get_header(hnd); 
    if (hnd->storage != NULL) { 
        assert(hnd->storage->db == storage->db);
        storage = hnd->storage;
    } else { 
        hnd->storage = storage;
    }
    class_descriptor* desc = &hnd->obj->cls; 
    obj->cpid = storage->get_cpid_by_descriptor(desc);

    opid_t opid;
    
    opid = storage->allocate(obj->cpid, desc->packed_size((char*)obj, hnd->obj->size),
                             (desc->class_attr & class_descriptor::cls_aligned) != 0);
    object_handle::assign_persistent_oid(hnd, opid);

    obj->state |= object_header::modified; 
    insert_in_transaction_list(hnd, parent_hnd);
}

void basic_metaobject::put_in_cache(hnd_t hnd)
{
    object_header* obj = get_header(hnd); 
    cache_manager* mng = cache_manager::get();
    int i = (obj->state & object_header::useful) ? 1 : 0;

    mng->used_cache_size[i] += hnd->obj->size; 

    obj->state &= ~object_header::fixed; 
    if (!(obj->state & object_header::notifier)) { 
        mng->lru[i].put_at_head(obj);
    }
    while (mng->used_cache_size[i] > mng->cache_size_limit[i]) {  
        hnd_t victim = mng->lru[i].tail;
        if (victim != 0 && victim != hnd) { 
            get_header(victim)->remove_object();
        } else { 
            break;
        }
    }
}    

void basic_metaobject::get_from_cache(hnd_t hnd)
{ 
    object_header* obj = get_header(hnd); 
    cache_manager* mng = cache_manager::get();
    int i = (obj->state & object_header::useful) ? 1 : 0; 
    mng->used_cache_size[i] -= hnd->obj->size;
    if ((obj->state & object_header::accessed) && mng->lru[i].head != hnd) { 
        obj->state |= object_header::useful;
    }
    obj->state |= object_header::fixed|object_header::accessed; 
    if (!(obj->state & object_header::notifier)) { 
        mng->lru[i].unlink(obj);
    }
}

void basic_metaobject::insert_in_transaction_list(hnd_t hnd, hnd_t parent_hnd) 
{
    cache_manager* mng = cache_manager::get();
    object_header* obj = get_header(hnd); 
    //
    // Prevent object involved in transaction from GC until end of transaction
    // 
    hnd->access += 1;
    //
    // Group all modified objects together 
    //
    if (parent_hnd) { 
        mng->trans_objects.put_after(get_header(parent_hnd), obj);  
    } else { 
        if (obj->state & object_header::modified) { 
            mng->trans_objects.put_at_head(obj); // modified objects first
        } else { 
            mng->trans_objects.put_at_tail(obj); 
        } 
    }
    obj->state |= object_header::in_trans; 
}

void basic_metaobject::remove_from_transaction_list(hnd_t hnd) 
{
    cache_manager* mng = cache_manager::get();
    object_header* obj = get_header(hnd); 
    mng->trans_objects.unlink(obj);
    obj->state &= ~object_header::in_trans; 
    hnd->access -= 1;
}

void basic_metaobject::update_object(hnd_t hnd)
{
    object_header* obj = get_header(hnd);
    internal_assert(!(obj->state & object_header::in_trans));
    if (obj->state & object_header::notifier) { 
        notify_item *np, **npp;
        npp = &notification_hash[size_t(hnd) % itemsof(notification_hash)];
        while ((np = *npp) != NULL && np->hnd != hnd) { 
            npp = &np->chain;
        }
        assert(np != NULL);
        np->e.signal();
        *npp = np->chain;
        delete np;
        obj->state = (obj->state & ~object_header::notifier) | object_header::fixed;
    }
    obj->remove_object();
}

void basic_metaobject::reload_object(hnd_t hnd)
{
    object_header* obj = get_header(hnd);
    internal_assert(obj->state & object_header::fixed);
    obj->state |= object_header::reloading;  
    obj->state &= ~object_header::invalidated;  
    hnd->storage->load(hnd, lof_none);     
    obj = get_header(hnd);
    object_monitor::unlock_global();
    obj->on_loading();
    object_monitor::lock_global();
}



void basic_metaobject::invalidate(hnd_t hnd)
{
    object_header* obj = get_header(hnd);
    internal_assert(cache_manager::get()->trans_commit_in_progress || 
        !(obj->state & (object_header::exl_locked|object_header::shr_locked)));
    if (!(obj->state & (object_header::in_trans|object_header::fixed)) 
        && !turnstile.is_locked(obj->monitor))
    {
        update_object(hnd);
    } else { 
        obj->state |= object_header::invalidated; 
    }
}

void basic_metaobject::commit_transaction()
{
    commit_transaction(cache_manager::get());
}


void basic_metaobject::abort_transaction(const database* dbs, 
                                         abort_reason reason)
{
    abort_transaction(dbs, cache_manager::get(), reason);
}

void basic_metaobject::commit_transaction(cache_manager* mng) { 
    object_l2_list& trans_objects = mng->trans_objects;
    mng->trans_commit_in_progress = True;
    
    while (!trans_objects.empty()) { 
        hnd_t hnd = trans_objects.head;

        if (!(get_header(hnd)->state & object_header::modified)) { 
            //
            // No more modified objects (because modified objects
            // are always placed at the end of transaction list)
            //
            do { 
                object_header* obj = get_header(hnd); 
                hnd_t next = obj->next; 
                internal_assert(!(obj->state & object_header::modified));
                hnd->obj->mop->release_transaction_object(hnd); 
                hnd = next; 
            } while (hnd != 0); 
            break; 
        } 

        obj_storage*    coordinator = NULL;
        int             n_trans_servers = 0;
        database const* dbs = hnd->storage->db;
        dnm_array<obj_storage*> trans_servers;
        
        do { 
            object_header* obj = get_header(hnd); 
            if (hnd->storage->db == dbs) { 
                if (obj->state & object_header::invalidated) {
                    abort_transaction(dbs, mng, aborted_by_server); 
                    break;
                }
                stid_t sid = hnd->storage->id; 
                if (trans_servers[sid] == NULL) {
                    hnd->storage->begin_transaction();
                    n_trans_servers += 1; 
                    trans_servers[sid] = hnd->storage; 
                }

                int flags = hnd->obj->mop->get_transaction_object_flags(hnd); 

                if (obj->state & object_header::modified) { 
                    flags |= tof_update; 
                }
                hnd->storage->include_object_in_transaction(hnd, flags); 
                if (coordinator == NULL || coordinator->id > sid) { 
                    coordinator = hnd->storage;  
                }
            } 
            hnd = obj->next; 
        } while (hnd != 0); 

        if (hnd != 0) { // transaction was aborted
            continue;
        }

        if (n_trans_servers != 0) {
            assert(n_trans_servers < MAX_TRANS_SERVERS);
            stid_t trans_sids[MAX_TRANS_SERVERS];
            obj_storage** server = &trans_servers;
            trid_t tid;
            int i, j, n = trans_servers.size();

            for (i = 0, j = 0; i < n; i++) { 
                if (server[i] != NULL) { 
                    trans_sids[j++] = server[i]->id;
                }
            }

            if (!coordinator->commit_coordinator_transaction(n_trans_servers, 
                                                             trans_sids,
                                                             tid))
            {
                abort_transaction(dbs, mng, aborted_by_server); 
                continue;
            } else { // transaction was successefully completed
                if (n_trans_servers > 1) { // global transaction 
                    for (i = 0; i < n; i++) { 
                        if (server[i] != NULL && server[i] != coordinator)
                        {
                            server[i]->commit_transaction(coordinator->id, 
                                                          n_trans_servers,
                                                          trans_sids,
                                                          tid);
                        }
                    }
                    if (!coordinator->wait_global_transaction_completion()) { 
                        abort_transaction(dbs, mng, aborted_by_server);
                        continue;
                    }
                }
            }
            // 
            // Transaction was successfully commited
            //
            hnd = trans_objects.head;
            while (hnd != 0) { 
                object_header* obj = get_header(hnd);
                hnd_t next = obj->next; 
                if (hnd->storage->db == dbs) { 
                    hnd->obj->mop->commit_object_changes(hnd);
                }
                hnd = next; 
            }
        }
    }
    mng->trans_commit_in_progress = False;
    if (mng->trans_commit_wait_flag) { 
        mng->trans_commit_event.signal();
        mng->trans_commit_wait_flag = False;
    }
} 


void basic_metaobject::abort_transaction(const database* dbs, cache_manager* mng, abort_reason reason)
{
    object_l2_list& trans_objects = mng->trans_objects;
    hnd_t hnd = trans_objects.head;
    while (hnd != 0) {
        hnd_t next = get_header(hnd)->next;
        if (hnd->storage->db == dbs || dbs == NULL) { 
            hnd->obj->mop->abort_object_changes(hnd, reason);
        } 
        hnd = next;
    }
    if (mng->abort_hook != NULL && reason == aborted_by_server) { 
        (*mng->abort_hook)(this);
    }
}

void basic_metaobject::set_cache_limit(size_t limit0, size_t limit1)
{
    cache_manager* mng = cache_manager::get();
    mng->cache_size_limit[0] = limit0; 
    mng->cache_size_limit[1] = limit1; 
}

void basic_metaobject::set_abort_transaction_hook(abort_hook_t hook)
{
    cache_manager::get()->abort_hook = hook;
}


void basic_metaobject::signal_on_modification(hnd_t hnd, event& e) 
{
    object_header* obj = get_header(hnd); 
    obj->state |= object_header::notifier;      
    unsigned h = size_t(hnd) % itemsof(notification_hash);      
    notification_hash[h] = NEW notify_item(hnd, e, notification_hash[h]);
}
        
//
// Optimistic metaobject
// 

void optimistic_metaobject::begin_read(hnd_t hnd)
{
    object_header* obj = get_header(hnd); 
    cache_manager* mng = cache_manager::get();
    if (mng->n_nested_transactions == 0) { 
	if (mng->begin_transaction() == PER_THREAD_TRANSACTION) { 
	    obj = get_header(hnd); 
	    if (!IS_VALID_OBJECT(obj)) {
		internal_assert(hnd->storage != NULL); 
		hnd->storage->load(hnd, lof_none);
		obj = get_header(hnd);
		internal_assert(IS_VALID_OBJECT(obj));          
	    }
	}
    }
    mng->n_nested_transactions += 1;
#ifdef GOODS_SUPPORT_EXCEPTIONS
    try {
#endif
    while (mng->trans_commit_in_progress) { 
        mng->trans_commit_event.reset();
        mng->trans_commit_wait_flag = True;
        object_monitor::unlock_global();
        mng->trans_commit_event.wait();
        object_monitor::lock_global();
        obj = get_header(hnd); 
        if (!IS_VALID_OBJECT(obj)) {
            internal_assert(hnd->storage != NULL); 
            hnd->storage->load(hnd, lof_none);
            obj = get_header(hnd);
            internal_assert(IS_VALID_OBJECT(obj));          
        }
    }
#ifdef GOODS_SUPPORT_EXCEPTIONS
    } catch(...) { 
        if (--mng->n_nested_transactions == 0) { 
	    mng->end_transaction();
	}
        throw;
    }
    try { 
#endif
    lock(hnd); // intertask locking
    obj = get_header(hnd);
    obj->n_invokations += 1; 
    if ((obj->state & (object_header::persistent|object_header::fixed)) 
        == object_header::persistent)
    {
        get_from_cache(hnd);
    }
    if ((obj->state & (object_header::persistent|object_header::initialized))
        == object_header::persistent)
    {
        obj->state |= object_header::initialized;
        object_monitor::unlock_global();
        obj->on_loading();
        object_monitor::lock_global();
    }
#ifdef GOODS_SUPPORT_EXCEPTIONS
    } catch(...) { 
        if (--mng->n_nested_transactions == 0) { 
	    mng->end_transaction();
	}
        obj->n_invokations -= 1; 
        unlock(hnd);
        throw;
    }
#endif
}

void optimistic_metaobject::end_read(hnd_t hnd)
{
    cache_manager* mng = cache_manager::get();
    unlock(hnd);
    object_header* obj = get_header(hnd);
    if (--obj->n_invokations == 0) { 
        if ((obj->state & (object_header::in_trans|object_header::fixed)) 
            == object_header::fixed) 
        {
            put_in_cache(hnd);
        }
        if ((obj->state & (object_header::in_trans|object_header::invalidated))
            == object_header::invalidated && !turnstile.is_locked(obj->monitor))
        { 
            update_object(hnd);
        }
    }
    if (--mng->n_nested_transactions == 0) {
#ifdef GOODS_SUPPORT_EXCEPTIONS
        if (dbException::stackUnwindCount != 0) { 
            abort_transaction(NULL, aborted_by_user);
        } else
#endif
        commit_transaction();
	mng->end_transaction();
    }
}

void optimistic_metaobject::begin_write(hnd_t hnd)
{ 
    optimistic_metaobject::begin_read(hnd);
    object_header* obj = get_header(hnd);
    if ((obj->state & (object_header::persistent|object_header::modified))
         == object_header::persistent)
    {
        obj->state |= object_header::modified;
        if (obj->state & object_header::in_trans) { 
            //
            // Relink object in transaction list according to
            // the changed state of object (modified)
            //
            remove_from_transaction_list(hnd);
        }
        insert_in_transaction_list(hnd);
    }
} 

void optimistic_metaobject::end_write(hnd_t hnd) 
{
    optimistic_metaobject::end_read(hnd);
}

void optimistic_metaobject::begin_transaction()
{
    cache_manager* mng = cache_manager::get();
    object_monitor::lock_global();
    mng->n_nested_transactions += 1;
    object_monitor::unlock_global();
}
   
void optimistic_metaobject::end_transaction()
{
    cache_manager* mng = cache_manager::get();
    object_monitor::lock_global();
    assert(mng->n_nested_transactions > 0); 
    if (--mng->n_nested_transactions == 0) { 
        commit_transaction();
	mng->end_transaction();
    }
    object_monitor::unlock_global();
}
    
int  optimistic_metaobject::get_transaction_object_flags(hnd_t)
{
    return tof_validate; 
}

void optimistic_metaobject::commit_object_changes(hnd_t hnd)
{
    object_header* obj = get_header(hnd);

    remove_from_transaction_list(hnd);
    //
    // Setting of 'persistent', 'fixed', 'initialized' and 'accessed' flags is
    // necessary for transient object included in persistent closure
    //
    obj->state = object_header::persistent | object_header::fixed 
        | object_header::initialized | object_header::accessed 
        | (obj->state & ~object_header::modified); 
    
    put_in_cache(hnd); 
    
    if (obj->state & object_header::invalidated) {
        update_object(hnd);
    }
}


void optimistic_metaobject::abort_object_changes(hnd_t hnd, abort_reason)
{
    remove_from_transaction_list(hnd);

    object_header* obj = get_header(hnd);
    if (!(obj->state & object_header::persistent)) { 
        obj->state &= ~object_header::modified;
        object_handle::remove_from_storage(hnd);
    } else { 
        if (obj->n_invokations == 0) { 
            put_in_cache(hnd);
        }
        if (obj->state & object_header::modified) { 
            obj->state |= object_header::invalidated; 
            obj->state &= ~object_header::modified;
        }
    }
    if (obj->n_invokations == 0
        && ((obj->state & object_header::invalidated) 
            || (hnd->access == 0 
                && !(obj->state & object_header::persistent))))
    {
        update_object(hnd);
    }
}

void optimistic_metaobject::release_transaction_object(hnd_t hnd)
{
    commit_object_changes(hnd);
}

optimistic_metaobject optimistic_scheme; 

//
// Optimistic metaobject with repeatable read facility
//

void optimistic_repeatable_read_metaobject::begin_read(hnd_t hnd)
{
    optimistic_metaobject::begin_read(hnd);
    object_header* obj = get_header(hnd);
    if ((obj->state & (object_header::persistent|object_header::in_trans))
        == object_header::persistent) 
    { 
        insert_in_transaction_list(hnd);
    }
} 

optimistic_repeatable_read_metaobject optimistic_repeatable_read_scheme;

//
// Pessimistic metaobject
//
    
boolean pessimistic_metaobject::solve_write_conflict(hnd_t hnd)
{ 
    console::error("Write conflict for object %x of class \"%s\"\n", 
                   hnd->opid, hnd->obj->cls.name);
    return False;
} 

boolean pessimistic_metaobject::solve_lock_conflict(hnd_t hnd, lck_t lck)
{ 
    object_monitor::unlock_global();
    boolean solved = get_header(hnd)->on_lock_failed(lck);
    object_monitor::lock_global();
    return solved; 
} 

int pessimistic_metaobject::get_transaction_object_flags(hnd_t)
{
    return tof_unlock;
}

boolean pessimistic_metaobject::storage_lock(hnd_t hnd, lck_t lck)
{
    while (!hnd->storage->lock(hnd->opid, lck, lck_attr)) { 
        if (!solve_lock_conflict(hnd, lck)) { 
            return False;
        }
    }
    return True;
}

void pessimistic_metaobject::begin_write(hnd_t hnd)
{
    int was_modified = (get_header(hnd)->state | ~object_header::modified);
    
    optimistic_metaobject::begin_write(hnd); 

    object_header* obj = get_header(hnd); 
    if ((obj->state & (object_header::persistent|object_header::exl_locked)) 
        == object_header::persistent) 
    { 
        if (storage_lock(hnd, lck_exclusive)) { 
            obj = get_header(hnd); 
            obj->state |= object_header::exl_locked; 
            if (obj->state & object_header::invalidated) { 
                if (obj->n_invokations == 1) {
                    reload_object(hnd); 
                    internal_assert(!(get_header(hnd)->state 
                                      & object_header::invalidated));
                } else {
                    if (!solve_write_conflict(hnd)) { 
                        get_header(hnd)->state &= was_modified;
                        abort_transaction(hnd->storage->db,
                                          aborted_by_user);
                    } else { 
                        get_header(hnd)->state &= ~object_header::invalidated;
                    } 
                }
            } 
        } else { 
            get_header(hnd)->state &= was_modified;
            abort_transaction(hnd->storage->db, aborted_by_user);
        }
    } 
}


void pessimistic_metaobject::commit_object_changes(hnd_t hnd)
{
    object_header* obj = get_header(hnd);
    // Locks are automatically remove when transactoin is committed
    obj->state &= ~(object_header::shr_locked|object_header::exl_locked);
    optimistic_metaobject::commit_object_changes(hnd);
}

void pessimistic_metaobject::abort_object_changes(hnd_t hnd, 
                                                  abort_reason reason) 
{
    object_header* obj = get_header(hnd);
    if (obj->state & (object_header::shr_locked|object_header::exl_locked)) { 
        hnd->storage->unlock(hnd->opid, lck_none);
        obj->state &= ~(object_header::shr_locked|object_header::exl_locked);  
    } 
    optimistic_metaobject::abort_object_changes(hnd, reason);
}
    

void pessimistic_metaobject::release_transaction_object(hnd_t hnd)
{
    object_header* obj = get_header(hnd); 
    if (obj->state & (object_header::exl_locked|object_header::shr_locked)) {
        hnd->storage->unlock(hnd->opid, lck_none);
        obj->state &= ~(object_header::shr_locked|object_header::exl_locked);  
    }
    optimistic_metaobject::commit_object_changes(hnd);
}


pessimistic_metaobject pessimistic_scheme;

pessimistic_metaobject nowait_pessimistic_scheme(lckattr_nowait);

//
// Lazy pessimistic metaobject. This metaobject combines
// characterstics of both optimistic and pessimistic protocols. 
// Lock is set after object has been modified.
//

boolean lazy_pessimistic_metaobject::solve_write_conflict(hnd_t hnd)
{ 
    object_monitor::unlock_global();
    boolean solved = get_header(hnd)->on_write_conflict();
    object_monitor::lock_global();
    return solved; 
} 

void lazy_pessimistic_metaobject::begin_write(hnd_t hnd)
{
    optimistic_metaobject::begin_write(hnd);
}

void lazy_pessimistic_metaobject::end_write(hnd_t hnd)
{
    object_header* obj = get_header(hnd);
    if ((obj->state & (object_header::persistent|object_header::modified)) ==
        (object_header::persistent|object_header::modified))
    {
        if (!(obj->state & object_header::exl_locked)) {
            if (storage_lock(hnd, lck_exclusive)) { 
                obj = get_header(hnd);
                obj->state |= object_header::exl_locked; 
                if (obj->state & object_header::invalidated) { 
                    if (!solve_write_conflict(hnd)) {   
                        abort_transaction(hnd->storage->db, aborted_by_user);
                        obj = get_header(hnd);
                    } else { 
                        obj = get_header(hnd);
                        obj->state &= ~object_header::invalidated;
                    } 
                }
            }
        } else { 
            abort_transaction(hnd->storage->db, aborted_by_user);
            obj = get_header(hnd);
        }
    }
    optimistic_metaobject::end_read(hnd);
}

lazy_pessimistic_metaobject lazy_pessimistic_scheme;

lazy_pessimistic_metaobject lazy_nowait_pessimistic_scheme(lckattr_nowait);



//
// Pessimistic repeatable read metaobject. No conflicts are possible in this
// scheme. However this scheme is not deadlock free and(or) significantly
// reduces concurrency. 
//

void pessimistic_repeatable_read_metaobject::begin_read(hnd_t hnd)
{
    optimistic_metaobject::begin_read(hnd); 
    object_header* obj = get_header(hnd); 

    if (obj->state & object_header::persistent) { 
        if (!(obj->state & object_header::exl_locked) 
            && (read_lock != lck_shared  
                || !(obj->state & object_header::shr_locked))) 
        { 
            if (storage_lock(hnd, read_lock)) { 
                obj = get_header(hnd); 
                obj->state |= (read_lock == lck_shared)
                    ? object_header::shr_locked : object_header::exl_locked; 
                if (obj->state & object_header::invalidated) {
                    // 
                    // As far as object is always locked before access, 
                    // there are no other invokations for invalidated object 
                    //
                    reload_object(hnd); 
                    obj = get_header(hnd); 
                    internal_assert(!(obj->state&object_header::invalidated));
                }
                if (!(obj->state & object_header::in_trans)) { 
                    insert_in_transaction_list(hnd);
                }
            } else { 
                abort_transaction(hnd->storage->db, aborted_by_user);
            } 
        }
    }
}


pessimistic_repeatable_read_metaobject pessimistic_repeatable_read_scheme;

pessimistic_repeatable_read_metaobject 
    pessimistic_exclusive_scheme(0, lck_exclusive);

pessimistic_repeatable_read_metaobject 
    nowait_pessimistic_repeatable_read_scheme(lckattr_nowait);

pessimistic_repeatable_read_metaobject 
    nowait_pessimistic_exclusive_scheme(lckattr_nowait, lck_exclusive);

//
// Pessimistic metaobject for child objects accessed always through parent 
// object. So locking parent object effectivly syncronize access to child
// and no explicit child locking is required (parent and child
// should be located in the same storage, otherwise we can access
// deteriorated instance of child object).
//

int hierarchical_access_metaobject::get_transaction_object_flags(hnd_t)
{
    return 0; // validation is not necessary
}

hierarchical_access_metaobject hierarchical_access_scheme;

transient_metaobject::transient_metaobject() {}

void transient_metaobject::abort_object_changes(hnd_t, abort_reason) {}
void transient_metaobject::abort_transaction(const database*, abort_reason) {}
void transient_metaobject::begin_read(hnd_t) {}
void transient_metaobject::begin_transaction() {}
void transient_metaobject::begin_write(hnd_t) {}
void transient_metaobject::commit_object_changes(hnd_t) {}
void transient_metaobject::commit_transaction() {}
void transient_metaobject::destroy(hnd_t) {}
void transient_metaobject::end_read(hnd_t) {}
void transient_metaobject::end_transaction() {}
void transient_metaobject::end_write(hnd_t) {}
int  transient_metaobject::get_transaction_object_flags(hnd_t) { return 0; }
void transient_metaobject::lock(hnd_t) {}
void transient_metaobject::notify(hnd_t hnd) {}
void transient_metaobject::release_transaction_object(hnd_t) {}
void transient_metaobject::unlock(hnd_t) {}
void transient_metaobject::wait(hnd_t) {}
boolean transient_metaobject::wait(hnd_t, time_t) { return True; }

void transient_metaobject::make_persistent(hnd_t, hnd_t)
{
    // Object instances using the transient_metaobject should NEVER be
    // persisted to the database.  Referring to a transient object
    // instance from a persistance object is a programmer error.
    assert(False);
}

void transient_metaobject::put_in_cache(hnd_t)
{
    // Object instances using the transient_metaobject should NEVER be
    // persisted to the database.  Referring to a transient object
    // instance from a persistance object is a programmer error.
    assert(False);
}    

void transient_metaobject::get_from_cache(hnd_t)
{ 
    // Object instances using the transient_metaobject should NEVER be
    // persisted to the database.  Referring to a transient object
    // instance from a persistance object is a programmer error.
    assert(False);
}

void transient_metaobject::insert_in_transaction_list(hnd_t, hnd_t)
{
//  hnd->access += 1;
}

void transient_metaobject::remove_from_transaction_list(hnd_t) 
{
//  hnd->access -= 1;
}

void transient_metaobject::invalidate(hnd_t)
{
    // Since transient objects can only be accessed by a single process,
    // it should never be required (or possible) to invalidate them.
    assert(False);
}


void transient_metaobject::signal_on_modification(hnd_t, event&) 
{
    // Since transient objects can only be accessed by a single process,
    // it should never be required (or possible) to notify anyone that
    // another process has modified the object.
    assert(False);
}

void transient_metaobject::set_cache_limit(size_t, size_t) {}
void transient_metaobject::set_abort_transaction_hook(abort_hook_t) {}

transient_metaobject transient_scheme;

END_GOODS_NAMESPACE