mserver.c

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

	THRprintf(GDKout, "#SERVERlistenSSL:Network started at %d\n", port);
#endif
	do {
		int retval;
		struct timeval tv;
		fd_set fds;

		FD_ZERO(&fds);
		FD_SET(sock, &fds);
		/* Wait up to 0.5 seconds. */
		tv.tv_sec = 0;
		tv.tv_usec = 500;

		retval = select(sock + 1, &fds, &fds, &fds, &tv);
		if (retval == 0) {
			/* nothing interesting has happened */
			continue;
		}
		if (retval < 0) {
			GDKsyserror("SERVERlistenSSL:select failed\n");
			goto fail;
		}
		if ((msgsock = accept(sock, (SOCKPTR) 0, (SOCKLEN *) 0)) < 0) {
			if (MT_geterrno() != EINTR) {
				msg = "Accept failed";
				goto fail;
			}
			continue;
		}
		if ((ssl = SSL_new(ctx)) == 0) {
			msg = "SSL_new failed";
			goto fail;
		}

		if (!SSL_set_fd(ssl, msgsock)) {
			msg = "SSL_set_fd failed";
			goto fail;
		}

		for (;;) {
			int ret, err;
			char *errstr;

			ret = SSL_accept(ssl);
			err = SSL_get_error(ssl, ret);
			switch (err) {
			case SSL_ERROR_WANT_READ:
			case SSL_ERROR_WANT_WRITE:
				/* try again */
				continue;
			case SSL_ERROR_NONE:
				/* successful connect */
				break;
			case SSL_ERROR_WANT_CONNECT:
			case SSL_ERROR_WANT_ACCEPT:
			case SSL_ERROR_WANT_X509_LOOKUP:
			default:
				/* some error occurred */
				errstr = ssl_error("SERVERlistenSSL", err, ret);
				GDKsyserror(errstr);
				GDKfree(errstr);
				SSL_free(ssl);
				close(msgsock);
				goto end_loop;
			}
			break;
		}

		doChallenge(cmd, msgsock);
end_loop:;
	} while (1);
	SSL_CTX_free(ctx);
	close(sock);
	if (cmd)
		GDKfree(cmd);
	SERVERannounce(server.sin_addr, port, 1);
	return MAL_SUCCEED;
fail:;
	if (sock >= 0)
		close(sock);
	if (ctx)
		SSL_CTX_free(ctx);
	if (cmd)
		GDKfree(cmd);
	throw(IO, "mserver.listenSSL", msg);
#else
	(void) Port;
	(void) Maxusers;
	(void) keyfile;
	(void) certfile;
	(void) cmd;
	throw(MAL, "mserver.listenSSL", "No SSL support");
#endif /* HAVE_OPENSSL */
}



#line 1035 "/export/scratch0/monet/monet.GNU.64.64.d.14791/MonetDB5/src/modules/mal/mserver.mx"
str
SERVERlisten_default(int *ret)
{
	int port = SERVERPORT;
	str p;
	int maxusers = SERVERMAXUSERS;

	(void) ret;
	p = GDKgetenv("mapi_port");
	if (p)
		port = (int) strtol(p, NULL, 10);
	return SERVERlisten(&port, 0, &maxusers, 0);
}

str
SERVERlisten2(int *ret, int *port, int *maxusers)
{
	(void) ret;
	return SERVERlisten(port, 0, maxusers, 0);
}

str
SERVERlisten3(int *ret, int *port, int *maxusers, str *cmd)
{
	(void) ret;
	return SERVERlisten(port, 0, maxusers, cmd);
}

str
SERVERlisten_port(int *ret, int *pid)
{
	int port = *pid;
	int maxusers = SERVERMAXUSERS;

	(void) ret;
	return SERVERlisten(&port, 0, &maxusers, 0);
}


#line 1080 "/export/scratch0/monet/monet.GNU.64.64.d.14791/MonetDB5/src/modules/mal/mserver.mx"

str
SERVERstop(int *ret)
{
	int i;
	for( i=0; i< lastlistener; i++)
		MT_kill_thread(listener[i]);
	lastlistener = 0;
	(void) ret;		/* fool compiler */
	return MAL_SUCCEED;
}


str
SERVERsuspend(int *res)
{
	(void) res;
	serveractive= FALSE;
	return MAL_SUCCEED;
}

str
SERVERresume(int *res)
{
	serveractive= TRUE;
	(void) res;
	return MAL_SUCCEED;
}

str
SERVERclient(int *res, stream **In, stream **Out) 
{
	int i = 0;
	char *name = (char *) GDKmalloc(BLOCK), *user = name, *s;
	char *challenge;
	int blocked = 0;
	bstream *bs;
	stream *fdin = *In, *fdout = *Out;

	(void) res;
	assert(name);
	memset(name, 0, BLOCK);

	/* is this routine used?  Where? */
	assert(0);

	/* write challenge string, the challenge is currently empty (disabled) */
	challenge = "::mserver_mal:4\n";

	/* temporarily misuse name */
	snprintf(name, BLOCK, "%2u%s", (unsigned int) strlen(challenge), challenge);
	stream_write(fdout, name, strlen(name), 1);
	/* get user name */
	*user = 0;
	stream_read(fdin, user, 1, 1);
	for (i = 0; !stream_errnr(fdin) && i < BLOCK && *user != '\n'; i++) {
		user++;
		stream_read(fdin, user, 1, 1);
	}
	if (stream_errnr(fdin)) {
		throw(IO, "SERVERclient", "fdin write problems");
	}
	*user = 0;
	/* in embedded mode we allow just one client */
#ifdef DEBUG_SERVER
	printf("mserver:Client accepted %s\n", name);
	fflush(stdout);
#endif
	s = strrchr(name, ':');
	blocked = s && strncmp(s, ":blocked", 8) == 0;
	if (blocked) {
		/* printf("mserver:use blocked mode\n");fflush(stdout); */
		*s = 0;
		fdin = block_stream(stream_rstream(fdin));
		fdout = block_stream(stream_wstream(fdout));
	}
#ifdef DEBUG_SERVER
	THRprintf(GDKout, "#SERVERlisten:client accepted %d\n", msgsock);
	THRprintf(GDKout, "#SERVERlisten:client string %s\n", name);
#endif
	bs = bstream_create(fdin, blocked ? 128 * BLOCK : 0);

	if (blocked)
		bs->eof = 1;
	MSscheduleClient(name, challenge, bs, fdout);
	return MAL_SUCCEED;
}

void
SERVERexit(){
	int ret;
	SERVERstop(&ret);
	/* remove any port identity file */
	system("rm -rf .*_port");
}


#line 1226 "/export/scratch0/monet/monet.GNU.64.64.d.14791/MonetDB5/src/modules/mal/mserver.mx"
#define MAXSESSIONS 32
struct{
	int key;
	str dbalias;	/* logical name of the session */
	Client c;
	Mapi mid;		/* communication channel */
	MapiHdl hdl;	/* result set handle */
} SERVERsessions[MAXSESSIONS];

static int sessionkey=0;

/* #define MAPI_TEST*/

str
SERVERconnectAll(int *key, str *host, int *port, str *username, str *password, str *lang, int ssl){
	Mapi mid;
	int i;

	mal_set_lock(mal_contextLock,"SERVERconnect");
	for(i=1; i< MAXSESSIONS; i++)
	if( SERVERsessions[i].c ==0 ) break;

	if( i==MAXSESSIONS){
		mal_unset_lock(mal_contextLock,"SERVERconnect");
		throw(IO, "mserver.connect", "Too many sessions");
	}
	SERVERsessions[i].c= MCgetClient();
	SERVERsessions[i].key= ++sessionkey;
	mal_unset_lock(mal_contextLock,"SERVERconnect");

	if (ssl)
		mid = mapi_connect_ssl(*host, *port, *username, *password, *lang, NULL);
	else
		mid = mapi_connect(*host, *port, *username, *password, *lang, NULL);

	if (mapi_error(mid)) {
		str err = mapi_error_str(mid);
		if (err == NULL)
			err = "(no reason given)";
		if (err[0] == '!')
			err = err + 1;
		SERVERsessions[i].c = NULL;
		throw(IO, "mserver.connect", "Could not connect: %s", err);
	}

#ifdef MAPI_TEST
	stream_printf(GDKout,"Succeeded to establish session\n");
#endif
	SERVERsessions[i].mid= mid;
	*key = SERVERsessions[i].key;
	return MAL_SUCCEED;
}

str
SERVERdisconnectALL(int *key){
	int i;

	mal_set_lock(mal_contextLock,"SERVERdisconnect");

	for(i=1; i< MAXSESSIONS; i++)
		if( SERVERsessions[i].c != 0 ) {
			SERVERsessions[i].c = NULL;
			*key = SERVERsessions[i].key;
			mapi_disconnect(SERVERsessions[i].mid);
		}

	mal_unset_lock(mal_contextLock,"SERVERdisconnect");

#ifdef MAPI_TEST
	stream_printf(GDKout,"Succeeded to close all session\n");
#endif
	return MAL_SUCCEED;
}

str
SERVERdisconnectWithAlias(int *key, str *dbalias){
	int i;

	mal_set_lock(mal_contextLock,"SERVERdisconnectWithAlias");

	for(i=0; i<MAXSESSIONS; i++)
		 if( SERVERsessions[i].dbalias &&
			 strcmp(SERVERsessions[i].dbalias, *dbalias)==0){
				SERVERsessions[i].c = NULL;
				*key = SERVERsessions[i].key;
				mapi_disconnect(SERVERsessions[i].mid);
				break;
		}

	if( i==MAXSESSIONS){
		mal_unset_lock(mal_contextLock,"SERVERdisconnectWithAlias");
		throw(IO, "mserver.disconnect", "Impossible to close session for db_alias: '%s'", dbalias);
	}

	mal_unset_lock(mal_contextLock,"SERVERdisconnectWithAlias");

#ifdef MAPI_TEST
	stream_printf(GDKout,"Succeeded to close all session\n");
#endif

	return MAL_SUCCEED;
}

str
SERVERconnect(int *key, str *host, int *port, str *username, str *password, str *lang){
	return SERVERconnectAll(key,host,port,username,password,lang, FALSE);
}
str
SERVERconnectssl(int *key, str *host, int *port, str *username, str *password, str *lang){
	return SERVERconnectAll(key,host,port,username,password,lang, TRUE);
}


str
SERVERreconnectAlias(int *key, str *host, int *port, str  * dbalias, str *username, str *password, str *lang){
	int i;
	str msg=MAL_SUCCEED;

	for(i=0; i<MAXSESSIONS; i++)
	 if( SERVERsessions[i].key && 
		 SERVERsessions[i].dbalias &&
		 strcmp(SERVERsessions[i].dbalias, *dbalias)==0){
			*key = SERVERsessions[i].key;
			return msg;
	}

	msg= SERVERconnectAll(key, host, port, username, password, lang, FALSE);
	if( msg == MAL_SUCCEED)
		msg = SERVERsetAlias(&i, key, dbalias);
	return msg;
}
str
SERVERreconnectWithoutAlias(int *key,str *host, int *port, str *username, str *password, str *lang){
	int i;
	str msg=MAL_SUCCEED, nme= "anonymous";

	for(i=0; i<MAXSESSIONS; i++)
	 if( SERVERsessions[i].key ){
			*key = SERVERsessions[i].key;
			return msg;
	}

	msg= SERVERconnectAll(key, host, port, username, password, lang, FALSE);
	if( msg == MAL_SUCCEED)
		msg = SERVERsetAlias(&i, key, &nme);
	return msg;
}


#line 1384 "/export/scratch0/monet/monet.GNU.64.64.d.14791/MonetDB5/src/modules/mal/mserver.mx"
str
SERVERsetAlias(int *ret, int *key, str *dbalias){
	int i;
	Mapi mid;
	
#line 1375 "/export/scratch0/monet/monet.GNU.64.64.d.14791/MonetDB5/src/modules/mal/mserver.mx"
 for(i=0; i< MAXSESSIONS; i++)
 if( SERVERsessions[i].c &&
     SERVERsessions[i].key== *key ) break;
 if( i== MAXSESSIONS)
	throw(MAL, "mserver.setAlias","Access violation,"
		" could not find matching session descriptor");
  mid= SERVERsessions[i].mid;
  (void) mid; /* silence compilers */


#line 1388 "/export/scratch0/monet/monet.GNU.64.64.d.14791/MonetDB5/src/modules/mal/mserver.mx"

    SERVERsessions[i].dbalias= GDKstrdup(*dbalias);
	*ret = 0;
	return MAL_SUCCEED;
}
str
SERVERlookup(int *ret, str *dbalias)
{
	int i;
	for(i=0; i< MAXSESSIONS; i++)
	if( SERVERsessions[i].dbalias &&
		strcmp(SERVERsessions[i].dbalias, *dbalias)==0){
		*ret= SERVERsessions[i].key; 
		return MAL_SUCCEED;
	}
	throw(MAL, "mserver.lookup", "Could not find database connection");
}
str
SERVERtrace(int *ret, int *key, int *flag){
	(void )ret;
	mapi_trace(SERVERsessions[*key].mid,*flag);
	return MAL_SUCCEED;
}
str
SERVERdisconnect(int *ret, int *key){
	int i;
	Mapi mid;
	
#line 1375 "/export/scratch0/monet/monet.GNU.64.64.d.14791/MonetDB5/src/modules/mal/mserver.mx"
 for(i=0; i< MAXSESSIONS; i++)
 if( SERVERsessions[i].c &&
     SERVERsessions[i].key== *key ) break;
 if( i== MAXSESSIONS)
	throw(MAL, "mserver.disconnect","Access violation,"
		" could not find matching session descriptor");
  mid= SERVERsessions[i].mid;
  (void) mid; /* silence compilers */


#line 1415 "/export/scratch0/monet/monet.GNU.64.64.d.14791/MonetDB5/src/modules/mal/mserver.mx"

	mapi_disconnect(mid);
	if( SERVERsessions[i].dbalias)
		GDKfree(SERVERsessions[i].dbalias);
	SERVERsessions[i].dbalias= 0;
	*ret = 0;
	return MAL_SUCCEED;
}
str
SERVERdestroy(int *ret, int *key){
	int i;
	Mapi mid;
	
#line 1375 "/export/scratch0/monet/monet.GNU.64.64.d.14791/MonetDB5/src/modules/mal/mserver.mx"
 for(i=0; i< MAXSESSIONS; i++)
 if( SERVERsessions[i].c &&
     SERVERsessions[i].key== *key ) break;
 if( i== MAXSESSIONS)
	throw(MAL, "mserver.destroy","Access violation,"
		" could not find matching session descriptor");
  mid= SERVERsessions[i].mid;
  (void) mid; /* silence compilers */


#line 1427 "/export/scratch0/monet/monet.GNU.64.64.d.14791/MonetDB5/src/modules/mal/mserver.mx"

	mapi_destroy(mid);
	SERVERsessions[i].c= 0;
	if( SERVERsessions[i].dbalias)
		GDKfree(SERVERsessions[i].dbalias);
	SERVERsessions[i].dbalias= 0;
	*ret = 0;
	return MAL_SUCCEED;
}

str
SERVERreconnect(int *ret, int *key){
	int i;
	Mapi mid;
	
#line 1375 "/export/scratch0/monet/monet.GNU.64.64.d.14791/MonetDB5/src/modules/mal/mserver.mx"
 for(i=0; i< MAXSESSIONS; i++)
 if( SERVERsessions[i].c &&
     SERVERsessions[i].key== *key ) break;
 if( i== MAXSESSIONS)
	throw(MAL, "mserver.destroy","Access violation,"
		" could not find matching session descriptor");
  mid= SERVERsessions[i].mid;
  (void) mid; /* silence compilers */


#line 1441 "/export/scratch0/monet/monet.GNU.64.64.d.14791/MonetDB5/src/modules/mal/mserver.mx"

	mapi_reconnect(mid);
	*ret = 0;
	return MAL_SUCCEED;
}
str
SERVERping(int *ret, int *key){
	int i;
	Mapi mid;
	
#line 1375 "/export/scratch0/monet/monet.GNU.64.64.d.14791/MonetDB5/src/modules/mal/mserver.mx"
 for(i=0; i< MAXSESSIONS; i++)
 if( SERVERsessions[i].c &&
     SERVERsessions[i].key== *key ) break;
 if( i== MAXSESSIONS)
	throw(MAL, "mserver.destroy","Access violation,"
		" could not find matching session descriptor");
  mid= SERVERsessions[i].mid;
  (void) mid; /* silence compilers */


#line 1450 "/export/scratch0/monet/monet.GNU.64.64.d.14791/MonetDB5/src/modules/mal/mserver.mx"

	*ret= mapi_ping(mid);
	return MAL_SUCCEED;
}
str 
SERVERquery(int *ret, int *key, str *qry){
	Mapi mid;
	MapiHdl hdl=0;
	int i;
	
#line 1375 "/export/scratch0/monet/monet.GNU.64.64.d.14791/MonetDB5/src/modules/mal/mserver.mx"
 for(i=0; i< MAXSESSIONS; i++)
 if( SERVERsessions[i].c &&
     SERVERsessions[i].key== *key ) break;
 if( i== MAXSESSIONS)
	throw(MAL, "mserver.query","Access violation,"
		" could not find matching session descriptor");
  mid= SERVERsessions[i].mid;
  (void) mid; /* silence compilers */


#line 1459 "/export/scratch0/monet/monet.GNU.64.64.d.14791/MonetDB5/src/modules/mal/mserver.mx"

	if( SERVERsessions[i].hdl)
		mapi_close_handle(SERVERsessions[i].hdl);
	SERVERsessions[i].hdl = mapi_query(mid, *qry);
	
#line 1197 "/export/scratch0/monet/monet.GNU.64.64.d.14791/MonetDB5/src/modules/mal/mserver.mx"
{int rn;
	if ( ( (rn=mapi_error(mid)) == -4 && hdl && mapi_result_error(hdl))  || rn){
		str err,newerr;
		size_t l;
		char *e,*f;

		if( hdl && mapi_result_error(hdl))
			err= mapi_result_error(hdl);
		else err= mapi_result_error(SERVERsessions[i].hdl);
		l= 2*strlen(err)+8192;
		newerr= (str) alloca(l);

		if (err == NULL)
			err = "(no additional error message)";