Browse code

- more tcp stuff and a lot of merging w/ latest cvs - content-length is appended automatically to messages that cross from udp to tcp - tcp2udp and udp2tcp now work under heavy stress (e.g.: throttle 200, 10 ser processes on dual cpu) - tcp performance still sucks, some things like disabling Nagle are still not in yet (for better debugging)

Andrei Pelinescu-Onciul authored on 07/02/2003 17:02:15
Showing 14 changed files
... ...
@@ -8,7 +8,7 @@
8 8
 VERSION = 0
9 9
 PATCHLEVEL = 8
10 10
 SUBLEVEL =   11
11
-EXTRAVERSION = pre5-tcp1-locking
11
+EXTRAVERSION = pre6-tcp2
12 12
 
13 13
 RELEASE=$(VERSION).$(PATCHLEVEL).$(SUBLEVEL)$(EXTRAVERSION)
14 14
 OS = $(shell uname -s | sed -e s/SunOS/solaris/ | tr "[A-Z]" "[a-z]")
... ...
@@ -38,6 +38,7 @@ New features
38 38
 - powerpc fast locking support
39 39
 - netbsd support
40 40
 - 64 bits arch. support (e.g. netbsd/sparc64).
41
+- tcp2udp and udp2tcp stateless forwarding (see forward_udp & forward_tcp)
41 42
 
42 43
 Changes to use of ser scripts
43 44
 =============================
... ...
@@ -10,7 +10,7 @@ x fix 0 parameter module f. call
10 10
 x better Via parsing (handle ' ' in uri, eg: foo.bar : 1234 ; received=) and
11 11
  ipv6 addresses ([fec0:aa::01]).
12 12
 - fix format string vulnerability in log()
13
-- fix alignement access problems (warning on Sun)
13
+- fix alignment access problems (warning on Sun)
14 14
 x (different way) add request header bitmap field for the modules
15 15
 - introduce variables & function in the script language (cfg. file)
16 16
 
... ...
@@ -69,4 +69,5 @@ x freopen stdin, stdout, stderr to /dev/null
69 69
 x generic locking lib
70 70
 - convert tm to use new locking lib
71 71
 - tcp disable nagle & other socket stuff (close()?)
72
+- force add rport (setflag(rport)???)
72 73
 
... ...
@@ -43,7 +43,7 @@
43 43
 #include <dmalloc.h>
44 44
 #endif
45 45
 
46
-/* WARNING: all lump add/insert operations excpect a pkg_malloc'ed char* 
46
+/* WARNING: all lump add/insert operations expect a pkg_malloc'ed char* 
47 47
  * pointer the will be DEALLOCATED when the sip_msg is destroyed! */
48 48
 
49 49
 enum lump_dir { LD_NEXT, LD_BEFORE, LD_AFTER };
... ...
@@ -312,7 +312,7 @@ int forward_request( struct sip_msg* msg, struct proxy_l * p, int proto)
312 312
 		goto error1;
313 313
 	}
314 314
 	 /* send it! */
315
-	DBG("Sending:\n%s.\n", buf);
315
+	DBG("Sending:\n%.*s.\n", (int)len, buf);
316 316
 	DBG("orig. len=%d, new_len=%d, proto=%d\n", msg->len, len, proto );
317 317
 	
318 318
 	
... ...
@@ -510,8 +510,9 @@ int forward_reply(struct sip_msg* msg)
510 510
 		STATS_TX_RESPONSE(  (msg->first_line.u.reply.statuscode/100) );
511 511
 #endif
512 512
 
513
-	DBG(" reply forwarded to %s:%d\n", msg->via2->host.s,
514
-		(unsigned short) msg->via2->port);
513
+	DBG(" reply forwarded to %.*s:%d\n", 
514
+			msg->via2->host.len, msg->via2->host.s,
515
+			(unsigned short) msg->via2->port);
515 516
 
516 517
 	pkg_free(new_buf);
517 518
 	free(to);
... ...
@@ -154,25 +154,11 @@ void hashtest_cycle( int hits[TABLE_ENTRIES+5], char *ip )
154 154
 				}
155 155
 }
156 156
 
157
-int init_hash()
158
-{
159
-	if (TABLE_ENTRIES != (1<<10)) {
160
-		LOG(L_WARN, "WARNING: hash function optimized for %d entries\n",
161
-			1<<10);
162
-		LOG(L_WARN, "WARNING: use of %d entries may lead "
163
-			"to unflat distribution\n", TABLE_ENTRIES );
164
-	} else {
165
-		DBG("DEBUG: hash function initialized with optimum table size\n");
166
-	}
167
-	return 1;
168
-}
169
-
170 157
 void hashtest()
171 158
 {
172 159
 	int hits[TABLE_ENTRIES+5];
173 160
 	int i;
174 161
 
175
-	init_hash();	
176 162
 	memset( hits, 0, sizeof hits );
177 163
 	hashtest_cycle( hits, "192.168.99.100" );
178 164
 	hashtest_cycle( hits, "172.168.99.100" );
... ...
@@ -39,7 +39,6 @@
39 39
 int new_hash( str  call_id, str cseq_nr );
40 40
 int new_hash2( str  call_id, str cseq_nr );
41 41
 
42
-int init_hash();
43 42
 
44 43
 #define hash( cid, cseq) new_hash2( cid, cseq )
45 44
 
... ...
@@ -880,8 +880,10 @@ static void sig_usr(int signo)
880 880
 	}else{
881 881
 		/* process the important signals */
882 882
 		switch(signo){
883
-			case SIGINT:
884 883
 			case SIGPIPE:
884
+					LOG(L_INFO, "INFO: signal %d received\n", signo);
885
+				break;
886
+			case SIGINT:
885 887
 			case SIGTERM:
886 888
 					LOG(L_INFO, "INFO: signal %d received\n", signo);
887 889
 					/* print memory stats for non-main too */
... ...
@@ -1280,11 +1282,6 @@ try_again:
1280 1280
 	DBG("test random number %u\n", rand());
1281 1281
 	
1282 1282
 	
1283
-	/* init hash fucntion */
1284
-	if (init_hash()<0) {
1285
-		LOG(L_ERR, "ERROR: init_hash failed\n");
1286
-		goto error;
1287
-	}
1288 1283
 
1289 1284
 	/*init mallocs (before parsing cfg !)*/
1290 1285
 	if (init_mallocs()==-1)
... ...
@@ -342,6 +342,44 @@ char* id_builder(struct sip_msg* msg, unsigned int *id_len)
342 342
 
343 343
 
344 344
 
345
+char* clen_builder(struct sip_msg* msg, unsigned int *clen_len)
346
+{
347
+	char* buf;
348
+	int len;
349
+	int value;
350
+	char* value_s;
351
+	int value_len;
352
+	char* body;
353
+	
354
+	
355
+	body=get_body(msg);
356
+	if (body==0){
357
+		ser_error=E_BAD_REQ;
358
+		LOG(L_ERR, "ERROR: clen_builder: no message body found"
359
+					" (missing crlf?)");
360
+		return 0;
361
+	}
362
+	value=msg->len-(int)(body-msg->buf);
363
+	value_s=int2str(value, &value_len);
364
+	DBG("clen_builder: content-length: %d (%s)\n", value, value_s);
365
+		
366
+	len=CONTENT_LENGTH_LEN+value_len+CRLF_LEN;
367
+	buf=pkg_malloc(sizeof(char)*(len+1));
368
+	if (buf==0){
369
+		ser_error=E_OUT_OF_MEM;
370
+		LOG(L_ERR, "ERROR: clen_builder: out of memory\n");
371
+		return 0;
372
+	}
373
+	memcpy(buf, CONTENT_LENGTH, CONTENT_LENGTH_LEN);
374
+	memcpy(buf+CONTENT_LENGTH_LEN, value_s, value_len);
375
+	memcpy(buf+CONTENT_LENGTH_LEN+value_len, CRLF, CRLF_LEN);
376
+	buf[len+1]=0; /* null terminate it */
377
+	*clen_len=len;
378
+	return buf;
379
+}
380
+
381
+
382
+
345 383
 /* computes the "unpacked" len of a lump list,
346 384
    code moved from build_req_from_req */
347 385
 static inline int lumps_len(struct lump* l)
... ...
@@ -548,10 +586,14 @@ char * build_req_buf_from_sip_req( struct sip_msg* msg,
548 548
 #ifdef USE_TCP
549 549
 	char* id_buf;
550 550
 	int id_len;
551
+	char* clen_buf;
552
+	int clen_len;
551 553
 	
552 554
 	
553 555
 	id_buf=0;
554 556
 	id_len=0;
557
+	clen_buf=0;
558
+	clen_len=0;
555 559
 #endif
556 560
 	extra_params.len=0;
557 561
 	extra_params.s=0;
... ...
@@ -580,6 +622,24 @@ char * build_req_buf_from_sip_req( struct sip_msg* msg,
580 580
 		extra_params.s=id_buf;
581 581
 		extra_params.len=id_len;
582 582
 	}
583
+	/* if sending proto == tcp, check if Content-Length needs to be added*/
584
+	if (proto==PROTO_TCP){
585
+		/* first of all parse content-length */
586
+		if (parse_headers(msg, HDR_CONTENTLENGTH, 0)==-1){
587
+			LOG(L_ERR, "build_req_buf_from_sip_req:"
588
+							" error parsing content-length\n");
589
+			goto skip_clen;
590
+		}
591
+		if (msg->content_length==0){
592
+			/* we need to add it */
593
+			if ((clen_buf=clen_builder(msg, &clen_len))==0){
594
+				LOG(L_ERR, "build_req_buf_from_sip_req:" 
595
+								" clen_builder failed\n");
596
+				goto skip_clen;
597
+			}
598
+		}
599
+	}
600
+skip_clen:
583 601
 #endif
584 602
 	branch.s=msg->add_to_branch_s;
585 603
 	branch.len=msg->add_to_branch_len;
... ...
@@ -652,6 +712,19 @@ char * build_req_buf_from_sip_req( struct sip_msg* msg,
652 652
 		if (insert_new_lump_after(anchor, rport_buf, rport_len, HDR_VIA)==0)
653 653
 			goto error03; /* free rport_buf*/
654 654
 	}
655
+#ifdef USE_TCP
656
+	/* if clen needs to be added, add it */
657
+	if (clen_len){
658
+		/* msg->unparsed should point just before the final crlf,
659
+		 * parse_headers is called from clen_builder */
660
+		anchor=anchor_lump(&(msg->add_rm), msg->unparsed-buf, 0,
661
+							 HDR_CONTENTLENGTH);
662
+		if (anchor==0) goto error04; /* free clen_buf*/
663
+		if (insert_new_lump_after(anchor, clen_buf, clen_len,
664
+					HDR_CONTENTLENGTH)==0)
665
+			goto error04; /* free clen_buf*/
666
+	}
667
+#endif
655 668
 
656 669
 	/* compute new msg len and fix overlapping zones*/
657 670
 	new_len=len+lumps_len(msg->add_rm);
... ...
@@ -716,12 +789,17 @@ error02:
716 716
 	if (received_buf) pkg_free(received_buf);
717 717
 error03:
718 718
 	if (rport_buf) pkg_free(rport_buf);
719
+#ifdef USE_TCP
720
+error04:
721
+	if (clen_buf) pkg_free(clen_buf);
722
+#endif
719 723
 error00:
720 724
 	*returned_len=0;
721 725
 	return 0;
722 726
 }
723 727
 
724 728
 
729
+
725 730
 char * build_res_buf_from_sip_res( struct sip_msg* msg,
726 731
 				unsigned int *returned_len)
727 732
 {
... ...
@@ -733,7 +811,14 @@ char * build_res_buf_from_sip_res( struct sip_msg* msg,
733 733
 #endif
734 734
 	char* buf;
735 735
 	unsigned int len;
736
-
736
+#ifdef USE_TCP
737
+	struct lump* anchor;
738
+	char* clen_buf;
739
+	int clen_len;
740
+	
741
+	clen_buf=0;
742
+	clen_len=0;
743
+#endif
737 744
 #ifdef SCRATCH
738 745
 	orig=msg->orig;
739 746
 #endif
... ...
@@ -763,12 +848,53 @@ char * build_res_buf_from_sip_res( struct sip_msg* msg,
763 763
 		via_offset=msg->h_via1->name.s-buf;
764 764
 	}
765 765
 #endif
766
+
767
+#ifdef USE_TCP
768
+
769
+	/* if sending proto == tcp, check if Content-Length needs to be added*/
770
+	if (msg->via2 && (msg->via2->proto==PROTO_TCP)){
771
+		DBG("build_res_from_sip_res: checking content-length for \n%.*s\n",
772
+				(int)msg->len, msg->buf);
773
+		/* first of all parse content-length */
774
+		if (parse_headers(msg, HDR_CONTENTLENGTH, 0)==-1){
775
+			LOG(L_ERR, "build_res_buf_from_sip_res:"
776
+							" error parsing content-length\n");
777
+			goto skip_clen;
778
+		}
779
+		if (msg->content_length==0){
780
+			DBG("build_res_from_sip_res: no content_length hdr found\n");
781
+			/* we need to add it */
782
+			if ((clen_buf=clen_builder(msg, &clen_len))==0){
783
+				LOG(L_ERR, "build_res_buf_from_sip_res:" 
784
+								" clen_builder failed\n");
785
+				goto skip_clen;
786
+			}
787
+		}
788
+	}
789
+skip_clen:
790
+#endif
791
+	
766 792
 	/* remove the first via*/
767 793
 	if (del_lump( &(msg->repl_add_rm), via_offset, via_len, HDR_VIA)==0){
768 794
 		LOG(L_ERR, "build_res_buf_from_sip_res: error trying to remove first"
769 795
 					"via\n");
770 796
 		goto error;
771 797
 	}
798
+#ifdef USE_TCP
799
+	/* if clen needs to be added, add it */
800
+	if (clen_len){
801
+		/* msg->unparsed should point just before the final crlf,
802
+		 * parse_headers is called from clen_builder */
803
+		anchor=anchor_lump(&(msg->repl_add_rm), msg->unparsed-buf, 0, 
804
+							HDR_CONTENTLENGTH);
805
+		DBG("build_res_from_sip_res: adding content-length: %.*s\n",
806
+				clen_len, clen_buf);
807
+		if (anchor==0) goto error_clen; /* free clen_buf*/
808
+		if (insert_new_lump_after(anchor, clen_buf, clen_len,
809
+					HDR_CONTENTLENGTH)==0)
810
+			goto error_clen; /* free clen_buf*/
811
+	}
812
+#endif
772 813
 	new_len=len+lumps_len(msg->repl_add_rm);
773 814
 
774 815
 	DBG(" old size: %d, new size: %d\n", len, new_len);
... ...
@@ -796,12 +922,15 @@ char * build_res_buf_from_sip_res( struct sip_msg* msg,
796 796
 #endif
797 797
 		len-s_offset);
798 798
 	 /* send it! */
799
-	DBG(" copied size: orig:%d, new: %d, rest: %d\n",
800
-			s_offset, offset,
801
-			len-s_offset );
799
+	DBG("build_res_from_sip_res: copied size: orig:%d, new: %d, rest: %d"
800
+			" msg=\n%s\n", s_offset, offset, len-s_offset, new_buf);
802 801
 
803 802
 	*returned_len=new_len;
804 803
 	return new_buf;
804
+#ifdef USE_TCP
805
+error_clen:
806
+	if (clen_buf) pkg_free(clen_buf);
807
+#endif
805 808
 error:
806 809
 	*returned_len=0;
807 810
 	return 0;
... ...
@@ -28,6 +28,7 @@
28 28
  * ---------
29 29
  * 2003-01-29 transport-independent message zero-termination in
30 30
  *            receive_msg (jiri)
31
+ * 2003-02-07 undoed jiri's zero term. changes (they break tcp) (andrei)
31 32
  */
32 33
 
33 34
 
... ...
@@ -81,7 +82,6 @@ int receive_msg(char* buf, unsigned int len, struct receive_info* rcv_info)
81 81
 	/* zero termination (termination of orig message bellow not that
82 82
 	   useful as most of the work is done with scrath-pad; -jiri  */
83 83
 	/* buf[len]=0; */ /* WARNING: zero term removed! */
84
-	buf[len]=0; /* transport-independent zero-termination */
85 84
 	msg->rcv=*rcv_info;
86 85
 	msg->id=msg_no;
87 86
 #ifdef SCRATCH
... ...
@@ -36,10 +36,12 @@
36 36
 #define _tcp_conn_h
37 37
 
38 38
 #include "ip_addr.h"
39
+#include "locking.h"
39 40
 
40 41
 
41 42
 #define TCP_BUF_SIZE 65535
42 43
 #define TCP_CON_TIMEOUT 60 /* in  seconds */
44
+#define TCP_CON_SEND_TIMEOUT 30 /* timeout after a send */
43 45
 #define TCP_CHILD_TIMEOUT 5 /* after 5 seconds, the child "returns" 
44 46
 							 the connection to the tcp master process */
45 47
 #define TCP_MAIN_SELECT_TIMEOUT 5 /* how often "tcp main" checks for timeout*/
... ...
@@ -82,6 +84,7 @@ struct tcp_req{
82 82
 struct tcp_connection{
83 83
 	int s; /*socket, used by "tcp main" */
84 84
 	int fd; /* used only by "children", don't modify it! private data! */
85
+	lock_t write_lock;
85 86
 	int id; /* id (unique!) used to retrieve a specific connection when
86 87
 	           reply-ing*/
87 88
 	struct receive_info rcv; /* src & dst ip, ports, proto a.s.o*/
... ...
@@ -92,9 +95,10 @@ struct tcp_connection{
92 92
 	union sockaddr_union su;
93 93
 #endif
94 94
 	struct tcp_req req; /* request data */
95
-	int refcnt;
95
+	volatile int refcnt;
96
+	int bad; /* if set this is a "bad" connection */
96 97
 	int timeout; /* connection timeout, after this it will be removed*/
97
-	unsigned addr_hash; /* hash indexes in thge 2 tables */
98
+	unsigned addr_hash; /* hash indexes in the 2 tables */
98 99
 	unsigned id_hash;
99 100
 	struct tcp_connection* next; /* next, prev in hash table, used by "main" */
100 101
 	struct tcp_connection* prev;
... ...
@@ -38,6 +38,7 @@
38 38
 #include <sys/time.h>
39 39
 #include <sys/types.h>
40 40
 #include <sys/socket.h>
41
+#include <sys/uio.h>  /* writev*/
41 42
 
42 43
 #include <unistd.h>
43 44
 
... ...
@@ -102,9 +103,15 @@ struct tcp_connection* tcpconn_new(int sock, union sockaddr_union* su,
102 102
 	}
103 103
 	c->s=sock;
104 104
 	c->fd=-1; /* not initialized */
105
+	if (lock_init(&c->write_lock)==0){
106
+		LOG(L_ERR, "ERROR: tcpconn_add: init lock failed\n");
107
+		goto error;
108
+	}
109
+	
105 110
 	c->rcv.src_su=*su;
106 111
 	
107 112
 	c->refcnt=0;
113
+	c->bad=0;
108 114
 	su2ip_addr(&c->rcv.src_ip, su);
109 115
 	c->rcv.src_port=su_getport(su);
110 116
 	c->rcv.proto=PROTO_TCP;
... ...
@@ -171,6 +178,16 @@ struct tcp_connection*  tcpconn_add(struct tcp_connection *c)
171 171
 }
172 172
 
173 173
 
174
+/* unsafe tcpconn_rm version (nolocks) */
175
+void _tcpconn_rm(struct tcp_connection* c)
176
+{
177
+	tcpconn_listrm(tcpconn_addr_hash[c->addr_hash], c, next, prev);
178
+	tcpconn_listrm(tcpconn_id_hash[c->id_hash], c, id_next, id_prev);
179
+	lock_destroy(&c->write_lock);
180
+	shm_free(c);
181
+}
182
+
183
+
174 184
 
175 185
 void tcpconn_rm(struct tcp_connection* c)
176 186
 {
... ...
@@ -178,6 +195,7 @@ void tcpconn_rm(struct tcp_connection* c)
178 178
 	tcpconn_listrm(tcpconn_addr_hash[c->addr_hash], c, next, prev);
179 179
 	tcpconn_listrm(tcpconn_id_hash[c->id_hash], c, id_next, id_prev);
180 180
 	TCPCONN_UNLOCK;
181
+	lock_destroy(&c->write_lock);
181 182
 	shm_free(c);
182 183
 }
183 184
 
... ...
@@ -198,7 +216,7 @@ struct tcp_connection* _tcpconn_find(int id, struct ip_addr* ip, int port)
198 198
 			DBG("c=%p, c->id=%d, ip=",c, c->id);
199 199
 			print_ip(&c->rcv.src_ip);
200 200
 			DBG(" port=%d\n", ntohs(c->rcv.src_port));
201
-			if (id==c->id) return c;
201
+			if ((id==c->id)&&(!c->bad)) return c;
202 202
 		}
203 203
 	}else if (ip){
204 204
 		hash=tcp_addr_hash(ip, port);
... ...
@@ -206,7 +224,8 @@ struct tcp_connection* _tcpconn_find(int id, struct ip_addr* ip, int port)
206 206
 			DBG("c=%p, c->id=%d, ip=",c, c->id);
207 207
 			print_ip(&c->rcv.src_ip);
208 208
 			DBG(" port=%d\n", ntohs(c->rcv.src_port));
209
-			if ( (port==c->rcv.src_port) && (ip_addr_cmp(ip, &c->rcv.src_ip)) )
209
+			if ( (!c->bad) && (port==c->rcv.src_port) &&
210
+					(ip_addr_cmp(ip, &c->rcv.src_ip)) )
210 211
 				return c;
211 212
 		}
212 213
 	}
... ...
@@ -215,13 +234,17 @@ struct tcp_connection* _tcpconn_find(int id, struct ip_addr* ip, int port)
215 215
 
216 216
 
217 217
 
218
-/* _tcpconn_find with locks */
219
-struct tcp_connection* tcpconn_get(int id, struct ip_addr* ip, int port)
218
+/* _tcpconn_find with locks and timeout */
219
+struct tcp_connection* tcpconn_get(int id, struct ip_addr* ip, int port,
220
+									int timeout)
220 221
 {
221 222
 	struct tcp_connection* c;
222 223
 	TCPCONN_LOCK;
223 224
 	c=_tcpconn_find(id, ip, port);
224
-	if (c) c->refcnt++;
225
+	if (c){ 
226
+			c->refcnt++;
227
+			c->timeout=get_ticks()+timeout;
228
+	}
225 229
 	TCPCONN_UNLOCK;
226 230
 	return c;
227 231
 }
... ...
@@ -249,9 +272,9 @@ int tcp_send(char* buf, unsigned len, union sockaddr_union* to, int id)
249 249
 	if (to){
250 250
 		su2ip_addr(&ip, to);
251 251
 		port=su_getport(to);
252
-		c=tcpconn_get(id, &ip, port); /* lock ;inc refcnt; unlock */
252
+		c=tcpconn_get(id, &ip, port, TCP_CON_SEND_TIMEOUT); 
253 253
 	}else if (id){
254
-		c=tcpconn_get(id, 0, 0);
254
+		c=tcpconn_get(id, 0, 0, TCP_CON_SEND_TIMEOUT);
255 255
 	}else{
256 256
 		LOG(L_CRIT, "BUG: tcp_send called with null id & to\n");
257 257
 		return -1;
... ...
@@ -260,7 +283,8 @@ int tcp_send(char* buf, unsigned len, union sockaddr_union* to, int id)
260 260
 	if (id){
261 261
 		if (c==0) {
262 262
 			if (to){
263
-				c=tcpconn_get(0, &ip, port); /* try again w/o id */
263
+				/* try again w/o id */
264
+				c=tcpconn_get(0, &ip, port, TCP_CON_SEND_TIMEOUT);
264 265
 				goto no_id;
265 266
 			}else{
266 267
 				LOG(L_ERR, "ERROR: tcp_send: id %d not found, dropping\n",
... ...
@@ -323,8 +347,27 @@ get_fd:
323 323
 	
324 324
 send_it:
325 325
 	DBG("tcp_send: sending...\n");
326
-	n=write(fd, buf, len);
326
+	lock_get(&c->write_lock);
327
+	n=send(fd, buf, len, MSG_NOSIGNAL);
328
+	lock_release(&c->write_lock);
327 329
 	DBG("tcp_send: after write: c= %p n=%d fd=%d\n",c, n, fd);
330
+	DBG("tcp_send: buf=\n%.*s\n", (int)len, buf);
331
+	if (n<0){
332
+		LOG(L_ERR, "ERROR: tcpsend: failed to send, n=%d: %s (%d)\n",
333
+				n, strerror(errno), errno);
334
+		/* error on the connection , mark it as bad and set 0 timeout */
335
+		c->bad=1;
336
+		c->timeout=0;
337
+		/* tell "main" it should drop this (optional it will t/o anyway?)*/
338
+		response[0]=(long)c;
339
+		response[1]=CONN_ERROR;
340
+		n=write(unix_tcp_sock, response, sizeof(response));
341
+		if (n<0){
342
+			LOG(L_ERR, "BUG: tcp_send: failed to get fd(write):%s (%d)\n",
343
+					strerror(errno), errno);
344
+			goto release_c;
345
+		}
346
+	}
328 347
 end:
329 348
 	close(fd);
330 349
 release_c:
... ...
@@ -343,6 +386,7 @@ void tcpconn_timeout(fd_set* set)
343 343
 	
344 344
 	
345 345
 	ticks=get_ticks();
346
+	TCPCONN_LOCK; /* fixme: we can lock only on delete IMO */
346 347
 	for(h=0; h<TCP_ADDR_HASH_SIZE; h++){
347 348
 		c=tcpconn_addr_hash[h];
348 349
 		while(c){
... ...
@@ -354,11 +398,12 @@ void tcpconn_timeout(fd_set* set)
354 354
 					FD_CLR(c->s, set);
355 355
 					close(c->s);
356 356
 				}
357
-				tcpconn_rm(c);
357
+				_tcpconn_rm(c);
358 358
 			}
359 359
 			c=next;
360 360
 		}
361 361
 	}
362
+	TCPCONN_UNLOCK;
362 363
 }
363 364
 
364 365
 
... ...
@@ -515,8 +560,12 @@ void tcp_main_loop()
515 515
 					if(send2child(tcpconn)<0){
516 516
 						LOG(L_ERR,"ERROR: tcp_main_loop: no children "
517 517
 								"available\n");
518
-						close(tcpconn->s);
519
-						tcpconn_rm(tcpconn);
518
+						TCPCONN_LOCK;
519
+						if (tcpconn->refcnt==0){
520
+							close(tcpconn->s);
521
+							_tcpconn_rm(tcpconn);
522
+						}else tcpconn->timeout=0; /* force expire */
523
+						TCPCONN_UNLOCK;
520 524
 					}
521 525
 				}
522 526
 			}
... ...
@@ -536,8 +585,12 @@ void tcp_main_loop()
536 536
 					if (send2child(tcpconn)<0){
537 537
 						LOG(L_ERR,"ERROR: tcp_main_loop: no "
538 538
 									"children available\n");
539
-						close(tcpconn->s);
540
-						tcpconn_rm(tcpconn);
539
+						TCPCONN_LOCK;
540
+						if (tcpconn->refcnt==0){
541
+							close(tcpconn->s);
542
+							_tcpconn_rm(tcpconn);
543
+						}else tcpconn->timeout=0; /* force expire*/
544
+						TCPCONN_UNLOCK;
541 545
 					}
542 546
 				}
543 547
 			}
... ...
@@ -578,13 +631,14 @@ read_again:
578 578
 						}
579 579
 						tcpconn=(struct tcp_connection*)response[0];
580 580
 						if (tcpconn){
581
-							tcpconn->refcnt--;
582
-							DBG("tcp_main_loop: %p refcnt= %d\n", 
583
-									tcpconn, tcpconn->refcnt);
581
+								if (tcpconn->bad) goto tcpconn_destroy;
584 582
 								FD_SET(tcpconn->s, &master_set);
585 583
 								if (maxfd<tcpconn->s) maxfd=tcpconn->s;
586 584
 								/* update the timeout*/
587 585
 								tcpconn->timeout=get_ticks()+TCP_CON_TIMEOUT;
586
+								tcpconn_put(tcpconn);
587
+								DBG("tcp_main_loop: %p refcnt= %d\n", 
588
+									tcpconn, tcpconn->refcnt);
588 589
 						}
589 590
 						break;
590 591
 					case CONN_ERROR:
... ...
@@ -597,14 +651,23 @@ read_again:
597 597
 						}
598 598
 						tcpconn=(struct tcp_connection*)response[0];
599 599
 						if (tcpconn){
600
+							if (tcpconn->s!=-1)
601
+								FD_CLR(tcpconn->s, &master_set);
602
+		tcpconn_destroy:
603
+							TCPCONN_LOCK; /*avoid races w/ tcp_send*/
600 604
 							tcpconn->refcnt--;
601
-							if (tcpconn->refcnt==0){
605
+							if (tcpconn->refcnt==0){ 
602 606
 								DBG("tcp_main_loop: destroying connection\n");
603 607
 								close(tcpconn->s);
604
-								tcpconn_rm(tcpconn);
608
+								_tcpconn_rm(tcpconn);
605 609
 							}else{
610
+								/* force timeout */
611
+								tcpconn->timeout=0;
612
+								tcpconn->bad=1;
606 613
 								DBG("tcp_main_loop: delaying ...\n");
614
+								
607 615
 							}
616
+							TCPCONN_UNLOCK;
608 617
 						}
609 618
 						break;
610 619
 					case CONN_GET_FD:
... ...
@@ -46,9 +46,9 @@
46 46
 #include "globals.h"
47 47
 #include "receive.h"
48 48
 #include "timer.h"
49
+#include "ut.h"
49 50
 
50 51
 
51
-#define q_memchr memchr
52 52
 
53 53
 /* reads next available bytes
54 54
  * return number of bytes read, 0 on EOF or -1 on error,
... ...
@@ -77,6 +77,7 @@ again:
77 77
 			return -1;
78 78
 		}
79 79
 	}
80
+	DBG("tcp_read: read %d bytes:\n%.*s\n", bytes_read, bytes_read, r->pos);
80 81
 	
81 82
 	r->pos+=bytes_read;
82 83
 	return bytes_read;
... ...
@@ -133,9 +134,13 @@ int tcp_read_headers(struct tcp_req *r, int fd)
133 133
 							  break
134 134
 
135 135
 
136
-	
137
-	bytes=tcp_read(r, fd);
138
-	if (bytes<=0) return bytes;
136
+	/* if we still have some unparsed part, parse it first, don't do the read*/
137
+	if (r->parsed<r->pos){
138
+		bytes=0;
139
+	}else{
140
+		bytes=tcp_read(r, fd);
141
+		if (bytes<=0) return bytes;
142
+	}
139 143
 	p=r->parsed;
140 144
 	
141 145
 	while(p<r->pos && r->error==TCP_REQ_OK){
... ...
@@ -154,7 +159,7 @@ int tcp_read_headers(struct tcp_req *r, int fd)
154 154
 			case H_SKIP:
155 155
 				/* find lf, we are in this state if we are not interested
156 156
 				 * in anything till end of line*/
157
-				p=q_memchr(p, '\n', r->pos-r->parsed);
157
+				p=q_memchr(p, '\n', r->pos-p);
158 158
 				if (p){
159 159
 					p++;
160 160
 					r->state=H_LF;
... ...
@@ -172,14 +177,18 @@ int tcp_read_headers(struct tcp_req *r, int fd)
172 172
 					case '\n':
173 173
 						/* found LF LF */
174 174
 						r->state=H_BODY;
175
+						DBG("tcp_read_headers: switching to H_BODY (lflf)\n");
175 176
 						if (r->has_content_len){
176 177
 							r->body=p+1;
177 178
 							r->bytes_to_go=r->content_len;
178 179
 							if (r->bytes_to_go==0){
179 180
 								r->complete=1;
181
+								p++;
180 182
 								goto skip;
181 183
 							}
182 184
 						}else{
185
+							DBG("tcp_read_headers: ERROR: no clen, p=%X\n",
186
+									*p);
183 187
 							r->error=TCP_REQ_BAD_LEN;
184 188
 						}
185 189
 						break;
... ...
@@ -193,14 +202,18 @@ int tcp_read_headers(struct tcp_req *r, int fd)
193 193
 				if (*p=='\n'){
194 194
 					/* found LF CR LF */
195 195
 					r->state=H_BODY;
196
+					DBG("tcp_read_headers: switching to H_BODY (lfcrlf)\n");
196 197
 					if (r->has_content_len){
197 198
 						r->body=p+1;
198 199
 						r->bytes_to_go=r->content_len;
199 200
 						if (r->bytes_to_go==0){
200 201
 							r->complete=1;
202
+							p++;
201 203
 							goto skip;
202 204
 						}
203 205
 					}else{
206
+						DBG("tcp_read_headers: ERROR: no clen, p=%X\n",
207
+									*p);
204 208
 						r->error=TCP_REQ_BAD_LEN;
205 209
 					}
206 210
 				}else r->state=H_SKIP;
... ...
@@ -342,17 +355,21 @@ int tcp_read_req(struct tcp_connection* con)
342 342
 		resp=CONN_RELEASE;
343 343
 		s=con->fd;
344 344
 		req=&con->req;
345
+		size=0;
346
+again:
345 347
 		if(req->complete==0 && req->error==TCP_REQ_OK){
346 348
 			bytes=tcp_read_headers(req, s);
347 349
 						/* if timeout state=0; goto end__req; */
348 350
 			DBG("read= %d bytes, parsed=%d, state=%d, error=%d\n",
349
-					bytes, req->parsed-req->buf, req->state, req->error );
351
+					bytes, req->parsed-req->start, req->state, req->error );
352
+			DBG("tcp_read_req: last char=%X, parsed msg=\n%.*s\n",
353
+					*(req->parsed-1), req->parsed-req->start, req->start);
350 354
 			if (bytes==-1){
351 355
 				LOG(L_ERR, "ERROR: tcp_read_req: error reading \n");
352 356
 				resp=CONN_ERROR;
353 357
 				goto end_req;
354 358
 			}
355
-			if (bytes==0){
359
+			if ((size==0) && (bytes==0)){
356 360
 				DBG( "tcp_read_req: EOF\n");
357 361
 				resp=CONN_EOF;
358 362
 				goto end_req;
... ...
@@ -360,15 +377,21 @@ int tcp_read_req(struct tcp_connection* con)
360 360
 		
361 361
 		}
362 362
 		if (req->error!=TCP_REQ_OK){
363
-			LOG(L_ERR,"ERROR: tcp_read_req: bad request, state=%d, error=%d\n",
364
-					req->state, req->error);
363
+			LOG(L_ERR,"ERROR: tcp_read_req: bad request, state=%d, error=%d "
364
+					  "buf:\n%.*s\nparsed:\n%.*s\n", req->state, req->error,
365
+					  req->pos-req->buf, req->buf,
366
+					  req->parsed-req->start, req->start);
367
+			DBG("- received from: port %d, ip -", ntohs(con->rcv.src_port));
368
+			print_ip(&con->rcv.src_ip); DBG("-\n");
365 369
 			resp=CONN_ERROR;
366 370
 			goto end_req;
367 371
 		}
368 372
 		if (req->complete){
369 373
 			DBG("tcp_read_req: end of header part\n");
374
+			DBG("- received from: port %d, ip - ", ntohs(con->rcv.src_port));
375
+			print_ip(&con->rcv.src_ip); DBG("-\n");
370 376
 			DBG("tcp_read_req: headers:\n%.*s.\n",
371
-					req->body-req->buf, req->buf);
377
+					req->body-req->start, req->start);
372 378
 			if (req->has_content_len){
373 379
 				DBG("tcp_read_req: content-length= %d\n", req->content_len);
374 380
 				DBG("tcp_read_req: body:\n%.*s\n", req->content_len,req->body);
... ...
@@ -383,7 +406,7 @@ int tcp_read_req(struct tcp_connection* con)
383 383
 			resp=CONN_RELEASE;
384 384
 			/* just for debugging use sendipv4 as receiving socket */
385 385
 			DBG("calling receive_msg(%p, %d, )\n",
386
-					req->buf, (int)(req->parsed-req->start));
386
+					req->start, (int)(req->parsed-req->start));
387 387
 			bind_address=sendipv4; /*&tcp_info[con->sock_idx];*/
388 388
 			con->rcv.proto_reserved1=con->id; /* copy the id */
389 389
 			if (receive_msg(req->start, req->parsed-req->start, &con->rcv)<0){
... ...
@@ -404,6 +427,8 @@ int tcp_read_req(struct tcp_connection* con)
404 404
 			req->state=H_SKIP_EMPTY;
405 405
 			req->complete=req->content_len=req->has_content_len=0;
406 406
 			req->bytes_to_go=0;
407
+			/* if we still have some unparsed bytes, try to  parse them too*/
408
+			if (size) goto again;
407 409
 			
408 410
 		}
409 411
 		
... ...
@@ -496,11 +521,13 @@ void tcp_receive_loop(int unix_sock)
496 496
 					LOG(L_ERR, "ERROR: tcp_receive_loop: read_fd:"
497 497
 									"no fd read\n");
498 498
 					resp=CONN_ERROR;
499
+					con->bad=1;
499 500
 					release_tcpconn(con, resp, unix_sock);
500 501
 				}
501 502
 				if (con==0){
502 503
 					LOG(L_ERR, "ERROR: tcp_receive_loop: null pointer\n");
503 504
 					resp=CONN_ERROR;
505
+					con->bad=1;
504 506
 					release_tcpconn(con, resp, unix_sock);
505 507
 				}
506 508
 				con->timeout=get_ticks()+TCP_CHILD_TIMEOUT;
... ...
@@ -524,6 +551,7 @@ void tcp_receive_loop(int unix_sock)
524 524
 					if (resp<0){
525 525
 						FD_CLR(con->fd, &master_set);
526 526
 						tcpconn_listrm(list, con, c_next, c_prev);
527
+						con->bad=1;
527 528
 						release_tcpconn(con, resp, unix_sock);
528 529
 					}else{
529 530
 						/* update timeout */
... ...
@@ -34,7 +34,8 @@
34 34
 
35 35
 /* "public" functions*/
36 36
 
37
-struct tcp_connection* tcpconn_get(int id, struct ip_addr* ip, int port);
37
+struct tcp_connection* tcpconn_get(int id, struct ip_addr* ip, int port, 
38
+									int timeout);
38 39
 void tcpconn_put(struct tcp_connection* c);
39 40
 int tcp_send(char* buf, unsigned len, union sockaddr_union* to, int id);
40 41