Difference between revisions of "Server.c"

From Organic Design wiki
m
(need to make iBuf circular to avoid msg movement in clearing)
Line 21: Line 21:
 
// struct for a stream-node's state to point to
 
// struct for a stream-node's state to point to
 
typedef struct siStruct {
 
typedef struct siStruct {
int fd, inptr, outptr;
+
int fd, inPtr, inMsg, outPtr;
char *inbuf, *outbuf;
+
char *inBuf, *outBuf;
 
} streamInfo;
 
} streamInfo;
  
Line 94: Line 94:
 
// Process message and clear it from the input buffer
 
// Process message and clear it from the input buffer
 
void processMessage(streamInfo *info) {
 
void processMessage(streamInfo *info) {
char *msg = info->inbuf;
 
 
msg[svrBUFSIZE-1] = '\0';
 
msg[svrBUFSIZE-1] = '\0';
 
printf("\n\nCONTENT:\n%s\n\n",msg);
 
printf("\n\nCONTENT:\n%s\n\n",msg);
 
// Handle buffer overrun
 
// Handle buffer overrun
if (info->inptr>svrBUFSIZE-svrPAKSIZE) logErr("This message exceeded svrBUFSIZE!");
+
if (info->iPtr>svrBUFSIZE-svrPAKSIZE) logErr("This message exceeded svrBUFSIZE!");
  
 
// Handle GET requests
 
// Handle GET requests
Line 119: Line 118:
 
}
 
}
 
else logAdd("Peerd only supports GET requests currently!");
 
else logAdd("Peerd only supports GET requests currently!");
info->inptr = 0; // clear input buffer for this stream
+
info->iPtr = 0; // clear input buffer for this stream
 
}
 
}
  
Line 135: Line 134:
 
streamInfo *newsi = malloc(sizeof(streamInfo));
 
streamInfo *newsi = malloc(sizeof(streamInfo));
 
newsi->fd = fd;
 
newsi->fd = fd;
newsi->inbuf = malloc(svrBUFSIZE);
+
newsi->iBuf = malloc(svrBUFSIZE+svrPAKSIZE);
newsi->outbuf = malloc(svrBUFSIZE);
+
newsi->oBuf = malloc(svrBUFSIZE+svrPAKSIZE);
newsi->inptr = 0;
+
newsi->iPtr = newsi->iMsg = newsi->oPtr = 0;
newsi->outptr = 0;
 
 
node n = nodeGetValue(nodeCLIENTS,nodeLOOP);
 
node n = nodeGetValue(nodeCLIENTS,nodeLOOP);
 
n = nodeLoopInsert(n,nodeNULL);
 
n = nodeLoopInsert(n,nodeNULL);
Line 155: Line 153:
 
// Get the info for this stream ready for reading/writing a packet if necessary
 
// Get the info for this stream ready for reading/writing a packet if necessary
 
streamInfo *info = *nodeState(this, nodeSTREAMINFO);
 
streamInfo *info = *nodeState(this, nodeSTREAMINFO);
char *inbuf = info->inbuf, *outbuf = info->outbuf;
+
char *iBuf = info->iBuf, *oBuf = info->oBuf;
 
int i,fd = info->fd;
 
int i,fd = info->fd;
  
Line 163: Line 161:
 
to.tv_sec = to.tv_usec = svrDELAY;
 
to.tv_sec = to.tv_usec = svrDELAY;
 
if (select(fd+1,&fdset,NULL,NULL,&to)>0) {
 
if (select(fd+1,&fdset,NULL,NULL,&to)>0) {
if ((i = recv(fd,inbuf+info->inptr,svrPAKSIZE,0)) > 0) {
+
if ((i = recv(fd,iBuf+info->iPtr,svrPAKSIZE,0)) > 0) {
 
printf("i=%d\n",i);
 
printf("i=%d\n",i);
while (i--) // Scan newly read packet for msg terminator (or buf-end)
+
while (i--)
if ((info->inptr>svrBUFSIZE-svrPAKSIZE)||(strncmp("\r\n\r\n",inbuf+info->inptr++,4) == 0))
+
if (strncmp("\r\n\r\n",iBuf+info->iPtr++,4) == 0)  
processMessage(info); // process & clear if complete or too big
+
processMessage(info);
 +
info->iPtr %= svrBUFSIZE;
 
}
 
}
 
else if (i == 0) {
 
else if (i == 0) {
Line 173: Line 172:
 
printf("Stream%d closed.\n",fd);
 
printf("Stream%d closed.\n",fd);
 
//nodeLoopRemove(this); // NOTE: reduction not handling PREV
 
//nodeLoopRemove(this); // NOTE: reduction not handling PREV
free(inbuf);
+
free(iBuf);
free(outbuf);
+
free(oBuf);
 
free(info);
 
free(info);
 
}
 
}
Line 188: Line 187:
 
// write a packet from outbuf
 
// write a packet from outbuf
 
//int len = svrPAKSIZE; // or less
 
//int len = svrPAKSIZE; // or less
//if (send(fd,outbuf+outptr,len,0)<0) logErr("send() failed!");
+
//if (send(fd,oBuf+oPtr,len,0)<0) logErr("send() failed!");
 
}
 
}
  
 
}
 
}

Revision as of 07:21, 1 August 2006

// 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


  1. define svrDELAY 0 // normal operation is 0 for no delay
  2. define svrPAKSIZE 10 // keep packet-size small for non-multithreaded design
  3. define svrBUFSIZE 100 // dictates max message size
  4. define svrMAXCLIENTS 1000 // used by listen()
  5. ifndef __WIN32__
  6. define WSAGetLastError() errno
  7. endif

// Socket structures and variables struct sockaddr_in addr; struct timeval to; fd_set fdset; unsigned long int sockopt_on; int sock;

// struct for a stream-node's state to point to typedef struct siStruct { int fd, inPtr, inMsg, outPtr; char *inBuf, *outBuf; } streamInfo;

// Function prototypes //void server(); // declared earlier in nodalHusk since called in root loop //void client(); int server_init(); int server_exit(); int nonblocking(int socket);

int server_init() {

#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;

// struct for a stream-node's state to point to typedef struct siStruct { int fd, inptr, outptr; char *inbuf, *outbuf; } streamInfo;

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

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

// Perform any cleanup for socket int server_exit() { // 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; }

// Process message and clear it from the input buffer void processMessage(streamInfo *info) { msg[svrBUFSIZE-1] = '\0'; printf("\n\nCONTENT:\n%s\n\n",msg); // Handle buffer overrun if (info->iPtr>svrBUFSIZE-svrPAKSIZE) logErr("This message exceeded svrBUFSIZE!");

// Handle GET requests if (strncmp(msg,"GET /",5)==0) {

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

// Exit if /stop requested if (strncmp(msg,"/stop",5)==0) { logAdd("/stop requested, stopping."); exit(EXIT_SUCCESS); }

// Send test response back send(info->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 logAdd("Peerd only supports GET requests currently!"); info->iPtr = 0; // clear input buffer for this stream }

// Function called by each stream-node in network's reduction loop // - there is also a streams-container node containing zero or more stream nodes void server() { FD_ZERO(&fdset); FD_SET(sock,&fdset); to.tv_sec = to.tv_usec = svrDELAY; if (select(sock+1,&fdset,NULL,NULL,&to)>0) { // New connection request, accpet and create new stream-node int fd = nonblocking(accept(sock,NULL,NULL)); if (fd>0) { printf("New connection: Stream%d.\n",fd); streamInfo *newsi = malloc(sizeof(streamInfo)); newsi->fd = fd; newsi->iBuf = malloc(svrBUFSIZE+svrPAKSIZE); newsi->oBuf = malloc(svrBUFSIZE+svrPAKSIZE); newsi->iPtr = newsi->iMsg = newsi->oPtr = 0; node n = nodeGetValue(nodeCLIENTS,nodeLOOP); n = nodeLoopInsert(n,nodeNULL); nodeSetValue(nodeCLIENTS,nodeLOOP,n); nodeSetValue(n,nodeCODE,nodeTRUE); *nodeState(n,nodeNULL) = &client; *nodeState(n,nodeSTREAMINFO) = newsi; } else logErr("accept(): failed!"); } }


// Each of the stream nodes points to this function in its State void client() { // Get the info for this stream ready for reading/writing a packet if necessary streamInfo *info = *nodeState(this, nodeSTREAMINFO); char *iBuf = info->iBuf, *oBuf = info->oBuf; int i,fd = info->fd;

// If any data to recieve, read a packet FD_ZERO(&fdset); FD_SET(fd,&fdset); to.tv_sec = to.tv_usec = svrDELAY; if (select(fd+1,&fdset,NULL,NULL,&to)>0) { if ((i = recv(fd,iBuf+info->iPtr,svrPAKSIZE,0)) > 0) { printf("i=%d\n",i); while (i--) if (strncmp("\r\n\r\n",iBuf+info->iPtr++,4) == 0) processMessage(info); info->iPtr %= svrBUFSIZE; } else if (i == 0) { // zero bytes to read, do orderly termination printf("Stream%d closed.\n",fd); //nodeLoopRemove(this); // NOTE: reduction not handling PREV free(iBuf); free(oBuf); free(info); } else logErrNum("recv() failed returning %d",WSAGetLastError()); }

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

}