pager.c

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

	assert(pager && pgno);

	pg = pager_recycle_page(pager);
	if(NULL == pg)
		return NULL;

	assert(NULL == pg->pg_all_next && NULL == pg->pg_all_prev &&
		!pager_page_is_in_free_list(pg) &&
		!pager_page_is_in_syncjf_list(pg) &&
		!pager_page_is_in_dirty_list(pg) &&
		!pager_page_is_dirty(pg));

	pager_init_page(pager, pg, pgno);

	return pg;
}

/**
 * @brief 将某一个页缓存置成脏,并写入jf
 */
static __INLINE__ void * pager_make_page_writable(pghd_t * pg)
{
	pager_t * pager = NULL;
	void * buf_write = NULL;

	/*
	* 判断jfd机制是否已启用,如果未启用则应该打开jfd文件,并分配着色表
	* 判断此页是否已经处在jfd中了(从着色表里查),
	*    如果没有则需要写入jfd,先页号,再写页缓存内容,最后写入校验和,并且给该页缓存做上injournal标识,并在着色表相应的位置置标识
	*    将页缓存置成dirty,加入dirty链表.
	* 如果页号比当前总页数大,
	* 此时应将页文件的总页数增1
	* 返回页缓存的首地址供上层写入(注意越界检测代码加入)
	*/

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

	buf_write = pager_get_page_bufdata(pg);

	pager = pg->pager;

	/* 如果未启用jf机制,则应启用 */
	if(NULL == pager->hjf)
	{
		if(0 != pager_open_journal(pager))
			return NULL;
	}

	/* 如果此页已在jf里做备份(pg->page_state != PAGE_CLEAN) */
	if(pager_page_is_in_journal(pg))
	{
		assert(pg->page_num <= pager_get_total_pages_count(pager));

		/* 将页置成dirty,并加入dirty list */
		if(!pager_page_is_dirty(pg))
			pager_make_page_dirty(pager, pg, PAGE_EVT_WRITE_BUF_BK);

		return buf_write;
	}

	/* 如果(pg->page_state == PAGE_CLEAN)将调用以下代码 */
	assert(!pager_page_is_dirty(pg));

	assert(pager->orig_pages_count != (unsigned int)-1);

	if(pg->page_num > pager->orig_pages_count)
	{
		/*
		* 页在修改之前本身是不存在的,所以也不用真正地写入,做个标记即可
		* 这里应处理当有多个页的超出页文件大小的情况,因为超出文件大小情况下,只需要同步
		* jf头就可以了,这种情况只需要同步一次,即之后再发生这种情况,因为不存在写入的问题
		* jf是不用步的,因为已经同步过了
		*/
		pager_set_total_pages_count(pager, ((unsigned int)pg->page_num));

		pager_make_page_dirty(pager, pg, PAGE_EVT_WRITE_BUF_BK);

		/* 如果已经做过同步,则需要修改状态 */
		if(pager->bsync_journal_has_done)
			pager_change_page_state(pg, PAGE_EVT_SYNC_JOURNAL);
	}
	else
	{
		unsigned int cksum = pager_cal_page_cksum(pg->pg_buf_start, pager->page_size, pager->cur_cksum_init);
		/*
		* 如果没有则需要写入jfd, <页缓存内容,先页号,最后写入校验和>,并且给该页缓存做上injournal标识,并在着色表相应的位置置标识
		*/
		uint_to_big_endian((unsigned int)pg->page_num, (pg->pg_buf_start + pager->page_size), sizeof(unsigned int));
		uint_to_big_endian(cksum,
			pg->pg_buf_start + pager->page_size + sizeof(unsigned int),
			sizeof(unsigned int));

		if(0 != OsFileWrite(pager->hjf, pg->pg_buf_start, pager_get_journal_rec_len(pager), NULL))
			return NULL;

		pager->in_journal_pages_count ++;
		pager->journal_end += pager_get_journal_rec_len(pager);

		/* 将页置成dirty,并加入dirty list */
		pager_make_page_dirty(pager, pg, PAGE_EVT_WRITE_BUF_BK);

		assert(!pager_page_is_sync_journal(pg));
	}

	pager_mark_page_in_journal(pager, pg);

	if(!pager_page_is_sync_journal(pg))
		pager->need_sync_journal = 1;

	return buf_write;
}

/**
 * @brief 从jf里回滚
 */
static __INLINE__ int pager_file_truncate(pager_t * pager, unsigned int page_count)
{
	assert(pager && pager->hf);

	if(0 != OsFileTruncate(pager->hf, page_count * pager->page_size))
		return -1;

	return 0;
}

/**
 * @brief 从jf里回滚
 */
static __INLINE__ int pager_roll_back_one_page(pager_t * pager,
											   unsigned int orig_page_count,
											   unsigned int cksum_init,
											   unsigned char * data_buf,
											   const unsigned int data_buf_len)
{
	int ret = 0;
	unsigned int cksum = 0;
	unsigned int pg_no = 0;
	pghd_t * pg = NULL;

	assert(pager && pager->hjf && pager->page_size);
	assert(pager_get_journal_rec_len(pager) == data_buf_len && data_buf);
	assert(pager_get_journal_rec_len(pager) >= pager->page_size);

	if(0 != OsFileRead(pager->hjf, data_buf, data_buf_len, NULL))
		return RHAPSODY_FAIL;

	pager->journal_end += data_buf_len;

	/* 计算校验和,判断此页是否是正确的 */
	array_to_uint_as_big_endian(data_buf + pager->page_size + sizeof(unsigned int), sizeof(unsigned int), cksum);
	if(cksum != pager_cal_page_cksum(data_buf, pager->page_size, cksum_init))
		return RHAPSODY_DONE;

	/* 在页缓存哈希表里查找 */
	array_to_uint_as_big_endian(data_buf + pager->page_size, sizeof(unsigned int), pg_no);
	pg = pager_hash_lookup(pager, pg_no);

	if(NULL == pg || pager_page_is_written_before_page_sync(pg))
	{
		/*
		* 找不到,需要写回文件
		* 找到,则判断是否在提交之前写回过文件,如果有也要写回文件
		*/
		if(0 != pager_write_page_data(pager, data_buf, pager->page_size, pg_no))
			return RHAPSODY_FAIL;
	}

	/* 重刷内存 */
	if(pg)
	{
		assert(PAGE_CLEAN != pg->pg_state);

		memcpy(pg->pg_buf_start, data_buf, pager->page_size);

		if(pager_page_is_dirty(pg))
		{
			assert(pager_page_is_in_dirty_list(pg));
			pager_page_out_of_dirty_list(pager, pg);
		}

		pager_change_page_state(pg, PAGE_EVT_ROLL_BACK);

		/* 回调用户的reload回调 */
		if(pager->page_reload_cb)
			pager->page_reload_cb(pg);
	}

	return RHAPSODY_OK;
}

/**
 * @brief 从jf里回滚
 */
static __INLINE__ int pager_roll_back_from_journal(pager_t * pager)
{
	int64 file_size = 0;
	unsigned int injournal_count = 0;
	unsigned int orig_page_count = 0;
	unsigned int cksum_init = 0;
	int ret = 0;
	int bfirst = 1;
	int i = 0;
	unsigned char * data_buf = NULL;
	unsigned int data_buf_len = 0;

	assert(pager && pager->hjf);

	/* 分配处理回滚过程中需要的内存 */
	data_buf_len = pager_get_journal_rec_len(pager);
	data_buf = MyMemPoolMalloc(pager->hm, data_buf_len);
	if(NULL == data_buf)
		return -1;

	if(0 != OsFileSize(pager->hjf, &file_size))
	{
		ret = -1;
		goto pager_roll_back_from_journal_end_;
	}

	if(0 != OsFileSeek(pager->hjf, 0))
	{
		ret = -1;
		goto pager_roll_back_from_journal_end_;
	}

	pager->journal_end = 0;

	while(1)
	{
		/* 先读取jf头 */
		ret = pager_read_jf_head(pager, file_size, &injournal_count, &orig_page_count, &cksum_init);
		if(RHAPSODY_OK != ret)
			break;

		if(bfirst)
		{
			/* 如果读出的第一个jf头,则要将页文件栽减成原来的大小 */
			if(0 != pager_file_truncate(pager, orig_page_count))
			{
				ret = -1;
				goto pager_roll_back_from_journal_end_;
			}

			pager->total_page_count = orig_page_count;

			bfirst = 0;
		}

		/* 如果是最后一个jf头,并且记录数显示为零,则有可能jf还未同步 */
		if(0 == injournal_count && pager->journal_end == (pager->jouranl_prev_hdr + pager->sector_size))
		{
			/* 说明此时有可能jf本身还同步,剩余的脏页可以直接于页文件里取,因为没有写回 */
			assert(pager->journal_end + pager->in_journal_pages_count * pager_get_journal_rec_len(pager) == file_size
				|| 0 == pager->ref_count);

			assert((pager->journal_end < file_size && pager->need_sync_journal) || 0 == pager->ref_count);

			/*
			* 此时外存对的页文件的页是干净的,但此时仍然从jf里取
			* 代价是额外的计算校验和的开销,
			* 如果改成直接从外页页文件中,则看外设的特点,以点os对文件的组织形式.
			* 如果外设是flash,此种方法效率较低.
			* 如果外设是机械磁盘,则要看页文件的大小,以及os对文件的组织.也许从jf里读,可以减少磁盘io时的寻道时间.
			*/
			injournal_count = pager->in_journal_pages_count;
		}

		/* 根据jf头,循环读出每一条备份记录 */
		for(i = injournal_count; i > 0; i --)
		{
			ret = pager_roll_back_one_page(pager, orig_page_count, cksum_init, data_buf, data_buf_len);
			if(RHAPSODY_OK != ret)
				break;
		}
	}

	{
		/*
		* 还需要还原内存中那些超过页文件大小的页缓存,
		* 将它们的数据刷回0
		* todo:优先此段代码
		*/
		pghd_t * pg = pager->pg_all;

		for(; pg; pg = pg->pg_all_next)
		{
			assert(!pager_page_is_dirty(pg));

			if(orig_page_count <= pg->page_num)
				memset(pg->pg_buf_start, 0, pager->page_size);

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

				pager_out_syncjf_list(pager, pg);
			}

			assert(!pager_page_is_in_syncjf_list(pg));

			if(pager_page_is_dirty(pg))
			{
				assert(pager_page_is_in_dirty_list(pg));
				pager_page_out_of_dirty_list(pager, pg);
			}

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

			assert(!pager_page_is_sync_journal(pg));
		}

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

	/* 如果ret != RHAPSODY_FAIL 删除备份日志文件 */
	if(RHAPSODY_FAIL != ret)
	{
		assert(NULL == pager->pg_dirty);
		pager->page_file_is_integrity = 1;

		if(0 != pager_close_and_del_journal(pager))
		{
			ret = -1;
			goto pager_roll_back_from_journal_end_;
		}

		ret = 0;
	}

pager_roll_back_from_journal_end_:

	MyMemPoolFree(pager->hm, data_buf);

	return ret;
}

/**
 * @brief 第一页获取页缓存引用,判断是否存在jf文件,存在则执行回滚操作
 */
static __INLINE__ int pager_check_and_roll_back(pager_t * pager)
{
	assert(pager && pager->hf);

	/* 判断jf文件是否存在,不存在,返回成功 */
	if(!OsFileExists((char *)MyBufferGet(pager->hb_jf_name, NULL)))
		return 0;

	/*
	* 存在,需要打开它进行回滚
	* 以只读方式打开jf文件
	*/
	if(0 != pager_open_journal_readonly(pager))
		return -1;

	/* 回滚 */
	if(0 != pager_roll_back_from_journal(pager))
		return -1;

	return 0;
}

/**
 * @brief 重新加载页缓存中的脏页
 */
static __INLINE__ int pager_reload_cache(pager_t * pager)
{
	unsigned char * buf = NULL;
	pghd_t * pg = NULL;

	assert(pager && pager->hf);
	assert(pager->pg_dirty);
	assert(pager->page_size);
	assert(pager->page_file_is_integrity);
	assert(NULL == pager->pg_syncjf_first && NULL == pager->pg_syncjf_last);

	buf = MyMemPoolMalloc(pager->hm, pager->page_size);
	if(NULL == buf)
		return -1;

	for(pg = pager->pg_dirty; pg; pg = pager->pg_dirty)
	{
		assert(pager_page_is_dirty(pg));
		assert(!pager_page_is_in_syncjf_list(pg));
		assert(!pager_page_is_sync_journal(pg));

		/* 从外存文件读取相应的页 */
		if(pg->page_num <= pager->orig_pages_count)
		{
			if(0 != pager_read_page_data(pager, pg->page_num, buf, pager->page_size))
				return -1;

			/* 拷贝至页缓存 */
			memcpy(pg->pg_buf_start, buf, pager->page_size);
		}
		else
			memset(pg->pg_buf_start, 0, pager->page_size);

		/* 改变页的状态 */
		pager_page_out_of_dirty_list(pager, pg);

		pager_change_page_state(pg, PAGE_EVT_ROLL_BACK);
		assert(PAGE_CLEAN == pg->pg_state);

		if(pager->page_reload_cb)
			pager->page_reload_cb(pg);
	}

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

	assert(NULL == pager->pg_dirty);

	/* 删除备份的jf文件 */
	if(0 != pager_close_and_del_journal(pager))
		return -1;

	return 0;
}

/**
 * @brief 取消所有的页的更改
 */
static __INLINE__ int pager_rollback(pager_t * pager)
{
	int ret = -1;

	assert(pager);

	/*
	* 在PagerSyn之前,RollBack是有效的,PagerSyn之后,RollBack没有意义了.
	* RollBack判断页文件是否已经同步过jfd,并且向fd里写了数据.此时应从jfd里回滚所有页
	* 如果没有,则重新加载fd里相应的页至脏页缓存即可,
	* 删除份的jfd文件.
	* jfd着色表以及页的injournal标识均置成0,因为数据恢复成完整状态.
	*/

	if(!pager->page_file_is_integrity)/* 如果在提交之前,有部分页缓存有写回至页文件,从jf里还原 */
		ret = pager_roll_back_from_journal(pager);
	else if(NULL == pager->pg_dirty)/* 如果没有脏的缓存页,把jf删除即可 */
		ret = pager_close_and_del_journal(pager);
	else/* 如果在提交之前,没有页缓存写回页文件,重新将脏缓存加载一次即可 */
		ret = pager_reload_cache(pager);

	/* 完成回滚之后,jf应删除 */
	return ret;
}

/**
 * @brief 按页号递增进行排序的比较回调函数
 *  > 0  表示 key1 大于 key2
 *  == 0 表示 key1 等于 key2