mal_client.c

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

#line 273 "/export/scratch0/monet/monet.GNU.64.64.d.14791/MonetDB5/src/mal/mal_client.mx"
#include "mal_config.h"
#include "mal_client.h"
#include "mal_readline.h"
#include "mal_import.h"
#include "mal_parser.h"
#include "mal_namespace.h"

/* This should be in src/mal/mal.h, as the function is implemented in 
 * src/mal/mal.c; however, it cannot, as "Client" isn't known there ... |-( 
 * For now, we move the prototype here, as it it only used here.
 * Maybe, we should concider also moving the implementation here...

int streamClient(Client c, str prompt);
int bstreamClient(Client c, str prompt);
 */

static void     freeClient  (Client c);

ClientRec   mal_clients[MAL_MAXCLIENTS+1];

void THRsuspend(Thread h){
	(void) h;
	/* printf("suspend Thread %d %d\n", h->tid, (int) h->pid);*/
}
void THRawake(Thread h){
	(void) h;
	/* printf("awake Thread %d %d\n", h->tid, (int) h->pid);*/
}

bstream *MCgetConsole(Client c){
	ClientInput *x;
	if( c->bak == 0) return c->fdin;
	x= c->bak;
	while(x->next) x= x->next;
	return x->fdin;
	/* return c->console;*/
}
void MCpushClientInput(Client c, bstream *new_input, int listing, char *prompt)
{
	ClientInput *x = (ClientInput*)GDKmalloc(sizeof(ClientInput));
	x->fdin = c->fdin;
	x->yycur = c->yycur;
	x->listing = c->listing;
	x->prompt = c->prompt;
	x->next = c->bak;
	c->bak = x;
	c->fdin = new_input;
	c->listing = listing;
	c->prompt = GDKstrdup(prompt);
	c->promptlength = strlen(c->prompt);
	c->yycur = 0;
}

void MCpopClientInput(Client c)
{
	ClientInput *x = c->bak;
	if (c->fdin) {
		/* missing protection against closing stdin stream */
		(void) stream_close(c->fdin->s);
		(void) stream_destroy(c->fdin->s);
		(void) bstream_destroy(c->fdin);
	}
	GDKfree(c->prompt);
	c->fdin = x->fdin;
	c->yycur = x->yycur;
	c->listing = x->listing;
	c->prompt = x->prompt;
	c->promptlength = strlen(c->prompt);
	c->bak = x->next;
	GDKfree(x);
}

Client MCnewClient(){
	Client  c;
	if( mal_clients[CONSOLE].user)
		GDKprotect();
	mal_set_lock(mal_contextLock, "newClient");
	if( mal_clients[CONSOLE].user &&
			mal_clients[CONSOLE].mode == FINISHING){
		showException(MAL,"newClient", "system shutdown in progress");
		mal_unset_lock(mal_contextLock, "newClient");
		return NULL;
	}
	for(c = mal_clients; c < mal_clients+MAL_MAXCLIENTS; c++) {
		if (c->mode == FREECLIENT) {
			c->mode= CLAIMED;
			break;
		}
	}
	mal_unset_lock(mal_contextLock, "newClient");

	if (c == mal_clients+MAL_MAXCLIENTS) 
		return NULL;
	c->idx= c - mal_clients;
#ifdef MAL_CLIENT_DEBUG
	printf("New client created %d\n",c-mal_clients);
#endif
	return c;
}


#line 379 "/export/scratch0/monet/monet.GNU.64.64.d.14791/MonetDB5/src/mal/mal_client.mx"
Client MCgetClient()
{
	Client c;
	size_t pid= MT_getpid();
	Thread tid;

	for(c = mal_clients+1; c < mal_clients+MAL_MAXCLIENTS; c++) 
		if( c->mypid == pid) return c;
	if( mal_clients->mypid== pid) return mal_clients;

	tid= THRget(THRgettid());
	for(c = mal_clients; c < mal_clients+MAL_MAXCLIENTS; c++) 
		if( c->mythread == tid) return c;

	for(c = mal_clients; c < mal_clients+MAL_MAXCLIENTS; c++) 
		if( c->mythread) stream_printf(c->fdout,"client %d\n",c->mythread);
	showException(MAL, "getClient", "unexpected call to getClient %d", tid);
	return NULL;
}

Client MCinitClient(oid user, bstream *fin, stream *fout) {
	Client c;
	str prompt;

	if ((c = MCnewClient()) == NULL)
		return NULL;

	c->user = user; 
	c->scenario = NULL;
	c->oldscenario = NULL;
	c->srcFile = NULL;
	c->sessionkey = 0;
	c->blkmode = 0;

	c->fdin = fin ? fin : bstream_create(GDKin,0);
	c->console = c->fdin;
	c->yycur = 0;
	c->bak = NULL;

	c->listing = 0;
	c->fdout = fout ? fout : GDKstdout;
	c->sysmon = mal_clients[0].sysmon ? mal_clients[0].fdout : 0;
	c->journal = NULL;
	c->history = 0;


#line 427 "/export/scratch0/monet/monet.GNU.64.64.d.14791/MonetDB5/src/mal/mal_client.mx"
	c->output = 0;
	if(c->nspace==0)
		c->nspace = newModule(NULL, putName("guest",5));
	c->curprg = c->backup = 0;
	c->glb = 0;

	c->father = NULL;
	c->login = time(0); c->delay= TIMEOUT;
	c->logout= c->login+c->delay;
	c->mode = AWAITING;
	c->itrace = 0;
	c->debugOptimizer = c->debugScheduler= 0;
	c->timer = 0;
	c->errbuf = 0;

	c->cwd= GDKstrdup(monet_cwd);
	prompt = !fin? GDKgetenv("monet_prompt"): PROMPT1;
	c->prompt= GDKstrdup(prompt);
	c->promptlength= strlen(prompt);
	if( isAdministrator(c)) MCinitClientThread(c);
	return c;
}


#line 454 "/export/scratch0/monet/monet.GNU.64.64.d.14791/MonetDB5/src/mal/mal_client.mx"
void MCinitClientThread(Client c)
{
	Thread t;
	char cname[11 + 1];
/*	still unsafe, race conditions, garbage left over??
	if( c->mythread){
		t= c->mythread;
	} else
*/
		snprintf(cname, 11, SZFMT, (size_t)(c->user));
		cname[11] = '\0';
		c->mythread = t=  THRnew(c->mypid=MT_getpid(),cname);
	c->errbuf = GDKerrbuf;
	if (c->errbuf == NULL) {
		GDKsetbuf( GDKmalloc(GDKMAXERRLEN));
		c->errbuf = GDKerrbuf;
	}
	c->errbuf[0]=0;


#line 476 "/export/scratch0/monet/monet.GNU.64.64.d.14791/MonetDB5/src/mal/mal_client.mx"
	if ( t==0) {
		showException(MAL, "initClientThread", "Failed to initialize client");
		MPresetProfiler(c->fdout);
		assert(c->bak==NULL);
		if( c->fdin){
			/* missing protection against closing stdin stream */
			(void) stream_close(c->fdin->s);
			(void) stream_destroy(c->fdin->s);
			(void) bstream_destroy(c->fdin);
		}
		if( c->fdout && c->fdout != GDKstdout) {
			(void) stream_close(c->fdout);
			(void) stream_destroy(c->fdout);
		}
		/* if( c->socket) close(c->socket);*/
		return ;
	}
	t->data[1] = c->fdin;
	t->data[0] = c->fdout;
} 


#line 505 "/export/scratch0/monet/monet.GNU.64.64.d.14791/MonetDB5/src/mal/mal_client.mx"
Client MCforkClient(Client father){
	Client son = NULL;
	if( father == NULL) return NULL;
	if (father->father != NULL) father = father->father;
	if ((son = MCinitClient(father->user, father->fdin, father->fdout))) {
		son->fdin= NULL;
		son->bak= NULL;
		son->yycur=0;
		son->father = father;
		son->scenario = father->scenario;
		son->prompt = GDKstrdup(father->prompt);
		son->promptlength= strlen(father->prompt);
		/* reuse the scopes wherever possible */
		son->nspace->name= GDKstrdup("Client-child");
		son->nspace->outer = mal_scope;
	}
	return son;
}



#line 538 "/export/scratch0/monet/monet.GNU.64.64.d.14791/MonetDB5/src/mal/mal_client.mx"
void freeClient(Client c)
{
	c->mode = FINISHING;

	/* epilogue actions have been done already */
	if (c->father == NULL) { /* normal client */
		if( c->fdout && c->fdout != GDKstdout){
			MPresetProfiler(c->fdout);
			(void) stream_close(c->fdout);
			(void) stream_destroy(c->fdout);
		}
		assert(c->bak==NULL);
		if(c->fdin){
			/* missing protection against closing stdin stream */
			(void) stream_close(c->fdin->s);
			(void) stream_destroy(c->fdin->s);
			(void) bstream_destroy(c->fdin);
		}
		c->fdout = NULL;
		c->fdin= NULL;
	} 
	/* forked client also */

	/* scope list and curprg can not be removed, 
	   because the client may reside in a
	   quit() command. Therefore the scopelist is re-used.
	if( c->curprg ) {
		freeSymbol(c->curprg); 
		c->curprg=0;
	}
	if( c->nspace) {
		freeModule(c->nspace);
		c->nspace=0;
	}
	   */
	/* re-use output buffer */
	c->scenario = NULL;
	if(c->prompt) GDKfree(c->prompt); c->prompt = NULL;
	if(c->cwd) GDKfree(c->cwd); c->cwd = NULL;
	c->promptlength=-1;
	if(c->errbuf){
		GDKsetbuf(0);
		GDKfree(c->errbuf);
		c->errbuf=0;
	}
	c->father = 0;
	c->login = c->delay = c->logout = 0;


#line 588 "/export/scratch0/monet/monet.GNU.64.64.d.14791/MonetDB5/src/mal/mal_client.mx"
	THRdel(c->mythread);
	c->mythread= 0;
	c->mypid= 0;
	/* THRsuspend(c->mythread); still unsafe */
	c->user = oid_nil;
	c->mode = FREECLIENT;
}



#line 608 "/export/scratch0/monet/monet.GNU.64.64.d.14791/MonetDB5/src/mal/mal_client.mx"
void MCcloseClient(Client c) {
	Client k;

#ifdef MAL_DEBUG_CLIENT
	printf("closeClient %d %d\n",c-mal_clients,c->user);
#endif
	/* kill living background clients */
	for (k = mal_clients; k < mal_clients+MAL_MAXCLIENTS; k++) {
		if ((k->mode > FINISHING) && (k->father == c)) 
			MCkillClient(k, PROCESSTIMEOUT);
	}

	/* free resources of a single thread */
	if( !isAdministrator(c)) {
		freeClient(c);
		return;
	} 

	/* adm is set to disallow new clients entering */
	mal_clients[CONSOLE].mode= FINISHING;

	for (k = mal_clients+1; k < mal_clients+MAL_MAXCLIENTS; k++) 
		if ((k->mode > FINISHING) ){
			showException(MAL,"closeClient", "client '%d' is still active", k->user);
			MCkillClient(k, PROCESSTIMEOUT);
		}
	MCkillClient(CONSOLE, PROCESSTIMEOUT);
	mal_exit();
}


#line 641 "/export/scratch0/monet/monet.GNU.64.64.d.14791/MonetDB5/src/mal/mal_client.mx"
void MCshutdown(Client c, int timeout){
	Client k;

	for (k = mal_clients; k < mal_clients+MAL_MAXCLIENTS; k++) {
		if ((k->mode > FINISHING) && (k != c)) 
			MCkillClient(k,timeout);
	}
	MCkillClient(c,timeout);
}



#line 668 "/export/scratch0/monet/monet.GNU.64.64.d.14791/MonetDB5/src/mal/mal_client.mx"
void MCkillClient(Client c, int timeout) {
	if (c && c->user != oid_nil ) {

#ifdef MAL_DEBUG_CLIENT
		printf("killClient %d\n", c - mal_clients);
#endif	/* MAL_DEBUG_CLIENT */

		/* GDKwarning("killClient:forcing client to stop ...\n");*/
		/* the process may hang upon gaining access to the file descriptors */ 
		c->mode = FINISHING;
		MPresetProfiler(c->fdout);
		if (c == mal_clients) {
			/* die the hard way */
			c->curprg->def->stop = 0;
		} else if (c->mythread) {
			(void) stream_close(c->fdout);
			(void) stream_destroy(c->fdout);
			c->fdout = NULL;
			(void) stream_close(c->fdin->s);
			(void) stream_destroy(c->fdin->s);
			(void) bstream_destroy(c->fdin);
			c->fdin = NULL;
			assert(c->bak == NULL);
			/* GDKwarning("killClient:about to kill process ...\n");*/
			MT_sleep_ms(1000 * timeout);
			MT_kill_thread(c->mythread->pid);
			MT_sleep_ms(1000 * timeout);
			/*GDKwarning("killClient:process killed\n");*/
		}
	}
} 



#line 704 "/export/scratch0/monet/monet.GNU.64.64.d.14791/MonetDB5/src/mal/mal_client.mx"
void MCcleanupClients(){
	Client c;
	for(c = mal_clients; c < mal_clients+MAL_MAXCLIENTS; c++) {
		/* if( c->nspace){ freeModuleList(c->nspace); c->nspace=0;}*/
		if( c->cwd){ 
			GDKfree(c->cwd); 
			c->cwd = NULL;
		}
		if( c->prompt){ 
			GDKfree(c->prompt); 
			c->prompt = NULL;
		}
		c->user = oid_nil;
		assert(c->bak==NULL);
		if( c->fdin){
			(void) stream_close(c->fdin->s);
			(void) stream_destroy(c->fdin->s); 
			(void) bstream_destroy(c->fdin);
			c->fdin= NULL;
		}
		if( c->fdout && c->fdout != GDKstdout){
			MPresetProfiler(c->fdout);
			(void) stream_destroy(c->fdout); 
			c->fdout= NULL;
		}
	}
}


#line 735 "/export/scratch0/monet/monet.GNU.64.64.d.14791/MonetDB5/src/mal/mal_client.mx"
str MCstopClientIndex(Client c, int id)
{
	if( ! isAdministrator(c))
		throw(PERMD, "stop","Only permitted from server console");
	if( id<=0 || id>=MAL_MAXCLIENTS || mal_clients[id].mode==FREECLIENT)
		throw(MAL, "stop","Illegal client index");
	MCkillClient(mal_clients+id, PROCESSTIMEOUT);
	return MAL_SUCCEED;
}
str MCstopClient(Client c, oid which){
	if( ! isAdministrator(c))
		throw(PERMD, "stop","Only permitted from server console");

	mal_set_lock(mal_contextLock, "stopClient");
	for(c = mal_clients; c < mal_clients+MAL_MAXCLIENTS; c++) {
		if (c->mode > FREECLIENT && c->user == which) {
			MCkillClient(c, PROCESSTIMEOUT);
			mal_unset_lock(mal_contextLock, "stopClient");
			return MAL_SUCCEED;
		}
	}
	mal_unset_lock(mal_contextLock, "stopClient");
	throw(ILLARG, "stop","Illegal client name");
}


#line 764 "/export/scratch0/monet/monet.GNU.64.64.d.14791/MonetDB5/src/mal/mal_client.mx"
int MCcountClients(){
	int cnt=0;
	Client c;
	for(c = mal_clients; c < mal_clients+MAL_MAXCLIENTS; c++) 
		if (c->mode != FREECLIENT) cnt++;
	return cnt;
}
int MCrunEmbedded(Client c){
	(void) c;
	/* to be defined */
	return 0;
}


#line 793 "/export/scratch0/monet/monet.GNU.64.64.d.14791/MonetDB5/src/mal/mal_client.mx"
void MCtraceClient(oid which, int flag){
	Client c;
	for(c = mal_clients; c < mal_clients+MAL_MAXCLIENTS; c++) 
		if( c->user == which ){
#ifdef MAL_CLIENT_DEBUG
			printf("trace client %d\n",c-mal_clients);
#endif
			if( flag) c->sysmon= mal_clients[0].fdout;
			else c->sysmon= NULL;
		}
}
void MCtraceAllClients(int flag){
	Client c;
	for(c = mal_clients; c < mal_clients+MAL_MAXCLIENTS; c++) 
		if( c->user != oid_nil ){
			if( flag) c->sysmon= mal_clients[0].fdout;
			else c->sysmon= NULL;
		}
}


#line 813 "/export/scratch0/monet/monet.GNU.64.64.d.14791/MonetDB5/src/mal/mal_client.mx"


#line 850 "/export/scratch0/monet/monet.GNU.64.64.d.14791/MonetDB5/src/mal/mal_client.mx"

int MCreadClient(Client c){
	bstream *in = c->fdin;

#ifdef MAL_CLIENT_DEBUG
	printf("# streamClient %d\n",c->user);
#endif

	while (in->pos < in->len && 
			(isspace((int)(in->buf[in->pos])) ||
			 in->buf[in->pos] == ';' || !in->buf[in->pos]))
		in->pos++;

	if (in->pos >= in->len) {
		ssize_t rd, sum = 0;
		assert(in->pos == in->len);

		if (in->eof || !isa_block_stream(in->s)) {
			
#line 844 "/export/scratch0/monet/monet.GNU.64.64.d.14791/MonetDB5/src/mal/mal_client.mx"
	if(c->promptlength>=0) { 
		if (!isa_block_stream(c->fdout))
			stream_write(c->fdout,c->prompt,c->promptlength,1); 
		stream_flush(c->fdout);
	}


#line 868 "/export/scratch0/monet/monet.GNU.64.64.d.14791/MonetDB5/src/mal/mal_client.mx"

			in->eof = 0;
		}
		while ((rd = bstream_next(in)) > 0 && !in->eof) {
			sum += rd;
			if (!in->mode)/* read one line at a time in line mode */
				break;
		}
		if (sum == 0 && in->eof && isa_block_stream(in->s)) {
			/* we hadn't seen the EOF before, so just try again
			   (this time with prompt) */
			return MCreadClient(c);
		}
#ifdef MAL_CLIENT_DEBUG
		printf("# simple stream received:");
#endif
	}
	if (in->pos >= in->len) {
		/* end of stream reached */
		if (c->bak) {
			MCpopClientInput(c);
			if( c->fdin == NULL)
				return 0;
			return MCreadClient(c);
		}
		return 0;
	}
	if( *CURRENT(c) == '<'){
		bstream *fdin;
		str nme= GDKstrdup(CURRENT(c)+1);
		str err;

		nme[strlen(CURRENT(c)+1)]=0;
		printf("# loading from file `%s`\n",nme);
		if ((err = malLoadScript(nme,&fdin)) != MAL_SUCCEED) {
			fprintf(stderr, "!%s\n", err);
			GDKfree(err);
			return 0;
		}
		GDKfree(nme);
		MCpushClientInput(c, fdin, c->listing, c->prompt);
		in->pos= in->len;
		return MCreadClient(c);
	} else 
	if( *CURRENT(c) == '?'){
		showHelp(c->nspace, CURRENT(c)+1, c->fdout);
		in->pos= in->len;
		return MCreadClient(c);
	} else {
#ifdef MAL_CLIENT_DEBUG
		printf("#finished\n");
#endif
		
#line 786 "/export/scratch0/monet/monet.GNU.64.64.d.14791/MonetDB5/src/mal/mal_client.mx"
	if( c->sysmon){
		stream_printf(c->sysmon,"#[%d]",c->user);
		stream_printf(c->sysmon, CURRENT(c));
		stream_printf(c->sysmon,"\n");
	}


#line 920 "/export/scratch0/monet/monet.GNU.64.64.d.14791/MonetDB5/src/mal/mal_client.mx"

	}
	return 1;
}


#line 925 "/export/scratch0/monet/monet.GNU.64.64.d.14791/MonetDB5/src/mal/mal_client.mx"