gdk_bbp.mx

这个是内存数据库中的一个管理工具

	return buf;
}

@}
@+ BBP Update Interface
Operations to insert, delete, clear, and modify BBP entries.
Our policy for the BBP is to provide unlocked BBP access for 
speed, but still write operations have to be locked.
@{
#ifdef DEBUG_THREADLOCAL_BATS
Create the shadow version (reversed) of a bat.
@- 
An existing BAT is inserted into the BBP 
@c
static INLINE str
BBPsubdir_recursive(str s, bat i)
{
	i >>= 6;
	if (i >= 64) {
		s = BBPsubdir_recursive(s, i);
		*s++ = DIR_SEP;
	}
	i &= 63;
	*s++ = '0' + (i >> 3);
	*s++ = '0' + (i & 7);
	return s;
}

static INLINE void
BBPgetsubdir(str s, bat i)
{
	if (i >= 64) {
		s = BBPsubdir_recursive(s, i);
	}
	*s = 0;
}

bat
BBPinsert(BAT *b)
{
	int lock = locked_by ? BBP_getpid() != locked_by : 1;
	str s;
	long_str dirname;
	bat i;

	/* ciritical section: get a new BBP entry */
	if (lock) {
	        gdk_set_lock(GDKtrimLock, "BBPreplace");
		gdk_set_lock(GDKcacheLock, "BBPinsert");
        }

	/* find an empty slot */
	if (BBP_free <= 0) {
		if (++BBPsize >= BBPlimit) {
			BBPextend(BATMARGIN, TRUE);
		} else {
			BBP_free = BBPsize - 1;
		}
	}
	i = BBP_free;
	BBP_free = BBP_next(BBP_free);

	if (lock) {
		gdk_unset_lock(GDKcacheLock, "BBPinsert");
	        gdk_unset_lock(GDKtrimLock, "BBPreplace");
        }
	/* rest of the work outside the lock , as GDKstrdup/GDKmalloc may trigger a BBPtrim */

	/* fill in basic BBP fields for the new bat */

	if (++BBP_curstamp < 0)
		BBP_curstamp = 0;
	b->batCacheid = i;
	b->batStamp = BBP_curstamp;
	b->creator_tid = BBP_getpid();

	BBP_status_set(i, BBPDELETING, "BBPentry");
	BBP_cache(i) = NULL;
	BBP_desc(i) = NULL;
	BBP_refs(i) = 1;	/* new bats have 1 pin */
	BBP_lrefs(i) = 0;	/* ie. no logical refs */

	if( BBP[ABS(i)].bak[(i)<0] == NULL ){
		s = BBPtmpname(dirname, 64, i);
		BBP_logical(i) = GDKstrdup(s);
		BBP[ABS(i)].bak[(i)<0] = BBP_logical(i);
	} else 
		BBP_logical(i) =  BBP[ABS(i)].bak[(i)<0];
	BBP_logical(-i) = NULL;

	/* Keep the physical location around forever */
	if( BBP_physical(i) == NULL){
		BBPgetsubdir(dirname, i);
		BBP_physical(i) = (str) GDKmalloc(strlen(dirname) + strLen(BATgetId(b)) - 4 + 1);
		GDKfilepath(BBP_physical(i), dirname, BATgetId(b) + 4, NULL);

	BATDEBUG THRprintf(GDKout, "#%d = new %s(%s,%s)\n", (int) i, BATgetId(b), ATOMname(b->htype), ATOMname(b->ttype));
	}

	return i;
}

void
BBPcacheit(BAT *b)
{
	int lock = locked_by ? BBP_getpid() != locked_by : 1;
	bat i = b->batCacheid;
	int mode;
	BAT *bm = (BAT *) GDKmalloc(sizeof(BAT));	/* alloc before lock to prevent trim problems */

	if (i) {
		assert(i > 0);
	} else {
		i = BBPinsert(b);	/* bat was not previously entered */
	}
	if (lock)
		gdk_set_lock(GDKswapLock[i & BBPLOCKMASK], "BBPcacheit");
	mode = (BBP_status(i) | BBPLOADED) & ~(BBPLOADING|BBPDELETING);
	BBP_status_set(i, mode, "BBPcacheit");
	BBP_lastused(i) = BBPLASTUSED(BBPstamp() + ((mode == BBPLOADED) ? 150 : 0));
	BBP_desc(i) = (BATstore *) b;

	/* fill in the mirror record */
	bm->GDKversion = b->GDKversion;
	bm->batCacheid = -i;
	bm->batBuns = b->batBuns;
	bm->dims.headtype = b->dims.tailtype;
	bm->dims.tailtype = b->dims.headtype;
	bm->dims.headloc = b->dims.tailloc;
	bm->dims.tailloc = b->dims.headloc;
	bm->dims.headkey = b->dims.tailkey;
	bm->dims.tailkey = b->dims.headkey;
	bm->dims.headvarsized = b->dims.tailvarsized;
	bm->dims.tailvarsized = b->dims.headvarsized;
	bm->dims.bunwidth = b->dims.bunwidth;
	bm->dims.bunshift = b->dims.bunshift;
	bm->dims.hseq = b->dims.tseq;
	bm->dims.tseq = b->dims.hseq;
	bm->H = b->T;
	bm->T = b->H;
	bm->P = b->P;
	bm->U = b->U;
	bm->hhash = b->thash;
	bm->thash = b->hhash;

	/* cache it! */
	BBP_cache(i) = b;
	BBP_cache(-i) = bm;

	if (lock)
		gdk_unset_lock(GDKswapLock[i & BBPLOCKMASK], "BBPcacheit");
}

@
@%BBPuncacheit@ changes the BBP status to swapped out.  Currently only 
used in BBPfree (bat swapped out) and BBPclear (bat destroyed forever).
@c

void
BBPuncacheit(bat i)
{
	BBPuncacheit_(i, FALSE);
}

static void
BBPuncacheit_(bat i, int unloaddesc)
{
	if (i < 0)
		i = -i;
	if (BBPcheck(i, "BBPuncacheit")) {
		BAT *b = (BAT *) BBP_desc(i);

		if (b) {
			if (BBP_cache(i)) {
				BATDEBUG THRprintf(GDKout, "#uncache %d (%s)\n", (int) i, BBPname(i));

				BBP_cache(i) = BBP_cache(-i) = NULL;

				/* clearing bits can be done without the lock */
				BBP_status_off(i, BBPLOADED, "BBPuncacheit");
			}
			if (unloaddesc) {
				BBP_desc(i) = NULL;
				BATdestroy(b);
			}
		}
	}
}

@- BBPclear
@%BBPclear@ removes a BAT from the BBP directory forever.
@c
static INLINE void 
bbpclear(bat i) 
{
		BATDEBUG {
			THRprintf(GDKout, "#clear %d (%s)\n", (int) i, BBPname(i));
		}
		BBPuncacheit_(i, TRUE);
		if (BBPtmpcheck(BBP_logical(i)) == 0) {
			BBP_delete(i);
		}
		if (BBP_logical(-i)) {
			BBP_delete(-i);
			if( BBP_logical(-i) != BBP[ABS(i)].bak[(-i)<0])
				GDKfree(BBP_logical(-i));
			BBP_logical(-i) = NULL;
		}
		if( BBP_logical(i) != BBP[ABS(i)].bak[(i)<0])
			GDKfree(BBP_logical(i));
		BBP_logical(i) = NULL;
		BBP_status_set(i, 0, "BBPclear");
		BBP_refs(i) = 0;
		BBP_lrefs(i) = 0;
		BBP_next(i) = BBP_free;
		BBP_free = i;
}

void
BBPclear(bat i)
{
	int lock = locked_by ? BBP_getpid() != locked_by : 1;

	if (BBPcheck(i, "BBPclear")) {
		if (lock)
			gdk_set_lock(GDKcacheLock, "BBPclear");
		bbpclear(ABS(i));
		if (lock)
			gdk_unset_lock(GDKcacheLock, "BBPclear");
	}
}

@}
@- BBP rename

Each BAT has a logical name that is globally unique. Its reverse view can
also be assigned a name, that also has to be globally unique.  The batId is 
the same as the logical BAT name.

The default logical name of a BAT is tmp_X, where X is the batCacheid.
Apart from being globally unique, new logical bat names cannot be of the 
form tmp_X, unless X is the batCacheid.

Physical names consist of a directory name followed by a logical name suffix. 
The directory name is derived from the batCacheid, and is currently organized 
in a hierarchy that puts max 64 bats in each directory (see BBPgetsubdir). 

Concerning the physical suffix: it is almost always bat_X. This saves us
a whole lot of trouble, as bat_X is always unique and no conflicts can occur.
Other suffixes are only supported in order just for backward compatibility with 
old repositories (you won't see them anymore in new repositories).
@{
@c
int
BBPrename(bat bid, str nme)
{
	BAT *b = BBPdescriptor(bid);
	long_str dirname;
	bat tmpid = 0, i;

	if (b == NULL)
		return 0;

	/* If name stays same, do nothing */
	if (BBP_logical(bid) && strcmp(BBP_logical(bid), nme) == 0)
		return 0;

	BBPgetsubdir(dirname, ABS(bid));

	if ((tmpid = BBPnamecheck(nme)) && (bid < 0 || tmpid != bid)) {
		return BBPRENAME_ILLEGAL;
	}
	if (strlen(dirname) + strLen(nme) + 1 >= IDLENGTH) {
		return BBPRENAME_LONG;
	}
	gdk_set_lock(GDKtrimLock, "BBPrename");
	gdk_set_lock(GDKcacheLock, "BBPrename");
	i = BBP_find(nme, FALSE);
	if (i != 0) {
		gdk_unset_lock(GDKcacheLock, "BBPrename");
		gdk_unset_lock(GDKtrimLock, "BBPrename");
		return BBPRENAME_ALREADY;
	}
	BBP_notrim = BBP_getpid();

	/* carry through the name change */
	if (BBP_logical(bid) && BBPtmpcheck(BBP_logical(bid)) == 0) {
		BBP_delete(bid);
	}
	if( BBP_logical(bid) != BBP[ABS(bid)].bak[(bid)<0])
		GDKfree(BBP_logical(bid));
	BBP_logical(bid) = GDKstrdup(nme);
	if (tmpid == 0) {
		BBP_insert(bid);
	}
	b->batDirtydesc = 1;
	if (b->batPersistence == PERSISTENT) {
		int lock = locked_by ? BBP_getpid() != locked_by : 1;

		if (lock)
			gdk_set_lock(GDKswapLock[i & BBPLOCKMASK], "BBPrename");
		BBP_status_on(ABS(bid), BBPRENAMED, "BBPrename");
		if (lock)
			gdk_unset_lock(GDKswapLock[i & BBPLOCKMASK], "BBPrename");
		BBPdirty(1);
	}
	gdk_unset_lock(GDKcacheLock, "BBPrename");
	BBP_notrim = 0;
	gdk_unset_lock(GDKtrimLock, "BBPrename");
	return 0;
}

@}

@+ BBP swapping Policy
The BAT can be moved back to disk using the routine @%BBPfree@.
It frees the storage for other BATs. After this call BAT* references
maintained for the BAT are wrong.
We should keep track of dirty unloaded BATs. They may have to be committed
later on, which may include reading them in again.

BBPswappable: may this bat be unloaded?
Only real bats without memory references can be unloaded.
@{
@h
#define BBPswappable(b) ((b) && BBP_refs((b)->batCacheid) == 0)
#define BBPtrimmable(b) (BBPswappable(b) && b->batParentid == 0 && (BBP_status((b)->batCacheid)&BBPWAITING) == 0)

#endif /* _GDK_BBP_H_ */
@- 
The @%BBP_ref@ contains the amount of live references to a BAT.
These might be in recursive BATs, C or MIL variables.  The count is 
incremented with @%BBPfix@ and decremented with @%BBPunfix@.
@c
static INLINE void
BBPspin(bat i, str s, int event)
{
	if (BBPcheck(i, "BBPspin") && (BBP_status(i) & event)) {
		lng spin = LL_CONSTANT(0);

		while (BBP_status(i) & event) {
			MT_sleep_ms(1);
			spin++;
		}
		BATDEBUG THRprintf(GDKout, "#BBPspin(%d,%s,%d): " LLFMT " loops\n", (int) i, s, event, spin);
	}
}

static INLINE int
incref(bat i, int logical, int lock)
{
	int refs = 0;

	if (i == bat_nil) {
		/* Stefan: May this happen? Or should we better call GDKerror(), here? */
		/* GDKerror("BBPincref() called with bat_nil!\n"); */
		return refs;
	}
	if (i < 0)
		i = -i;
	if (BBPcheck(i, "BBPincref")) {
		if (lock) {
			gdk_set_lock(GDKswapLock[i & BBPLOCKMASK], "BBPincref");

			while (BBP_status(i) & BBPUNSTABLE) {
				gdk_unset_lock(GDKswapLock[i & BBPLOCKMASK], "BBPincref spin wait");
				MT_sleep_ms(1);
				gdk_set_lock(GDKswapLock[i & BBPLOCKMASK], "BBPincref spin wait");
			}
		}

		/* got the lock */
		if (logical) {
			refs = ++BBP_lrefs(i);
		} else {
			refs = ++BBP_refs(i);
		}
		if (lock)
			gdk_unset_lock(GDKswapLock[i & BBPLOCKMASK], "BBPfix");
	}
	return refs;
}

int
BBPincref(bat i, int logical)
{
	int lock = locked_by ? BBP_getpid() != locked_by : 1;
	return incref(i, logical, lock);
}

void
BBPshare(bat parent)
{
	int lock = locked_by ? BBP_getpid() != locked_by : 1;
	int fix = 0;

	if (parent < 0)
		parent = -parent;
	if (lock)
		gdk_set_lock(GDKswapLock[parent & BBPLOCKMASK], "BBPshare");
	if (++BBP_cache(parent)->batSharecnt == 1)
		fix = 1;
	if (lock)
		gdk_unset_lock(GDKswapLock[parent & BBPLOCKMASK], "BBPshare");
	if (fix)
		BBPincref(parent, FALSE);
}

static INLINE int
decref(bat i, int logical, int lock)
{
	int refs = 0, swap = 0, swap_unloading = 0;
	BAT *b;

	/* decrement references by one */
	if (logical) {
		if (BBP_lrefs(i) == 0) {
			GDKerror("BBPdecref: %s does not have logical references.\n", BBPname(i));
		} else {
			refs = --BBP_lrefs(i);
		}
	} else {
		if (BBP_refs(i) == 0) {
			GDKerror("BBPdecref: %s does not have pointer fixes.\n", BBPname(i));
		} else {
			refs = --BBP_refs(i);
		}
	}

	/* we destroy transients asap and unload persistent bats only if they have been made cold */
	b = BBP_cache(i);
	if (BBP_refs(i) > 0 || (BBP_lrefs(i) > 0 && BBP_lastused(i) != 0)) {
		/* bat cannot be swapped out. renew its last usage stamp for the BBP LRU policy */
		int sec = BBPLASTUSED(BBPstamp());
		if (sec > BBPLASTUSED(BBP_lastused(i)))
			BBP_lastused(i) = sec;
	} else if (b || (BBP_status(i) & BBPTMP)) {
		/* bat will be unloaded now. set the UNLOADING bit while locked so no other thread thinks its available anymore */
                if (BBP_status(i) & BBPUNLOADING) {