pager.c

基于btree索引算法的数据库代码

 *  < 0  表示 key1 小于 key2
 *
 * @param context:用户自定义的上下文数据
 */
static int pager_sort_dirty_compare(const void * data1, const void * data2, const void * context)
{
	pghd_t * pg1 = *((pghd_t **)data1);
	pghd_t * pg2 = *((pghd_t **)data2);
	assert(pg1 && pg2);
	assert(pg2->page_num && pg1->page_num);

	/*return pg1->page_num - pg2->page_num */
	if(pg1->page_num >= pg2->page_num)
		return pg1->page_num - pg2->page_num;
	else
		return -1;
}

/**
 * @brief 同步所有脏缓存页
 */
static __INLINE__ int pager_sort_dirty_list(pager_t * pager)
{
	/*
	* sqlite有用的是归并排序
	* 对脏页进行排序,保证从最小的页开始写回外存
	* 在将脏页写入外存页文件之前,按页号进行排序是很重要的
	*
	* 分配一块内存数组
	* 将所有的页依次填写这个内存数组
	* 以页号为比较关键字对这个数组进行快速排序
	* 将排序好的数组按次序添加至dirty list
	*/

	pghd_t * pg = NULL;
	pghd_t ** pg_array = NULL;
	unsigned int dirty_pages_count = 0;
	unsigned int i = 0;
	assert(pager && pager->pg_dirty);

	for(pg = pager->pg_dirty; pg; pg = pg->pg_dirty_next)
		dirty_pages_count ++;

	if(dirty_pages_count <= 1)
		return 0;

	pg_array = MyMemPoolMalloc(pager->hm, sizeof(pg_array[0]) * dirty_pages_count);
	if(NULL == pg_array)
		return -1;

	for(i = 0, pg = pager->pg_dirty; pg; pg = pg->pg_dirty_next, i ++)
	{
		assert(pager_page_is_dirty(pg));
		assert(pg->page_num);

		assert(i < dirty_pages_count);
		pg_array[i] = pg;
	}

	pg = NULL;
	MyAlgQuickSort(pg_array, dirty_pages_count, sizeof(pg_array[0]), pager_sort_dirty_compare, NULL, NULL, NULL, NULL, &pg, sizeof(pg));
	pager->pg_dirty = NULL;

	for(i = 0; i < dirty_pages_count; i ++)
	{
		pg_array[i]->pg_dirty_next = NULL;
		pg_array[i]->pg_dirty_prev = NULL;

		pager_add_page_to_dirty_list(pager, pg_array[i]);
	}

	assert(pg_array);
	MyMemPoolFree(pager->hm, pg_array);

	return 0;
}

/**
 * @brief 同步所有脏缓存页
 */
static __INLINE__ int pager_syn(pager_t * pager)
{
	pghd_t * pg = NULL;

	/*
	* 首先将jfd文件同步至外存,将所有页缓存置成同步过jfd
	* 将同步过的空闲页缓存加入指定的链表(用于页缓存满时,换页优先从这个链表里取空闲页缓存)
	* 将所有的dirty页写入fd中,将dirty链表置空.每页的dirty标识也置空
	* 删除份的jfd文件.
	* jfd着色表以及页的injournal标识均置成0,因为一个完整提交已经完成了.
	*
	* 成功之后,将文件的完整性置成1
	*
	*/

	assert((pager->pg_dirty && pager->hjf) || 
		(!pager->page_file_is_integrity && pager->hjf) ||
		(!pager->hjf && NULL == pager->pg_dirty && pager->page_file_is_integrity));

	/* 如果不存在脏页,并且外页的数据是完整的,同步就不需要进行 */
	if(NULL == pager->pg_dirty && pager->page_file_is_integrity)
		return 0;

	/* 将jf同步至外存 */
	if(0 != pager_sync_journal(pager))
		return -1;

	/* 将脏页写回页文件 */
	if(pager->pg_dirty)
		pager_sort_dirty_list(pager);
	for(pg = pager->pg_dirty; pg; pg = pager->pg_dirty)
	{
		assert(pager_page_is_dirty(pg));

		/* 写页缓存写回页文件 */
		if(0 != pager_write_page(pager, pg))
			return -1;

		/* 从dirty list里脱链 */
		pager_page_out_of_dirty_list(pager, pg);

		/* 改变页状态 */
		pager_change_page_state(pg, PAGE_EVT_WRITE_BACK_SYNCJF);
	}

	assert(NULL == pager->pg_dirty);

	/* 同步页文件至外存 */
	if(0 != OsFileSyn(pager->hf))
		return -1;

	/*
	* 清除所有页的clean_sync标记
	* todo 优先此段代码
	*/
	for(pg = pager->pg_all; pg; pg = pg->pg_all_next)
	{
		assert(!pager_page_is_dirty(pg));

		if(0 == pg->ref_count)
		{
			assert(pager_page_is_in_free_list(pg));

			if(pager_page_is_sync_journal(pg))
			{
				assert(pager_page_is_in_syncjf_list(pg));
				pager_out_syncjf_list(pager, pg);
			}
			else
				assert(!pager_page_is_in_syncjf_list(pg));
		}
		else
		{
			assert(!pager_page_is_in_free_list(pg));
			assert(!pager_page_is_in_syncjf_list(pg));
		}

		assert(!pager_page_is_in_syncjf_list(pg));

		if(!pager_page_is_clean_not_in_jf(pg))
			pager_change_page_state(pg, PAGE_EVT_SYNC);

		assert(pager_page_is_clean_not_in_jf(pg));
	}

	assert(NULL == pager->pg_syncjf_first && NULL == pager->pg_syncjf_last);

	pager->page_file_is_integrity = 1;

	/* 关闭jouranl */
	if(0 != pager_close_and_del_journal(pager))
		return -1;

	return 0;
}

/**
 * @brief 销毁缓存管理
 */
static __INLINE__ int pager_destroy(pager_t * pager)
{
	assert(pager);

	/* 做回滚操作 */
	if(pager->hf && !pager->page_file_is_integrity)
		pager_rollback(pager);

	if(pager->hjf)
		pager_close_and_del_journal(pager);

	if(pager->hf)
		OsFileClose(pager->hf);

	if(pager->hash_page)
		MyHashTableDestruct(pager->hash_page);

	if(pager->hash_pgno_journal)
		MyHashTableDestruct(pager->hash_pgno_journal);

	if(pager->hrand)
		myrandDestruct(pager->hrand);

	if(pager->hb_file_name)
		MyBufferDestruct(pager->hb_file_name);

	if(pager->hb_jf_name)
		MyBufferDestruct(pager->hb_jf_name);

	{
		/* 依次释放页缓存 */
		pghd_t * pg = pager->pg_all;
		while(pg)
		{
			pghd_t * pg_temp = pg->pg_all_next;
			MyMemPoolFree(pager->hm, pg);
			pg = pg_temp;
		}
	}

	MyMemPoolFree(pager->hm, pager);

	return 0;
}

/**
 * @brief 页缓存哈希表哈希函数
 */
static size_t pg_hash_fun(const void * key)
{
	return (size_t)key;
}

/**
 * @brief 页缓存哈希表,比较键值是否相等 1:相等 0:不相等
 */
static int pg_hash_keyequl_fun(const void * key1, const void * key2)
{
	return key1 == key2;
}

/**
 * @brief 根据页号,获取一页缓存
 */
static __INLINE__ pghd_t * pager_get_page(pager_t * pager, unsigned int pgno)
{
	/*
	* 如果是第一次获取页,则应判断journal日志文是否存,如果存在则需要执行回滚操作
	* 将journal日志文件中的备份,依次读入,检查校验和,如果正确,写入页文件fd
	* 回滚结束后,应删除jfd文件
	* 首先从页缓存的哈希表里查找,如果找到,页引用计数加1,函数返回
	* 如果在缓存中找不到相应的页,则需要分配页缓存,从外存读取,初始化是否injournal的标识
	* 此时应注意页缓存满的情况:
	* 如果不存在引用计数为零的页缓存,则仍然分配页缓存.
	* 如果存在引用计数为零的页缓存,则淘汰该页缓存,用于装载新的页,此时需要将此页写回fd,
	*　优先取那些同步过jfd的空闲页缓存,如果没有,则同步jfd,同时从哈希表里删除相应的记录
	*  将页号与页缓存指针添加进哈希表
	*  如果页处在空闲链表中,脱链
	*  如果页处在同步过jfd的空闲链表中,脱链
	*  页引用计数加1
	*/

	pghd_t * pg = NULL;

	assert(pager && pager->hash_page);

	if(0 == pager->ref_count && NULL == pager->hjf)
	{
		/*
		* 如果是第一次获取页,则应判断journal日志文是否存,如果存在则需要执行回滚操作
		* 将journal日志文件中的备份,依次读入,检查校验和,如果正确,写入页文件fd
		* 回滚结束后,应删除jfd文件
		*/
		if(0 != pager_check_and_roll_back(pager))
			return NULL;
	}

	pg = pager_hash_lookup(pager,  pgno);

	if(NULL == pg)
	{
		if((pager->pg_free_first || pager->pg_syncjf_first) && pager_get_cached_pages_count(pager) >= pager->max_cache_pages)
			pg = pager_recycle_and_init_page(pager, pgno);/* 回收一页缓存 */
		else
			pg = pager_allocate_page(pager, pgno);/* 分配一页缓存 */

		if(NULL == pg)
			goto pager_get_page_err_;

		assert(pg->page_num == pgno);

		/* 读取页 */
		if(0 != pager_read_page(pager, pgno, pg))
			goto pager_get_page_err_;

		/* 进入哈希表,以及总的页缓存链表 */
		if(0 != pager_add_page_to_hash_and_list(pager, pg))
			goto pager_get_page_err_;
	}

	pager_ref_page(pager, pg);
	return pg;

pager_get_page_err_:

	/* 由于没有记录在案,此页缓存必需释放 */
	if(pg)
		pager_free_page(pager, pg);

	return NULL;
}

/**
 * @brief 取消对某个页缓存的引用
 */
static __INLINE__ int pager_release_page(pghd_t * pg)
{
	pager_t * pager = NULL; 

	assert(pg && pg->pager);
	assert(pg->ref_count > 0);

	pager = pg->pager;

	/*
	* 页引用计数减1,
	* 如果页的引用计数为零,则可以将它加入空闲页缓存链表,如果该页并且已经同步过jfd,则可以加入空闲的jfd缓存链表,
	*   并且需要调用用户的release回调函数通知上层
	*/

	/* 引用计数减1 */
	assert(pg->ref_count);
	pg->ref_count --;

	assert(pager->ref_count);
	pager->ref_count --;

	if(pg->ref_count > 0)
		return 0;

	assert(!pager_page_is_in_free_list(pg));
	assert(!pager_page_is_in_syncjf_list(pg));

	/* 如果引用计数为零 */
	/* 如果需要同步至jf,加入的free list */
	if(pager_page_is_sync_journal(pg))
		pager_add_to_syncjf_list(pager, pg);/* 如果不需要同步至jf,加入到syncjf_list,以便优先回收 */
	else
		pager_add_to_free_list(pager, pg);

	/* 调用用户的回调函数 */
	if(pager->page_release_cb)
		pager->page_release_cb(pg);

	return 0;
}

/**
 * @brief 写入页头信息
 */
static __INLINE__ int pager_write_file_page_info(pager_t * pager)
{
	/*
	* 页的第一页用于存放页文件本身的信息
	* 第一页
	*  offset bytes        des
	*  0      16           格式版本
	*  16     4            页大小
	*  20     4            第一个空闲页
	*  24     40           保留
	*  64   pg_sz - 64     用户数据     
	*/

	pghd_t * first_pg = NULL;

	assert(pager && pager->hf);
	assert(0 == pager->ref_count);

	first_pg = pager_get_page(pager, PAGER_FIRST_PAGE_NO);
	if(NULL == first_pg)
		return -1;

	if(NULL == pager_make_page_writable(first_pg))
		return -1;

	memcpy(first_pg->pg_buf_start + PAGER_HDR_VER_OFFSET, VERSION, PAGER_HDR_VER_LEN);

	uint_to_big_endian(pager->page_size, first_pg->pg_buf_start + PAGER_HDR_PAGE_SIZE_OFFSET, sizeof(unsigned int));
	uint_to_big_endian(0, first_pg->pg_buf_start + PAGER_HDR_FIRST_FREE_OFFSET, sizeof(unsigned int));

	pager_release_page(first_pg);

	assert(0 == pager->ref_count);
	pager_syn(pager);

	return 0;
}

/**
 * @brief 读取页文件信息,读取页大小,以及包含多少页,并判断文件版本
 */
static __INLINE__ int pager_read_file_page_info(pager_t * pager)
{
	/*
	* 页的第一页用于存放页文件本身的信息
	* 第一页
	*  offset bytes        des
	*  0      16           格式版本
	*  16     4            页大小
	*  20     4            第一个空闲页
	*  24     40           保留
	*  64   pg_sz - 64     用户数据     
	*/
	char ver[sizeof(VERSION)];
	unsigned int pg_sz = 0;
	int64 pg_file_sz = 0;

	assert(pager && pager->hf);

	/* 判断页文件大小是否满足一页,否则需要写入页文件头信息 */
	if(0 != OsFileSize(pager->hf, &pg_file_sz))
		return -1;

	if(pg_file_sz < sizeof(ver))
	{
		/* 无法读到足够的信息,则应认为此页文件是无效的,需要写入头信息 */
		if(0 != pager_write_file_page_info(pager))
			return -1;

		return 0;
	}

	if(0 != OsFileSeek(pager->hf, 0))
		return -1;

	if(0 != OsFileRead(pager->hf, ver, sizeof(ver), NULL))
		return -1;

	if(strncmp(ver, VERSION, strlen(VERSION)) != 0)
	{
		/* 页文件格式可能是当前程序不支持的 */
		LOG_WARN(("page file version err,this page maybe create by the later pager verion,[file version:%s], [program ver:%s]",
			ver, VERSION));

		return -1;
	}

	if(0 != OsFileSeek(pager->hf, PAGER_HDR_VER_LEN))
		return -1;

	/* 读出页的大小 */
	if(0 != pager_read