Browse code

core: improved de-buffering for websockets

- This should handle the case that the full TCP packet hasn't been received
when the read function is called. Not sure how to explicitly test this
though.

Peter Dunkley authored on 16/06/2012 21:58:36
Showing 1 changed files
... ...
@@ -110,11 +110,6 @@ int is_msg_complete(struct tcp_req* r);
110 110
 #define HTTP11CONTINUE_LEN	(sizeof(HTTP11CONTINUE)-1)
111 111
 #endif
112 112
 
113
-#ifdef READ_WS
114
-static int ws_process_msg(char* tcpbuf, unsigned int len,
115
-		struct receive_info* rcv_info, struct tcp_connection* con);
116
-#endif
117
-
118 113
 #define TCPCONN_TIMEOUT_MIN_RUN  1 /* run the timers each new tick */
119 114
 
120 115
 /* types used in io_wait* */
... ...
@@ -444,10 +439,6 @@ int tcp_read_headers(struct tcp_connection *c, int* read_flags)
444 444
 		if (bytes<=0) return bytes;
445 445
 	}
446 446
 	p=r->parsed;
447
-#ifdef READ_WS
448
-	if (c->flags & F_CONN_WS)
449
-		return ws_process_msg(p, bytes, &c->rcv, c);
450
-#endif
451 447
 
452 448
 	while(p<r->pos && r->error==TCP_REQ_OK){
453 449
 		switch((unsigned char)r->state){
... ...
@@ -1025,6 +1016,110 @@ int msrp_process_msg(char* tcpbuf, unsigned int len,
1025 1025
 #endif
1026 1026
 
1027 1027
 #ifdef READ_WS
1028
+static int tcp_read_ws(struct tcp_connection *c, int* read_flags)
1029
+{
1030
+	int bytes, pos, mask_present;
1031
+	unsigned long len;
1032
+	char *p;
1033
+	struct tcp_req *r;
1034
+
1035
+	r=&c->req;
1036
+	if (unlikely(r->parsed < r->pos))
1037
+		bytes = 0;
1038
+	else
1039
+	{
1040
+#ifdef USE_TLS
1041
+		if (unlikely(c->type == PROTO_TLS))
1042
+			bytes = tls_read(c, read_flags);
1043
+		else
1044
+#endif
1045
+			bytes = tcp_read(c, read_flags);
1046
+
1047
+		if (bytes <= 0)
1048
+			return 0;
1049
+	}
1050
+
1051
+	p = r->parsed;
1052
+	pos = 0;
1053
+
1054
+	/*
1055
+	 0                   1                   2                   3
1056
+	 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
1057
+	+-+-+-+-+-------+-+-------------+-------------------------------+
1058
+	|F|R|R|R| opcode|M| Payload len |    Extended payload length    |
1059
+	|I|S|S|S|  (4)  |A|     (7)     |             (16/64)           |
1060
+	|N|V|V|V|       |S|             |   (if payload len==126/127)   |
1061
+	| |1|2|3|       |K|             |                               |
1062
+	+-+-+-+-+-------+-+-------------+ - - - - - - - - - - - - - - - +
1063
+	|     Extended payload length continued, if payload len == 127  |
1064
+	+ - - - - - - - - - - - - - - - +-------------------------------+
1065
+	|                               |Masking-key, if MASK set to 1  |
1066
+	+-------------------------------+-------------------------------+
1067
+	| Masking-key (continued)       |          Payload Data         |
1068
+	+-------------------------------- - - - - - - - - - - - - - - - +
1069
+	:                     Payload Data continued ...                :
1070
+	+ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - +
1071
+	|                     Payload Data continued ...                |
1072
+	+---------------------------------------------------------------+
1073
+
1074
+	Do minimal parse required to make sure the full message has been
1075
+	received (websocket module will do full parse and validation).
1076
+	*/
1077
+
1078
+	/* Process first two bytes */
1079
+	if (bytes < pos + 2)
1080
+		goto skip;
1081
+	pos++;
1082
+	mask_present = p[pos] & 0x80;
1083
+	len = (p[pos++] & 0xff) & ~0x80;
1084
+
1085
+	/* Work out real length */
1086
+	if (len == 126)
1087
+	{
1088
+		if (bytes < pos + 2)
1089
+			goto skip;
1090
+
1091
+		len = 0;
1092
+		len |= (p[pos++] & 0xff) <<  8;
1093
+		len |= (p[pos++] & 0xff) <<  0;
1094
+	}
1095
+	else if (len == 127)
1096
+	{
1097
+		if (bytes < pos + 8)
1098
+			goto skip;
1099
+
1100
+		/* Only decoding the last four bytes of the length...
1101
+		   This limits the size of WebSocket messages that can be
1102
+		   handled to 2^32 - which should be plenty for SIP! */
1103
+		len = 0;
1104
+		pos += 4;
1105
+		len |= (p[pos++] & 0xff) << 24;
1106
+		len |= (p[pos++] & 0xff) << 16;
1107
+		len |= (p[pos++] & 0xff) <<  8;
1108
+		len |= (p[pos++] & 0xff) <<  0;
1109
+	}
1110
+
1111
+	/* Skip mask */
1112
+	if (mask_present)
1113
+	{
1114
+		if (bytes < pos + 4)
1115
+			goto skip;
1116
+		pos += 4;
1117
+	}
1118
+
1119
+	/* Now check the whole message has been received */
1120
+	if (bytes < pos + len)
1121
+		goto skip;
1122
+
1123
+	pos += len;
1124
+	r->bytes_to_go = bytes - pos;
1125
+	r->flags |= F_TCP_REQ_COMPLETE;
1126
+	r->parsed = &p[pos];
1127
+
1128
+skip:
1129
+	return bytes;
1130
+}
1131
+
1028 1132
 static int ws_process_msg(char* tcpbuf, unsigned int len,
1029 1133
 		struct receive_info* rcv_info, struct tcp_connection* con)
1030 1134
 {
... ...
@@ -1145,7 +1240,12 @@ int tcp_read_req(struct tcp_connection* con, int* bytes_read, int* read_flags)
1145 1145
 
1146 1146
 again:
1147 1147
 		if (likely(req->error==TCP_REQ_OK)){
1148
-			bytes=tcp_read_headers(con, read_flags);
1148
+#ifdef READ_WS
1149
+			if (unlikely(con->flags&F_CONN_WS))
1150
+				bytes=tcp_read_ws(con, read_flags);
1151
+			else
1152
+#endif
1153
+				bytes=tcp_read_headers(con, read_flags);
1149 1154
 #ifdef EXTRA_DEBUG
1150 1155
 						/* if timeout state=0; goto end__req; */
1151 1156
 			DBG("read= %d bytes, parsed=%d, state=%d, error=%d\n",
... ...
@@ -1173,7 +1273,6 @@ again:
1173 1173
 				resp=CONN_EOF;
1174 1174
 				goto end_req;
1175 1175
 			}
1176
-		
1177 1176
 		}
1178 1177
 		if (unlikely(req->error!=TCP_REQ_OK)){
1179 1178
 			LOG(L_ERR,"ERROR: tcp_read_req: bad request, state=%d, error=%d "
... ...
@@ -1261,6 +1360,12 @@ again:
1261 1261
 						&con->rcv, con);
1262 1262
 			}else
1263 1263
 #endif
1264
+#ifdef READ_WS
1265
+			if (unlikely(con->flags&F_CONN_WS)){
1266
+				ret = ws_process_msg(req->start, req->parsed-req->start,
1267
+									&con->rcv, con);
1268
+			}else
1269
+#endif
1264 1270
 				ret = receive_tcp_msg(req->start, req->parsed-req->start,
1265 1271
 									&con->rcv, con);
1266 1272