mserver.mx

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

@' The contents of this file are subject to the MonetDB Public License
@' Version 1.1 (the "License"); you may not use this file except in
@' compliance with the License. You may obtain a copy of the License at
@' http://monetdb.cwi.nl/Legal/MonetDBLicense-1.1.html
@'
@' Software distributed under the License is distributed on an "AS IS"
@' basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See the
@' License for the specific language governing rights and limitations
@' under the License.
@'
@' The Original Code is the MonetDB Database System.
@'
@' The Initial Developer of the Original Code is CWI.
@' Portions created by CWI are Copyright (C) 1997-2007 CWI.
@' All Rights Reserved.

@f mserver
@a N.J. Nes P. Boncz, S. Mullender, M. Kersten
@v 1.1
@+ MonetDB server interface
This module contains a simple interface for setting up
internet connections and to initialize a client session.

Clients may initialize a private listener to implement
specific services. For example, in an OLTP environment
it may make sense to have a listener for each transaction
type, which simply parses a sequence of transaction parameters.

Authorization of access to the server is handled as part
of the client record initialization phase.

The complete Mapi library is also available to setup
communication with another Mserver.
This library internally uses pointer handles, which we replace with
an index in a locally maintained table. It provides a handle
to easily detect havoc clients.
@mal
module mserver;

command listen():int
address SERVERlisten_default
comment "Start a Mapi server with the default settings.";
command listen(port:int):int
address SERVERlisten_port
comment "Start a Mapi server on the given port";
command listen(port:int, maxusers:int):int
address SERVERlisten2
comment "Start a Mapi server";

command listen(port:int, maxusers:int, cmd:str):int
address SERVERlisten3
comment "Start the Mapi listener on <port> for <maxusers>. 
	For a new client connection MAL procedure <cmd>(Stream s_in, Stream s_out)
	is called.If no <cmd> is specified a new client thread is forked.";

command listen_ssl(port:int, maxusers:int,
	keyfile:str,certfile:str, cmd:str):int
address SERVERlistenSSL
comment "Start the Mapi listener on <port> for <maxusers> using SSL. 
	<keyfile> and <certfile> give the path names for files with the 
	server key and certificates in PEM format. For a new client connection 
	MAL procedure <cmd>(Stream s_in, Stream s_out) is called.
	If no <cmd> is specified a new client thread is forked.";

command stop():void
address SERVERstop
comment "Terminate connection listeners"; 

command suspend():void
address SERVERsuspend
comment "Suspend accepting connections";

command resume():void
address SERVERresume
comment "Resume connection listeners";

command malclient(in:streams, out:streams):void
address SERVERclient
comment "Start a Mapi client.";


command trace(mid:int,flag:int):void
address SERVERtrace
comment "Toggle the Mapi library tracer";

command reconnect(host:str, port:int, usr:str, passwd:str,lang:str):int
address SERVERreconnectWithoutAlias
comment "Re-establish connection with a remote mserver";

command reconnect(host:str, port:int, db_alias:str, usr:str, passwd:str,lang:str):int
address SERVERreconnectAlias
comment "Re-establish connection with a remote mserver";

command reconnect(mid:int):void
address SERVERreconnect
comment "Re-establish a connection";

command connect(host:str, port:int, usr:str, passwd:str,lang:str):int
address SERVERconnect
comment "Establish connection with a remote mserver";

command connect_ssl(host:str, port:int, usr:str, passwd:str,lang:str):int
address SERVERconnectssl
comment "Establish connection with a remote mserver using the
secure socket layer";

command disconnect(dbalias:str):int
address SERVERdisconnectWithAlias
comment "Close connection with a remote mserver";

command disconnect():int
address SERVERdisconnectALL
comment "Close connections with all remote mserver";

command setAlias(dbalias:str)
address SERVERsetAlias
comment "Give the channel a logical name";

command lookup(dbalias:str):int
address SERVERlookup
comment "Retrieve the connection identifier";

command disconnect(mid:int):void
address SERVERdisconnect
comment "Terminate the session";

command destroy(mid:int):void
address SERVERdestroy
comment "Destroy the handle";

command ping(mid:int):int
address SERVERping
comment "Test availability of server";

command query(mid:int, qry:str):int
address SERVERquery
comment "Sent the query for execution";

command query_handle(mid:int, qry:str):int
address SERVERquery_handle
comment "Sent the query for execution";

pattern query_array(mid:int, qry:str, arg:str...):int
address SERVERquery_array
comment "Sent the query for execution replacing '?' by arguments";

command prepare(mid:int, qry:str):int
address SERVERprepare
comment "Prepare a query for execution";

command finish(hdl:int):int
address SERVERfinish
comment "Remove all remaining answers";

command get_field_count(hdl:int):int
address SERVERget_field_count
comment "Return number of fields";

command get_row_count(hdl:int):int
address SERVERget_row_count
comment "Return number of rows";

command fetch_row(hdl:int):int
address SERVERrows_affected
comment "Return number of affected rows";

command fetch_row(hdl:int):int
address SERVERfetch_row
comment "Retrieve the next row for analysis";

command fetch_all_rows(hdl:int):int
address SERVERfetch_all_rows
comment "Retrieve all rows into the cache";

command fetch_field(hdl:int,fnr:int):str
address SERVERfetch_field_str
comment "Retrieve a single field";

command fetch_field(hdl:int,fnr:int):int
address SERVERfetch_field_int
comment "Retrieve a single int field";

command fetch_field(hdl:int,fnr:int):lng
address SERVERfetch_field_lng
comment "Retrieve a single lng field";

command fetch_field(hdl:int,fnr:int):sht
address SERVERfetch_field_sht
comment "Retrieve a single sht field";

command fetch_field(hdl:int,fnr:int):void
address SERVERfetch_field_void
comment "Retrieve a single void field";

command fetch_field(hdl:int,fnr:int):oid
address SERVERfetch_field_oid
comment "Retrieve a single void field";

command fetch_field(hdl:int,fnr:int):chr
address SERVERfetch_field_chr
comment "Retrieve a single chr field";

command fetch_field_array(hdl:int):bat[:int,:str]
address SERVERfetch_field_bat
comment "Retrieve all fields for a row";

command fetch_line(hdl:int):str
address SERVERfetch_line
comment "Retrieve a complete line";

command fetch_reset(hdl:int):int
address SERVERfetch_reset
comment "Reset the cache read line.";

command next_result(hdl:int):int
address SERVERnext_result
comment "Go to next result set";

command error(mid:int):int
address SERVERerror
comment "Check for an error in the communication";

command getError(mid:int):str
address SERVERgetError
comment "Get error message";

command explain(mid:int):str
address SERVERexplain
comment "Turn the error seen into a string";

pattern put(mid:int, nme:str, val:any_1):void
address SERVERput
comment "Send a value to a remote site";

pattern put(nme:str, val:any_1):str
address SERVERputLocal
comment "Prepare sending a value to a remote site";

pattern rpc(key:int,qry:str...):any
address SERVERmapi_rpc_single_row
comment "Sent a simple query for execution and fetch result";

pattern rpc(key:int,qry:str):bat[:any_1,:any_2]
address SERVERmapi_rpc_bat;

command rpc(key:int,qry:str):void
address SERVERquery
comment "Sent a simple query for execution";

pattern bind(key:int,rschema:str,rtable:str,rcolumn:str,i:int):bat[:any_1,:any_2]
address SERVERbindBAT
comment "Bind a remote variable to a local one";

pattern bind(key:int,rschema:str,rtable:str,i:int):bat[:any_1,:any_2]
address SERVERbindBAT
comment "Bind a remote variable to a local one";

pattern bind(key:int,remoteName:str):bat[:any_1,:any_2]
address SERVERbindBAT
comment "Bind a remote variable to a local one";


mserver.listen();
@-
@{
#command listen(open:bit):int
#address SERVERlisten_open
#comment "Start a Mapi server on the default port global bound or not";
#command receive(key:int,nme:str):bat[:any_1,:any_2]
#address SERVERreceive
#comment "Fetch a bat from a remote site";

#command send(key:int,nme:str, r:bat[:any_1,:any_2]):void
#address SERVERsend
#comment "Sent a bat to a remote site";
command prelude()
address SERVERprelude;
command epilogue()
address SERVERepilogue;

@h
#ifndef SERVER_H
#define SERVER_H
/* #define DEBUG_SERVER */

#include "mal_client.h"
#include "mal_session.h"
#include "mal_exception.h"
#include "mal_interpreter.h"
#include "mal_authorize.h"
#include "mal_sabaoth.h"
#include <Mapi.h>

#ifdef WIN32
#ifndef LIBMSERVER
#define m_export extern __declspec(dllimport)
#else
#define m_export extern __declspec(dllexport)
#endif
#else
#define m_export extern
#endif


#define NEW_ARRAY( type, size )	(type*)GDKmalloc((size)*sizeof(type))
#define STREQ(a, b) 		(strcmp(a, b)==0)

#define SERVERPORT		50000
#define SERVERSSLPORT		40000
#define SERVERMAXUSERS 		5

m_export str SERVERlistenThread(int *Sock);
m_export str SERVERlisten(int *Port, str *Usockfile, int *Maxusers, str *Cmd);
m_export bat * SERVERprelude(void);
m_export void SERVERepilogue(void);
m_export str SERVERlistenSSL(int *Port, int *Maxusers, str keyfile, str certfile, str cmd);
m_export str SERVERlisten_default(int *ret);
m_export str SERVERlisten2(int *ret, int *port, int *maxusers);
m_export str SERVERlisten3(int *ret, int *port, int *maxusers, str *cmd);
m_export str SERVERlisten_port(int *ret, int *pid);
m_export str SERVERstop(int *ret);
m_export str SERVERsuspend(int *ret);
m_export str SERVERresume(int *ret);
m_export void SERVERexit(void);

m_export str SERVERconnect(int *ret, str *host, int *port, str *username, str *password, str *lang);
m_export str SERVERdisconnectWithAlias(int *ret, str *db_alias);
m_export str SERVERdisconnectALL(int *ret);
m_export str SERVERreconnectAlias(int *ret, str *host, int *port, str *dbalias, str *username, str *password, str *lang);
m_export str SERVERreconnectWithoutAlias(int *ret, str *host, int *port, str *username, str *password, str *lang);
m_export str SERVERconnectssl(int *ret, str *host, int *port, str *username, str *password, str *lang);
m_export str SERVERtrace(int *ret, int *mid, int *flag);
m_export str SERVERdisconnect(int *ret, int *mid);
m_export str SERVERsetAlias(int *ret, int *mid, str *dbalias);
m_export str SERVERlookup(int *ret, str *dbalias);
m_export str SERVERdestroy(int *ret, int *mid);
m_export str SERVERreconnect(int *ret, int *mid);
m_export str SERVERping(int *ret, int *mid);
m_export str SERVERquery(int *ret, int *mid, str *qry);
m_export str SERVERquery_handle(int *ret, int *mid, str *qry);
m_export str SERVERquery_array(MalBlkPtr mb, MalStkPtr stk, InstrPtr pc);
m_export str SERVERprepare(int *ret, int *key, str *qry);
m_export str SERVERexecute(int *ret, int *idx);
m_export str SERVERfinish(int *ret, int *idx);
m_export str SERVERrows_affected(int *ret, int *idx);
m_export str SERVERget_row_count(int *ret, int *idx);
m_export str SERVERget_field_count(int *ret, int *idx);
m_export str SERVERfetch_row(int *ret, int *idx);
m_export str SERVERfetch_all_rows(int *ret, int *idx);
m_export str SERVERfetch_field_str(str *ret, int *idx, int *fnr);
m_export str SERVERfetch_field_int(int *ret, int *idx, int *fnr);
m_export str SERVERfetch_field_lng(lng *ret, int *idx, int *fnr);
m_export str SERVERfetch_field_sht(sht *ret, int *idx, int *fnr);
m_export str SERVERfetch_field_void(oid *ret, int *idx, int *fnr);
m_export str SERVERfetch_field_oid(oid *ret, int *idx, int *fnr);
m_export str SERVERfetch_field_chr(chr *ret, int *idx, int *fnr);
m_export str SERVERfetch_line(str *ret, int *key);
m_export str SERVERnext_result(int *ret, int *key);
m_export str SERVERfetch_reset(int *ret, int *key);
m_export str SERVERfetch_field_bat(int *bid, int *idx);
m_export str SERVERerror(int *ret, int *idx);
m_export str SERVERgetError(str *ret, int *idx);
m_export str SERVERexplain(str *ret, int *idx);
m_export str SERVERmapi_rpc_single_row(MalBlkPtr mb, MalStkPtr stk, InstrPtr pci);
m_export str SERVERmapi_rpc_single_bat(MalBlkPtr mb, MalStkPtr stk, InstrPtr pci);
m_export str SERVERput(MalBlkPtr mb, MalStkPtr stk, InstrPtr pci);
m_export str SERVERputLocal(MalBlkPtr mb, MalStkPtr stk, InstrPtr pci);
m_export str SERVERbindBAT(MalBlkPtr mb, MalStkPtr stk, InstrPtr pci);
m_export str SERVERclient(int *res, stream **In, stream **Out);
m_export str SERVERmapi_rpc_single_row(MalBlkPtr mb, MalStkPtr stk, InstrPtr pci);
m_export str SERVERmapi_rpc_bat(MalBlkPtr mb, MalStkPtr stk, InstrPtr pci);

#endif /* SERVER_H */
@c
#include "mal_config.h"
#include "mserver.h"

#include  <sys/types.h>
#ifdef HAVE_SYS_SOCKET_H
# include <sys/socket.h>
#endif
#ifdef HAVE_WIN32
# include <winsock.h>
#endif
#ifdef HAVE_SYS_UN_H
#include <sys/un.h>
#endif

#define SOCKPTR struct sockaddr *
#if HAVE_SOCKLEN_T
#define SOCKLEN socklen_t
#else
#define SOCKLEN int
#endif

#ifdef NATIVE_WIN32
#define s_close(s)	closesocket(s)
#else
#define s_close(s)	close(s)
#endif

#ifdef HAVE_OPENSSL
#include <openssl/ssl.h>
#include <openssl/err.h>
#endif

static char seedChars[] = {'a', 'b', 'c', 'd', 'e', 'f', 'g', 'h', 'i', 'j',
	'k', 'l', 'm', 'n', 'o', 'p', 'q', 'r', 's', 't', 'u', 'v', 'w', 'x',
	'y', 'z', 'A', 'B', 'C', 'D', 'E', 'F', 'G', 'H', 'I', 'J', 'K', 'L',
	'M', 'N', 'O', 'P', 'Q', 'R', 'S', 'T', 'U', 'V', 'W', 'X', 'Y', 'Z',
	'1', '2', '3', '4', '5', '6', '7', '8', '9', '0'};


static void generateChallenge(str buf, int min, int max) {
	size_t size;
	size_t chr;
	size_t i;

	/* don't seed the randomiser, or you get the same challenge during
	 * the same second */
	/* srand(time(NULL)); */
	size = rand();
	size = (size % (max - min)) + min;
	for (i = 0; i < size; i++) {
		chr = rand();
		chr %= 62;
		buf[i] = seedChars[chr];
	}
	buf[i] = '\0';
}

static void 
doChallenge(str cmd, int msgsock) {
	if (cmd == NULL) {
		char *buf = (char *) GDKmalloc(BLOCK);
		char *challenge;
		char *algos;
		stream *fdin = block_stream( socket_rastream(msgsock, "Server read"));
		stream *fdout = block_stream( socket_wastream(msgsock, "Server write"));
		bstream *bs;

		if (!fdin || !fdout) {
			GDKsyserror("SERVERlisten:fdin or fdout problems");
			return;
		}

		assert(buf);

		/* generate the challenge string */
		challenge = alloca(sizeof(char) * (12 + 1));
		generateChallenge(challenge, 8, 12);
		if (AUTHgetHashAlgorithms(&algos) != MAL_SUCCEED) assert (0);
		/* note that we claim to speak proto 8 here */
		stream_printf(fdout, "%s:mserver:8:%s:%s",
				challenge,
				algos,
#ifdef WORDS_BIGENDIAN
				"BIG"
#else
				"LIT"
#endif
				);
		stream_flush(fdout);
		/* get response */
		stream_read_block(fdin, buf, BLOCK, 1);

		/* in embedded mode we allow just one client */
		if (GDKembedded && MCcountClients() > 1) {
			stream_close(fdin);
			stream_destroy(fdin);
			stream_close(fdout);
			stream_destroy(fdout);
			return;
		}
#ifdef DEBUG_SERVER
		printf("mserver:Client accepted %s\n", name);
		fflush(stdout);

		THRprintf(GDKout, "#SERVERlisten:client accepted %d\n", msgsock);
		THRprintf(GDKout, "#SERVERlisten:client string %s\n", name);
#endif
		bs = bstream_create(fdin, 128 * BLOCK);

		bs->eof = 1;
		MSscheduleClient(buf, challenge, bs, fdout);
	} else {
		/* in M4 it is possible to execute a command directly */
	}
}

static char *threadcommand;
static int usock = -1;
static MT_Id listener[8]; 
static int lastlistener=0;
static int serveractive=TRUE;

str
SERVERlistenThread(int *Sock)
{
	char *msg = 0;
	char *cmd = threadcommand;

	if( lastlistener < 8 )
		listener[lastlistener++] = MT_getpid();
	(void)usock;
	do {
		int retval;
		struct timeval tv;
		fd_set fds;
		int msgsock;
		int sock = *Sock;

		FD_ZERO(&fds);