mal_client.mx

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

		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;
@-
The GDK thread administration should be set to reflect use of
the proper IO descriptors.
@c
	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;
} 
@- Client decendants
Forking is a relatively cheap way to create a new client.
The new client record shares the IO descriptors.
To avoid interference, we limit children to only produce 
output by closing the input-side.

If the father itself is a temporary client, let 
the new child depend on the grandfather.
@c
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;
}

@-

When a client needs to be terminated then the file descriptors for
its input/output are simply closed. 
This leads to a graceful degradation, but may take some time
when the client is busy.
A more forcefull method is to kill the client thread, but this
may leave locks and semaphores in an undesirable state.

The routine freeClient ends a single client session,
but through side effects of sharing IO descriptors,
also its children. Conversely, a child can not close a parent.
@{
@c
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;
@-
The threads may not be removed, but should become dormant
@c
	THRdel(c->mythread);
	c->mythread= 0;
	c->mypid= 0;
	/* THRsuspend(c->mythread); still unsafe */
	c->user = oid_nil;
	c->mode = FREECLIENT;
}

@-
If a client disappears from the scene (eof on stream), we should
terminate all its children. This is in principle a forcefull action,
because the children may be ignoring the primary IO streams.
(Instead they may be blocked in an infinite loop)

Special care should be taken by closing the 'adm' thread.
It is permitted to leave only when it is the sole user of the system.

Furthermore, once we enter closeClient, the process in which it is
raised has already lost its file descriptors.
@c
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();
}
@-
Shutting down a server should be limited to console operations
and authorized remote admin calls.
@c
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);
}

@-
Killing a (child) client is a delicate action, because the state of
the client is unknown. The current solution is to issue
a soft-termination request first by setting its mode flag. 
The client main interpreter loop should look at it upon each
cycle and terminate.

After the PROCESSTIMEOUT delay the thread is forcefully killed,
assuming it was either waiting for a lock, or external
event in any way.

The current code is not correct. It should selectively kill
a pthread, rather waiting for them to finish the request being
handled.  Since they may be waiting for IO this can take a
very long time, since we have access to the file descriptor it may
work by closing them.
@c 
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");*/
		}
	}
} 

@-
At the end of the server session all remaining structured are
explicitly released to simplify detection of memory leakage problems.
@c
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;
		}
	}
}
@-
An internet connection may be terminated from the server console.
And killing the administrator this way is prohibited.
@c
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");
}
@-
In embedded mode there can be at most one console client and one 
Mapi connection. Moreover, the Mapi connection should disable the administrator
console.
@c
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;
}
@-
@}
@+ Spying on clients
During system monitoring with multiple clients active it is
often hard to detect who or what is causing a system degradation. 
Therefore a spying scheme for the system administrator is
available. He can enable tracing interactions from the internet connections
to the system console.
@{
@= sysTrace
	if( c->sysmon){
		stream_printf(c->sysmon,"#[%d]",c->user);
		stream_printf(c->sysmon, @1);
		stream_printf(c->sysmon,"\n");
	}
@-
@c
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;
		}
}
@}
@+ Client input
Input to be processed is collected in a Client specific buffer.
It is filled by reading information from a stream, a terminal,
or by scheduling strings constructed internally. 
The latter involves removing any escape character needed to 
manipulate the string within the kernel.
The buffer space is automatically expanded to accommodate new 
information and the read pointers are adjusted.
@-
The input is read from a (blocked) stream and stored in the client record
input buffer. The storage area grows automatically upon need.
The origin of the input stream depends on the connectivity mode.

Most interactions should be regulated through the Mclient
front-end. This will also take care of buffering a request before
submission. This avoids a significant number of network interactions.

@{
@-
Each operation received from a front-end consists of at least one line.
To simplify misaligned communication with front-ends, we use different
prompts structures. [think, can we avoid acks]

@-
The default action is to read information from an ascii-stream
one line at a time. This is the preferred mode for reading
from terminal.
@-
The next statement block is to be read. Send a prompt to warn
the front-end to issue the request.
@= sendPrompt
	if(c->promptlength>=0) { 
		if (!isa_block_stream(c->fdout))
			stream_write(c->fdout,c->prompt,c->promptlength,1); 
		stream_flush(c->fdout);
	}
@c

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)) {
			@:sendPrompt@
			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
		@:sysTrace(CURRENT(c))@
	}
	return 1;
}
@}