PX4模块设计之四十六：dataman模块

1. dataman模块简介

### Description
Module to provide persistent storage for the rest of the system in form of a simple database through a C API.
Multiple backends are supported:
- a file (eg. on the SD card)
- RAM (this is obviously not persistent)

It is used to store structured data of different types: mission waypoints, mission state and geofence polygons.
Each type has a specific type and a fixed maximum amount of storage items, so that fast random access is possible.

### Implementation
Reading and writing a single item is always atomic. If multiple items need to be read/modified atomically, there is
an additional lock per item type via `dm_lock`.

**DM_KEY_FENCE_POINTS** and **DM_KEY_SAFE_POINTS** items: the first data element is a `mission_stats_entry_s` struct,
which stores the number of items for these types. These items are always updated atomically in one transaction (from
the mavlink mission manager). During that time, navigator will try to acquire the geofence item lock, fail, and will not
check for geofence violations.

dataman <command> [arguments...]
 Commands:
   start
     [-f <val>]  Storage file
                 values: <file>
     [-r]        Use RAM backend (NOT persistent)

 The options -f and -r are mutually exclusive. If nothing is specified, a file
 'dataman' is used

   stop

   status        print status info

注：print_usage函数是具体对应实现。

2. 模块入口函数dataman_main

该模块纯C语言风格，提供类似数据库的API接口，后端可以采用固化文件存储或者RAM存储(类似Memcache)方式。

模块仅支持start/stop/status子命令，start子命令含f/r选项，其中f是文件后端保存数据，r是ram后端保存数据。

dataman_main
 ├──> <argc < 2>
 │   ├──> usage()
 │   └──> return -1
 ├──> <!strcmp(argv[1], "start")>
 │   ├──> <is_running()>
 │   │   ├──> PX4_WARN("dataman already running")
 │   │   └──> return -1
 │   ├──> [jump over start and look at options first]
 │   │   ├──> <case 'f'> <backend_check()>
 │   │   │   ├──> backend = BACKEND_FILE
 │   │   │   ├──> k_data_manager_device_path = strdup(dmoptarg)
 │   │   │   └──> PX4_INFO("dataman file set to: %s", k_data_manager_device_path)
 │   │   ├──> <case 'r'> <backend_check()>
 │   │   │   └──> backend = BACKEND_RAM
 │   │   └──> <default>
 │   │       └──> usage() // return
 │   ├──> <backend == BACKEND_NONE)>
 │   │   └──> backend = BACKEND_FILE
 │   │       └──> k_data_manager_device_path = strdup(default_device_path)
 │   ├──> start()
 │   ├──> <!is_running()>
 │   │   ├──> PX4_ERR("dataman start failed")
 │   │   ├──> free(k_data_manager_device_path)
 │   │   ├──> k_data_manager_device_path = nullptr
 │   │   └──> return -1
 │   └──> return 0
 ├──> <!is_running()>   // Worker thread should be running for all other commands
 │   ├──> PX4_WARN("dataman worker thread not running")
 │   ├──> usage()
 │   └──> return -1
 ├──> <!strcmp(argv[1], "stop")>
 │   ├──> stop()
 │   ├──> free(k_data_manager_device_path)
 │   └──> k_data_manager_device_path = nullptr
 ├──> <!strcmp(argv[1], "status")>
 │   └──> status()
 └──> <else>
     └──> usage()

3. dataman模块重要函数

3.1 start

启动task_main任务来完成dataman的业务逻辑。

start
 ├──> [g_init_sema use case is a signal]
 │   ├──> px4_sem_init(&g_init_sema, 1, 0)
 │   └──> px4_sem_setprotocol(&g_init_sema, SEM_PRIO_NONE)
 ├──> <task = px4_task_spawn_cmd("dataman", SCHED_DEFAULT, SCHED_PRIORITY_DEFAULT - 10,PX4_STACK_ADJUSTED(TASK_STACK_SIZE), task_main,nullptr)) < 0>  //start the worker thread with low priority for disk IO
 │   ├──> px4_sem_destroy(&g_init_sema)
 │   ├──> PX4_ERR("task start failed")
 │   └──> return -1
 ├──> [wait for the thread to actually initialize]
 │   ├──> px4_sem_wait(&g_init_sema)
 │   └──> px4_sem_destroy(&g_init_sema)
 └──> return 0

3.2 stop

这里采用的是优雅退出的方式，通过全局变量以flag的方式通知运行的任务。

stop
 ├──> g_task_should_exit = true;  //Tell the worker task to shut down
 └──> px4_sem_post(&g_work_queued_sema);

3.3 status

将dataman模块性能和统计数据打印输出。

status  // display usage statistics
 ├──> PX4_INFO("Writes   %u", g_func_counts[dm_write_func])
 ├──> PX4_INFO("Reads    %u", g_func_counts[dm_read_func])
 ├──> PX4_INFO("Clears   %u", g_func_counts[dm_clear_func])
 ├──> PX4_INFO("Max Q lengths work %u, free %u", g_work_q.max_size, g_free_q.max_size)
 ├──> perf_print_counter(_dm_read_perf)
 └──> perf_print_counter(_dm_write_perf)

3.4 task_main

主要业务流程是三个方法：

1. dm_write_func
2. dm_read_func
3. dm_clear_func

task_main
 ├──> [dataman初始化逻辑：操作方法，工作队列等]
 │   ├──> [Dataman can use disk or RAM 操作方法初始化: g_dm_ops]
 │   ├──> [Initialize global variables: g_key_offsets/g_func_counts/_dm_read_perf/_dm_write_perf/g_work_q/g_free_q/g_work_queued_sema/g_sys_state_mutex_fence/g_sys_state_mutex_mission/g_item_locks]
 │   ├──> int ret = g_dm_ops->initialize(max_offset); //操作方法初始化
 │   ├──> px4_sem_post(&g_init_sema);  //Tell startup that the worker thread has completed its initialization
 │   ├──> while (true)  // Start the endless loop, waiting for then processing work requests
 │   │   ├──> <!g_task_should_exit> // do we need to exit ??? or wait for work
 │   │   │   └──> g_dm_ops->wait(&g_work_queued_sema);
 │   │   ├──> while ((work = dequeue_work_item())) // Empty the work queue, handle each work item with the appropriate handler
 │   │   │   ├──> <case dm_write_func>
 │   │   │   │   ├──> g_func_counts[dm_write_func]++;
 │   │   │   │   └──> work->result = g_dm_ops->write(work->write_params.item, work->write_params.index, work->write_params.buf, work->write_params.count);
 │   │   │   ├──> <case dm_read_func>
 │   │   │   │   ├──> g_func_counts[dm_read_func]++;
 │   │   │   │   └──> work->result = g_dm_ops->read(work->read_params.item, work->read_params.index, work->read_params.buf, work->read_params.count);
 │   │   │   ├──> <case dm_clear_func>
 │   │   │   │   ├──> g_func_counts[dm_clear_func]++;
 │   │   │   │   └──> work->result = g_dm_ops->clear(work->clear_params.item);
 │   │   │   └──> <defaul>    // should never happe
 │   │   │       └──> work->result = -1;
 │   │   ├──> px4_sem_post(&work->wait_sem);  // Inform the caller that work is done
 │   │   └──> <g_task_should_exit> // time to go???? 
 │   │       └──> break;
 │   ├──> g_dm_ops->shutdown();  // 操作方法关闭
 ├──> [清空工作队列]
 └──> [优雅退出处理]  // 函数内部出现业务流程失败，则goto到该位置，典型的C语言优化处理逻辑流程
     ├──> backend = BACKEND_NONE;
     ├──> destroy_q(&g_work_q);
     ├──> destroy_q(&g_free_q);
     ├──> px4_sem_destroy(&g_work_queued_sema);
     ├──> px4_sem_destroy(&g_sys_state_mutex_mission);
     ├──> px4_sem_destroy(&g_sys_state_mutex_fence);
     ├──> perf_free(_dm_read_perf);
     ├──> _dm_read_perf = nullptr;
     ├──> perf_free(_dm_write_perf);
     ├──> _dm_write_perf = nullptr;
     └──> return 0;

4. API接口

数据一致性和有效性的角度，模块提供了lock/trylock/unlock方法来保证read/write/clear。

1. dm_read
2. dm_write
3. dm_lock
4. dm_trylock
5. dm_unlock
6. dm_clear

/** Retrieve from the data manager store */
__EXPORT ssize_t
dm_read(
	dm_item_t item,			/* The item type to retrieve */
	unsigned index,			/* The index of the item */
	void *buffer,			/* Pointer to caller data buffer */
	size_t buflen			/* Length in bytes of data to retrieve */
);

/** write to the data manager store */
__EXPORT ssize_t
dm_write(
	dm_item_t  item,		/* The item type to store */
	unsigned index,			/* The index of the item */
	const void *buffer,		/* Pointer to caller data buffer */
	size_t buflen			/* Length in bytes of data to retrieve */
);

/**
 * Lock all items of a type. Can be used for atomic updates of multiple items (single items are always updated
 * atomically).
 * Note that this lock is independent from dm_read & dm_write calls.
 * @return 0 on success and lock taken, -1 on error (lock not taken, errno set)
 */
__EXPORT int
dm_lock(
	dm_item_t item			/* The item type to lock */
);

/**
 * Try to lock all items of a type (@see sem_trywait()).
 * @return 0 if lock is taken, -1 otherwise (on error or if already locked. errno is set accordingly)
 */
__EXPORT int
dm_trylock(
	dm_item_t item			/* The item type to lock */
);

/** Unlock all items of a type */
__EXPORT void
dm_unlock(
	dm_item_t item			/* The item type to unlock */
);

/** Erase all items of this type */
__EXPORT int
dm_clear(
	dm_item_t item			/* The item type to clear */
);

5. 存储内容

dataman模块可存储的数据类型。

/** Types of items that the data manager can store */
typedef enum {
	DM_KEY_SAFE_POINTS = 0,		/* Safe points coordinates, safe point 0 is home point */
	DM_KEY_FENCE_POINTS,		/* Fence vertex coordinates */
	DM_KEY_WAYPOINTS_OFFBOARD_0,	/* Mission way point coordinates sent over mavlink */
	DM_KEY_WAYPOINTS_OFFBOARD_1,	/* (alternate between 0 and 1) */
	DM_KEY_MISSION_STATE,		/* Persistent mission state */
	DM_KEY_COMPAT,
	DM_KEY_NUM_KEYS			/* Total number of item types defined */
} dm_item_t;

#if defined(MEMORY_CONSTRAINED_SYSTEM)
enum {
	DM_KEY_SAFE_POINTS_MAX = 8,
	DM_KEY_FENCE_POINTS_MAX = 16,
	DM_KEY_WAYPOINTS_OFFBOARD_0_MAX = NUM_MISSIONS_SUPPORTED,
	DM_KEY_WAYPOINTS_OFFBOARD_1_MAX = NUM_MISSIONS_SUPPORTED,
	DM_KEY_MISSION_STATE_MAX = 1,
	DM_KEY_COMPAT_MAX = 1
};
#else
/** The maximum number of instances for each item type */
enum {
	DM_KEY_SAFE_POINTS_MAX = 8,
	DM_KEY_FENCE_POINTS_MAX = 64,
	DM_KEY_WAYPOINTS_OFFBOARD_0_MAX = NUM_MISSIONS_SUPPORTED,
	DM_KEY_WAYPOINTS_OFFBOARD_1_MAX = NUM_MISSIONS_SUPPORTED,
	DM_KEY_MISSION_STATE_MAX = 1,
	DM_KEY_COMPAT_MAX = 1
};
#endif

6. 参考资料

【1】PX4开源软件框架简明简介
【2】PX4模块设计之十一：Built-In框架
【3】PX4模块设计之十二：High Resolution Timer设计
【4】PX4模块设计之十三：WorkQueue设计
【5】PX4模块设计之十七：ModuleBase模块
【6】PX4模块设计之三十：Hysteresis类
【7】PX4 modules_main
【8】PX4模块设计之四十一：I2C/SPI Bus Instance基础知识