pqueue.mx

一个内存数据库的源代码这是服务器端还有客户端

  hBUNsize = BUNsize(h);

  cur = hbase;
  num_elems = BATcount(h);
  posel = 0;
  
  /*while posel is not a leaf and pqueue[posel].tail > any of childen*/
  while (posel*2+1 < num_elems) { /*there exists a left son*/
	if (posel*2+2< num_elems) { /*there exists a right son*/
	  if (atom_CMP(
		BUNtail(h,hbase+(posel*2+1)*hBUNsize),
		BUNtail(h,hbase+(posel*2+2)*hBUNsize), tpe) @2 0 ) {
		swap = posel*2+1;
	  } else {
		swap = posel*2+2;
	  }
	} else
		swap = posel*2+1;

	swp = hbase+swap*hBUNsize;

	if (atom_CMP( BUNtail(h,swp), BUNtail(h,cur), tpe) @2 0 ) {
		/*swap elements*/
		for(i=0;i<hBUNsize; i++){
			ch= cur[i]; cur[i]=swp[i]; swp[i]=ch;
		}
	  cur = swp;
	  posel = swap;
	} else 
		break;
  }
  h->hsorted = h->tsorted = FALSE;

  return GDK_SUCCEED;
}

/* removes the root element, puts the last element as root and moves it down */
int pqueue_dequeue_any@1(BAT *h)
{
  BUN hbase;
  int hBUNsize;

  size_t num_elements;

  if (!(num_elements = BATcount(h))) {
	printf("pqueue_dequeue: Cannot dequeue from empty queue\n");
	return GDK_FAIL;
  }

  hbase = BUNfirst(h);
  hBUNsize = BUNsize(h);

  /* copy last element to the first position*/
  memcpy(hbase, hbase+(num_elements-1)*hBUNsize, hBUNsize);

  /*delete last element*/
  BUNdelete(h, hbase+(num_elements-1)*hBUNsize, FALSE);

  pqueue_movedowntop_any@1(h);
  return GDK_SUCCEED;
}

/* replaces the top element with the input if it is larger (smaller) and
 * updates the heap */
int pqueue_topreplace_any@1(BAT *h, oid *idx, ptr el, int tpe)
{
  BUN hbase;

  hbase = BUNfirst(h);

  if (atom_CMP(BUNtail(h,hbase), el, tpe) @2 0) {
	BUNinplace(h, hbase, idx, el, 0);
	*(oid*)hbase = *idx;
	pqueue_movedowntop_any@1(h);
    	h->hsorted = h->tsorted = FALSE;
  }
  
  return GDK_SUCCEED;
}

int pqueue_topn_voidany@1(BAT **H, BAT *t, int *N)
{
  size_t i, n = BATcount(t);
  oid idx = t->hseqbase;
  int bs = BUNsize(t);
  char *p = BUNfirst(t);
  int tpe = BATttype(t);

  if ((size_t) *N < n)
   	n = (size_t) *N;
  do_pqueue_init(H,t,n);

  for(i=0; i<n; i++, idx++, p+=bs) {
	pqueue_enqueue_any@1(*H, &idx, BUNtvar(t,p), tpe); 
  }
  n = BATcount(t);
  for(; i<n; i++, idx++, p+=bs) {
	pqueue_topreplace_any@1(*H, &idx, BUNtvar(t,p), tpe); 
  }
  return GDK_SUCCEED;
}

int pqueue_topn_any@1(BAT **H, BAT *t, int *N)
{
  size_t i, n = BATcount(t);
  int bs = BUNsize(t);
  char *p = BUNfirst(t);
  int tpe = BATttype(t);

  if ((size_t) *N < n)
   	n = (size_t) *N;
  do_pqueue_init(H,t,n);

  for(i=0; i<n; i++, p+=bs) {
	pqueue_enqueue_any@1(*H, (oid*)BUNhloc(t,p), BUNtail(t,p), tpe); 
  }
  n = BATcount(t);
  for(; i<n; i++, p+=bs) {
	pqueue_topreplace_any@1(*H, (oid*)BUNhloc(t,p), BUNtail(t,p), tpe); 
  }
  return GDK_SUCCEED;
}

@c
@:pqueueimpl_any(min,<)@
@:pqueueimpl_any(max,>)@

@-
old stuff

proc pqueue_peek( BAT[oid,any] h ) : any {
	return h.fetch(0);
}

proc pqueue_topn( BAT[oid,any] t, int n, int direction ) : BAT[oid,any] {
	if (direction > 0) {
		return pqueue_topn_max(t,n);
	} else {
		return pqueue_topn_min(t,n);
	}
}

PROC test_pqueue_str() : void {
	#
	# Simple example of using the heap
	# A min-heap is used to find the k-largest elements 
	# in an (oid,str) BAT with m random values.
	#

	module("alarm");

	# table t has m elements
	var m := 100;
	var t := new(void,str,m);
	t.seqbase(oid(1));

	# We want the k largest elements of t
	var k := 10; 

	var h := pqueue_init(t, k); 

	srand(time());

	var i := 1;
	while (i<=k) {
		var val := str(rand()%1000);
		t.insert(oid(i), val);
		pqueue_enqueue_min(h, oid(i), val);
		i :+= 1;
	}
	while (i<=m) {
		var val := str(rand()%1000);
		t.insert(oid(i), val);
		pqueue_topreplace_min(h, oid(i), val);
		i :+= 1;
	}

	var s := sort_rev(t.reverse()).reverse().slice(0,k); # topn the old way
	var tpn := pqueue_topn(t,k,0);
	print([=](s,h));
	print([=](s,tpn));
}
PROC test_pqueue() : void {
	#
	# Simple example of using the heap
	# A min-heap is used to find the k-largest elements 
	# in an (oid,int) BAT with m random values.
	#

	module("alarm");

	# table t has m elements
	var m := 100;
	var t := new(void,int,m);
	t.seqbase(oid(1));

	# We want the k largest elements of t
	var k := 10; 

	var h := pqueue_init(t, k); 

	srand(time());

	var i := 1;
	while (i<=k) {
		var val := int(rand()%1000);
		t.insert(oid(i), val);
		pqueue_enqueue_min(h, oid(i), val);
		i :+= 1;
	}
	while (i<=m) {
		var val := int(rand()%1000);
		t.insert(oid(i), val);
		pqueue_topreplace_min(h, oid(i), val);
		i :+= 1;
	}

	var s := sort_rev(t.reverse()).reverse().slice(0,k); # topn the old way
	var tpn := pqueue_topn(t,k,0);
	print([=](s,h));
	print([=](s,tpn));
}
@-
The M5 wrapper code
@c
str
PQinit(int *ret, int *bid, int *maxsize){
	BAT *b,*bn;
	if( (b= BATdescriptor(*bid)) == NULL)
		throw(MAL, "pqueue.init","Could not access BAT");
	pqueue_init(&bn,b,maxsize);
	*ret = bn->batCacheid;
	BBPkeepref(b->batCacheid);
	BBPreleaseref(bn->batCacheid);
	return MAL_SUCCEED;
}
@-
@= PQimpl1
pqueue_export str PQenqueue_@1@2(int *ret, int *bid, oid *idx, @1 *el);
str
PQenqueue_@1@2(int *ret, int *bid, oid *idx, @1 *el){
	BAT *b;
	if( (b= BATdescriptor(*bid)) == NULL)
		throw(MAL, "pqueue.enqueue","Could not access BAT");
	pqueue_enqueue_@1@2(b,idx,el);
	(void) ret;
	return MAL_SUCCEED;
}

pqueue_export str PQtopreplace_@1@2(int *ret, int *bid, oid *idx, @1 *el);
str
PQtopreplace_@1@2(int *ret, int *bid, oid *idx, @1 *el){
	BAT *b;
	if( (b= BATdescriptor(*bid)) == NULL)
		throw(MAL, "pqueue.init","Could not access BAT");
	pqueue_topreplace_@1@2(b,idx,el);
	(void) ret;
	return MAL_SUCCEED;
}

@= PQimpl2
pqueue_export str PQmovedowntop_@1@2(int *ret, int *bid);
str
PQmovedowntop_@1@2(int *ret, int *bid){
	BAT *b;
	if( (b= BATdescriptor(*bid)) == NULL)
		throw(MAL, "pqueue.movedowntop","Could not access BAT");
	pqueue_movedowntop_@1@2(b);
	(void) ret;
	return MAL_SUCCEED;
}

pqueue_export str PQdequeue_@1@2(int *ret, int *bid);
str
PQdequeue_@1@2(int *ret, int *bid){
	BAT *b;
	if( (b= BATdescriptor(*bid)) == NULL)
		throw(MAL, "pqueue.init","Could not access BAT");
	if( pqueue_dequeue_@1@2(b) == GDK_FAIL)
		throw(MAL, "pqueue.dequeue","Cannot dequeue from empty queue");
	(void) ret;
	return MAL_SUCCEED;
}

pqueue_export str PQtopn_void@1@2(int *ret, int *bid, int *N);
str
PQtopn_void@1@2(int *ret, int *bid, int *N){
	BAT *b,*bn;
	if( (b= BATdescriptor(*bid)) == NULL)
		throw(MAL, "pqueue.topN","Could not access BAT");
	if (b->htype == TYPE_void)
		pqueue_topn_void@1@2(&bn,b,N);
	else
		pqueue_topn_@1@2(&bn,b,N);
	if( bn){
		*ret= bn->batCacheid;
		BBPkeepref(*ret);
		BBPreleaseref(b->batCacheid);
		return MAL_SUCCEED;
	}
	BBPreleaseref(b->batCacheid);
	throw(MAL, "pqueue.topN","Failed to create BAT");
}
pqueue_export str PQtopn_@1@2(int *ret, int *bid, int *N);
str
PQtopn_@1@2(int *ret, int *bid, int *N){
	BAT *b,*bn;
	if( (b= BATdescriptor(*bid)) == NULL)
		throw(MAL, "pqueue.topN","Could not access BAT");
	if (b->htype == TYPE_void)
		pqueue_topn_void@1@2(&bn,b,N);
	else
		pqueue_topn_@1@2(&bn,b,N);
	if( bn){
		*ret= bn->batCacheid;
		BBPkeepref(*ret);
		BBPreleaseref(b->batCacheid);
		return MAL_SUCCEED;
	}
	BBPreleaseref(b->batCacheid);
	throw(MAL, "pqueue.topN","Failed to create BAT");
}
@= PQminmax1
 @:PQimpl1(@1,min)@
 @:PQimpl1(@1,max)@
@= PQminmax2
 @:PQimpl2(@1,min)@
 @:PQimpl2(@1,max)@
@= PQminmax
 	@:PQminmax1(@1)@
 	@:PQminmax2(@1)@
@c
	@:PQminmax(chr)@
	@:PQminmax(sht)@
	@:PQminmax(int)@
	@:PQminmax(oid)@
	@:PQminmax(ptr)@
	@:PQminmax(lng)@
	@:PQminmax(flt)@
	@:PQminmax(dbl)@
	@:PQimpl2(any,min)@
	@:PQimpl2(any,max)@

@= PQany
/* int PQenqueue_any@1(BAT *h, oid *idx, ptr el, int tpe)*/
pqueue_export str PQenqueue_any@1(MalBlkPtr mb, MalStkPtr stk, InstrPtr p);
str  PQenqueue_any@1(MalBlkPtr mb, MalStkPtr stk, InstrPtr p)
{
	int tpe;
	BAT *h;
	oid *idx;
	ptr el;

        if( p->argc != 4 || 
	    getArgType(mb,p,1) != TYPE_bat || 
	    getArgType(mb,p,2) != TYPE_oid) 
		throw(MAL, "enqueue_@1", "type mismatch\n");
	tpe = getArgType(mb,p,3);

	h = BATdescriptor(*(bat*) getArgReference(stk,p,1));
	if (!h)
		throw(MAL, "enqueue_@1", "BAT range error\n");
	idx = (oid*) getArgReference(stk,p,2);
	el = (ptr) getArgReference(stk,p,3);
		
	pqueue_enqueue_any@1(h, idx, el, tpe);
	return MAL_SUCCEED;
}

/* int pqueue_topreplace_any@1(BAT *h, oid *idx, ptr el, int tpe) */
pqueue_export str PQtopreplace_any@1(MalBlkPtr mb, MalStkPtr stk, InstrPtr p);
str  PQtopreplace_any@1(MalBlkPtr mb, MalStkPtr stk, InstrPtr p)
{
	int tpe;
	BAT *h;
	oid *idx;
	ptr el;

        if( p->argc != 4 || 
	    getArgType(mb,p,1) != TYPE_bat || 
	    getArgType(mb,p,2) != TYPE_oid) 
		throw(MAL, "topreplace_@1", "type mismatch\n");
	tpe = getArgType(mb,p,3);

	h = BATdescriptor(*(bat*) getArgReference(stk,p,1));
	if (!h)
		throw(MAL, "topreplace_@1", "BAT range error\n");
	idx = (oid*) getArgReference(stk,p,2);
	el = (ptr) getArgReference(stk,p,3);
		
	pqueue_topreplace_any@1(h, idx, el, tpe);
	return MAL_SUCCEED;
}
@c
	@:PQany(min)@
	@:PQany(max)@
@}