linux 下多线程读写，_fdb_flash_read函数出现异常问题

[maxwell-hub]

基于linux demo 例程，增加读写线程，读线程每次遍历一边数据库，写线程每秒插入1条bob记录，但是运行一段时间后，出现如下异常

---

[FlashDB][kv][env] (../../src/fdb_kvdb.c:328) Error: The KV @0x00000817 length has an error.

于是，在read_kv函数中的_fdb_flash_read函数后面增加kv_hdr.len 长度的打印，发现出现问题时候，读取的kv_hdr.len 长度是个异常值
如下这种

_fdb_flash_read  -2107555323 
_fdb_flash_read -1
_fdb_flash_read 1288869888
_fdb_flash_read 589823
_fdb_flash_read 1288869888
_fdb_flash_read -2107555323

最近排查了几天，实在没发现有什么问题，特来请教下大佬们，谢谢 @armink @Spider84 @DavidLin1577 @eggcar

/*

#include <string.h>
#include <stdio.h>
#include <pthread.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <flashdb.h>

#define FDB_LOG_TAG "[main]"

struct file_status_handler_t
{
    char filePath[256]; //文件的全路径包括文件名
    char filePathSms4[256];
    FILE *fout_aac;
    FILE *fout_aac_sms4;
    int64_t createTime; //  创建文件时间 ms单位的时间戳
    bool isFirst;
    int writeFlag; // 负数 表示文件不可写  0可写
    char fileName[64];
    int16_t recordTimeS;  // aac file time(s)
    int16_t recordTimeSC; //保存用户传入参数
    int leftSpace;        // 最小内存限制
    // int        fragmentNum;    // 分割文件数量，此为标志， 达到此数量后新建文件夹另存(用于非第一次判断)
    bool fragmentFlag;
    char fileBase[64]; //文件保存的基础路径
    char fileDir[64];  //在fileBase的基础上创建以时间为单位的文件夹，即当前文件所在的目录
};

typedef struct self_data
{
    struct fdb_kvdb     kvdb ;
    pthread_mutex_t     kv_locker;
    pthread_mutex_t     ts_locker;
   
}self_kv;



static uint32_t boot_count = 0;
static time_t boot_time[10] = {0, 1, 2, 3};
/* default KV nodes */
static struct fdb_default_kv_node default_kv_table[] = {
        {"username", "armink", 0}, /* string KV */
        {"password", "123456", 0}, /* string KV */
        // {"boot_count", &boot_count, sizeof(boot_count)}, /* int type KV */
        // {"boot_time", &boot_time, sizeof(boot_time)},    /* int array type KV */
};
/* KVDB object */

/* TSDB object */
struct fdb_tsdb tsdb = { 0 };
/* counts for simulated timestamp */
static int counts = 0;

extern void kvdb_basic_sample(fdb_kvdb_t kvdb);
extern void kvdb_type_string_sample(fdb_kvdb_t kvdb);
extern void kvdb_type_blob_sample(fdb_kvdb_t kvdb);
extern void tsdb_sample(fdb_tsdb_t tsdb);

static void lock(fdb_db_t db)
{
    pthread_mutex_lock((pthread_mutex_t *)db->user_data);
}

static void unlock(fdb_db_t db)
{
    pthread_mutex_unlock((pthread_mutex_t *)db->user_data);
}

static fdb_time_t get_time(void)
{
    return time(NULL);
}


void *write_thread(void * arg)
{
    self_kv *data = (self_kv *) arg;
    struct fdb_blob blob;
    char file_name[64]; 
    char file_path[128];
    unsigned int time_cnt;
     
    while(1)    
    {   
        sprintf(file_name, "cnt_%d", time_cnt);
        sprintf(file_path,"path_%d",time_cnt);
        time_cnt ++ ;
        fdb_kv_set_blob(&data->kvdb, file_name, fdb_blob_make(&blob, file_path, strlen(file_path)));
        // printf("set 'file_name':%s value to %s\n", file_name, file_path);
        usleep(1000*1000);
    }
    printf("exit write_thread\n ");
}

int read_test(self_kv *kv)
{
    self_kv *data = (self_kv *) kv;
    struct fdb_kv_iterator iterator;
    fdb_kv_t cur_kv;
    struct fdb_blob blob;
    size_t data_size;
    uint8_t data_buf[128];
    int file_size = 0;
    fdb_kv_iterator_init(&iterator);
    
    while (fdb_kv_iterate(&data->kvdb, &iterator))
    {
        // LOG_DEBUG("enter record_from_database ...。\n");
        cur_kv = &(iterator.curr_kv);
        data_size = (size_t)cur_kv->value_len;
        memset(data_buf, 0, 128);
        fdb_blob_read((fdb_db_t)&data->kvdb, fdb_kv_to_blob(cur_kv, fdb_blob_make(&blob, data_buf, data_size)));
/*
        * balabala do what ever you like with blob...
        */
        // FDB_INFO("get kv  key--%s value---%s   value_size is %d  \n", cur_kv->name, data_buf, (int)data_size);
        printf("get kv  key--%s value---%s   value_size is %d \n", cur_kv->name, data_buf, (int)data_size);
    }
}

void *read_thread(void * arg)
{
    
    while (1)
    {
        read_test(arg);
        sleep(1);
    }
        
}

int main(void)
{
    fdb_err_t result;
    bool file_mode = true;
    uint32_t sec_size = 4096, db_size = sec_size * 40;   
    self_kv  data ;
   
    pthread_t read_tid ,write_tid;
    { /* KVDB Sample */
        struct fdb_default_kv default_kv;

        default_kv.kvs = default_kv_table;
        default_kv.num = sizeof(default_kv_table) / sizeof(default_kv_table[0]);
        /* set the lock and unlock function if you want */
        pthread_mutex_init(&data.kv_locker, NULL);
        fdb_kvdb_control(&data.kvdb, FDB_KVDB_CTRL_SET_LOCK, (void *)lock);
        fdb_kvdb_control(&data.kvdb, FDB_KVDB_CTRL_SET_UNLOCK, (void *)unlock);
        /* set the sector and database max size */
        fdb_kvdb_control(&data.kvdb, FDB_KVDB_CTRL_SET_SEC_SIZE, &sec_size);
        fdb_kvdb_control(&data.kvdb, FDB_KVDB_CTRL_SET_MAX_SIZE, &db_size);
        /* enable file mode */
        fdb_kvdb_control(&data.kvdb, FDB_KVDB_CTRL_SET_FILE_MODE, &file_mode);
        /* create database directory */
        mkdir("fdb_kvdb1", 0777);
        /* Key-Value database initialization
         *
         *       &kvdb: database object
         *       "env": database name
         * "fdb_kvdb1": The flash partition name base on FAL. Please make sure it's in FAL partition table.
         *              Please change to YOUR partition name.
         * &default_kv: The default KV nodes. It will auto add to KVDB when first initialize successfully.
         *  &kv_locker: The locker object.
         */
        result = fdb_kvdb_init(&data.kvdb, "env", "fdb_kvdb1", &default_kv, &data.kv_locker);

        if (result != FDB_NO_ERR) {
            return -1;
        }
       
        pthread_create(&write_tid, NULL , write_thread , &data);
        // sleep(5);
        pthread_create(&read_tid, NULL , read_thread , &data);

        // /* run basic KV samples */
        // kvdb_basic_sample(&data.kvdb);
        // /* run string KV samples */
        // kvdb_type_string_sample(&data.kvdb);
        // /* run blob KV samples */
        // kvdb_type_blob_sample(&data.kvdb);
    }
    while(1)
    {
        sleep(1);
    }



    return 0;
}

[armink]

fdb_kv_iterate 迭代器内部是没有加锁的，这里是个坑，你使用迭代器前后加一下锁，再试试

[armink]

1、lock/unlock 就是你前面定义的给到 FlashDB 的两个锁函数

static void unlock(fdb_db_t db)
{
    pthread_mutex_unlock((pthread_mutex_t *)db->user_data);
}

static fdb_time_t get_time(void)
{
    return time(NULL);
}

2、你的业务是每次在读取线程中需要遍历整个数据库的 KV 吗？可否说一下业务场景？

3、read_kv() 是不给外部提供调用的，内部使用时都会加锁进行使用，目前只发现 fdb_kv_iterate() 里使用的 read_kv() 没有加锁，所以我说这里有点坑

[maxwell-hub]

感谢朱工细心指导🙏
2. 我们的业务场景也很简单，就是未上传的文件存储记录在falshdb数据库中，间歇性遍历数据库上传文件到服务器；具体思路是这样：一个线程写数据到磁盘，写完之后在flashdb插入一条记录；另一个线程去间歇性遍历数据库，找到一条记录后开启文件上传线程将文件上传到服务器后删除改条flashdb记录；这块遍历整个数据库的操作等下优化下，每次遍历一个即退出开始上传，这样阻塞时间会短点
3. 我之前尝试在read_kv()前加锁，也会导致_fdb_flash_read函数异常，但是没尝试fdb_kv_iterate前加锁的方法，我马上测试下，有结果反馈给朱工，感谢朱工

[armink]

大概理解你的需求了，我感觉 FlashDB 里的时序数据库功能可能更适合你，速度还快

时序数据库可以有不同的状态，可以按照不同状态（如：未同步、已删除）对各个记录进行标识，同时还支持按照时间检索

[maxwell-hub]

谢谢朱工，tsdb我来熟悉下

[eggcar]

感谢朱工回复🙏，我还有几个疑问？

1. lock(); 和unlock();是另外的锁还是说是申请给数据库的线程锁？
2. 在我们自己的业务代码里面，因为数据库记录数有点多，如果在迭代器外面加锁的话，会导致其他线程很长时间拿不到资源的呀
3. flashdb数据库实现复杂，没梳理清楚内部思路，但是看到写数据库会调用dblock()和dbunlock(),但是read_kv()中没有加锁，意思是写加锁，读取不加锁嘛？如果一个线程正在写(但没写完)，这时候读取到最后一条不完整记录会是什么样的处理呢？

不好意思我的邮箱把github的邮件都屏蔽掉了
不整体加锁的话，flashdb可能不太好解决这个问题，主要是在遍历过程中如果在当前遍历的sector上触发了gc会彻底破坏遍历进度，或许可以在iterator里增加一个是否被gc打断的标志，如果当前sector被gc 跳去下一个sector继续遍历？但如果连续两个sector触发gc问题也没有得到本质解决...个人感觉不加整体锁很难实现
受限于嵌入式的资源，flashdb也不能用快照机制解决，使用kvdb做这个功能可能不太合适