bal_release/src/common/db_engine/bcm_db_engine.c

/******************************************************************************
 *
 *  <:copyright-BRCM:2016:DUAL/GPL:standard
 *  
 *     Copyright (c) 2016 Broadcom
 *     All Rights Reserved
 *  
 *  Unless you and Broadcom execute a separate written software license
 *  agreement governing use of this software, this software is licensed
 *  to you under the terms of the GNU General Public License version 2
 *  (the "GPL"), available at http://www.broadcom.com/licenses/GPLv2.php,
 *  with the following added to such license:
 *  
 *     As a special exception, the copyright holders of this software give
 *     you permission to link this software with independent modules, and
 *     to copy and distribute the resulting executable under terms of your
 *     choice, provided that you also meet, for each linked independent
 *     module, the terms and conditions of the license of that module.
 *     An independent module is a module which is not derived from this
 *     software.  The special exception does not apply to any modifications
 *     of the software.
 *  
 *  Not withstanding the above, under no circumstances may you combine
 *  this software in any way with any other Broadcom software provided
 *  under a license other than the GPL, without Broadcom's express prior
 *  written consent.
 *  
 *  :>
 *
 *****************************************************************************/
 
/*
 * Data Base Engine
 *
 * Proprietary and confidential.
 */
#include <bcmos_system.h>
#include <bcm_db_engine.h>

/* DB trace level */
bcmos_trace_level bcmdb_trace_level = BCMOS_TRACE_LEVEL_ERROR;

#define BCMDB_TRACE(level, fmt, args...)        \
    do {                                        \
        if (level <= bcmdb_trace_level)     \
             bcmos_trace(level, "%s#%d> " fmt, __FUNCTION__, __LINE__, ## args);    \
    } while (0)

/* Print error trace conditionally, depending on the current trace level */
#define BCMDB_TRACE_ERROR(fmt, args...)        \
    BCMDB_TRACE(BCMOS_TRACE_LEVEL_ERROR, "DB ERR: %s#%d> " fmt, __FUNCTION__, __LINE__, ## args)

/* Print warning trace conditionally, depending on the current trace level */
#define BCMDB_TRACE_WARNING(fmt, args...)        \
    BCMDB_TRACE(BCMOS_TRACE_LEVEL_WARNING, "DB WARN: %s#%d> " fmt, __FUNCTION__, __LINE__, ## args)

/* Print info trace conditionally, depending on the current trace level */
#define BCMDB_TRACE_INFO(fmt, args...)        \
    BCMDB_TRACE(BCMOS_TRACE_LEVEL_INFO, "DB INFO: %s#%d> " fmt, __FUNCTION__, __LINE__, ## args)

/* Enable debug prints */
#define BCMDB_DEBUG

#ifdef BCMDB_DEBUG

/* Print debug trace conditionally, depending on the current trace level */
#define BCMDB_TRACE_DEBUG(fmt, args...)        \
    BCMDB_TRACE(BCMOS_TRACE_LEVEL_DEBUG, "DB DBG: %s#%d> " fmt, __FUNCTION__, __LINE__, ## args)

#else

#define BCMDB_TRACE_DEBUG(fmt, args...)

#endif



/** Data base entry
 * \ingroup bcmdb
 */
struct bcmdb_entry
{
    void *data;       /* Set or record */
    uint8_t flags;
#define BCMDB_FLAG_VALID    0x01  /**< Entry is valid */
#define BCMDB_FLAG_RECORD   0x02  /**< Record */
#define BCMDB_FLAG_SOS      0x10  /**< Set of sets */
    uint8_t read_count;
    uint8_t write_pending;
    uint8_t ffu;
};

/** Set/Record change notification
 * \ingroup bcmdb
 */
typedef struct bcmdb_notify
{
    struct bcmdb_notify *next;
    bcmdb_notify_cb cb;
    long cb_priv;
} bcmdb_notify;

/** Data Base Set control block
 * \ingroup bcmdb
 */
struct bcmdb_set
{
    bcmdb_entry entry;        /**< Common fields for set and record */
    char *name;                 /**< Set name */
    bcmdb_set *parent;        /**< Set parent */
    bcmdb_key my_key;         /**< Key in the parent set */
    int max_entries;            /**< Max number of elements in the set. -1=inlimited */
    int num_entries;            /**< Current number of elements in the set */
    int entry_size;             /**< Set entry size. */
    int magic;                  /**< Magic number */
#define BCMDB_MAGIC_ACTIVE_SET         (('a'<<24) | ('s'<<16) | ('e'<<8) | 't')
#define BCMDB_MAGIC_FREE_SET           (('f'<<24) | ('s'<<16) | ('e'<<8) | 't')

    /** Get next element */
    bcmdb_key (*entry_get_next)(const bcmdb_set *this, bcmdb_key cur);

    /** Add new entry. returns 0 if ok */
    int (*entry_new)(bcmdb_set *this, bcmdb_key key, const void *data);

    /** Delete entry */
    int (*entry_delete)(bcmdb_set *this, bcmdb_entry *entry);

    /*
     * Handle – key mapping
     */

    /** Convert entry handle to entry key */
    bcmdb_key (*handle_to_key)(const bcmdb_set *this, const bcmdb_entry *entry);

    /** Convert entry key to entry handle */
    bcmdb_entry *(*key_to_handle)(const bcmdb_set *this, bcmdb_key key);

    /*
     * Set/Record locking
     * entry is handle of the set or record to be locked/unlocked
     */

    /** Lock the whole set for reading */
    void (*lock_set_read)(bcmdb_set *set);

    /** Unlock set locked for reading */
    void (*unlock_set_read)(bcmdb_set *set);

    /** Lock the whole set for update */
    long (*lock_set_modify)(bcmdb_set *set);

    /** Unlock set locked for update */
    void (*unlock_set_modify)(bcmdb_set *set, long fastlock_flags);

    /** Lock the set recursively for update (SoS only) */
    void (*lock_set_recursively_modify)(bcmdb_set *set);

    /** Unlock recursively set locked for update (SoS only)  */
    void (*unlock_set_recursively_modify)(bcmdb_set *set);

    /** Lock record for reading */
    void *(*lock_record_read)(bcmdb_set *set, bcmdb_key key);

    /** Release read lock */
    void (*unlock_record_read)(bcmdb_set *set, bcmdb_key key);

    /** Lock record for modification. */
    void *(*lock_record_write)(bcmdb_set *set, bcmdb_key key, int is_deletion);

    /** Release modify lock */
    void (*unlock_record_write)(bcmdb_set *set, int is_deletion, int is_cancellation);

    /** Format function that converts record data to human-readable form */
    bcmdb_format_cb format;

    /** Update notification mechanism.\n
     * Note that notification functions are called before the actual update of the data base, so that
     * there is an option to abort the update if needed.
     */
    bcmdb_notify *notify_list_head;

    /** Shadow data record */
    void *shadow_data;

    /** holds the pointer to a write-locked entry */
    bcmdb_entry *write_locked_entry ;

    /** mutex that prevents multiple simultaneous updates */
    bcmos_mutex mutex_update;

    /** semaphore that holds update while there are unfinished reads */
    bcmos_sem sem_wait_read_to_finish;

    /** fastlock */
    bcmos_fastlock fastlock;
};


/**< Data base record
 * \ingroup bcmdb
 */
struct bcmdb_record
{
    struct bcmdb_entry e;       /**< Entry - common part for set and record */
};

/*
 * DB backend callbacks
 */

/*
 * Array-based backend
 */

/** Get next element */
static bcmdb_key _bcmdb_array_entry_get_next(const bcmdb_set *this, bcmdb_key key)
{
    BUG_ON(!this->entry.data);

    if (key < 0)
        key = 0;
    else
        ++key;

    while((unsigned)key<this->max_entries)
    {
        bcmdb_entry *entry = (bcmdb_entry *)this->entry.data + key;
        if ((entry->flags & BCMDB_FLAG_VALID))
            break;
        ++key;
    }

    if ((unsigned)key >= this->max_entries)
        return BCMDB_KEY_NO_MORE;  /* no more */

    return key;
}

/*
 * Handle – key mapping
 */

/** Convert entry handle to entry key */
static inline bcmdb_key _bcmdb_array_handle_to_key(const bcmdb_set *this, const bcmdb_entry *entry)
{
    bcmdb_key key;

    BUG_ON(!this);
    BUG_ON(!entry);
    BUG_ON(!this->magic == BCMDB_MAGIC_ACTIVE_SET);
    BUG_ON(!this->entry.data);

    key = entry - (bcmdb_entry *)this->entry.data;
    if ((unsigned)key >= this->max_entries)
        return BCMDB_KEY_INVAL;

    return key;
}


/** Convert entry key to entry handle */
static inline bcmdb_entry *_bcmdb_array_key_to_handle(const bcmdb_set *this, bcmdb_key key)
{
    BUG_ON(!this);
    BUG_ON(!this->magic == BCMDB_MAGIC_ACTIVE_SET);
    BUG_ON(!this->entry.data);

    if ((unsigned long)key >= this->max_entries)
        return NULL;

    return (bcmdb_entry *)this->entry.data + key;
}


/** sem-based set read-lock */
static void bcmdb_set_lock_read_sem(bcmdb_set *set)
{
    bcmos_mutex_lock(&set->mutex_update, BCMOS_WAIT_FOREVER);
}


/** sem-based set read-unlock */
static void bcmdb_set_unlock_read_sem(bcmdb_set *set)
{
    bcmos_mutex_unlock(&set->mutex_update);
}


/** sem-based set modify-lock */
static long bcmdb_set_lock_modify_sem(bcmdb_set *set)
{
    bcmos_mutex_lock(&set->mutex_update, BCMOS_WAIT_FOREVER);
    return 0;
}


/** sem-based set modify-unlock */
static void bcmdb_set_unlock_modify_sem(bcmdb_set *set, long fastlock_flags)
{
    bcmos_mutex_unlock(&set->mutex_update);
}


/** helper function which recursively locks all subsets of SoS
 *  for modify */
static void bcmdb_recursive_subsets_lock_modify(bcmdb_set *sos)
{
    int key;
    int left_entries = sos->num_entries;
    bcmdb_entry *entry;
    bcmdb_set *subset;

    BUG_ON(!(sos->entry.flags & BCMDB_FLAG_SOS));

    for (key = 0; key < sos->max_entries; ++key)
    {
        entry = sos->key_to_handle(sos, key);

        if ((entry->flags & BCMDB_FLAG_VALID))
        {
            subset = (bcmdb_set *)entry->data;
            subset->lock_set_recursively_modify(subset);

            --left_entries;
            /* if we handled all subsets we can break the "for" */
            if (left_entries==0)
            {
                break;
            }
        }
    }
}


/** helper function which recursively unlocks all subsets of SoS
 *  for modify */
static void bcmdb_recursive_subsets_unlock_modify(bcmdb_set *sos)
{
    int key;
    int left_entries = sos->num_entries;
    bcmdb_entry *entry;
    bcmdb_set *subset;

    BUG_ON(!(sos->entry.flags & BCMDB_FLAG_SOS));

    for (key = 0; key < sos->max_entries; ++key)
    {
        entry = sos->key_to_handle(sos, key);

        if ((entry->flags & BCMDB_FLAG_VALID))
        {
            subset = (bcmdb_set *)entry->data;
            subset->unlock_set_recursively_modify(subset);

            --left_entries;
            /* if we handled all subsets we can break the "for" */
            if (left_entries==0)
            {
                break;
            }
        }
    }
}


/** sem-based set modify-lock recursively */
static void bcmdb_set_lock_recursively_modify_sem(bcmdb_set *set)
{
    BCMDB_TRACE_DEBUG("lock set modify recursively: %s\n", set->name);
    bcmos_mutex_lock(&set->mutex_update, BCMOS_WAIT_FOREVER);
    BCMDB_TRACE_DEBUG("mutex was locked\n");

    if ((set->entry.flags & BCMDB_FLAG_SOS))
    {
        BCMDB_TRACE_DEBUG("going to lock the subsets\n");
        bcmdb_recursive_subsets_lock_modify(set);
    }
}


/** sem-based set modify-unlock recursively */
static void bcmdb_set_unlock_recursively_modify_sem(bcmdb_set *set)
{
    BCMDB_TRACE_DEBUG("unlock set modify recursively: %s\n", set->name);

    if ((set->entry.flags & BCMDB_FLAG_SOS))
    {
        BCMDB_TRACE_DEBUG("going to unlock the subsets\n");
        bcmdb_recursive_subsets_unlock_modify(set);
    }

    bcmos_mutex_unlock(&set->mutex_update);
    BCMDB_TRACE_DEBUG("mutex was unlocked\n");
}


/** NB-read-sem-write policy: set read lock */
static void bcmdb_set_lock_read__nb_read_sem_write(bcmdb_set *set)
{
    long flags;

    BCMDB_TRACE_DEBUG("lock set read: %s\n", set->name);
    flags = bcmos_fastlock_lock(&set->fastlock);
    ++set->entry.read_count;
    BCMDB_TRACE_DEBUG("read_count is now: %u\n", set->entry.read_count);
    bcmos_fastlock_unlock(&set->fastlock, flags);
}


/** NB-read-sem-write policy: set read unlock */
static void bcmdb_set_unlock_read__nb_read_sem_write(bcmdb_set *set)
{
    long flags;

    BCMDB_TRACE_DEBUG("unlock set read: %s\n", set->name);

    flags = bcmos_fastlock_lock(&set->fastlock);
    BUG_ON(!set->entry.read_count);
    if (!(--set->entry.read_count) &&
        set->entry.write_pending)
    {
        BCMDB_TRACE_DEBUG("going to wake pending writer\n");

        set->entry.write_pending = 0;
        bcmos_fastlock_unlock(&set->fastlock, flags);
        bcmos_sem_post(&set->sem_wait_read_to_finish);
    }
    else
    {
        bcmos_fastlock_unlock(&set->fastlock, flags);
    }
}


/** NB-read-sem-write policy: set modify lock */
static long bcmdb_set_lock_modify__nb_read_sem_write(bcmdb_set *set)
{
    long flags;

    BCMDB_TRACE_DEBUG("lock set modify: %s\n", set->name);
    bcmos_mutex_lock(&set->mutex_update, BCMOS_WAIT_FOREVER);
    BCMDB_TRACE_DEBUG("mutex was locked\n");

    while(1)
    {
        flags = bcmos_fastlock_lock(&set->fastlock);
        if (!set->entry.read_count)
            break;
        /* Wait until read is completed */
        set->entry.write_pending = 1;
        bcmos_fastlock_unlock(&set->fastlock, flags);
        bcmos_sem_wait(&set->sem_wait_read_to_finish, BCMOS_WAIT_FOREVER);
    }
    /* At this point fastlock is taken and there are no pending reads */

    BCMDB_TRACE_DEBUG("fastlock is taken, no active reads\n");

    return flags;
}


/** NB-read-sem-write policy: set modify unlock */
static void bcmdb_set_unlock_modify__nb_read_sem_write(bcmdb_set *set, long fastlock_flags)
{
    BCMDB_TRACE_DEBUG("unlock set modify: %s\n", set->name);
    bcmos_fastlock_unlock(&set->fastlock, fastlock_flags);
    bcmos_mutex_unlock(&set->mutex_update);
    BCMDB_TRACE_DEBUG("mutex was unlocked\n");
}


/** sem-read/sem-write policy: lock entry */
static void *bcmdb_sem_read_sem_write_lock(bcmdb_set *set, bcmdb_key key, int is_deletion)
{
    bcmdb_entry *entry;
    bcmos_mutex_lock(&set->mutex_update, BCMOS_WAIT_FOREVER);

    if (is_deletion)
    {
        /* there is nothing to return in deletion case */
        return NULL;
    }

    entry = set->key_to_handle(set, key);
    if (!entry || !(entry->flags & BCMDB_FLAG_VALID))
    {
        bcmos_mutex_unlock(&set->mutex_update);
        return NULL;
    }
    return entry->data;
}

/** sem-read/sem-write policy: unlock entry */
static void bcmdb_sem_read_sem_write_unlock(bcmdb_set *set)
{
    bcmos_mutex_unlock(&set->mutex_update);
}

/** sem-read/sem-write policy: lock entry for reading */
static void *bcmdb_sem_read_sem_write_lock_read(bcmdb_set *set, bcmdb_key key)
{
    return bcmdb_sem_read_sem_write_lock(set, key, 0) ;
}

/** sem-read/sem-write policy: unlock entry for reading */
static void bcmdb_sem_read_sem_write_unlock_read(bcmdb_set *set, bcmdb_key key)
{
    bcmdb_sem_read_sem_write_unlock(set);
}

/** sem-read/sem-write policy: lock entry for writing */
static void *bcmdb_sem_read_sem_write_lock_write(bcmdb_set *set, bcmdb_key key, int is_deletion)
{
    return bcmdb_sem_read_sem_write_lock(set, key, is_deletion) ;
}

/** sem-read/sem-write policy: unlock entry for writing */
static void bcmdb_sem_read_sem_write_unlock_write(bcmdb_set *set, int is_deletion, int is_cancellation)
{
    bcmdb_sem_read_sem_write_unlock(set);
}

/** non-blocking-read/shadow write policy: Lock entry for reading */
static void *bcmdb_nb_read_shadow_write_lock_read(bcmdb_set *set, bcmdb_key key)
{
    bcmdb_entry *entry;
    long flags;

    flags = bcmos_fastlock_lock(&set->fastlock);
    entry = set->key_to_handle(set, key);
    if (!entry || !(entry->flags & BCMDB_FLAG_VALID))
    {
        bcmos_fastlock_unlock(&set->fastlock, flags);
        return NULL;
    }
    ++entry->read_count;

    BCMDB_TRACE_DEBUG("lock record read: %s, key=%d new_read_count=%d\n", set->name, key, entry->read_count);

    bcmos_fastlock_unlock(&set->fastlock, flags);
    return entry->data;
}


/** non-blocking-read/shadow write policy: Unlock entry for reading */
static void bcmdb_nb_read_shadow_write_unlock_read(bcmdb_set *set, bcmdb_key key)
{
    bcmdb_entry *entry=set->key_to_handle(set, key);
    long flags;
    BUG_ON(!entry);

    BCMDB_TRACE_DEBUG("unlock record read: %s, key=%d\n", set->name, key);

    flags = bcmos_fastlock_lock(&set->fastlock);
    /* If write is pending - finish it and release write lock */
    BUG_ON(!entry->read_count);
    if (!(--entry->read_count) && set->entry.write_pending)
    {
        BCMDB_TRACE_DEBUG("going to wake pending writer\n");

        /* Write was pending. Release write task */
        set->entry.write_pending = 0;
        bcmos_fastlock_unlock(&set->fastlock, flags);
        bcmos_sem_post(&set->sem_wait_read_to_finish);
    }
    else
    {
        BCMDB_TRACE_DEBUG("read_count is now: %u\n", entry->read_count);
        bcmos_fastlock_unlock(&set->fastlock, flags);
    }
}

/** non-blocking-read/shadow write policy: Lock entry for
 *  writing/deletion.
 *  returned value of NULL means error only in case that
 *  is_deletion is 0 */
static void *bcmdb_nb_read_shadow_write_lock_write(bcmdb_set *set, bcmdb_key key, int is_deletion)
{
    bcmdb_entry *entry;

    BCMDB_TRACE_DEBUG("lock record write: %s, key=%d\n", set->name, key);

    bcmos_mutex_lock(&set->mutex_update, BCMOS_WAIT_FOREVER);
    BCMDB_TRACE_DEBUG("mutex was locked\n");

    if (is_deletion)
    {
        /* there is nothing to return in deletion case */
        return NULL;
    }

    /* this check is needed since mutex_update is task-aware.
       it is not allowed for a task to lock for writing a 2nd record before unlocking the first one. */
    if (set->write_locked_entry)
    {
        BCMDB_TRACE_ERROR("there is already an entry locked for writing\n");
        bcmos_mutex_unlock(&set->mutex_update);
        return NULL;
    }

    entry = set->key_to_handle(set, key);
    if (!entry || !(entry->flags & BCMDB_FLAG_VALID))
    {
        bcmos_mutex_unlock(&set->mutex_update);
        return NULL;
    }
    /* Copy data to shadow entry */
    memcpy(set->shadow_data, entry->data, set->entry_size);
    set->write_locked_entry = entry;
    return set->shadow_data;
}

/** non-blocking-read/shadow write policy: Unlock entry for
 *  writing/deletion */
static void bcmdb_nb_read_shadow_write_unlock_write(bcmdb_set *set, int is_deletion, int is_cancellation)
{
    bcmdb_entry *entry = set->write_locked_entry;
    long flags;
    void *old_data;

    /* no entry is locked */
    BUG_ON(!entry);

    BCMDB_TRACE_DEBUG("unlock record write: %s\n", set->name);

    /* cancellation: no need to update the entry from the shadow (or to delete the entry in case of deletion). */
    if (is_cancellation)
    {
        set->write_locked_entry = NULL;
        bcmos_mutex_unlock(&set->mutex_update);
        return;
    }

    while(1)
    {
        flags = bcmos_fastlock_lock(&set->fastlock);
        /* Wait until neither record nor set are locked for reading */
        if (!entry->read_count && !set->entry.read_count)
            break;

        /* Read lock is active. wait */

        BCMDB_TRACE_DEBUG("read lock is active. going to sleep.\n");

        set->entry.write_pending = 1;
        bcmos_fastlock_unlock(&set->fastlock, flags);
        bcmos_sem_wait(&set->sem_wait_read_to_finish, BCMOS_WAIT_FOREVER);
    }

    /* At this point there is no read lock and fastlock is taken. */

    BCMDB_TRACE_DEBUG("fastlock is taken, no active reads\n");

    if (is_deletion)
    {
        /* delete the entry */
        set->entry_delete(set, entry);
    }
    else
    {
        /* Exchange record data with shadow and release all locks. */
        old_data = entry->data;
        entry->data = set->shadow_data;
        set->shadow_data = old_data;
        set->write_locked_entry = NULL;
    }

    bcmos_fastlock_unlock(&set->fastlock, flags);
    bcmos_mutex_unlock(&set->mutex_update);
    BCMDB_TRACE_DEBUG("mutex was unlocked\n");
}

/** none policy: set read-lock */
static inline void bcmdb_set_lock_read_dummy(bcmdb_set *set)
{
}

/** none policy: set read-unlock */
static inline void bcmdb_set_unlock_read_dummy(bcmdb_set *set)
{
}

/** none policy: set modify-lock */
static inline long bcmdb_set_lock_modify_dummy(bcmdb_set *set)
{
    return 0;
}

/** none policy: set modify-unlock */
static inline void bcmdb_set_unlock_modify_dummy(bcmdb_set *set, long fastlock_flags)
{
}

/** none policy: set modify-lock recursively  */
static void bcmdb_set_lock_recursively_modify_dummy(bcmdb_set *set)
{
}

/** none policy: set modify-unlock recursively  */
static void bcmdb_set_unlock_recursively_modify_dummy(bcmdb_set *set)
{
}

/** none policy: record lock */
static inline void *bcmdb_dummy_lock(bcmdb_set *set, bcmdb_key key, int is_deletion)
{
    bcmdb_entry *entry;

    /* there is nothing to return in deletion case */
    if (is_deletion)
        return NULL;

    entry = set->key_to_handle(set, key);
    if (!entry || !(entry->flags & BCMDB_FLAG_VALID))
        return NULL;
    return entry->data;
}

/** none policy: record unlock */
static inline void bcmdb_dummy_unlock(bcmdb_set *set)
{
}

/** none policy: record lock for reading */
static inline void *bcmdb_dummy_lock_read(bcmdb_set *set, bcmdb_key key)
{
    return bcmdb_dummy_lock(set, key, 0);
}

/** none policy: record unlock for reading */
static inline void bcmdb_dummy_unlock_read(bcmdb_set *set, bcmdb_key key)
{
    bcmdb_dummy_unlock(set);
}

/** none policy: record lock for writing */
static inline void *bcmdb_dummy_lock_write(bcmdb_set *set, bcmdb_key key, int is_deletion)
{
    return bcmdb_dummy_lock(set, key, is_deletion);
}

/** none policy: record unlock for writing */
static inline void bcmdb_dummy_unlock_write(bcmdb_set *set, int is_deletion, int is_cancellation)
{
    bcmdb_dummy_unlock(set);
}



/** Add new sub-set. returns 0 if ok
 * data contains new set handle
 */
static int bcmdb_set_new(bcmdb_set *this, bcmdb_key key, const void *data)
{
    /* Although this callback takes "const void *" parameter,
     * it is just for compatibility fith SoR->new_entry interface.
     * For SoS this parameter is not constant on application level
     * (see bcmdb_set_add())
     */
    bcmdb_entry *entry = this->key_to_handle(this, key);
    bcmdb_set *new_set = (bcmdb_set *)(long)data;

    if (!entry)
        return BCM_ERR_PARM;
    if ((entry->flags & BCMDB_FLAG_VALID))
        return BCM_ERR_ALREADY;
    ++this->num_entries;
    entry->data = (void *)(long)data;
    entry->flags |= BCMDB_FLAG_VALID;
    new_set->my_key = key;
    new_set->parent = this;
    return 0;
}

/** Add new record. returns 0 if ok
 * data contains record data pointer
 */
static int bcmdb_record_new(bcmdb_set *this, bcmdb_key key, const void *data)
{
    bcmdb_record *record = (bcmdb_record *)this->key_to_handle(this, key);
    if (!record || !record->e.data)
        return BCM_ERR_PARM;
    if ((record->e.flags & BCMDB_FLAG_VALID))
        return BCM_ERR_ALREADY;
    ++this->num_entries;
    memcpy(record->e.data, data, this->entry_size);
    record->e.flags |= BCMDB_FLAG_VALID;
    return 0;
}

/** Delete entry */
static int bcmdb_entry_delete(bcmdb_set *this, bcmdb_entry *entry)
{
    if (!entry)
        return BCM_ERR_PARM;
    if (!(entry->flags & BCMDB_FLAG_VALID))
        return BCM_ERR_ALREADY;
    entry->flags &= ~BCMDB_FLAG_VALID;
    --this->num_entries;
    return 0;
}


/*
 * External APIs
 */


/** Initialize data base engine
 *
 * \return
 *      0   - OK\n
 *      <0  - error code
 * \ingroup bcmdb
 */
int bcmdb_module_init(void)
{
    return 0;
}


/** Make set-of-sets control block.
 *
 * Helper function that creates a set of sets with reasonable defaults for all callbacks and fields.
 * Once created, the set control block can be tuned before adding the new set to its parent set.
 * \param[in]   init            set parameters
 * \param[out]  new_set         set control block
 * \return
 *      0   - OK\n
 *      <0  - error code
 */
int bcmdb_make_set_of_sets(const bcmdb_sos_init *init, bcmdb_set **new_set)
{
    bcmdb_set *sos;
    bcmdb_entry *entries;
    int rc;

    /* Parameter check */
    if (!init || !init->name || !new_set)
        return BCM_ERR_PARM;
    if ((init->backend_type == BCMDB_BACKEND_ARRAY) && !init->max_entries)
        return BCM_ERR_PARM;

    /* Allocate set control block and set records */
    sos = bcmos_calloc(sizeof(bcmdb_set) + strlen(init->name) + 1);
    if (!sos)
        return BCM_ERR_NOMEM;
    sos->name = (char *)(sos + 1);
    strcpy(sos->name, init->name);
    sos->entry_size = sizeof(bcmdb_set);
    sos->max_entries = init->max_entries;
    sos->my_key = BCMDB_KEY_INVAL;
    sos->entry.flags = BCMDB_FLAG_SOS;
    sos->magic = BCMDB_MAGIC_ACTIVE_SET;

    /* Set backend callbacks */
    switch(init->backend_type)
    {
        case BCMDB_BACKEND_ARRAY:
            entries = bcmos_calloc(sizeof(bcmdb_set)*init->max_entries);
            if (!entries)
            {
                bcmos_free(sos);
                return BCM_ERR_NOMEM;
            }
            sos->entry.data = entries;
            sos->entry_get_next = _bcmdb_array_entry_get_next;
            sos->handle_to_key = _bcmdb_array_handle_to_key;
            sos->key_to_handle = _bcmdb_array_key_to_handle;
            sos->entry_new = bcmdb_set_new;
            sos->entry_delete = bcmdb_entry_delete;
            break;

        default:
            printf("Only array-based DB backend is supported\n");
            bcmos_free(sos);
            return BCM_ERR_NOT_SUPPORTED;
    }

    /* Set locking callbacks. SoS locking policy is always SEMAPHORE */

    /* in SoS, locking for read is same as for write (and is done recursively). */
    sos->lock_set_read = bcmdb_set_lock_recursively_modify_sem;
    sos->unlock_set_read = bcmdb_set_unlock_recursively_modify_sem;
    sos->lock_set_modify = bcmdb_set_lock_modify_sem;
    sos->unlock_set_modify = bcmdb_set_unlock_modify_sem;
    sos->lock_set_recursively_modify = bcmdb_set_lock_recursively_modify_sem ;
    sos->unlock_set_recursively_modify = bcmdb_set_unlock_recursively_modify_sem ;

    /* create mutex_update */
    rc = bcmos_mutex_create(&sos->mutex_update, init->os_flags);
    if (rc)
    {
        bcmos_free(entries);
        bcmos_free(sos);
        return BCM_ERR_NOMEM;
    }

    bcmos_fastlock_init(&sos->fastlock, init->os_flags);

    *new_set = sos;

    return 0;
}




/** Make set-of-records control block.
 *
 * Helper function that creates a set of records with reasonable defaults for all callbacks and fields.
 * Once created, the set control block can be tuned before adding the new set to its parent set.
 * \param[in]   init            set parameters
 * \param[in]   alloc_records   true (1) - allocate memory for all records.
 * \param[out]  new_set         set control block
 * \return
 *      0   - OK\n
 *      <0  - error code
 */
int bcmdb_make_set_of_records(const bcmdb_sor_init *init, int alloc_records, bcmdb_set **new_set)
{
    bcmdb_set *sor;
    bcmdb_entry *entries = NULL;
    void *data = NULL;
    int i;
    int rc ;

    /* Parameter check */
    if (!init || !init->name)
        return BCM_ERR_PARM;
    if ((init->backend_type == BCMDB_BACKEND_ARRAY) && !init->max_entries)
        return BCM_ERR_PARM;
    if (!init->record_size)
        return BCM_ERR_PARM;

    /* Allocate set control block and set records */
    sor = bcmos_calloc(sizeof(bcmdb_set) + strlen(init->name) + 1);
    if (!sor)
        return BCM_ERR_NOMEM;
    sor->name = (char *)(sor + 1);
    strcpy(sor->name, init->name);
    sor->entry_size = init->record_size;
    sor->max_entries = init->max_entries;
    sor->my_key = BCMDB_KEY_INVAL;
    sor->magic = BCMDB_MAGIC_ACTIVE_SET;
    sor->format = init->format;

    /* Set backend callbacks */
    switch(init->backend_type)
    {
        case BCMDB_BACKEND_ARRAY:
            entries = bcmos_calloc(sizeof(bcmdb_entry)*init->max_entries);
            if (!entries)
            {
                bcmos_free(sor);
                return BCM_ERR_NOMEM;
            }
            sor->entry.data = entries;
            sor->entry_get_next = _bcmdb_array_entry_get_next;
            sor->handle_to_key = _bcmdb_array_handle_to_key;
            sor->key_to_handle = _bcmdb_array_key_to_handle;
            sor->entry_new = bcmdb_record_new;
            sor->entry_delete = bcmdb_entry_delete;

            /* Preallocate data */
            if (alloc_records)
            {
                int size = init->max_entries * init->record_size;
                if (init->lock_policy == BCMDB_LOCK_NB_READ_SHADOW_WRITE)
                    size += init->record_size; /* room for shadow entry */
                /* Allocate data + 1 extra for shadow area */
                data = bcmos_calloc(size);
                if (!data)
                {
                    bcmos_free(entries);
                    bcmos_free(sor);
                    return BCM_ERR_NOMEM;
                }
                for(i=0; i<init->max_entries; i++)
                {
                    bcmdb_entry *entry = (bcmdb_entry *)sor->entry.data + i;
                    entry->data = (void *)((long)data + i * init->record_size);
                }
                if (init->lock_policy == BCMDB_LOCK_NB_READ_SHADOW_WRITE)
                {
                    sor->shadow_data = (void *)((long)data + i * init->record_size);
                }
            }

            /* Initialize records */
            for(i=0; i<init->max_entries; i++)
            {
                bcmdb_entry *entry = (bcmdb_entry *)sor->entry.data + i;
                entry->flags = BCMDB_FLAG_RECORD;
            }
            break;

        default:
            printf("Only array-based DB backend is supported\n");
            bcmos_free(sor);
            return BCM_ERR_NOT_SUPPORTED;
    }

    /* Set locking callbacks based on locking policy */
    switch(init->lock_policy)
    {
        case BCMDB_LOCK_SEM_READ_SEM_WRITE:
            sor->lock_record_write = bcmdb_sem_read_sem_write_lock_write;
            sor->lock_record_read = bcmdb_sem_read_sem_write_lock_read;
            sor->unlock_record_write = bcmdb_sem_read_sem_write_unlock_write;
            sor->unlock_record_read = bcmdb_sem_read_sem_write_unlock_read;
            sor->lock_set_read = bcmdb_set_lock_read_sem;
            sor->unlock_set_read = bcmdb_set_unlock_read_sem;
            sor->lock_set_modify = bcmdb_set_lock_modify_sem;
            sor->unlock_set_modify = bcmdb_set_unlock_modify_sem;
            sor->lock_set_recursively_modify = bcmdb_set_lock_recursively_modify_sem ;
            sor->unlock_set_recursively_modify = bcmdb_set_unlock_recursively_modify_sem ;
            break;

        case BCMDB_LOCK_NONE:
        case BCMDB_LOCK_OTHER:
            sor->lock_record_write = bcmdb_dummy_lock_write;
            sor->lock_record_read = bcmdb_dummy_lock_read;
            sor->unlock_record_write = bcmdb_dummy_unlock_write;
            sor->unlock_record_read = bcmdb_dummy_unlock_read;
            sor->lock_set_read = bcmdb_set_lock_read_dummy;
            sor->unlock_set_read = bcmdb_set_unlock_read_dummy;
            sor->lock_set_modify = bcmdb_set_lock_modify_dummy;
            sor->unlock_set_modify = bcmdb_set_unlock_modify_dummy;
            sor->lock_set_recursively_modify = bcmdb_set_lock_recursively_modify_dummy ;
            sor->unlock_set_recursively_modify = bcmdb_set_unlock_recursively_modify_dummy ;
            break;

        case BCMDB_LOCK_NB_READ_SHADOW_WRITE:
            sor->lock_record_write = bcmdb_nb_read_shadow_write_lock_write;
            sor->lock_record_read = bcmdb_nb_read_shadow_write_lock_read;
            sor->unlock_record_write = bcmdb_nb_read_shadow_write_unlock_write;
            sor->unlock_record_read = bcmdb_nb_read_shadow_write_unlock_read;
            sor->lock_set_read = bcmdb_set_lock_read__nb_read_sem_write;
            sor->unlock_set_read = bcmdb_set_unlock_read__nb_read_sem_write;
            sor->lock_set_modify = bcmdb_set_lock_modify__nb_read_sem_write;
            sor->unlock_set_modify = bcmdb_set_unlock_modify__nb_read_sem_write;
            sor->lock_set_recursively_modify = bcmdb_set_lock_recursively_modify_sem ;
            sor->unlock_set_recursively_modify = bcmdb_set_unlock_recursively_modify_sem ;
            break;

        default:
            printf("Lock policy %d is not supported\n", init->lock_policy);
            if (data)
                bcmos_free(data);
            if (entries)
                bcmos_free(entries);
            bcmos_free(sor);
            return BCM_ERR_NOT_SUPPORTED;
    }

    /* create mutex_update */
    rc = bcmos_mutex_create(&sor->mutex_update, init->os_flags);
    if (rc)
    {
        if (data)
            bcmos_free(data);
        if (entries)
            bcmos_free(entries);
        bcmos_free(sor);
        return BCM_ERR_NOMEM;
    }

    /* create sem_wait_read_to_finish. it is initialized to be taken */
    rc = bcmos_sem_create(&sor->sem_wait_read_to_finish, 0, init->os_flags);
    if (rc)
    {
        /* no point to check here the error code of bcmos_mutex_destroy */
        bcmos_mutex_destroy(&sor->mutex_update);
        if (data)
            bcmos_free(data);
        if (entries)
            bcmos_free(entries);
        bcmos_free(sor);
        return BCM_ERR_NOMEM;
    }

    bcmos_fastlock_init(&sor->fastlock, init->os_flags);

    *new_set = sor;

    return 0;
}


/** Lock data set for reading. When set is locked - it can't be
 *  modified.
 *
 * \param[in]   set             data base set to be locked
 *
 * \ingroup bcmdb
 */
void bcmdb_set_lock_read(bcmdb_set *set)
{
    BUG_ON(!set);
    BUG_ON(!set->magic == BCMDB_MAGIC_ACTIVE_SET);
    BUG_ON(!set->lock_set_read);
    set->lock_set_read(set);
}


/** Release data set lock
 *
 * Unlock set locked by \ref bcmdb_set_lock_read
 *
 * \param[in]   set             data base set to be unlocked
 *
 * \ingroup bcmdb
 */
void bcmdb_set_unlock_read(bcmdb_set *set)
{
    BUG_ON(!set);
    BUG_ON(!set->magic == BCMDB_MAGIC_ACTIVE_SET);
    BUG_ON(!set->unlock_set_read);
    set->unlock_set_read(set);
}


/** Lock data set for modify. If the set is SoS, the locking
 *  will be recursive.
 *
 * \param[in]   set             data base set to be locked
 *
 * \ingroup bcmdb
 */
void bcmdb_set_lock_modify(bcmdb_set *set)
{
    BUG_ON(!set);
    BUG_ON(!set->magic == BCMDB_MAGIC_ACTIVE_SET);
    BUG_ON(!set->lock_set_recursively_modify);
    set->lock_set_recursively_modify(set);
}


/** Release data set lock
 *
 * Unlock set locked by \ref bcmdb_set_lock_modify
 *
 * \param[in]   set             data base set to be unlocked
 *
 * \ingroup bcmdb
 */
void bcmdb_set_unlock_modify(bcmdb_set *set)
{
    BUG_ON(!set);
    BUG_ON(!set->magic == BCMDB_MAGIC_ACTIVE_SET);
    BUG_ON(!set->unlock_set_recursively_modify);
    set->unlock_set_recursively_modify(set);
}


/** Add set to the parent set with specific key.
 *
 * The function adds set to the parent set creating data base hierarchy.
 * The function automatically acquires modify lock and releases it
 * in the end of operation.
 * \param[in]   sos             parent set of sets
 * \param[in]   key             key to add new set at
 * \param[in]   new_set         set control block
 * \return
 *      =0  - OK\n
 *      <0  - error code
 * \ingroup bcmdb
 */
int bcmdb_set_add(bcmdb_set *sos, bcmdb_key key, bcmdb_set *new_set)
{
    int rc;
    long fastlock_flags;
    BUG_ON(!sos);
    BUG_ON(!sos->magic == BCMDB_MAGIC_ACTIVE_SET);
    BUG_ON(!(sos->entry.flags & BCMDB_FLAG_SOS));
    BUG_ON(!new_set);
    BUG_ON(!new_set->magic == BCMDB_MAGIC_ACTIVE_SET);
    BUG_ON(!new_set->my_key == BCMDB_KEY_INVAL);
    fastlock_flags = sos->lock_set_modify(sos);
    rc = sos->entry_new(sos, key, new_set);
    sos->unlock_set_modify(sos, fastlock_flags);
    return rc;
}


/** Get set handle given its key.
 *
 * \param[in]   sos             parent set of sets
 * \param[in]   key             set key.
 * \return
 *      !=0 - set handle
 *      NULL- doesn't exist
 * \ingroup bcmdb
 */
bcmdb_set *bcmdb_set_handle(const bcmdb_set *sos, bcmdb_key key)
{
    bcmdb_entry *entry;
    BUG_ON(!sos);
    BUG_ON(!sos->magic == BCMDB_MAGIC_ACTIVE_SET);
    BUG_ON(!(sos->entry.flags & BCMDB_FLAG_SOS));
    entry = sos->key_to_handle(sos, key);
    if (!entry || !(entry->flags & BCMDB_FLAG_VALID))
        return NULL;
    return (bcmdb_set *)entry->data;
}


/** Get set key given its handle.
 *
 * \param[in]   set             set handle
 * \param[in]   key             set key.
 * \return
 *      !=BCMDB_KEY_INVAL - set key
 *      BCMDB_KEY_INVAL - error
 * \ingroup bcmdb
 */
bcmdb_key bcmdb_set_key(const bcmdb_set *set)
{
    BUG_ON(!set);
    BUG_ON(!set->magic == BCMDB_MAGIC_ACTIVE_SET);
    return set->my_key;
}


/** Get set name
 *
 * \param[in]   set             set handle
 * \return set name
 * \ingroup bcmdb
 */
const char *bcmdb_set_name(const bcmdb_set *set)
{
    BUG_ON(!set);
    BUG_ON(!set->magic == BCMDB_MAGIC_ACTIVE_SET);
    return set->name;
}


/** Get number of records in the set.
 *
 * \param[in]   set             set handle
 * \return number of active records in the set
 * \ingroup bcmdb
 */
int bcmdb_set_num_records(const bcmdb_set *set)
{
    BUG_ON(!set);
    BUG_ON(!set->magic == BCMDB_MAGIC_ACTIVE_SET);
    return set->num_entries;
}


/** Get entry size
 *
 * \param[in]   set             set handle
 * \return set entry size
 * \ingroup bcmdb
 */
int bcmdb_set_entry_size(const bcmdb_set *set)
{
    BUG_ON(!set);
    BUG_ON(!set->magic == BCMDB_MAGIC_ACTIVE_SET);
    return set->entry_size;
}


/** Add record to the parent set with specific key.
 *
 * The function creates a new record and adds it to the parent set with specific key.
 * The function automatically acquires modify lock and releases it
 * in the end of operation.
 * \param[in]   sor             parent set of records
 * \param[in]   key             key to add new set at
 * \param[in]   data            record data. Data size is defined at parent SOR registration time.
 * \param[out]  p_record        new record handle
 * \return
 *      =0  - OK\n
 *      <0  - error code
 * \ingroup bcmdb
 */
int bcmdb_record_add(bcmdb_set *sor, bcmdb_key key, const void *data)
{
    int rc;
    long fastlock_flags;

    BUG_ON(!sor);
    BUG_ON(!data);
    BUG_ON(!sor->magic == BCMDB_MAGIC_ACTIVE_SET);
    BUG_ON(!(sor->entry.flags & BCMDB_FLAG_SOS)==0);

    fastlock_flags = sor->lock_set_modify(sor);
    rc=sor->entry_new(sor, key, data);
    sor->unlock_set_modify(sor, fastlock_flags);
    return rc;
}


/** Delete record from the parent SoR given the record key.
 *
 * The function automatically acquires modify lock and releases it
 * in the end of operation.
 *
 * \param[in]   sor             parent set of records
 * \param[in]   key             record key.
 */
void bcmdb_record_delete(bcmdb_set *sor, bcmdb_key key)
{
    BUG_ON(!sor);
    BUG_ON(!sor->magic == BCMDB_MAGIC_ACTIVE_SET);
    BUG_ON(!(sor->entry.flags & BCMDB_FLAG_SOS)==0);

    sor->lock_record_write(sor, key, 1);
    sor->unlock_record_write(sor, 1, 0);
}


/** Get record data pointer without locking.
 *
 * The function returns pointer to data structure stored in data base record.\n
 * Attention! The caller is required to aquire read or write lock - as appropriate
 * before calling this function and release the lock when processing is finished.
 *
 * \param[in]   sor             parent set of records
 * \param[in]   key             record key
 * \return
 *      data pointer. NULL if there is no record matching the key.
 * \ingroup bcmdb
 */
void *bcmdb_record_getraw_nolock(bcmdb_set *sor, bcmdb_key key)
{
    bcmdb_entry *entry;
    BUG_ON(!sor);
    BUG_ON(!sor->magic == BCMDB_MAGIC_ACTIVE_SET);
    BUG_ON(!(sor->entry.flags & BCMDB_FLAG_SOS)==0);
    entry = sor->key_to_handle(sor, key);
    if (!entry || !(entry->flags & BCMDB_FLAG_VALID))
        return NULL;
    return entry->data;
}


/** Lock record for reading and return record data pointer.
 *
 * The function aquires read-lock and returns pointer to data structure stored in data base record.\n
 * read-lock must be released separately when the pointer is no longer in use.
 * Note that the default record-read lock is non-blocking and counting.
 * That means that multiple processes cam read-lock the same record without blocking.
 * The function is low-level. It is recommended to use macro \ref bcmdb_record_get_read instead.
 *
 * \param[in]   sor             parent set of records
 * \param[in]   key             record key
 * \return
 *      data pointer. NULL if there is no record matching the key.
 * \ingroup bcmdb
 */
const void *bcmdb_record_getraw_read(bcmdb_set *sor, bcmdb_key key)
{
    BUG_ON(!sor);
    BUG_ON(!sor->magic == BCMDB_MAGIC_ACTIVE_SET);
    BUG_ON(!(sor->entry.flags & BCMDB_FLAG_SOS)==0);
    return sor->lock_record_read(sor, key);
}


/** Unlock record locked for reading.
 *
 * \param[in]   sor             parent set of records
 * \param[in]   key             record key
 *
 * \ingroup bcmdb
 */
void bcmdb_record_unlock_read(bcmdb_set *sor, bcmdb_key key)
{
    BUG_ON(!sor);
    BUG_ON(!sor->magic == BCMDB_MAGIC_ACTIVE_SET);
    BUG_ON(!(sor->entry.flags & BCMDB_FLAG_SOS)==0);
    sor->unlock_record_read(sor, key);
}


/** Read record data into user area.
 *
 * The function aquires read-lock, reads data into user area and releases read-lock.
 *
 * \param[in]   sor             parent set of records
 * \param[in]   key             record key
 * \param[in]   offset          offset in data record
 * \param[in]   size            data size. Note that offset+size must be <= record_size
 * \param[in]   data            data pointer.
 * \return
 *      =0-OK\n
 *      <0-error code
 * \ingroup bcmdb
 */
int bcmdb_record_read(bcmdb_set *sor, bcmdb_key key, int offset, int size, void *data)
{
    const void *d = bcmdb_record_getraw_read(sor, key);
    if (!d)
        return BCM_ERR_PARM;
    if ((unsigned)offset + (unsigned)size > sor->entry_size)
    {
        bcmdb_record_unlock_read(sor, key);
        return BCM_ERR_PARM;
    }
    memcpy(data, (const char *)d+(unsigned)offset, (unsigned)size);
    bcmdb_record_unlock_read(sor, key);
    return 0;
}


/** Lock record for writing and return record data pointer.
 *
 * The function aquires write-lock and returns pointer to data structure stored in data base record.\n
 * write-lock must be released separately when the pointer is no longer in use.
 * The function is low-level. It is recommended to use macro \ref bcmdb_record_get_write instead.
 *
 * \param[in]   sor             parent set of records
 * \param[in]   key             record key
 * \return
 *      data pointer. NULL if there is no record matching the key.
 * \ingroup bcmdb
 */
void *bcmdb_record_getraw_write(bcmdb_set *sor, bcmdb_key key)
{
    BUG_ON(!sor);
    BUG_ON(!sor->magic == BCMDB_MAGIC_ACTIVE_SET);
    BUG_ON(!(sor->entry.flags & BCMDB_FLAG_SOS)==0);
    return sor->lock_record_write(sor, key, 0);
}


/** Unlock record locked for writing.
 *
 * \param[in]   sor             parent set of records
 * \param[in]   is_cancellation TRUE=cancel transaction
 *
 * \ingroup bcmdb
 */
void bcmdb_record_unlock_write(bcmdb_set *sor, int is_cancellation)
{
    BUG_ON(!sor);
    BUG_ON(!sor->magic == BCMDB_MAGIC_ACTIVE_SET);
    BUG_ON(!(sor->entry.flags & BCMDB_FLAG_SOS)==0);
    sor->unlock_record_write(sor, 0, is_cancellation);
}


/** Write record data.
 *
 * The function aquires modify-lock, replaces data stored in data base record
 * and releses the lock.
 *
 * \param[in]   sor             parent set of records
 * \param[in]   key             record key
 * \param[in]   offset          offset in data record
 * \param[in]   size            data size. Note that offset+size must be <= record_size
 * \param[in]   data            data pointer.
 * \return
 *      =0-OK\n
 *      <0-error code
 * \ingroup bcmdb
 */
int bcmdb_record_write(bcmdb_set *sor, bcmdb_key key, int offset, int size, const void *data)
{
    void *d=bcmdb_record_getraw_write(sor, key);
    if (!d)
        return BCM_ERR_PARM;
    if ((unsigned)offset + (unsigned)size > sor->entry_size)
    {
        bcmdb_record_unlock_write(sor, 0);
        return BCM_ERR_PARM;
    }
    memcpy((char *)d+(unsigned)offset, data, (unsigned)size);
    bcmdb_record_unlock_write(sor, 0);
    return 0;
}


/** Register notification function to get informed
 * when data base set is modified.
 *
 * \param[in]   sor             parent set of records
 * \param[in]   cb              callback function pointer
 * \param[in]   cb_priv         private data that should be passed to the callback
 * \return
 *      =0  - OK\n
 *      <0  - error code
 * \ingroup bcmdb
 */
int bcmdb_set_notify_register(bcmdb_set *sor, bcmdb_notify_cb cb, long cb_priv)
{
    bcmdb_notify *nf_new, *nf, *nf_prev = NULL;

    BUG_ON(!sor);
    BUG_ON(!sor->magic == BCMDB_MAGIC_ACTIVE_SET);
    BUG_ON(!cb);

    nf_new = bcmos_calloc(sizeof(bcmdb_notify));
    if (!nf_new)
        return BCM_ERR_NOMEM;
    nf_new->cb = cb;
    nf_new->cb_priv = cb_priv;

    /* Add to set's notification list */
    nf = sor->notify_list_head;
    while(nf)
    {
        nf_prev = nf;
        nf = nf->next;
    }
    if (nf_prev)
        nf_prev->next = nf_new;
    else
        sor->notify_list_head = nf_new;

    return 0;
}


/** Data base iterator
 *
 * \param[in]   set             data base set
 * \param[in]   prev            last entry. BCMDB_KEY_ANY=start from the beginning
 * \return  data base entry key following prev or BCMDB_KEY_NO_MORE if end is reached.\n
 *          BCMDB_KEY_INVAL is reqturned if prev key is invalid
 * \ingroup bcmdb
 */
bcmdb_key bcmdb_set_iterate(const bcmdb_set *set, bcmdb_key prev)
{
    BUG_ON(!set);
    BUG_ON(!set->magic == BCMDB_MAGIC_ACTIVE_SET);
    return set->entry_get_next(set, prev);
}


/* Print database structure */
static void _bcmdb_print_structure(const bcmdb_set *set, int level)
{
    int i;

    if (!set)
        return;

    /* Indentation */
    for(i=0; i<level; i++)
        printf("\t");

    if ((set->entry.flags & BCMDB_FLAG_SOS))
    {
        bcmdb_key key = bcmdb_set_iterate(set, BCMDB_KEY_ANY);
        printf("%-16s SoS max_entries=%d entries=%d\n", set->name, set->max_entries, set->num_entries);
        while(key >= 0)
        {
            _bcmdb_print_structure(bcmdb_set_handle(set, key), level+1);
            key = bcmdb_set_iterate(set, key);
        }
    }
    else
    {
        printf("%-16s SoR max_entries=%d entries=%d record_size=%d total_size=%d\n",
               set->name, set->max_entries, set->num_entries, set->entry_size,
               set->entry_size*set->max_entries);
    }
}


/** Print database structure.
 *
 * \param[in]   set             root set
 * \ingroup bcmdb
 */
void bcmdb_set_print_structure(const bcmdb_set *set)
{
    _bcmdb_print_structure(set, 0);
}


/** Format record for printing.
 *
 * The function converts record data to human-readible format.
 *
 * \param[in]   sor             parent set of records
 * \param[in]   key             record key
 * \param[out]  buffer          output buffer
 * \param[in]   size            buffer size
 * \return
 *      >=0-amount of data placed in the buffer\n
 *      <0-error code
 */
int bcmdb_record_read_formatted(bcmdb_set *sor, bcmdb_key key, char *buffer, int size)
{
    const void *data;
    int len;
    if (!buffer || !size)
        return BCM_ERR_PARM;
    if (!sor->format)
        return BCM_ERR_NOT_SUPPORTED;
    *buffer=0;
    data = bcmdb_record_getraw_read(sor, key);
    if (!data)
        return 0;
    len = sor->format(data, buffer, size);
    bcmdb_record_unlock_read(sor, key);
    return len;
}