Browse code

- more tcp stuff (not sure it compiles though, so don't use -DUSE_TCP yet)

Andrei Pelinescu-Onciul authored on 09/12/2002 18:40:42
Showing 4 changed files
... ...
@@ -241,7 +241,7 @@ static inline int init_su( union sockaddr_union* su,
241 241
 
242 242
 
243 243
 
244
-/* inits a struct sockaddr_union from a struct hostent, an address index int
244
+/* inits a struct sockaddr_union from a struct hostent, an address index in
245 245
  * the hostent structure and a port no.
246 246
  * WARNING: no index overflow  checks!
247 247
  * returns 0 if ok, -1 on error (unknown address family) */
... ...
@@ -33,6 +33,13 @@
33 33
 
34 34
 
35 35
 #define TCP_BUF_SIZE 65535
36
+#define TCP_CON_TIMEOUT 60 /* in  seconds */
37
+#define TCP_CHILD_TIMEOUT 5 /* after 5 seconds, the child "returns" 
38
+							 the connection to the tcp master process */
39
+#define TCP_MAIN_SELECT_TIMEOUT 5 /* how often "tcp main" checks for timeout*/
40
+#define TCP_CHILD_SELECT_TIMEOUT 2 /* the same as above but for children */
41
+
42
+
36 43
 enum {	TCP_REQ_INIT, TCP_REQ_OK, TCP_READ_ERROR, TCP_REQ_OVERRUN, 
37 44
 	 	TCP_REQ_BAD_LEN };
38 45
 enum {	H_SKIP, H_LF, H_LFCR,  H_BODY, H_STARTWS,
... ...
@@ -44,7 +51,6 @@ enum {	H_SKIP, H_LF, H_LFCR,  H_BODY, H_STARTWS,
44 44
 
45 45
 struct tcp_req{
46 46
 	struct tcp_req* next;
47
-	int fd;
48 47
 	/* sockaddr ? */
49 48
 	char buf[TCP_BUF_SIZE]; /* bytes read so far*/
50 49
 	char* pos; /* current position in buf */
... ...
@@ -59,17 +65,51 @@ struct tcp_req{
59 59
 };
60 60
 
61 61
 
62
-#define init_tcp_req( r, f) \
62
+
63
+
64
+struct tcp_connection{
65
+	int s; /*socket, used by "tcp main" */
66
+	int fd; /* used only by "children" */
67
+	struct ip_addr ip; /* peer ip */
68
+	int sock_idx; /* receiving socket index in the tcp_info array */
69
+	union sockaddr_union su;
70
+	struct tcp_req req; /* request data */
71
+	int refcnt;
72
+	int timeout; /* connection timeout, after this it will be removed*/
73
+	struct tcp_connection* next;
74
+	struct tcp_connection* prev;
75
+};
76
+
77
+
78
+
79
+
80
+#define init_tcp_req( r) \
63 81
 	do{ \
64 82
 		memset( (r), 0, sizeof(struct tcp_req)); \
65
-		(r)->fd=(f); \
66 83
 		(r)->parsed=(r)->pos=(r)->buf; \
67 84
 		(r)->error=TCP_REQ_OK;\
68 85
 		(r)->state=H_STARTWS; \
69 86
 	}while(0)
70 87
 
71 88
 
89
+/* add a tcpconn to a list*/
90
+#define tcpconn_listadd(head, c) \
91
+	do{ \
92
+		/* add it at the begining of the list*/ \
93
+		(c)->next=(head); \
94
+		(c)->prev=0; \
95
+		if ((head)) (head)->prev=(c); \
96
+		(head)=(c); \
97
+	} while(0)
98
+
72 99
 
100
+/* remove a tcpconn from a list*/
101
+#define tcpconn_listrm(head, c) \
102
+	do{ \
103
+		if ((head)==(c)) (head)=(c)->next; \
104
+		if ((c)->next) (c)->next->prev=(c)->prev; \
105
+		if ((c)->prev) (c)->prev->next=(c)->next; \
106
+	}while(0)
73 107
 
74 108
 
75 109
 
... ...
@@ -28,6 +28,11 @@
28 28
 
29 29
 #ifdef USE_TCP
30 30
 
31
+
32
+#ifndef SHM_MEM
33
+#error "shared memory support needed (add -DSHM_MEM to Makefile.defs)"
34
+#endif
35
+
31 36
 #include <sys/select.h>
32 37
 
33 38
 #include <sys/time.h>
... ...
@@ -43,10 +48,12 @@
43 43
 
44 44
 #include "ip_addr.h"
45 45
 #include "pass_fd.h"
46
+#include "tcp_conn.h"
46 47
 #include "globals.h"
47 48
 #include "mem/mem.h"
48 49
 
49 50
 
51
+
50 52
 #define local_malloc pkg_malloc
51 53
 #define local_free   pkg_free
52 54
 
... ...
@@ -62,14 +69,6 @@ struct tcp_child{
62 62
 
63 63
 enum { CONN_OK, CONN_ERROR };
64 64
 
65
-struct tcp_connection{
66
-	int s; /*socket */
67
-	struct ip_addr ip;
68
-	union sockaddr_union su;
69
-	int refcnt;
70
-	struct tcp_connection* next;
71
-	struct tcp_connection* prev;
72
-};
73 65
 
74 66
 
75 67
 
... ...
@@ -83,7 +82,7 @@ struct tcp_connection*  tcpconn_add(int sock, union sockaddr_union* su)
83 83
 	struct tcp_connection *c;
84 84
 	
85 85
 
86
-	c=(struct tcp_connection*)local_malloc(sizeof(struct tcp_connection));
86
+	c=(struct tcp_connection*)shm_malloc(sizeof(struct tcp_connection));
87 87
 	if (c==0){
88 88
 		LOG(L_ERR, "ERROR: tcpconn_add: mem. allocation failure\n");
89 89
 		goto error;
... ...
@@ -92,12 +91,11 @@ struct tcp_connection*  tcpconn_add(int sock, union sockaddr_union* su)
92 92
 	c->su=*su;
93 93
 	c->refcnt=0;
94 94
 	su2ip_addr(&c->ip, su);
95
+	init_tcp_req(&c->req);
96
+	c->timeout=get_ticks()+TCP_CON_TIMEOUT;
95 97
 
96 98
 	/* add it at the begining of the list*/
97
-	c->next=conn_list;
98
-	c->prev=0;
99
-	if (conn_list) conn_list->prev=c;
100
-	conn_list=c;
99
+	tcpconn_listadd(conn_list, c);
101 100
 	return c;
102 101
 	
103 102
 error:
... ...
@@ -108,11 +106,29 @@ error:
108 108
 
109 109
 void tcpconn_rm(struct tcp_connection* c)
110 110
 {
111
+	tcpconn_listrm(conn_list, c);
112
+	shm_free(c);
113
+}
114
+
115
+
116
+/* very ineficient for now, use hashtable some day - FIXME*/
117
+void tcpconn_timeout()
118
+{
119
+	struct tcp_connection *c, *next;
120
+	int jiffies;;
111 121
 	
112
-	if (conn_list==c) conn_list=c->next;
113
-	if (c->next) c->next->prev=c->prev;
114
-	if (c->prev) c->prev->next=c->next;
115
-	local_free(c);
122
+	
123
+	jiffies=get_ticks();
124
+	c=conn_list;
125
+	while(c){
126
+		next=c->next;
127
+		if ((c->refcnt==0) && (jiffies<c->timeout)) {
128
+			DBG("tcpconn_timeout: timeout for %p (%d < %d)\n",
129
+					c, jiffies, c->timeout);
130
+			tcpconn_rm(c);
131
+		}
132
+		c=next;
133
+	}
116 134
 }
117 135
 
118 136
 
... ...
@@ -195,6 +211,7 @@ void tcp_main_loop()
195 195
 	int state;
196 196
 	int bytes;
197 197
 	socklen_t su_len;
198
+	struct timeval timeout;
198 199
 
199 200
 	/*init */
200 201
 	maxfd=0;
... ...
@@ -219,9 +236,14 @@ void tcp_main_loop()
219 219
 	
220 220
 	while(1){
221 221
 		sel_set=master_set;
222
-		n=select(maxfd+1, &sel_set, 0 ,0 , 0);
222
+		timeout->tv_sec=TCP_MAIN_SELECT_TIMEOUT;
223
+		timeout->tv_usec=0;
224
+		n=select(maxfd+1, &sel_set, 0 ,0 , &timeout);
223 225
 		if (n<0){
226
+			if (errno==EINTR) continue; /* just a signal */
224 227
 			/* errors */
228
+			LOG(L_ERR, "ERROR: tcp_main_loop: select:(%d) %s\n", errno,
229
+					strerror(errno));
225 230
 		}
226 231
 		
227 232
 		for (r=0; r<sock_no && n; r++){
... ...
@@ -301,6 +323,8 @@ read_again:
301 301
 						if (tcpconn->refcnt==0){
302 302
 							FD_SET(tcpconn->s, &master_set);
303 303
 							if (maxfd<tcpconn->s) maxfd=tcpconn->s;
304
+							/* update the timeout*/
305
+							tcpconn->timeout=get_ticks()+TCP_CON_TIMEOUT;
304 306
 						}
305 307
 					}else{
306 308
 						/*error, we should destroy it */
... ...
@@ -317,6 +341,9 @@ read_again:
317 317
 				}
318 318
 			}
319 319
 		}
320
+		
321
+		/* remove old connections */
322
+		tcpconn_timeout();
320 323
 	
321 324
 	}
322 325
 }
... ...
@@ -52,7 +52,7 @@
52 52
 /* reads next available bytes
53 53
  * return number of bytes read, 0 on EOF or -1 on error,
54 54
  * sets also r->error */
55
-int tcp_read(struct tcp_req *r)
55
+int tcp_read(struct tcp_req *r, int fd)
56 56
 {
57 57
 	int bytes_free, bytes_read;
58 58
 	
... ...
@@ -64,7 +64,7 @@ int tcp_read(struct tcp_req *r)
64 64
 		return -1;
65 65
 	}
66 66
 again:
67
-	bytes_read=read(r->fd, r->pos, bytes_free);
67
+	bytes_read=read(fd, r->pos, bytes_free);
68 68
 
69 69
 	if(bytes_read==-1){
70 70
 		if (errno == EWOULDBLOCK || errno == EAGAIN){
... ...
@@ -92,7 +92,7 @@ again:
92 92
  * when either r->body!=0 or r->state==H_BODY =>
93 93
  * all headers have been read. It should be called in a while loop.
94 94
  * returns < 0 if error or 0 if EOF */
95
-int tcp_read_headers(struct tcp_req *r)
95
+int tcp_read_headers(struct tcp_req *r, int fd)
96 96
 {
97 97
 	int bytes, remaining;
98 98
 	char *p;
... ...
@@ -133,7 +133,7 @@ int tcp_read_headers(struct tcp_req *r)
133 133
 
134 134
 
135 135
 	
136
-	bytes=tcp_read(r);
136
+	bytes=tcp_read(r, fd);
137 137
 	if (bytes<=0) return bytes;
138 138
 	p=r->parsed;
139 139
 	
... ...
@@ -313,52 +313,86 @@ skip:
313 313
 
314 314
 void tcp_receive_loop(int unix_sock)
315 315
 {
316
-	struct tcp_req req;
316
+	struct tcp_req* req;
317
+	struct tcp_connection* list; /* list with connections in use */
318
+	struct tcp_connection* con;
317 319
 	int bytes;
318 320
 	long size;
319 321
 	int n;
320
-	long id;
322
+	int nfds;
321 323
 	int s;
322 324
 	long state;
323 325
 	long response[2];
326
+	fd_set master_set;
327
+	fd_set sel_set;
328
+	int maxfd;
329
+	struct timeval timeout;
324 330
 	
325 331
 	
326
-	
332
+	/* init */
333
+	list=con=0;
334
+	FD_ZERO(&master_set);
335
+	FD_SET(unix_sock, &master_set);
336
+	maxfd=unix_sock;
327 337
 	
328 338
 	/* listen on the unix socket for the fd */
329 339
 	for(;;){
330
-			n=receive_fd(unix_sock, &id, sizeof(id), &s);
331
-			if (n<0){
332
-				if (errno == EWOULDBLOCK || errno == EAGAIN || errno == EINTR){
333
-					continue;
334
-				}else{
335
-					fprintf(stderr, "ERROR: tcp_receive_loop: read_fd: %s\n",
340
+			timeout->tv_sec=TCP_CHILD_SELECT_TIMEOUT;
341
+			timeout->tv_usec=0;
342
+			sel_set=master_set;
343
+			nfds=select(maxfd+1, &sel_set, 0 , 0 , &timeout);
344
+			if (nfds<0){
345
+				if (errno==EINTR) continue; /* just a signal */
346
+				/* errors */
347
+				LOG(L_ERR, "ERROR: tcp_receive_loop: select:(%d) %s\n", errno,
348
+					strerror(errno));
349
+				continue;
350
+			}
351
+			if (FD_ISSET(unix_sock, &sel_set)){
352
+				nfds--;
353
+				/* a new conn from "main" */
354
+				n=receive_fd(unix_sock, &con, sizeof(con), &s);
355
+				if (n<0){
356
+					if (errno == EWOULDBLOCK || errno == EAGAIN ||
357
+							errno == EINTR){
358
+						continue;
359
+					}else{
360
+						LOG(L_CRIT,"BUG: tcp_receive_loop: read_fd: %s\n",
336 361
 							strerror(errno));
337
-					abort(); /* big error*/
362
+						abort(); /* big error*/
363
+					}
338 364
 				}
339
-			}
340
-			if (n==0){
341
-					fprintf(stderr, 
342
-							"WARNING: tcp_receive_loop: 0 bytes read\n");
365
+				if (n==0){
366
+					LOG(L_ERR, "WARNING: tcp_receive_loop: 0 bytes read\n");
343 367
 					continue;
344
-			}
345
-			fprintf(stderr, "received n=%d id=%ld, fd=%d\n", n, id, s);
346
-			if (s==-1) {
347
-					fprintf(stderr, "ERROR: tcp_receive_loop: read_fd:"
368
+				}
369
+				DBG("received n=%d con=%ld, fd=%d\n", n, con, s);
370
+				if (s==-1) {
371
+					LOG(L_ERR, "ERROR: tcp_receive_loop: read_fd:"
348 372
 									"no fd read\n");
349 373
 					state=-1;
350 374
 					goto end_req; /* ?*/
375
+				}
376
+				if (con==0){
377
+					LOG(L_ERR, "ERROR: tcp_receive_loop: null pointer\n");
378
+					state=-1;
379
+					goto end_req;
380
+				}
381
+				con->fd=s;
382
+				FD_SET(s, &master_set);
383
+				if (maxfd<s) maxfd=s;
384
+				tcpconn_listadd(list, con);
351 385
 			}
352
-		
353
-		init_tcp_req(&req, s);
354
-		
355
-		
356
-	again:
357
-		while(req.complete==0 && req.error==TCP_REQ_OK){
358
-			bytes=tcp_read_headers(&req);
359
-			/* if timeout state=0; goto end__req; */
386
+			for (con=list; con && nfds ; con=con->next){
387
+				if (FD_ISSET(con->fd, &sel_set)){
388
+					nfds--;
389
+					req=&con->req;
390
+again:
391
+		while(req->complete==0 && req->error==TCP_REQ_OK){
392
+			bytes=tcp_read_headers(req, s);
393
+						/* if timeout state=0; goto end__req; */
360 394
 			fprintf(stderr, "read= %d bytes, parsed=%d, state=%d, error=%d\n",
361
-					bytes, req.parsed-req.buf, req.state, req.error );
395
+					bytes, req->parsed-req->buf, req->state, req->error );
362 396
 			if (bytes==-1){
363 397
 				fprintf(stderr, "ERROR!\n");
364 398
 				state=-1;
... ...
@@ -371,19 +405,19 @@ void tcp_receive_loop(int unix_sock)
371 371
 			}
372 372
 
373 373
 		}
374
-		if (req.error!=TCP_REQ_OK){
374
+		if (req->error!=TCP_REQ_OK){
375 375
 			fprintf(stderr, "bad request, state=%d, error=%d\n",
376
-					req.state, req.error);
376
+					req->state, req->error);
377 377
 			state=-1;
378 378
 			goto end_req;
379 379
 		}
380 380
 		fprintf(stderr, "end of header part\n");
381
-		fprintf(stderr, "headers:\n%.*s.\n",req.body-req.buf, req.buf);
382
-		if (req.has_content_len){
383
-			fprintf(stderr, "content-length= %d\n", req.content_len);
384
-			fprintf(stderr, "body:\n%.*s\n", req.content_len, req.body);
381
+		fprintf(stderr, "headers:\n%.*s.\n",req->body-req->buf, req->buf);
382
+		if (req->has_content_len){
383
+			fprintf(stderr, "content-length= %d\n", req->content_len);
384
+			fprintf(stderr, "body:\n%.*s\n", req->content_len, req->body);
385 385
 		}else{
386
-			req.error=TCP_REQ_BAD_LEN;
386
+			req->error=TCP_REQ_BAD_LEN;
387 387
 			fprintf(stderr, "content length not present or unparsable\n");
388 388
 			state=-1;
389 389
 			goto end_req;
... ...
@@ -393,21 +427,21 @@ void tcp_receive_loop(int unix_sock)
393 393
 		state=0;
394 394
 		/* just for debugging use sendipv4 as receiving socket */
395 395
 		DBG("calling receive_msg(%p, %d, %p)\n",
396
-				req.buf, (int)(req.parsed-req.buf), &sendipv4->su);
396
+				req->buf, (int)(req->parsed-req->buf), &sendipv4->su);
397 397
 		bind_address=sendipv4;
398
-		receive_msg(req.buf, req.parsed-req.buf, &sendipv4->su);
398
+		receive_msg(req->buf, req->parsed-req->buf, &sendipv4->su);
399 399
 
400 400
 		/* prepare for next request */
401
-		size=req.pos-req.body;
402
-		if (size) memmove(req.buf, req.body, size);
401
+		size=req->pos-req->body;
402
+		if (size) memmove(req->buf, req->body, size);
403 403
 		fprintf(stderr, "\npreparing for new request, kept %ld bytes\n", size);
404
-		req.pos=req.buf+size;
405
-		req.parsed=req.buf;
406
-		req.body=0;
407
-		req.error=TCP_REQ_OK;
408
-		req.state=H_STARTWS;
409
-		req.complete=req.content_len=req.has_content_len=0;
410
-		req.bytes_to_go=0;
404
+		req->pos=req->buf+size;
405
+		req->parsed=req->buf;
406
+		req->body=0;
407
+		req->error=TCP_REQ_OK;
408
+		req->state=H_STARTWS;
409
+		req->complete=req->content_len=req->has_content_len=0;
410
+		req->bytes_to_go=0;
411 411
 	
412 412
 		/* process last req. */
413 413
 		
... ...
@@ -418,7 +452,7 @@ void tcp_receive_loop(int unix_sock)
418 418
 		/* release req & signal the parent */
419 419
 		if (s!=-1) close(s);
420 420
 		/* errno==EINTR, EWOULDBLOCK a.s.o todo */
421
-		response[0]=id;
421
+		response[0]=con;
422 422
 		response[1]=state;
423 423
 		write(unix_sock, response, sizeof(response));
424 424