cpi.c

uClinux下用的数据库

		direction;
{
	cpi_cls	clusterBuf;
	cpi_ch	*cacheEntry;
	char	*cp,
		*cp1,
		*pagePtr;


	/*
	** Shuffle all the headers in the page one slot starting
	** at the specified header.  
	*/
	if (cpiDebug)
	{
		printf("DEBUG  shiftHeaderPage %s Cluster %d from %d\n",
			direction == CPI_SHIFT_UP?"up":"down",
			clusterNum, headerNum);
	}

	cacheEntry = readPage(idx, CPI_CLS_PAGE(idx, clusterNum));
	getCluster(idx, cacheEntry, clusterNum, &clusterBuf);
	pagePtr = cacheEntry->ptr;
	if (direction == CPI_SHIFT_UP)
	{
		cp = pagePtr + CPI_CLS_SIZE(idx) + 
			(headerNum * CPI_HDR_SIZE(idx));
		cp1 = cp + CPI_HDR_SIZE(idx);
		bcopy(cp,cp1,(CPI_SBK(idx)->hdrsPerPage-headerNum-1) * 
			CPI_HDR_SIZE(idx));
	}
	else
	{
		headerNum ++;
		if (headerNum < *clusterBuf.numHeaders)
		{
			cp = pagePtr + CPI_CLS_SIZE(idx) + 
				(headerNum * CPI_HDR_SIZE(idx));
			cp1 = cp - CPI_HDR_SIZE(idx);
			bcopy(cp,cp1,(CPI_SBK(idx)->hdrsPerPage-headerNum-1) * 
				CPI_HDR_SIZE(idx));
		}
	}
	*(clusterBuf.numHeaders) += direction;
}




static void setMaxValue(idx, clusterNum, headerNum, key, flag)
	cpi	*idx;
	int	headerNum,
		clusterNum;
	char	*key;
	int	flag;
{
	int	res = 0;
	cpi_cls	clusterBuf;
	cpi_hdr	headerBuf;
	cpi_ch	*cachePtr;


	/*
	** Reset the maxValue of the page header if required. If the
	** data page is empty we set the value regardless.
	*/
	cachePtr = readPage(idx, CPI_CLS_PAGE(idx,clusterNum));
	getHeader(idx, cachePtr, clusterNum, headerNum, &headerBuf);
	if (*headerBuf.numRecords > 0 && flag == CPI_IF_NEEDED)
	{
		compareValues(idx, key, headerBuf.maxValue, res);
		if (res <= 0)
		{
			return;
		}
	}
	if (cpiDebug)
	{
		printf("DEBUG  Max value for hdr %d, page %d set to ",
			headerBuf.headerNum, *headerBuf.pageNum);
		printValue(key, CPI_SBK(idx)->keyType);
		printf("\n");
	}
	bcopy(key, headerBuf.maxValue, CPI_SBK(idx)->keySize);

	/*
	** If this is the last header in the cluster then reset the
	** cluster's max value too
	*/
	getCluster(idx,cachePtr,clusterNum,&clusterBuf);
	if (headerNum == *(clusterBuf.numHeaders) - 1)
	{
		bcopy(key, clusterBuf.maxValue, CPI_SBK(idx)->keySize);
		if (cpiDebug)
		{
			printf("DEBUG  Max value for cluster %d set to ",
				clusterNum);
			printValue(key, CPI_SBK(idx)->keyType);
			printf("\n");
		}
	}
}





static void resetMaxValue(idx, clusterNum, headerNum)
	cpi	*idx;
	int	headerNum,
		clusterNum;
{
	cpi_nod	nodeBuf;
	cpi_hdr	headerBuf;
	cpi_ch	*cachePtr;
	int	pageNum;

	cachePtr = readPage(idx, CPI_CLS_PAGE(idx,clusterNum));	
	getHeader(idx, cachePtr, clusterNum, headerNum, &headerBuf);
	pageNum = *headerBuf.pageNum;
	cachePtr = readPage(idx, pageNum);
	getRecord(idx, cachePtr, clusterNum, headerNum,  
		*headerBuf.numRecords - 1, &nodeBuf);
	if (cpiDebug)
	{
		printf("DEBUG  Reset Max for hdr %d:%d page %d\n", 
			clusterNum,headerNum, pageNum);
	}
	setMaxValue(idx, clusterNum, headerNum, nodeBuf.key, CPI_FORCE);
}





static void splitClusters(idx, cluster1, cluster2)
	cpi	*idx;
	int	cluster1, cluster2;
{
	cpi_cls	cluster1Buf, cluster2Buf;
	cpi_ch	*cache1, *cache2;
	char	*ptr1, *ptr2, *cp;
	int	spare,
		splitDest;


	cache1 = readPage(idx, CPI_CLS_PAGE(idx, cluster1));
	cache2 = readPage(idx, CPI_CLS_PAGE(idx, cluster2));
	getCluster(idx, cache1, cluster1, &cluster1Buf);
	getCluster(idx, cache2, cluster2, &cluster2Buf);
	ptr1 = cache1->ptr + CPI_CLS_SIZE(idx);
	ptr2 = cache2->ptr + CPI_CLS_SIZE(idx);
	if (*cluster1Buf.numHeaders < *cluster2Buf.numHeaders)
	{
		spare = CPI_SBK(idx)->hdrsPerPage - *cluster1Buf.numHeaders;
		splitDest = 1;
	}
	else
	{
		spare = CPI_SBK(idx)->hdrsPerPage - *cluster2Buf.numHeaders;
		splitDest = 2;
	}
	if (cpiDebug)
	{
		printf("DEBUG  splitClusters : Cls1=%d, Cls2=%d, ",
			cluster1, cluster2);
		printf("spare slots=%d, dest=hdr%d\n",
			spare, splitDest);
	}

	/*
	** Copy (spare / 2) headers from the full cluster to the other
	*/
	if (splitDest == 1)
	{
		cp = ptr1 + (*cluster1Buf.numHeaders * CPI_HDR_SIZE(idx));
		bcopy(ptr2, cp, (spare / 2) * CPI_HDR_SIZE(idx));
		cp = ptr2 + ((spare / 2) * CPI_HDR_SIZE(idx));
		bcopy(cp,ptr2,(*cluster2Buf.numHeaders - 
			(spare / 2)) * CPI_HDR_SIZE(idx));
		*(cluster1Buf.numHeaders) += spare / 2;
		*(cluster2Buf.numHeaders) -= spare / 2;
	}
	else
	{
		cp = ptr2;
		bcopy(cp, cp + (spare / 2 * CPI_HDR_SIZE(idx)), 
			*cluster2Buf.numHeaders * CPI_HDR_SIZE(idx));
		cp = ptr1 + ((*cluster1Buf.numHeaders - (spare / 2)) 
			* CPI_HDR_SIZE(idx));
		bcopy(cp, ptr2, spare / 2 * CPI_HDR_SIZE(idx)); 
		*(cluster1Buf.numHeaders) -= spare / 2;
		*(cluster2Buf.numHeaders) += spare / 2;
	}


	/*
	** The maxValue of the clusters may have changed so force a reset
	*/
	resetMaxValue(idx, cluster1, *cluster1Buf.numHeaders - 1);
	resetMaxValue(idx, cluster2, *cluster2Buf.numHeaders - 1);

	if (cpiDebug)
	{
		printCluster(idx, &cluster1Buf);
		printCluster(idx, &cluster2Buf);
	}
}




static void splitPages(idx, cluster1, header1, cluster2, header2)
	cpi	*idx;
	int	cluster1, header1,
		cluster2, header2;
{
	cpi_hdr	header1Buf, header2Buf;
	cpi_ch	*cachePtr1, *cachePtr2,
		*nodeCachePtr1, *nodeCachePtr2;
	char	*page1Ptr, *page2Ptr,
		*cp;
	int	spare,
		splitDest;


	cachePtr1 = readPage(idx, CPI_CLS_PAGE(idx, cluster1));
	getHeader(idx, cachePtr1, cluster1, header1, &header1Buf);
	cachePtr2 = readPage(idx, CPI_CLS_PAGE(idx, cluster2));
	getHeader(idx, cachePtr2, cluster2, header2, &header2Buf);
	if (*header1Buf.numRecords < *header2Buf.numRecords)
	{
		spare = CPI_SBK(idx)->recsPerPage - *header1Buf.numRecords;
		splitDest = 1;
	}
	else
	{
		spare = CPI_SBK(idx)->recsPerPage - *header2Buf.numRecords;
		splitDest = 2;
	}
	nodeCachePtr1 = readPage(idx, *header1Buf.pageNum);
	nodeCachePtr2 = readPage(idx, *header2Buf.pageNum);
	page1Ptr = nodeCachePtr1->ptr;
	page2Ptr = nodeCachePtr2->ptr;
	if (cpiDebug)
	{
		printf("DEBUG  splitPages : hdr1=%d:%d, ", cluster1, header1);
		printf("hdr2=%d:%d, spare slots=%d, dest=hdr%d\n",
			cluster2, header2, spare, splitDest);
	}

	/*
	** Copy (spare / 2) records from the full page to the other
	*/
	if (splitDest == 1)
	{
		cp = page1Ptr + (*header1Buf.numRecords * CPI_REC_SIZE(idx));
		bcopy(page2Ptr, cp, (spare / 2) * CPI_REC_SIZE(idx));
		cp = page2Ptr + ((spare / 2) * CPI_REC_SIZE(idx));
		bcopy(cp,page2Ptr,(*header2Buf.numRecords - 
			(spare / 2)) * CPI_REC_SIZE(idx));
		*(header1Buf.numRecords) += spare / 2;
		*(header2Buf.numRecords) -= spare / 2;
	}
	else
	{
		cp = page2Ptr;
		bcopy(cp, cp + (spare / 2 * CPI_REC_SIZE(idx)), 
			*header2Buf.numRecords * CPI_REC_SIZE(idx));
		cp = page1Ptr + ((*header1Buf.numRecords - (spare / 2)) 
			* CPI_REC_SIZE(idx));
		bcopy(cp, page2Ptr, spare / 2 * CPI_REC_SIZE(idx)); 
		*(header1Buf.numRecords) -= spare / 2;
		*(header2Buf.numRecords) += spare / 2;
	}

	/*
	** The maxValue of the pages may have changed so
	** force a reset in the headers.
	*/
	resetMaxValue(idx, cluster1, header1);
	resetMaxValue(idx, cluster2, header2);

	if (cpiDebug)
	{
		printHeader(idx, &header1Buf);
		printHeader(idx, &header2Buf);
	}
}




static void createSpaceForCluster(idx, clusterNum)
	cpi	*idx;
	int	clusterNum;
{
	cpi_cls	clusterBuf;
	cpi_hdr	headerBuf;
	cpi_ch	*cachePtr;
	int	foundRoom = 0,
		curCluster,
		cluster1 = 0, 
		cluster2 = 0, 
		curHeader,
		freeSpace;
	cpi_ch	*cacheEntry1,
		*cacheEntry2;

	if (cpiDebug)
	{
		printf("DEBUG  createSpaceForCluster( %d)\n",clusterNum);
	}
	/*
	** If there's room in a neighbour cluster then grab half
	** of it.  Try the left neighbour first if there is one.
	*/
	if (clusterNum > 0)
	{
		curCluster = clusterNum - 1;
		cachePtr = readPage(idx, CPI_CLS_PAGE(idx,curCluster));
		getCluster(idx, cachePtr, curCluster,&clusterBuf);
		freeSpace = CPI_SBK(idx)->hdrsPerPage - *clusterBuf.numHeaders;
		if (freeSpace > CPI_SBK(idx)->hdrsPerPage / 8 && freeSpace > 3)
		{
			cluster1 = curCluster;
			cluster2 = clusterNum;
			foundRoom = 1;
		}
	}

	/*
	** How about the right neighbour
	*/
	if ( ! foundRoom)
	{
		if (clusterNum < CPI_SBK(idx)->numClusters -1 )
		{
			curCluster = clusterNum + 1;
			cachePtr=readPage(idx,CPI_CLS_PAGE(idx,curCluster));
			getCluster(idx, cachePtr, curCluster,&clusterBuf);
			freeSpace = CPI_SBK(idx)->hdrsPerPage - 
				*clusterBuf.numHeaders;
			if (freeSpace > CPI_SBK(idx)->hdrsPerPage / 8 && 
				freeSpace > 3)
			{
				cluster1 = clusterNum;
				cluster2 = curCluster;
				foundRoom = 1;
			}
		}
	}

	/*
	** If there's no room near here then we have to create a new cluster.
	** Drop a new cluser on the end of the file and roll the clusters
	** down one from clusterNum.
	*/
	if ( ! foundRoom)
	{
		curCluster = CPI_SBK(idx)->numClusters - 1;
		initialiseCluster(idx, curCluster + 1);
		cacheEntry2 = readPage(idx,CPI_CLS_PAGE(idx,curCluster+1));
		while(curCluster > clusterNum)
		{
			cacheEntry1 = readPage(idx,
				CPI_CLS_PAGE(idx,curCluster));
			bcopy(cacheEntry1->ptr, cacheEntry2->ptr, 
				CPI_SBK(idx)->pageSize);

			getCluster(idx,cacheEntry2,curCluster+1,&clusterBuf);
			*clusterBuf.pageNum=CPI_CLS_PAGE(idx,curCluster+1);
			cacheEntry2 = cacheEntry1;
			curCluster--;
		}
		cluster1 = clusterNum;
		cluster2 = clusterNum + 1;

		/*
		** Zero out the values in the old cluster so we can
		** reuse it as a fresh page
		*/
		cachePtr = readPage(idx, CPI_CLS_PAGE(idx,cluster2));
		getCluster(idx, cachePtr, cluster2, &clusterBuf);
		curHeader = 0;
		while(curHeader < *clusterBuf.numHeaders)
		{
			getHeader(idx, cachePtr, cluster2, curHeader, 
				&headerBuf);
			*headerBuf.numRecords = 0;
			*headerBuf.pageNum = 0;
			curHeader++;
		}
		*clusterBuf.numHeaders = 0;
	}
	splitClusters(idx, cluster1, cluster2);
}


static void createSpaceForHeader(idx, clusterNum, headerNum)
	cpi	*idx;
	int	clusterNum,
		headerNum;
{
	cpi_cls	clusterBuf;
	cpi_hdr	headerBuf;
	cpi_ch	*cachePtr;
	int	foundRoom = 0,
		curCluster, curHeader,
		cluster1 = 0, header1 = 0,
		cluster2 = 0, header2 = 0,
		freeSpace,
		validPage;

	if (cpiDebug)
	{
		printf("DEBUG  createSpaceForHeader( %d,%d)\n",clusterNum,
			headerNum);
	}
	/*
	** If there's room in a neighbour page then grab half
	** of it.  Try the left neighbour first.
	*/
	curCluster = clusterNum;
	curHeader = headerNum;
	validPage = 1;
	if (curHeader == 0)
	{
		if (curCluster > 0)
		{
			curCluster --;
			cachePtr = readPage(idx,CPI_CLS_PAGE(idx,curCluster));
			getCluster(idx,cachePtr,curCluster,&clusterBuf);
			curHeader = *clusterBuf.numHeaders - 1;
		}
		else
		{
			validPage = 0;
		}
	}
	else
	{
		curHeader --;
	}
	if (validPage)
	{
		cachePtr = readPage(idx,CPI_CLS_PAGE(idx,curCluster));
		getHeader(idx, cachePtr, curCluster, curHeader, &headerBuf);
		freeSpace = CPI_SBK(idx)->recsPerPage - *headerBuf.numRecords;
		if (freeSpace > CPI_SBK(idx)->recsPerPage / 4 &&
			freeSpace > 3)
		{
			cluster1 = curCluster;
			header1 = curHeader;
			cluster2 = clusterNum;
			header2 = headerNum;
			foundRoom = 1;
		}
	}

	/*
	** How about the right neighbour
	*/
	if ( ! foundRoom)
	{
		curCluster = clusterNum;
		curHeader = headerNum;
		validPage = 1;
		cachePtr = readPage(idx,CPI_CLS_PAGE(idx,curCluster));
		getCluster(idx,cachePtr,curCluster,&clusterBuf);
		if (curHeader == CPI_SBK(idx)->hdrsPerPage - 1)
		{
			curCluster ++;
			curHeader = 0;
			if (curCluster == CPI_SBK(idx)->numClusters)
			{
				validPage = 0;
			}
		}
		else
		{
			curHeader ++;
			if (curHeader == *clusterBuf.numHeaders)
			{
				validPage = 0;
			}
		}
		if (validPage)
		{
			cachePtr = readPage(idx,CPI_CLS_PAGE(idx,curCluster));
			getHeader(idx,cachePtr,curCluster,curHeader,&headerBuf);
			freeSpace = CPI_SBK(idx)->recsPerPage - 
				*headerBuf.numRecords;
			if (freeSpace > CPI_SBK(idx)->recsPerPage / 4 &&
				freeSpace > 3)
			{
				cluster2 = curCluster;
				header2 = curHeader;
				cluster1 = clusterNum;
				header1 = headerNum;
				foundRoom = 1;
			}
		}
	}

	/*
	** If there's no room near here then we have to create a new page.
	** Note, the above code will leave curCluster and curHeader set
	** to the header directly after the one we are working on.  If
	** the page wasn't valid before, it will be after the shiftHeader()
	*/
	if ( ! foundRoom)
	{
		int	tmpPageNum;

		getFreePage(idx, &tmpPageNum);
		shiftHeaderPage(idx, clusterNum, headerNum, CPI_SHIFT_UP);

		cachePtr = readPage(idx,CPI_CLS_PAGE(idx,clusterNum));
		getHeader(idx, cachePtr, clusterNum, headerNum, &headerBuf);
		*headerBuf.pageNum = tmpPageNum;
		*headerBuf.numRecords = 0;
		cluster2 = curCluster;
		header2 = curHeader;
		cluster1 = clusterNum;
		header1 = headerNum;
	}
	splitPages(idx, cluster1, header1, cluster2, header2);
}



/*************************************************************************
**************************************************************************
**
**                        INTERFACE ROUTINES
**
**************************************************************************
*************************************************************************/


/*
static void apiInit()
{
}
*/
#define	apiInit()


int cpiCreate(path, mode, keySize, keyType, flags, envPtr)
        char    *path;
        int     mode,
                keySize,
                keyType,
                flags;
	cpi_env	*envPtr;
{
        int     fd;
        cpi_sbk sbk;
	cpi	idx;

        /*
        ** Create the actual file
        */
	apiInit();
        fd = open(path,O_RDWR | O_CREAT, mode);
        if (fd < 0)
        {
                return(-1);
        }

        /*
        ** Setup the basic superblock info
        */
        strcpy(sbk.fileType,"#! Clustered Page Index Format");
        sbk.magic = CPI_MAGIC;
        sbk.version = CPI_VERSION;
        sbk.keyType = keyType;
        if (keyType == CPI_CHAR)
                keySize++;
        sbk.keySize = keySize;


	/*
	** Now the important stuff.  Set the page size and ensure the
	** mapping size is VM page aligned
	*/
	sbk.pageSize = 0;
	if (envPtr)
		sbk.pageSize = envPtr->pageSize;
	if (sbk.pageSize == 0)
		sbk.pageSize = getpagesize();
	sbk.cacheSize = 0;
	if (envPtr)
		sbk.cacheSize = envPtr->cacheSize;
	if (sbk.cacheSize == 0)
		sbk.cacheSize = CPI_DEFAULT_CACHE;

	sbk.vmPageSize = getpagesize();
	if (sbk.pageSize % sbk.vmPageSize != 0)
	{
		sbk.mapSize = ((sbk.pageSize / sbk.vmPageSize) + 1) *
			sbk.vmPageSize;
	}
	else
	{
		sbk.mapSize = sbk.pageSize;
	}


#	ifdef FORCE_SMALL_PAGES
	sbk.recsPerPage = 5;
	sbk.hdrsPerPage = 6;
#	else
	sbk.recsPerPage = (sbk.pageSize / (keySize + sizeof(u_int)));
	sbk.hdrsPerPage = (sbk.pageSize - (keySize + 2 * sizeof(u_int))) / 
		(keySize + 2 * sizeof(u_int));
#	endif

	sbk.numRecords = 0;
	sbk.numKeys = 0;
	sbk.numClusters = 1;
	sbk.cacheLookups = 0;
	sbk.cacheHits = 0;
	sbk.freeList = 2;
	
        if (write(fd,&sbk,sizeof(sbk)) < sizeof(sbk))
        { 
                close(fd);
                unlink(path);
                return(-1);
        }       


	/*
	** Initialise the first cluster
	*/
	idx.opaque = &sbk;
	idx.fd = fd;
	initialiseCluster(&idx, 0);
	close(fd);
	return(0);
}