s_persistency.c

Go to the documentation of this file.

/*
 * $Id: s_persistency.c 236 2007-04-18 12:53:40Z vingarzan $
 *
 * Copyright (C) 2004-2007 FhG Fokus
 *
 * This file is part of Open IMS Core - an open source IMS CSCFs & HSS
 * implementation
 *
 * Open IMS Core is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation; either version 2 of the License, or
 * (at your option) any later version.
 *
 * For a license to use the Open IMS Core software under conditions
 * other than those described here, or to purchase support for this
 * software, please contact Fraunhofer FOKUS by e-mail at the following
 * addresses:
 *     info@open-ims.org
 *
 * Open IMS Core is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * It has to be noted that this Open Source IMS Core System is not
 * intended to become or act as a product in a commercial context! Its
 * sole purpose is to provide an IMS core reference implementation for
 * IMS technology testing and IMS application prototyping for research
 * purposes, typically performed in IMS test-beds.
 *
 * Users of the Open Source IMS Core System have to be aware that IMS
 * technology may be subject of patents and licence terms, as being
 * specified within the various IMS-related IETF, ITU-T, ETSI, and 3GPP
 * standards. Thus all Open IMS Core users have to take notice of this
 * fact and have to agree to check out carefully before installing,
 * using and extending the Open Source IMS Core System, if related
 * patents and licences may become applicable to the intended usage
 * context.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 *
 */

#include "s_persistency.h"

extern persistency_mode_t scscf_persistency_mode;
extern char* scscf_persistency_location;        
extern auth_hash_slot_t *auth_data;             
extern int auth_data_hash_size;                 
extern int r_hash_size;                     
extern r_hash_slot *registrar;              
extern int s_dialogs_hash_size;                     
extern s_dialog_hash_slot *s_dialogs;               
/*****  DB related stuff *****/
extern db_con_t* scscf_db; 
extern db_func_t scscf_dbf; 
extern int* auth_snapshot_version;
extern int* auth_step_version;
extern int* dialogs_snapshot_version;
extern int* dialogs_step_version;
extern int* registrar_snapshot_version;
extern int* registrar_step_version;
extern gen_lock_t* db_lock; 
extern char* scscf_name;

int s_dump(bin_data* x, char* location, char* prepend_fname, data_type_t dt);
int bin_dump_to_db(bin_data *x, data_type_t dt);
int bin_dump_registrar_to_table(bin_data *x, int snapshot_version, int step_version);
int bin_dump_dialogs_to_table(bin_data *x, int snapshot_version, int step_version);
int bin_dump_auth_to_table(bin_data *x, int snapshot_version, int step_version);
int bin_bulk_dump_to_table(data_type_t dt, int snapshot_version, int step_version, bin_data *x);
int bin_cache_dump_registrar_to_table(int snapshot_version, int step_version);
int bin_cache_dump_dialogs_to_table(int snapshot_version, int step_version);
int bin_cache_dump_auth_to_table(int snapshot_version, int step_version);
int delete_older_snapshots(char* table, char* node_id, data_type_t dt, int current_snapshot);

int s_load(bin_data *x, char* location, char* prepend_fname, data_type_t dt);
int bin_load_from_db(bin_data *x, data_type_t dt);
int bin_load_registrar_from_table(bin_data *x);
int bin_load_dialogs_from_table(bin_data *x);
int bin_load_auth_from_table(bin_data *x);
int bin_bulk_load_from_table(data_type_t dt, bin_data* x);
int bin_cache_load_registrar_from_table();
int bin_cache_load_dialogs_from_table();
int bin_cache_load_auth_from_table();
int db_get_last_snapshot_version(char* table, char* node_id, data_type_t dt, int* version);
int set_versions(data_type_t dt, int snapshot_version, int step_version);
/*********************/

int make_snapshot_authdata()
{
    bin_data x;
    auth_userdata *aud;
    int i;
    
    /*In WITH_DATABASE_CACHE mode, serialize each hashtable element separately */
    if(scscf_persistency_mode!=WITH_DATABASE_CACHE){    
        if (!bin_alloc(&x,256)) goto error;
        for(i=0;i<auth_data_hash_size;i++){
            auth_data_lock(i);
            aud = auth_data[i].head;
            while(aud){
                if (!bin_encode_auth_userdata(&x,aud)) goto error;
                aud = aud->next;
            }
            auth_data_unlock(i);
        }
        //bin_print(&x);
    }
    i=s_dump(&x,scscf_persistency_location,"authdata", S_AUTH);
    //i = bin_dump(&x,scscf_persistency_mode,scscf_persistency_location,"authdata");        
    if(scscf_persistency_mode!=WITH_DATABASE_CACHE){
        bin_free(&x);
    }
    return i;
error:
    return 0;
}
  

int load_snapshot_authdata()
{
    bin_data x;
    auth_userdata *aud;
    if (!s_load(&x,scscf_persistency_location,"authdata", S_AUTH)) goto error;
    //if (!bin_load(&x,scscf_persistency_mode,scscf_persistency_location,"authdata")) goto error;
    
    if(scscf_persistency_mode!=WITH_DATABASE_CACHE){
        //bin_print(&x);
        x.max=0;
        LOG(L_INFO,"INFO:"M_NAME":load_snapshot_auth: max %d len %d\n",x.max,x.len);
        while(x.max<x.len){
            aud = bin_decode_auth_userdata(&x);     
            if (!aud) return 0;
            LOG(L_INFO,"INFO:"M_NAME":load_snapshot_auth: Loaded auth_userdata for <%.*s>\n",aud->private_identity.len,aud->private_identity.s);
            auth_data_lock(aud->hash);
            aud->prev = auth_data[aud->hash].tail;
            aud->next = 0;
            if (auth_data[aud->hash].tail) auth_data[aud->hash].tail->next = aud;
            auth_data[aud->hash].tail = aud;
            if (!auth_data[aud->hash].head) auth_data[aud->hash].head = aud;
            auth_data_unlock(aud->hash);
        }
        bin_free(&x);
    }
    return 1;
error:
    return 0;
}

void persistency_timer_authdata(unsigned int ticks, void* param)
{
    make_snapshot_authdata();
    
    if (auth_snapshot_version) (*auth_snapshot_version)++;  
}



int make_snapshot_dialogs()
{
    bin_data x;
    s_dialog *d;
    int i;  
    
    /*In WITH_DATABASE_CACHE mode, serialize each hashtable element separately */
    if(scscf_persistency_mode!=WITH_DATABASE_CACHE){
        if (!bin_alloc(&x,256)) goto error;     
        for(i=0;i<s_dialogs_hash_size;i++){
            d_lock(i);
            d = s_dialogs[i].head;
            while(d){
                if (!bin_encode_s_dialog(&x,d)) goto error;
                d = d->next;
            }
            d_unlock(i);
        }
        //bin_print(&x);
    }
    i=s_dump(&x,scscf_persistency_location,"sdialogs",S_DIALOGS);
    //i = bin_dump(&x,scscf_persistency_mode,scscf_persistency_location,"sdialogs");        
    if(scscf_persistency_mode!=WITH_DATABASE_CACHE){
        bin_free(&x);
    }
    return i;
error:
    return 0;
}  

int load_snapshot_dialogs()
{
    bin_data x;
    s_dialog *d;
    if (!s_load(&x,scscf_persistency_location,"sdialogs",S_DIALOGS)) goto error;
    //if (!bin_load(&x,scscf_persistency_mode,scscf_persistency_location,"sdialogs")) goto error;
    
    if(scscf_persistency_mode!=WITH_DATABASE_CACHE){
        //bin_print(&x);
        x.max=0;
        LOG(L_INFO,"INFO:"M_NAME":load_snapshot_dlg: max %d len %d\n",x.max,x.len);
        while(x.max<x.len){
            d = bin_decode_s_dialog(&x);
            if (!d) return 0;
            LOG(L_INFO,"INFO:"M_NAME":load_snapshot_dlg: Loaded s_dialog for <%.*s>\n",d->aor.len,d->aor.s);
            d_lock(d->hash);
            d->prev = s_dialogs[d->hash].tail;
            d->next = 0;
            if (s_dialogs[d->hash].tail) s_dialogs[d->hash].tail->next = d;
            s_dialogs[d->hash].tail = d;
            if (!s_dialogs[d->hash].head) s_dialogs[d->hash].head = d;
            d_unlock(d->hash);
        }
        bin_free(&x);
    }
    return 1;
error:
    return 0;
}


void persistency_timer_dialogs(unsigned int ticks, void* param)
{
    make_snapshot_dialogs();
    
    if (dialogs_snapshot_version) (*dialogs_snapshot_version)++;        
}


int make_snapshot_registrar()
{
    bin_data x;
    r_public *p;
    int i;
    
    /*In WITH_DATABASE_CACHE mode, serialize each hashtable element separately */
    if(scscf_persistency_mode!=WITH_DATABASE_CACHE){
        if (!bin_alloc(&x,256)) goto error;     
        for(i=0;i<r_hash_size;i++){
            r_lock(i);
            p = registrar[i].head;
            while(p){
                if (!bin_encode_r_public(&x,p)) goto error;
                p = p->next;
            }
            r_unlock(i);
        }
        //bin_print(&x);
    }
    i = s_dump(&x,scscf_persistency_location,"sregistrar",S_REGISTRAR);
    //i = bin_dump(&x,scscf_persistency_mode,scscf_persistency_location,"sregistrar");      
    if(scscf_persistency_mode!=WITH_DATABASE_CACHE){
        bin_free(&x);
    }
    return i;
error:
    return 0;
}  

int load_snapshot_registrar()
{
    bin_data x;
    r_public *p;
    if (!s_load(&x,scscf_persistency_location,"sregistrar",S_REGISTRAR)) goto error;
    //if (!bin_load(&x,scscf_persistency_mode,scscf_persistency_location,"sregistrar")) goto error;
    
    if(scscf_persistency_mode!=WITH_DATABASE_CACHE){
        //bin_print(&x);
        x.max=0;
        LOG(L_INFO,"INFO:"M_NAME":load_snapshot_registrar: max %d len %d\n",x.max,x.len);
        while(x.max<x.len){
            p = bin_decode_r_public(&x);
            if (!p) return 0;
            LOG(L_INFO,"INFO:"M_NAME":load_snapshot_registrar: Loaded r_public for <%.*s>\n",p->aor.len,p->aor.s);
            r_lock(p->hash);
            p->prev = registrar[p->hash].tail;
            p->next = 0;
            if (registrar[p->hash].tail) registrar[p->hash].tail->next = p;
            registrar[p->hash].tail = p;
            if (!registrar[p->hash].head) registrar[p->hash].head = p;
            r_unlock(p->hash);
        }
        bin_free(&x);
    }
    return 1;
error:
    return 0;
    
}


void persistency_timer_registrar(unsigned int ticks, void* param)
{
    make_snapshot_registrar();
    
    if (registrar_snapshot_version) (*registrar_snapshot_version)++;
}


/******  DB related functions  *****/

    /* Dump related functions*/
    
int s_dump(bin_data* x, char* location, char* prepend_fname, data_type_t dt){
    
    switch (scscf_persistency_mode){
        case NO_PERSISTENCY:
            LOG(L_ERR,"ERR:"M_NAME":s_dump: Snapshot done but persistency was disabled...\n");
            return 0;
        case WITH_FILES:
            return bin_dump_to_file(x,location,prepend_fname);
        case WITH_DATABASE_BULK:
            return bin_dump_to_db(x, dt);
        case WITH_DATABASE_CACHE:
            return bin_dump_to_db(NULL, dt);//ignore x, x is empty
        default:
            LOG(L_ERR,"ERR:"M_NAME":s_dump: Snapshot done but no such mode %d\n",scscf_persistency_mode);
            return 0;
    }
}

int bin_dump_to_db(bin_data *x, data_type_t dt){
    
    int snapshot_version;
    int step_version;
    
    switch(dt){
        case S_REGISTRAR:
            snapshot_version=*registrar_snapshot_version;
            step_version=*registrar_step_version;
            return bin_dump_registrar_to_table(x, snapshot_version, step_version);
        case S_DIALOGS:
            snapshot_version=*dialogs_snapshot_version;
            step_version=*dialogs_step_version;
            return bin_dump_dialogs_to_table(x, snapshot_version, step_version);
        case S_AUTH:
            snapshot_version=*auth_snapshot_version;
            step_version=*auth_step_version;
            return bin_dump_auth_to_table(x, snapshot_version, step_version);
        default:
            LOG(L_ERR,"ERR:"M_NAME":bin_dump_to_db: No such information to dump %d\n", dt);
            return 0;
    }
}

int bin_dump_registrar_to_table(bin_data *x, int snapshot_version, int step_version){
    
    if(x){//whole hashtable serialized to x
        return bin_bulk_dump_to_table(S_REGISTRAR, snapshot_version, step_version, x);
    }
    else{//serialize and dump each hashtable element separately
        return bin_cache_dump_registrar_to_table(snapshot_version, step_version);
    }
}

int bin_dump_dialogs_to_table(bin_data *x, int snapshot_version, int step_version){
    
    if(x){//whole hashtable serialized to x
        return bin_bulk_dump_to_table(S_DIALOGS, snapshot_version, step_version, x);
    }
    else{//serialize and dump each hashtable element separately
        return bin_cache_dump_dialogs_to_table(snapshot_version, step_version);
    }
}

int bin_dump_auth_to_table(bin_data *x, int snapshot_version, int step_version){
    
    if(x){//whole hashtable serialized to x
        return bin_bulk_dump_to_table(S_AUTH, snapshot_version, step_version, x);
    }
    else{//serialize and dump each hashtable element separately
        return bin_cache_dump_auth_to_table(snapshot_version, step_version);
    }
}

int bin_bulk_dump_to_table(data_type_t dt, int snapshot_version, int step_version, bin_data *x){
    db_key_t keys[5];
    db_val_t vals[5];
    int len;

    /* id auto incremented */
    keys[0] = "node_id";
    keys[1] = "data_type";
    keys[2] = "snapshot_version";
    keys[3] = "step_version";
    /* record_id_1/2/3/4=NULL */
    keys[4] = "data";
    
    vals[0].type = DB_STR;
    vals[0].nul = 0;
    vals[0].val.str_val.s=scscf_name;
    len = strlen(scscf_name);
    vals[0].val.str_val.len=MIN(len, 64);

    vals[1].type = DB_INT;
    vals[1].nul = 0;
    vals[1].val.int_val=dt;

    vals[2].type = DB_INT;
    vals[2].nul = 0;
    vals[2].val.int_val=snapshot_version;
    
    vals[3].type = DB_INT;
    vals[3].nul = 0;
    vals[3].val.int_val=step_version;

    str d = {x->s, x->len};
    vals[4].type = DB_BLOB;
    vals[4].nul = 0;
    vals[4].val.blob_val = d;

    //lock
    lock_get(db_lock);

    if (scscf_dbf.use_table(scscf_db, "snapshot") < 0) {
        LOG(L_ERR, "ERR:"M_NAME":bin_bulk_dump_to_table(): Error in use_table\n");
        lock_release(db_lock);//unlock
        return 0;
    }

    if (scscf_dbf.insert(scscf_db, keys, vals, 5) < 0) {
        LOG(L_ERR, "ERR:"M_NAME":bin_bulk_dump_to_table(): Error while inserting on snapshot table\n");
        lock_release(db_lock);//unlock
        return 0;
    }
    
    //delete older snapshots
    if (delete_older_snapshots("snapshot", scscf_name, dt, snapshot_version)!=1){
        LOG(L_ERR, "ERR:"M_NAME":bin_bulk_dump_to_table(): Error while deleting older snapshots from snapshot table\n");
        lock_release(db_lock);//unlock
        return 0;
    }
    
    //unlock
    lock_release(db_lock);
    
    return 1;
}

int bin_cache_dump_registrar_to_table(int snapshot_version, int step_version){
    bin_data x;
    r_public *p;
    int i;
    int len;
    
    //lock
    lock_get(db_lock);

    if (scscf_dbf.use_table(scscf_db, "snapshot") < 0) {
        LOG(L_ERR, "ERR:"M_NAME":bin_cache_dump_registrar_to_table(): Error in use_table\n");
        goto error;
    }
    
    for(i=0;i<r_hash_size;i++){
        r_lock(i);
        p = registrar[i].head;
        while(p){
            if (!bin_alloc(&x,128)){
                r_unlock(i);
                goto error;
            }
            if (!bin_encode_r_public(&x,p)){
                r_unlock(i);
                goto error;
            }
            
            db_key_t keys[6];
            db_val_t vals[6];
            
            /* id auto incremented */
            keys[0] = "node_id";
            keys[1] = "data_type";
            keys[2] = "snapshot_version";
            keys[3] = "step_version";
            keys[4] = "record_id_1"; /* aor */
            /* record_id_2/3/4=NULL */
            keys[5] = "data";
            
            vals[0].type = DB_STR;
            vals[0].nul = 0;
            vals[0].val.str_val.s=scscf_name;
            len = strlen(scscf_name);
            vals[0].val.str_val.len=MIN(len, 64);

            vals[1].type = DB_INT;
            vals[1].nul = 0;
            vals[1].val.int_val=S_REGISTRAR;

            vals[2].type = DB_INT;
            vals[2].nul = 0;
            vals[2].val.int_val=snapshot_version;
    
            vals[3].type = DB_INT;
            vals[3].nul = 0;
            vals[3].val.int_val=step_version;
    
            vals[4].type = DB_STR;
            vals[4].nul = 0;
            vals[4].val.str_val.s=p->aor.s;
            vals[4].val.str_val.len=MIN(p->aor.len, 256);
        
            vals[5].type = DB_BLOB;
            vals[5].nul = 0;
            str data={x.s, x.len};
            vals[5].val.blob_val = data;
            
            if (scscf_dbf.insert(scscf_db, keys, vals, 6) < 0) {
                LOG(L_ERR, "ERR:"M_NAME":bin_cache_dump_registrar_to_table(): Error while inserting on snapshot table\n");
                r_unlock(i);
                goto error;
            }
            bin_free(&x);
            
            p=p->next;
        }
        r_unlock(i);
    }   
    
    //delete older snapshots
    if (delete_older_snapshots("snapshot", scscf_name, S_REGISTRAR, snapshot_version)!=1){
        LOG(L_ERR, "ERR:"M_NAME":bin_cache_dump_registrar_to_table(): Error while deleting older snapshots from snapshot table\n");
        goto error;
    }
    
    //unlock
    lock_release(db_lock);
    
    return 1;

error:
    lock_release(db_lock);//unlock
    return 0;
}

int bin_cache_dump_dialogs_to_table(int snapshot_version, int step_version){
    bin_data x;
    s_dialog *dialog;
    int i;
    int len;
    
    //lock
    lock_get(db_lock);

    if (scscf_dbf.use_table(scscf_db, "snapshot") < 0) {
        LOG(L_ERR, "ERR:"M_NAME":bin_cache_dump_dialogs_to_table(): Error in use_table\n");
        goto error;
    }
    
    for(i=0;i<s_dialogs_hash_size;i++){
        d_lock(i);
        dialog = s_dialogs[i].head;
        while(dialog){
            if (!bin_alloc(&x,128)){
                d_unlock(i);
                goto error;
            }
            if (!bin_encode_s_dialog(&x,dialog)){
                d_unlock(i);
                goto error;
            }
            
            db_key_t keys[7];
            db_val_t vals[7];
            
            /* id auto incremented */
            keys[0] = "node_id";
            keys[1] = "data_type";
            keys[2] = "snapshot_version";
            keys[3] = "step_version";
            keys[4] = "record_id_1"; /* call-id */
            keys[5] = "record_id_2"; /* aor */
            /* record_id_3/4=NULL */
            keys[6] = "data";
            
            vals[0].type = DB_STR;
            vals[0].nul = 0;
            vals[0].val.str_val.s=scscf_name;
            len = strlen(scscf_name);
            vals[0].val.str_val.len=MIN(len, 64);

            vals[1].type = DB_INT;
            vals[1].nul = 0;
            vals[1].val.int_val=S_DIALOGS;

            vals[2].type = DB_INT;
            vals[2].nul = 0;
            vals[2].val.int_val=snapshot_version;
    
            vals[3].type = DB_INT;
            vals[3].nul = 0;
            vals[3].val.int_val=step_version;
    
            vals[4].type = DB_STR;
            vals[4].nul = 0;
            vals[4].val.str_val.s=dialog->call_id.s;
            vals[4].val.str_val.len=MIN(dialog->call_id.len, 256);
    
            vals[5].type = DB_STR;
            vals[5].nul = 0;
            vals[5].val.str_val.s=dialog->aor.s;
            vals[5].val.str_val.len=MIN(dialog->aor.len, 256);
    
            vals[6].type = DB_BLOB;
            vals[6].nul = 0;
            str data={x.s, x.len};
            vals[6].val.blob_val = data;
            
            if (scscf_dbf.insert(scscf_db, keys, vals, 7) < 0) {
                LOG(L_ERR, "ERR:"M_NAME":bin_cache_dump_dialogs_to_table(): Error while inserting on snapshot table\n");
                d_unlock(i);
                goto error;
            }
            bin_free(&x);
            
            dialog = dialog->next;
        }
        d_unlock(i);
    }   
    
    //delete older snapshots
    if (delete_older_snapshots("snapshot", scscf_name, S_DIALOGS, snapshot_version)!=1){
        LOG(L_ERR, "ERR:"M_NAME":bin_cache_dump_dialogs_to_table(): Error while deleting older snapshots from snapshot table\n");
        goto error;
    }
    
    //unlock
    lock_release(db_lock);
    
    return 1;

error:
    lock_release(db_lock);//unlock
    return 0;
}

int bin_cache_dump_auth_to_table(int snapshot_version, int step_version){
    bin_data x;
    auth_userdata *aud;
    int i;
    int len;
    
    //lock
    lock_get(db_lock);

    if (scscf_dbf.use_table(scscf_db, "snapshot") < 0) {
        LOG(L_ERR, "ERR:"M_NAME":bin_cache_dump_auth_to_table(): Error in use_table\n");
        goto error;
    }
    
    for(i=0;i<auth_data_hash_size;i++){
        auth_data_lock(i);
        aud = auth_data[i].head;
        while(aud){
            if (!bin_alloc(&x,128)){
                auth_data_unlock(i);
                goto error;
            }
            if (!bin_encode_auth_userdata(&x,aud)){
                auth_data_unlock(i);
                goto error;
            }
            
            db_key_t keys[7];
            db_val_t vals[7];
            
            /* id auto incremented */
            keys[0] = "node_id";
            keys[1] = "data_type";
            keys[2] = "snapshot_version";
            keys[3] = "step_version";
            keys[4] = "record_id_1"; /* private */
            keys[5] = "record_id_2"; /* public */
            /* record_id_3/4=NULL*/
            keys[6] = "data";
            
            vals[0].type = DB_STR;
            vals[0].nul = 0;
            vals[0].val.str_val.s=scscf_name;
            len = strlen(scscf_name);
            vals[0].val.str_val.len=MIN(len, 64);

            vals[1].type = DB_INT;
            vals[1].nul = 0;
            vals[1].val.int_val=S_AUTH;

            vals[2].type = DB_INT;
            vals[2].nul = 0;
            vals[2].val.int_val=snapshot_version;
    
            vals[3].type = DB_INT;
            vals[3].nul = 0;
            vals[3].val.int_val=step_version;
    
            vals[4].type = DB_STR;
            vals[4].nul = 0;
            vals[4].val.str_val.s=aud->private_identity.s;
            vals[4].val.str_val.len=MIN(aud->private_identity.len, 256);
            
            vals[5].type = DB_STR;
            vals[5].nul = 0;
            vals[5].val.str_val.s=aud->public_identity.s;
            vals[5].val.str_val.len=MIN(aud->public_identity.len, 256);
    
            vals[6].type = DB_BLOB;
            vals[6].nul = 0;
            str data={x.s, x.len};
            vals[6].val.blob_val = data;
            
            if (scscf_dbf.insert(scscf_db, keys, vals, 7) < 0) {
                LOG(L_ERR, "ERR:"M_NAME":bin_cache_dump_auth_to_table(): Error while inserting on snapshot table\n");
                auth_data_unlock(i);
                goto error;
            }
            bin_free(&x);
            
            aud = aud->next;
        }
        auth_data_unlock(i);
    }   
    
    //delete older snapshots
    if (delete_older_snapshots("snapshot", scscf_name, S_AUTH, snapshot_version)!=1){
        LOG(L_ERR, "ERR:"M_NAME":bin_cache_dump_auth_to_table(): Error while deleting older snapshots from snapshot table\n");
        goto error;
    }
    
    //unlock
    lock_release(db_lock);
    
    return 1;

error:
    lock_release(db_lock);//unlock
    return 0;
}


int bin_db_keep_count=1; 
int delete_older_snapshots(char* table, char* node_id, data_type_t dt, int current_snapshot){

    db_key_t query_cols[3];
    db_op_t  query_ops[3];
    db_val_t query_vals[3];
    int len;

    query_cols[0] = "snapshot_version";
    query_ops[0] = OP_LEQ;
    query_vals[0].type = DB_INT;
    query_vals[0].nul = 0;
    query_vals[0].val.int_val = current_snapshot - bin_db_keep_count;
    
    query_cols[1] = "node_id";
    query_ops[1] = OP_EQ;
    query_vals[1].type = DB_STR;
    query_vals[1].nul = 0;
    len = strlen(node_id);
    query_vals[1].val.str_val.s=node_id;
    query_vals[1].val.str_val.len=MIN(len, 256);
    
    query_cols[2] = "data_type";
    query_ops[2] = OP_EQ;
    query_vals[2].type = DB_INT;
    query_vals[2].nul = 0;
    query_vals[2].val.int_val = dt;

    if (scscf_dbf.use_table(scscf_db, table) < 0) {
        LOG(L_ERR, "ERR:"M_NAME":delete_older_snapshots(): Error in use_table\n");
        return 0;
    }

    if (scscf_dbf.delete(scscf_db, query_cols, query_ops, query_vals, 3) < 0) {
        LOG(L_ERR, "ERR:"M_NAME":delete_older_snapshots(): Error while deleting older snapshots\n");
        return 0;
    }
    
    return 1;
}


    /* Load related functions*/
    
int s_load(bin_data *x, char* location, char* prepend_fname, data_type_t dt){
    switch (scscf_persistency_mode){
        case NO_PERSISTENCY:
            LOG(L_ERR,"ERR:"M_NAME":s_load: Persistency support was disabled\n");
            return 0;
        case WITH_FILES:
            return bin_load_from_file(x,location,prepend_fname);        
        case WITH_DATABASE_BULK:
            return bin_load_from_db(x, dt);
        case WITH_DATABASE_CACHE:
            return bin_load_from_db(NULL, dt); //ignore x, x is empty
        default:
            LOG(L_ERR,"ERR:"M_NAME":s_load: Can't resume because no such mode %d\n",scscf_persistency_mode);
            return 0;
    }
}

int bin_load_from_db(bin_data *x, data_type_t dt){
    switch(dt){
        case S_REGISTRAR:
            return bin_load_registrar_from_table(x);
        case S_DIALOGS:
            return bin_load_dialogs_from_table(x);
        case S_AUTH:
            return bin_load_auth_from_table(x);
        default:
            LOG(L_ERR,"ERR:"M_NAME":bin_load_from_db: No such information to load %d\n", dt);
            return 0;
    }
}

int bin_load_registrar_from_table(bin_data *x){
    if(x){//whole hashtable dumped to db
        return bin_bulk_load_from_table(S_REGISTRAR, x);
    }
    else{//each hashtable element dumped to DB separately
        return bin_cache_load_registrar_from_table();
    }
}

int bin_load_dialogs_from_table(bin_data *x){
    if(x){//whole hashtable dumped to db
        return bin_bulk_load_from_table(S_DIALOGS, x);
    }
    else{//each hashtable element dumped to DB separately
        return bin_cache_load_dialogs_from_table();
    }
}

int bin_load_auth_from_table(bin_data *x){
    if(x){//whole hashtable dumped to db
        return bin_bulk_load_from_table(S_AUTH, x);
    }
    else{//each hashtable element dumped to DB separately
        return bin_cache_load_auth_from_table();
    }
}

int bin_bulk_load_from_table(data_type_t dt, bin_data* x){
    int snapshot_version;
    int r;
    int len;
    bin_alloc(x,1024);
    
    //lock
    lock_get(db_lock);
    
    if((r=db_get_last_snapshot_version("snapshot", scscf_name, dt, &snapshot_version))==0){
        LOG(L_ERR,"ERR:"M_NAME":bin_bulk_load_from_table: Error while getting snapshot version\n");
        lock_release(db_lock);//unlock
        return 0;
    }
    else if(r==-1){//empty table, nothing to load
            LOG(L_INFO,"INFO:"M_NAME":bin_bulk_load_from_table: snapshot table empty\n");
            lock_release(db_lock);//unlock
            return 1;
        }
        
    //set snapshot/step versions
    set_versions(dt, snapshot_version+1, 0);
        
    db_key_t keys[3];
    db_val_t vals[3];
    db_op_t ops[3];
    db_key_t result_cols[1];
    
    db_res_t *res = NULL;
    
    keys[0] = "node_id";
    keys[1] = "data_type";
    keys[2] = "snapshot_version";
    
    ops[0] = OP_EQ;
    ops[1] = OP_EQ;
    ops[2] = OP_EQ;

    vals[0].type = DB_STR;
    vals[0].nul = 0;
    vals[0].val.str_val.s = scscf_name;
    len = strlen(scscf_name);
    vals[0].val.str_val.len = MIN(len, 256);
    
    vals[1].type = DB_INT;
    vals[1].nul = 0;
    vals[1].val.int_val = dt;

    vals[2].type = DB_INT;
    vals[2].nul = 0;
    vals[2].val.int_val = snapshot_version;
    
    result_cols[0]="data";
    
    if (scscf_dbf.use_table(scscf_db, "snapshot") < 0) {
        LOG(L_ERR, "ERR:"M_NAME":bin_bulk_load_from_table(): Error in use_table\n");
        lock_release(db_lock);//unlock
        return 0;
    }
    
    if (scscf_dbf.query(scscf_db, keys, ops, vals, result_cols, 3, 1, 0, &res) < 0) {
        LOG(L_ERR, "ERR:"M_NAME":bin_bulk_load_from_table(): Error while querying snapshot table\n");
        lock_release(db_lock);//unlock