Difference between revisions of "Server.c"

From Organic Design wiki
m
m
 
(391 intermediate revisions by 3 users not shown)
Line 1: Line 1:
// good sock function ref at
+
{{legacy}}
// http://www.opengroup.org/onlinepubs/009695399/idx/networking.html
+
<source lang="c">
 +
// This article and all its includes are licenced under LGPL
 +
// GPL: http://www.gnu.org/copyleft/lesser.html
 +
// SRC: http://www.organicdesign.co.nz/server.c
  
// Set up socket and listening loop
+
#define svrDELAY 0        // normal operation is 0 for no delay
#define LISTENPORT 2012
+
#define svrPAKSIZE 128     // keep packet-size small for non-multithreaded design
#define PAKSIZE 128   // keep packet-size small for non-multithreaded design
+
#define svrPAKMAX  16384  // maximum packet size to allow for variable pak size at runtime
#define BUFSIZE 10000 // max message size
+
#define svrBUFSIZE 4096    // dictates max message size
#define MAXCLIENTS 100
+
#define svrMAXCLIENTS 1000 // used by listen()
 
+
#ifndef __WIN32__
// - todo: don't exit on connect errors, keep trying every 10s
+
#define WSAGetLastError() errno
// Includes for socket (trying to use one source cpp for osx,win32,*ux)
 
#ifdef WINDOWS
 
#include <winsock.h>
 
#else
 
#include <sys/socket.h>
 
#include <sys/select.h>
 
#include <netinet/in.h>
 
#include <fcntl.h> // needed for O_NONBLOCK option on server
 
#include <sys/time.h> // for select()
 
//#include <arpa/inet.h>
 
//#include <netdb.h>
 
 
#endif
 
#endif
  
int processMessage(char* msg);
+
// Socket structures, globals and prototypes
 +
void server();
 +
int serverInit();
 +
int serverExit();
 +
int nonblocking(int socket);
 +
fd_set *fdsetInit(int fd);
 +
fd_set fdset;
 +
int sock,fd;
 +
struct sockaddr_in addr;
 +
struct timeval to;
 +
unsigned long int sockopt_on;
  
// struct type to represent a currently connected stream
+
// Client/Stream info structure, globals and prototypes
typedef struct streamstruct {
+
char *pBuf;                // Buffer to read/write data packets
char* buf;
+
const char *term = "\r\n\r\n";
int fd;
+
typedef struct siStruct {
 +
int fd;                // File-descriptor for this stream
 +
char *iBuf, *oBuf;    // Message buffers for this stream
 +
int iPtr;             // Index into input-buffer for next packet to start at
 +
int tPtr;              // Index into term (atomised strcmp incase terminator spans packet-boundary)
 +
int oPtr;
 
} streamInfo;
 
} streamInfo;
 +
void client();
 +
int streamOpen(int fd,char *resource);
 +
void streamClose(streamInfo *stream);
 +
void streamProcess(streamInfo *stream);
 +
 +
 +
// Initialise socket listening on specified port
 +
int serverInit() {
 +
 +
#if __WIN32__
 +
WSADATA wsaData;
 +
if (WSAStartup(MAKEWORD(2,0),&wsaData)!=0) logErr("WSAStartup() failed!");
 +
#endif
  
// set up socket struct
+
// Set up structures for socket & select
struct sockaddr_in my_addr, client_addr;
+
sockopt_on = 1;
int sa_in_size = sizeof(struct sockaddr_in);
+
int szAddr = sizeof(struct sockaddr_in);
memset((char *)&my_addr, 0, sa_in_size); // zero the struct
+
memset((char*)&addr, 0, szAddr); // zero the struct
my_addr.sin_family = PF_INET;
+
addr.sin_family = AF_INET;
my_addr.sin_port = htons(LISTENPORT);
+
addr.sin_port = htons(port);
my_addr.sin_addr.s_addr = htonl(INADDR_ANY);
+
addr.sin_addr.s_addr = htonl(INADDR_ANY);
 +
errno = 0;
  
// get a socket
+
// Do the usual socket polava: create,options,bind,listen
int server;
+
if ((sock = socket(PF_INET,SOCK_STREAM,IPPROTO_TCP))<0)
if ((server = socket(PF_INET, SOCK_STREAM, IPPROTO_TCP)) < 0) logAdd("socket() failed!");
+
logErrNum("init/socket() failed returning %d",WSAGetLastError());
 +
if (setsockopt(sock,SOL_SOCKET,SO_REUSEADDR,(char*)&sockopt_on,sizeof(int))<0)
 +
logErrNum("init/setsockopt() failed returning %d",WSAGetLastError());
 +
if (bind(nonblocking(sock),(struct sockaddr*)&addr,szAddr)<0)
 +
logErrNum("init/bind() failed returning %d",WSAGetLastError());
 +
if (listen(sock,svrMAXCLIENTS)<0)
 +
logErrNum("init/listen() failed returning %d",WSAGetLastError());
  
// make it reusable
+
if (WSAGetLastError() == 0) {
int sockopt_on = 1;
+
char *msg = malloc(100);
if (setsockopt(server, SOL_SOCKET, SO_REUSEADDR, &sockopt_on, sizeof(int)) < 0) logAdd("setsockopt() failed!");
+
sprintf(msg,"Daemon \"%s\" started successfully and serving on port %d.",peer,port);
 +
logAdd(msg);
 +
free(msg);
 +
}
 +
else {
 +
logAdd("Server failed to start!");
 +
return WSAGetLastError();
 +
}
  
// bind our socket to the port
+
// Globals for server() and client() quantum-functions
if (bind(server, (struct sockaddr *)&my_addr, sa_in_size) < 0) logAdd("bind() failed!";
+
pBuf = malloc(svrPAKMAX);
 +
*nodeState(nodeSERVER,0) = &server;
 +
nodeSetValue(nodeSERVER,nodeCODE,nodeTRUE);
  
// start listening for incoming connections
+
return 0;
if (listen(server, MAXCLIENTS) < 0) logAdd("listen() failed!";
+
}
  
// make the server non-blocking so accept() returns straight away
+
 
 +
// Perform any cleanup for socket
 +
int serverExit() {
 +
free(pBuf);
 +
// todo: close streams and free buffers etc
 +
#if __WIN32__
 +
WSACleanup();
 +
#endif
 +
return WSAGetLastError();
 +
}
 +
 
 +
 
 +
// make the passed socket non-blocking so accept() returns straight away for multiplexed model
 
// - if no incoming requests, returns EAGAIN or EWOULDBLOCK state
 
// - if no incoming requests, returns EAGAIN or EWOULDBLOCK state
// - we need this for non-multithreaded design
+
int nonblocking(int socket) {
// - this may need to be done for each new incomming stream...?
+
#if __WIN32__
// - some examples use fcntl(server, F_SETFL, FNDELAY) or fcntl(server, F_SETFL, O_NONBLOCK);
+
ioctlsocket(socket,FIONBIO,&sockopt_on);
fcntl(server, O_NONBLOCK);
+
#else
 +
fcntl(socket,F_SETFL,fcntl(socket,F_GETFL)|O_NONBLOCK);
 +
#endif
 +
return socket;
 +
}
  
// Setup stuff for select()
 
struct timeval timeout;
 
fd_set *fdset;
 
  
// array of current client connections
+
// Return pointer to fdset filled with passed file-descriptor ready for a select() call
// - this may be replaced by blocks pointed to by select()'s fd array
+
fd_set *fdsetInit(int fd) {
streamInfo streams[MAXCLIENTS],si;
+
FD_ZERO(&fdset);
 +
FD_SET(fd,&fdset);
 +
to.tv_sec = to.tv_usec = svrDELAY;
 +
return &fdset;
 +
}
  
  
 +
// Process message in iBuf which should be a complete HTTP GET request
 +
// - iPtr is put back to 0 so next message fills from start of iBuf
 +
void streamProcess(streamInfo *stream) {
 +
char *rqst, *end, *msg = stream->iBuf;
 +
if (strncmp("GET /",msg,5)==0) {
  
// - server is in is in nodal ROOT loop
+
// Extract request path & query-string
void network(node stream) {
+
rqst = end = msg+4;
 +
while(*++end > ' ');
 +
*end = '\0';
 +
printf("Request:\"%s\"\n",rqst);
  
int i, stream, nStreams = 0;
+
// Exit if /stop requested
 +
if (strcmp("/stop",rqst)==0) nodeExit();
  
timeout.tv_sec = 1; // later use 0 for min time
+
// Send test response back
timeout.tv_usec = 0;
+
send(stream->fd,"HTTP/1.1 200 OK\r\nDate: Mon, 31 Jul 2006 22:27:14 NZST\r\nContent-Type: text/html\r\nConnection: keep-alive\r\nContent-Length: 9\r\n\r\nStink eh?",134,0);
 +
}
 +
else {
 +
send(stream->fd,"HTTP/1.1 200 OK\r\nDate: Mon, 31 Jul 2006 22:27:14 NZST\r\nContent-Type: text/html\r\nConnection: keep-alive\r\nContent-Length: 9\r\n\r\nNo GET!!!",134,0);
 +
logAdd("Peerd only supports GET requests currently!");
 +
}
 +
stream->iPtr = 0; // reset iPtr ready for next message
 +
}
  
// Check for new connection
 
FD_ZERO(fdset);
 
FD_SET(server,fdset);
 
if (select(maxfd+1, fdset, NULL, NULL, &timeout)>0) {
 
// new connection
 
}
 
  
// Data to receive?
+
// Create new client-node and streamInfo structure from passed file-descriptor
FD_ZERO(fdset);
+
// - if fd is NULL, then try and create one by resolving the resource string
i = 0;
+
void streamOpen(int fd, char *resource) {
while(si = streams[i++]) FD_SET(si.fd,fdset);
+
if (fd == NULL) {
if (select(maxfd+1, streams, NULL, NULL, &timeout)>0) {
+
// if resource is /^[./]/ then its a file
// read packet from each in fdset
+
// - file will have different packet size etc
 +
// - may need to include &recv/&read etc in streamInfo struct
 
}
 
}
 +
if (fd < 0) return logErrMsg("Couldn't open resource \"%s\"",resource);
 +
logAddNum("New connection: Stream%d.",fd);
 +
// Create new streamInfo structure and allocate buffers
 +
streamInfo *stream = malloc(sizeof(streamInfo));
 +
stream->fd = fd;
 +
stream->iBuf = malloc(svrBUFSIZE);
 +
stream->oBuf = malloc(svrBUFSIZE);
 +
stream->iPtr = stream->oPtr = stream->tPtr = 0;
  
// Data to send?
+
// Create new client-node in clients-loop and add the new streamInfo
FD_ZERO(fdset);
+
node nc = nodeGetValue(nodeCLIENTS,nodeLOOP);
i = 0;
+
nc = nodeLoopInsert(nc,0);
while(si = streams[i++]) FD_SET(si.fd,fdset);
+
nodeSetValue(nodeCLIENTS,nodeLOOP,nc);
if (select(maxfd+1, NULL, streams, NULL, &timeout)>0) {
+
nodeSetValue(nc,nodeCODE,nodeTRUE);
// write packet from each in fdset
+
*nodeState(nc,0) = &client;
}
+
*nodeState(nc,nodeSTREAM) = stream;
 +
}
  
  
 +
// Close the passed stream, free its resources and remove from clients-loop
 +
void streamClose(streamInfo *stream) {
 +
printf("Stream%d closed.\n",stream->fd);
 +
//nodeLoopRemove(this); // NOTE: reduction not handling PREV
 +
free(stream->iBuf);
 +
free(stream->oBuf);
 +
free(stream);
 +
}
  
  
 +
// Each of the stream nodes points to this function in its State
 +
void client() {
  
// ****** old loop ******
+
// Get pointer to the streamInfo structure in current nodal context (this)
 +
streamInfo *stream = *nodeState(this, nodeSTREAM);
 +
fd = stream->fd;
 +
char *iBuf = stream->iBuf, *oBuf = stream->oBuf;
 +
int n, i = 0;
  
// Wait for any incomming connection
+
// If any data to recieve, read a packet
// O_NONBLOCK is set so, EAGAIN or EWOULDBLOCK returned if no requests
+
if (select(fd+1,fdsetInit(fd),NULL,NULL,&to)>0) {
stream = accept(server, (struct sockaddr *)&client_addr, &sa_in_size)
+
if ((n = recv(fd,pBuf,svrPAKSIZE,0)) > 0) {
if (stream == EAGAIN) {
+
// Some bytes were read, append to current message (loop last packet if msg too big)
// No new connection requests - read any available data from streams
+
if (stream->iPtr > svrBUFSIZE-svrPAKSIZE) stream->iPtr -= svrPAKSIZE;
for (i = 0; i < nStreams; i++) {
+
while (n--)  // Append current msg, for each complete msg, process it & reset iPtr
char* buf = streams[i].buf;
+
if ((iBuf[stream->iPtr++] = pBuf[i++]) != term[stream->tPtr++]) stream->tPtr = 0;
int stream = streams[i].fileno;
+
else if (term[stream->tPtr] == 0) streamProcess(stream);
 
}
 
}
}
+
else if (n == 0) streamClose(stream); // Zero bytes were read, do orderly termination
else if (stream == ECONNABORTED) {
+
else logErrNum("client/recv() failed returning %d",WSAGetLastError());
logAdd("Stream closed by client");
 
close(stream);
 
}
 
else if (stream < 0) logAdd("accept() failed!");
 
else {
 
// New stream, create input buffer etc
 
streamInfo si;
 
si.buf = malloc(BUFSIZE);
 
si.fileno = stream;
 
streams[nStreams++] = &si;
 
 
}
 
}
  
 
+
// Data to send?
// log the connecter
+
//if (select(fd+1,NULL,fdsetInit(fd),NULL,&to)>0) {
// - should get info for this stream, incl buf
+
// write a packet from outbuf
printf("got connection from %s\n", inet_ntoa(client_addr.sin_addr));
+
//int len = svrPAKSIZE; // or less
 
+
//if (send(fd,oBuf+oPtr,len,0)<0) logErr("send() failed!");
// get the reply
+
// }
// - should keep receiving until \r\n\0?, then call serverProcessMessage()
 
if (recv(stream, buf, bufsize, 0) == -1) perror("recv");
 
else printf("The client says \"%s\"\n", buf);
 
  
 
}
 
}
  
// Parses a message content and responds to client
 
int processMessage(char* msg) {
 
// test if restart cmd first
 
  
// send response
+
// Function called by each stream-node in network's reduction loop
if (send(stream, MSG, strlen(MSG)+1, 0) == -1) logAdd("send() failed!");
+
// - Checks if any new connections waiting on listening socket,
else logAdd("Sent message");
+
//  if so accept and call streamOpen() with new file-descriptor
 +
void server() {
 +
if (select(sock+1,fdsetInit(sock),NULL,NULL,&to)>0)
 +
if ((fd = nonblocking(accept(sock,NULL,NULL)))>0) streamOpen(fd,NULL);
 +
else logErrNum("server/accept() failed returning %d",WSAGetLastError());
 
}
 
}
 +
</source>
 +
[[Category:C]]

Latest revision as of 15:22, 6 July 2015

Legacy.svg Legacy: This article describes a concept that has been superseded in the course of ongoing development on the Organic Design wiki. Please do not develop this any further or base work on this concept, now this page is for historic record only.
// This article and all its includes are licenced under LGPL
// GPL: http://www.gnu.org/copyleft/lesser.html
// SRC: http://www.organicdesign.co.nz/server.c

#define svrDELAY 0         // normal operation is 0 for no delay
#define svrPAKSIZE 128     // keep packet-size small for non-multithreaded design
#define svrPAKMAX  16384   // maximum packet size to allow for variable pak size at runtime
#define svrBUFSIZE 4096    // dictates max message size
#define svrMAXCLIENTS 1000 // used by listen()
#ifndef __WIN32__
#define WSAGetLastError() errno
#endif

// Socket structures, globals and prototypes
void server();
int serverInit();
int serverExit();
int nonblocking(int socket);
fd_set *fdsetInit(int fd);
fd_set fdset;
int sock,fd;
struct sockaddr_in addr;
struct timeval to;
unsigned long int sockopt_on;

// Client/Stream info structure, globals and prototypes
char *pBuf;                // Buffer to read/write data packets
const char *term = "\r\n\r\n";
typedef struct siStruct {
	int fd;                // File-descriptor for this stream
	char *iBuf, *oBuf;     // Message buffers for this stream
	int iPtr;              // Index into input-buffer for next packet to start at
	int tPtr;              // Index into term (atomised strcmp incase terminator spans packet-boundary)
	int oPtr;
	} streamInfo;
void client();
int streamOpen(int fd,char *resource);
void streamClose(streamInfo *stream);
void streamProcess(streamInfo *stream);


// Initialise socket listening on specified port
int serverInit() {

	#if __WIN32__
	WSADATA wsaData;
	if (WSAStartup(MAKEWORD(2,0),&wsaData)!=0) logErr("WSAStartup() failed!");
	#endif

	// Set up structures for socket & select
	sockopt_on = 1;
	int szAddr = sizeof(struct sockaddr_in);
	memset((char*)&addr, 0, szAddr); // zero the struct
	addr.sin_family = AF_INET;
	addr.sin_port = htons(port);
	addr.sin_addr.s_addr = htonl(INADDR_ANY);
	errno = 0;

	// Do the usual socket polava: create,options,bind,listen
	if ((sock = socket(PF_INET,SOCK_STREAM,IPPROTO_TCP))<0)
		logErrNum("init/socket() failed returning %d",WSAGetLastError());
	if (setsockopt(sock,SOL_SOCKET,SO_REUSEADDR,(char*)&sockopt_on,sizeof(int))<0)
		logErrNum("init/setsockopt() failed returning %d",WSAGetLastError());
	if (bind(nonblocking(sock),(struct sockaddr*)&addr,szAddr)<0)
		logErrNum("init/bind() failed returning %d",WSAGetLastError());
	if (listen(sock,svrMAXCLIENTS)<0)
		logErrNum("init/listen() failed returning %d",WSAGetLastError());

	if (WSAGetLastError() == 0) {
		char *msg = malloc(100);
		sprintf(msg,"Daemon \"%s\" started successfully and serving on port %d.",peer,port);
		logAdd(msg);
		free(msg);
		}
	else {
		logAdd("Server failed to start!");
		return WSAGetLastError();
		}

	// Globals for server() and client() quantum-functions
	pBuf = malloc(svrPAKMAX);
	*nodeState(nodeSERVER,0) = &server;
	nodeSetValue(nodeSERVER,nodeCODE,nodeTRUE);

	return 0;	
	}


// Perform any cleanup for socket
int serverExit() {
	free(pBuf);
	// todo: close streams and free buffers etc
	#if __WIN32__
	WSACleanup();
	#endif
	return WSAGetLastError();
	}


// make the passed socket non-blocking so accept() returns straight away for multiplexed model
// - if no incoming requests, returns EAGAIN or EWOULDBLOCK state
int nonblocking(int socket) {
	#if __WIN32__
	ioctlsocket(socket,FIONBIO,&sockopt_on);
	#else
	fcntl(socket,F_SETFL,fcntl(socket,F_GETFL)|O_NONBLOCK);
	#endif
	return socket;
	}


// Return pointer to fdset filled with passed file-descriptor ready for a select() call
fd_set *fdsetInit(int fd) {
	FD_ZERO(&fdset);
	FD_SET(fd,&fdset);
	to.tv_sec = to.tv_usec = svrDELAY;
	return &fdset;
	}


// Process message in iBuf which should be a complete HTTP GET request
// - iPtr is put back to 0 so next message fills from start of iBuf
void streamProcess(streamInfo *stream) {
	char *rqst, *end, *msg = stream->iBuf;
	if (strncmp("GET /",msg,5)==0) {

		// Extract request path & query-string
		rqst = end = msg+4;
		while(*++end > ' ');
		*end = '\0';
		printf("Request:\"%s\"\n",rqst);

		// Exit if /stop requested
		if (strcmp("/stop",rqst)==0) nodeExit();

		// Send test response back
		send(stream->fd,"HTTP/1.1 200 OK\r\nDate: Mon, 31 Jul 2006 22:27:14 NZST\r\nContent-Type: text/html\r\nConnection: keep-alive\r\nContent-Length: 9\r\n\r\nStink eh?",134,0);
		}
	else {
		send(stream->fd,"HTTP/1.1 200 OK\r\nDate: Mon, 31 Jul 2006 22:27:14 NZST\r\nContent-Type: text/html\r\nConnection: keep-alive\r\nContent-Length: 9\r\n\r\nNo GET!!!",134,0);
		logAdd("Peerd only supports GET requests currently!");
		}
	stream->iPtr = 0; // reset iPtr ready for next message
	}


// Create new client-node and streamInfo structure from passed file-descriptor
// - if fd is NULL, then try and create one by resolving the resource string
void streamOpen(int fd, char *resource) {
	if (fd == NULL) {
		// if resource is /^[./]/ then its a file
		// - file will have different packet size etc
		// - may need to include &recv/&read etc in streamInfo struct
		}
	if (fd < 0) return logErrMsg("Couldn't open resource \"%s\"",resource);
	logAddNum("New connection: Stream%d.",fd);
	// Create new streamInfo structure and allocate buffers
	streamInfo *stream = malloc(sizeof(streamInfo));
	stream->fd = fd;
	stream->iBuf = malloc(svrBUFSIZE);
	stream->oBuf = malloc(svrBUFSIZE);
	stream->iPtr = stream->oPtr = stream->tPtr = 0;

	// Create new client-node in clients-loop and add the new streamInfo
	node nc = nodeGetValue(nodeCLIENTS,nodeLOOP);
	nc = nodeLoopInsert(nc,0);
	nodeSetValue(nodeCLIENTS,nodeLOOP,nc);
	nodeSetValue(nc,nodeCODE,nodeTRUE);
	*nodeState(nc,0) = &client;
	*nodeState(nc,nodeSTREAM) = stream;
	}


// Close the passed stream, free its resources and remove from clients-loop
void streamClose(streamInfo *stream) {
	printf("Stream%d closed.\n",stream->fd);
	//nodeLoopRemove(this); // NOTE: reduction not handling PREV
	free(stream->iBuf);
	free(stream->oBuf);
	free(stream);
	}


// Each of the stream nodes points to this function in its State
void client() {

	// Get pointer to the streamInfo structure in current nodal context (this)
	streamInfo *stream = *nodeState(this, nodeSTREAM);
	fd = stream->fd;
	char *iBuf = stream->iBuf, *oBuf = stream->oBuf;
	int n, i = 0;

	// If any data to recieve, read a packet
	if (select(fd+1,fdsetInit(fd),NULL,NULL,&to)>0) {
		if ((n = recv(fd,pBuf,svrPAKSIZE,0)) > 0) {
			// Some bytes were read, append to current message (loop last packet if msg too big)
			if (stream->iPtr > svrBUFSIZE-svrPAKSIZE) stream->iPtr -= svrPAKSIZE;
			while (n--)  // Append current msg, for each complete msg, process it & reset iPtr
				if ((iBuf[stream->iPtr++] = pBuf[i++]) != term[stream->tPtr++]) stream->tPtr = 0;
				else if (term[stream->tPtr] == 0) streamProcess(stream);
			}
		else if (n == 0) streamClose(stream); // Zero bytes were read, do orderly termination
		else logErrNum("client/recv() failed returning %d",WSAGetLastError());
		}

	// Data to send?
	//if (select(fd+1,NULL,fdsetInit(fd),NULL,&to)>0) {
		// write a packet from outbuf
		//int len = svrPAKSIZE; // or less
		//if (send(fd,oBuf+oPtr,len,0)<0) logErr("send() failed!");
	//	}

	}


// Function called by each stream-node in network's reduction loop
// - Checks if any new connections waiting on listening socket,
//   if so accept and call streamOpen() with new file-descriptor
void server() {
	if (select(sock+1,fdsetInit(sock),NULL,NULL,&to)>0)
		if ((fd = nonblocking(accept(sock,NULL,NULL)))>0) streamOpen(fd,NULL);
		else logErrNum("server/accept() failed returning %d",WSAGetLastError());
	}