mserver.mx

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

			mapi_result_error(SERVERsessions[i].hdl));
	return MAL_SUCCEED;
}
str 
SERVERrows_affected(int *ret, int *key){
	Mapi mid;
	int i;
	@:accessTest(*key,rows_affected)@
	*ret= mapi_rows_affected(SERVERsessions[i].hdl);
	return MAL_SUCCEED;
}
str 
SERVERfetch_row(int *ret, int *key){
	Mapi mid;
	int i;
	@:accessTest(*key,fetch_row)@
	*ret= mapi_fetch_row(SERVERsessions[i].hdl);
	return MAL_SUCCEED;
}
str 
SERVERfetch_all_rows(int *ret, int *key){
	Mapi mid;
	int i;
	@:accessTest(*key,fetch_all_rows)@
	*ret= mapi_fetch_all_rows(SERVERsessions[i].hdl);
	return MAL_SUCCEED;
}
str 
SERVERfetch_field_str(str *ret, int *key, int *fnr){
	Mapi mid;
	int i;
	str fld;
	@:accessTest(*key,fetch_field)@
	fld= mapi_fetch_field(SERVERsessions[i].hdl,*fnr);
	*ret= GDKstrdup(fld? fld: str_nil);
	if( mapi_error(mid) ) 
		throw(MAL, "mserver.fetch_field_str",
			mapi_result_error(SERVERsessions[i].hdl));
	return MAL_SUCCEED;
}
str 
SERVERfetch_field_int(int *ret, int *key, int *fnr){
	Mapi mid;
	int i;
	str fld;
	@:accessTest(*key,fetch_field)@
	fld= mapi_fetch_field(SERVERsessions[i].hdl,*fnr);
	*ret= fld? (int) atol(fld): int_nil;
	if( mapi_error(mid) ) 
		throw(MAL, "mserver.fetch_field_int",
			mapi_result_error(SERVERsessions[i].hdl));
	return MAL_SUCCEED;
}
str 
SERVERfetch_field_lng(lng *ret, int *key, int *fnr){
	Mapi mid;
	int i;
	str fld;
	@:accessTest(*key,fetch_field)@
	fld= mapi_fetch_field(SERVERsessions[i].hdl,*fnr);
	*ret= fld? atol(fld): lng_nil;
	if( mapi_error(mid) ) 
		throw(MAL, "mserver.fetch_field_lng",
			mapi_result_error(SERVERsessions[i].hdl));
	return MAL_SUCCEED;
}
str 
SERVERfetch_field_sht(sht *ret, int *key, int *fnr){
	Mapi mid;
	int i;
	str fld;
	@:accessTest(*key,fetch_field)@
	fld= mapi_fetch_field(SERVERsessions[i].hdl,*fnr);
	*ret= fld? (sht) atol(fld): sht_nil;
	if( mapi_error(mid) ) 
		throw(MAL, "mserver.fetch_field",
			mapi_result_error(SERVERsessions[i].hdl));
	return MAL_SUCCEED;
}
str 
SERVERfetch_field_void(oid *ret, int *key, int *fnr){
	Mapi mid;
	int i;
	@:accessTest(*key,fetch_field)@
	(void) fnr;
	*ret = oid_nil;
	throw(MAL, "mserver.fetch_field_void","defaults to nil");
}
str 
SERVERfetch_field_oid(oid *ret, int *key, int *fnr){
	Mapi mid;
	int i;
	str fld;
	@:accessTest(*key,fetch_field)@
	fld= mapi_fetch_field(SERVERsessions[i].hdl,*fnr);
	if( mapi_error(mid) ) 
		throw(MAL, "mserver.fetch_field_oid",
			mapi_result_error(SERVERsessions[i].hdl));
	if(fld==0 || strcmp(fld,"nil")==0)
		*(oid*) ret= void_nil;
	else *(oid*) ret = (oid) atol(fld);
	return MAL_SUCCEED;
}
str 
SERVERfetch_field_chr(chr *ret, int *key, int *fnr){
	Mapi mid;
	int i;
	str fld;
	@:accessTest(*key,fetch_field)@
	fld= mapi_fetch_field(SERVERsessions[i].hdl,*fnr);
	if( mapi_error(mid) ) 
		throw(MAL, "mserver.fetch_field_chr",
			mapi_result_error(SERVERsessions[i].hdl));
	if(fld==0 || strcmp(fld,"nil")==0)
		*(chr*) ret= chr_nil;
	else *(chr*) ret = *fld;
	return MAL_SUCCEED;
}
str 
SERVERfetch_line(str *ret, int *key){
	Mapi mid;
	int i;
	str fld;
	@:accessTest(*key,fetch_line)@
	fld= mapi_fetch_line(SERVERsessions[i].hdl);
	if( mapi_error(mid) ) 
		throw(MAL, "mserver.fetch_line",
			mapi_result_error(SERVERsessions[i].hdl));
	*ret= GDKstrdup(fld? fld:str_nil);
	return MAL_SUCCEED;
}
str 
SERVERnext_result(int *ret, int *key){
	Mapi mid;
	int i;
	@:accessTest(*key,next_result)@
	mapi_next_result(SERVERsessions[i].hdl);
	if( mapi_error(mid) ) 
		throw(MAL, "mserver.next_result",
			mapi_result_error(SERVERsessions[i].hdl));
	*ret= *key;
	return MAL_SUCCEED;
}
str 
SERVERfetch_reset(int *ret, int *key){
	Mapi mid;
	int i;
	@:accessTest(*key,fetch_reset)@
	mapi_fetch_reset(SERVERsessions[i].hdl);
	if( mapi_error(mid) ) 
		throw(MAL, "mserver.fetch_reset",
			mapi_result_error(SERVERsessions[i].hdl));
	*ret= *key;
	return MAL_SUCCEED;
}
str
SERVERfetch_field_bat(int *bid, int *key){
	int i,j,cnt;
	Mapi mid;
	char *fld;
	int o=0; 
	BAT *b;

	@:accessTest(*key,rpc)@
	b= BATnew(TYPE_oid,TYPE_str,256);
	cnt= mapi_get_field_count(SERVERsessions[i].hdl);
	for(j=0; j< cnt; j++){
		fld= mapi_fetch_field(SERVERsessions[i].hdl,j);
		if( mapi_error(mid) ) {
			*bid = b->batCacheid;
			BBPkeepref(*bid);
			throw(MAL, "mserver.fetch_field_bat",
				mapi_result_error(SERVERsessions[i].hdl));
		}
		BUNins(b,&o,fld, FALSE);
		o++;
	}
	if (!(b->batDirty&2)) b = BATsetaccess(b, BAT_READ);
	*bid = b->batCacheid;
	BBPkeepref(*bid);
	return MAL_SUCCEED;
}
str
SERVERerror(int *ret, int *key){
	Mapi mid;
	int i;
	@:accessTest(*key,error)@
	*ret= mapi_error(mid);
	return MAL_SUCCEED;
}
str
SERVERgetError(str *ret, int *key){
	Mapi mid;
	int i;
	@:accessTest(*key,getError)@
	*ret= GDKstrdup(mapi_error_str(mid));
	return MAL_SUCCEED;
}

str 
SERVERexplain(str *ret, int *key){
	Mapi mid;
	int i;

	@:accessTest(*key,explain)@
	*ret= GDKstrdup(mapi_error_str(mid));
	return MAL_SUCCEED;
}
@-
The remainder should contain the wrapping of 
relevant SERVER functions. Furthermore, we 
should analyse the return value and update 
the stack trace.

Two routines should be 
mserver.rpc(key,"query") 
@-
The generic scheme for handling a remote MAL
procedure call with a single row answer.
@c
void SERVERfieldAnalysis(str fld, int tpe, ValPtr v){
	v->vtype= tpe;
	switch(tpe){
	case TYPE_void:
		v->val.oval = void_nil;
		break;
	case TYPE_oid:
		if(fld==0 || strcmp(fld,"nil")==0)
			v->val.oval= void_nil;
		else v->val.oval = (oid) atol(fld);
		break;
	case TYPE_bit:
		if(fld== 0 || strcmp(fld,"nil")==0)
			v->val.cval[0]= bit_nil;
		else
		if(strcmp(fld,"true")==0)
			v->val.cval[0]= TRUE;
		if(strcmp(fld,"false")==0)
			v->val.cval[0]= FALSE;
		v->val.cval[1]= 0;
		v->val.cval[2]= 0;
		v->val.cval[3]= 0;
		break;
	case TYPE_chr:
		if(fld==0 || strcmp(fld,"nil")==0)
			v->val.cval[0]= chr_nil;
		else
			v->val.cval[0]= *fld;
		v->val.cval[1]= 0;
		v->val.cval[2]= 0;
		v->val.cval[3]= 0;
		break;
	case TYPE_sht:
		if(fld==0 || strcmp(fld,"nil")==0)
			v->val.shval = sht_nil;
		else v->val.shval= (sht)  atol(fld);
		break;
	case TYPE_int:
		if(fld==0 || strcmp(fld,"nil")==0)
			v->val.ival = int_nil;
		else v->val.ival= (int)  atol(fld);
		break;
	case TYPE_lng:
		if(fld==0 || strcmp(fld,"nil")==0)
			v->val.lval= lng_nil;
		else v->val.lval= (lng)  atol(fld);
		break;
	case TYPE_flt:
		if(fld==0 || strcmp(fld,"nil")==0)
			v->val.fval= flt_nil;
		else v->val.fval= (flt)  atof(fld);
		break;
	case TYPE_dbl:
		if(fld==0 || strcmp(fld,"nil")==0)
			v->val.dval= dbl_nil;
		else v->val.dval= (dbl)  atof(fld);
		break;
	case TYPE_str:
		if(fld==0 || strcmp(fld,"nil")==0){
			v->val.sval= GDKstrdup(str_nil);
			v->len= strlen(v->val.sval);
		} else {
			v->val.sval= GDKstrdup(fld);
			v->len= strlen(fld);
		}
		break;
	}
}
str
SERVERmapi_rpc_single_row(MalBlkPtr mb, MalStkPtr stk, InstrPtr pci)
{
	int key,i,j;
	Mapi mid;
	MapiHdl hdl;
	char *s,*fld, *qry=0;

	key= * (int*) getArgReference(stk,pci,pci->retc);
	@:accessTest(key,rpc)@
#ifdef MAPI_TEST
	stream_printf(GDKout,"about to sent: %s\n",qry);
#endif
	/* glue all strings together */
	for(i= pci->retc+1; i<pci->argc; i++){
		fld= * (str*) getArgReference(stk,pci,i);
		if( qry == 0)
			qry= GDKstrdup(fld);
		else {	
			s= (char*) GDKmalloc(strlen(qry)+strlen(fld)+1);
			strcpy(s,qry);
			strcat(s,fld);
			GDKfree(qry);
			qry= s;
		}
	}
	hdl= mapi_query(mid, qry);
	@:catchErrors(mserver.rpc)@
	GDKfree(qry);

	i= 0;
	while( mapi_fetch_row(hdl)){
		for(j=0; j<pci->retc; j++){
			fld= mapi_fetch_field(hdl,j);
#ifdef MAPI_TEST
			stream_printf(GDKout,"Got: %s\n",fld);
#endif
			switch(getVarType(mb,getArg(pci,j)) ){
			case TYPE_void:
			case TYPE_oid:
			case TYPE_bit:
			case TYPE_chr:
			case TYPE_sht:
			case TYPE_int:
			case TYPE_lng:
			case TYPE_flt:
			case TYPE_dbl:
			case TYPE_str:
				SERVERfieldAnalysis(fld, 
					getVarType(mb,getArg(pci,j)),
					&stk->stk[getArg(pci,j)]);
				break;
			default:
				throw(MAL, "mapi.rpc",
						"Missing type implementation ");
			/* all the other basic types come here */
			}
		}
		i++;
	}
	if( i>1)
		throw(MAL, "mapi.rpc","Too many answers");
	return MAL_SUCCEED;
}
@-
Transport of the BATs is only slightly more complicated.
The generic implementation based on a pattern is the next
step. 
@c
str
SERVERmapi_rpc_bat(MalBlkPtr mb, MalStkPtr stk, InstrPtr pci){
	int *ret;
	int *key;
	str *qry,err= MAL_SUCCEED;
	int i;
	Mapi mid;
	MapiHdl hdl;
	char *fld1, *fld2;
	BAT *b;
	ValRecord hval,tval;
	int ht,tt;

	ret= (int*) getArgReference(stk,pci,0);
	key= (int*) getArgReference(stk,pci,pci->retc);
	qry= (str*) getArgReference(stk,pci,pci->retc+1);
	@:accessTest(*key,rpc)@
	ht= getHeadType(getVarType(mb,getArg(pci,0)));
	tt= getTailType(getVarType(mb,getArg(pci,0)));

	hdl= mapi_query(mid, *qry);
	@:catchErrors(mserver.rpc)@
	
	b= BATnew(ht,tt,256);
	i= 0;
	if ( mapi_fetch_row(hdl)){
		int oht = ht, ott = tt;

		fld1= mapi_fetch_field(hdl,0);
		fld2= mapi_fetch_field(hdl,1);
		if (fld1 && ht == TYPE_void) 
			ht = TYPE_oid;
		if (fld2 && tt == TYPE_void) 
			tt = TYPE_oid;
		SERVERfieldAnalysis(fld1, ht, &hval);
		SERVERfieldAnalysis(fld2, tt, &tval);
		if (oht != ht)
			BATseqbase(b, hval.val.oval);
		if (ott != tt)
			BATseqbase(BATmirror(b), tval.val.oval);
		BUNins(b,VALget(&hval),VALget(&tval), FALSE);
	}
	while( mapi_fetch_row(hdl)){
		fld1= mapi_fetch_field(hdl,0);
		fld2= mapi_fetch_field(hdl,1);
		SERVERfieldAnalysis(fld1, ht, &hval);
		SERVERfieldAnalysis(fld2, tt, &tval);
		BUNins(b,VALget(&hval),VALget(&tval), FALSE);
	}
	if (!(b->batDirty&2)) b = BATsetaccess(b, BAT_READ);
	*ret = b->batCacheid;
	BBPkeepref(*ret);

	return err;
}
str
SERVERput(MalBlkPtr mb, MalStkPtr stk, InstrPtr pci){
	int *key;
	str *nme;
	ptr val;
	int i,tpe;
	Mapi mid;
	MapiHdl hdl=0;
	char *w=0, buf[BUFSIZ];

	key= (int*) getArgReference(stk,pci,pci->retc);
	nme= (str*) getArgReference(stk,pci,pci->retc+1);
	val= (ptr) getArgReference(stk,pci,pci->retc+2);
	@:accessTest(*key,put)@
	switch( (tpe=getArgType(mb,pci, pci->retc+2)) ){
	case TYPE_bat:
	case TYPE_ptr:
		throw(MAL, "mserver.glue","Unsupported type");
        case TYPE_str:
                snprintf(buf,BUFSIZ,"%s:=%s;",*nme,*(char**)val);
                break;
	default:
		ATOMformat(tpe,val,&w);
		snprintf(buf,BUFSIZ,"%s:=%s;",*nme,w);
		GDKfree(w);
		break;
	}
	if( SERVERsessions[i].hdl)
		mapi_close_handle(SERVERsessions[i].hdl);
	SERVERsessions[i].hdl= mapi_query(mid, buf);
	@:catchErrors(mserver.put)@
	return MAL_SUCCEED;
}
str
SERVERputLocal(MalBlkPtr mb, MalStkPtr stk, InstrPtr pci){
	str *ret, *nme;
	ptr val;
	int tpe;
	char *w=0, buf[BUFSIZ];

	ret= (str*) getArgReference(stk,pci,0);
	nme= (str*) getArgReference(stk,pci,pci->retc);
	val= (ptr) getArgReference(stk,pci,pci->retc+1);
	switch( (tpe=getArgType(mb,pci, pci->retc+1)) ){
	case TYPE_bat:
	case TYPE_ptr:
		throw(MAL, "mserver.glue","Unsupported type");
        case TYPE_str:
                snprintf(buf,BUFSIZ,"%s:=%s;",*nme,*(char**)val);
                break;
	default:
		ATOMformat(tpe,val,&w);
		snprintf(buf,BUFSIZ,"%s:=%s;",*nme,w);
		GDKfree(w);
		break;
	}
	*ret= GDKstrdup(buf);
	return MAL_SUCCEED;
}
str 
SERVERbindBAT(MalBlkPtr mb, MalStkPtr stk, InstrPtr pci){
	int *key;
	str *nme,*tab,*col;
	int i;
	Mapi mid;
	MapiHdl hdl=0;
	char buf[BUFSIZ];
	key= (int*) getArgReference(stk,pci,pci->retc);
	nme= (str*) getArgReference(stk,pci,pci->retc+1);
	@:accessTest(*key,bind)@
	if( pci->argc == 6) {
		tab= (str*) getArgReference(stk,pci,pci->retc+2);
		col= (str*) getArgReference(stk,pci,pci->retc+3);
		i= *(int*) getArgReference(stk,pci,pci->retc+4);
		snprintf(buf,BUFSIZ,"%s:bat[:void,:%s]:=sql.bind(\"%s\",\"%s\",\"%s\",0,%d);",
			getVarName(mb,getDestVar(pci)),
			getTypeName(getTailType(getVarType(mb,getDestVar(pci)))),
			*nme, *tab,*col,i);
	} else if( pci->argc == 5) {
		tab= (str*) getArgReference(stk,pci,pci->retc+2);
		i= *(int*) getArgReference(stk,pci,pci->retc+3);
		snprintf(buf,BUFSIZ,"%s:bat[:void,:oid]:=sql.bind(\"%s\",\"%s\",0,%d);",
			getVarName(mb,getDestVar(pci)),*nme, *tab,i);
	} else {
		str hn,tn;
		int target= getArgType(mb,pci,0);
		hn= getTypeName(getHeadType(target));
		tn= getTypeName(getTailType(target));
		snprintf(buf,BUFSIZ,"%s:bat[%s,%s]:=bbp.bind(\"%s\");",
			getVarName(mb,getDestVar(pci)), hn,tn, *nme);
		GDKfree(hn);
		GDKfree(tn);
	}
	if( SERVERsessions[i].hdl)
		mapi_close_handle(SERVERsessions[i].hdl);
	SERVERsessions[i].hdl= mapi_query(mid, buf);
	@:catchErrors(mserver.bind)@
	return MAL_SUCCEED;
}
@}