1 #include <assert.h> 2 3 #include "mos_api.h" 4 #include "tcp_util.h" 5 #include "tcp_in.h" 6 #include "tcp_out.h" 7 #include "tcp_ring_buffer.h" 8 #include "eventpoll.h" 9 #include "debug.h" 10 #include "timer.h" 11 #include "ip_in.h" 12 #include "tcp_rb.h" 13 #include "config.h" 14 #include "scalable_event.h" 15 16 #define MAX(a, b) ((a)>(b)?(a):(b)) 17 #define MIN(a, b) ((a)<(b)?(a):(b)) 18 19 #define RECOVERY_AFTER_LOSS TRUE 20 #define SELECTIVE_WRITE_EVENT_NOTIFY TRUE 21 /*----------------------------------------------------------------------------*/ 22 static inline void 23 Handle_TCP_ST_ESTABLISHED (mtcp_manager_t mtcp, tcp_stream* cur_stream, 24 struct pkt_ctx *pctx); 25 /*----------------------------------------------------------------------------*/ 26 static inline int 27 FilterSYNPacket(mtcp_manager_t mtcp, uint32_t ip, uint16_t port) 28 { 29 struct sockaddr_in *addr; 30 31 /* TODO: This listening logic should be revised */ 32 33 /* if not listening, drop */ 34 if (!mtcp->listener) { 35 return FALSE; 36 } 37 38 /* if not the address we want, drop */ 39 addr = &mtcp->listener->socket->saddr; 40 if (addr->sin_port == port) { 41 if (addr->sin_addr.s_addr != INADDR_ANY) { 42 if (ip == addr->sin_addr.s_addr) { 43 return TRUE; 44 } 45 return FALSE; 46 } else { 47 int i; 48 49 for (i = 0; i < g_config.mos->netdev_table->num; i++) { 50 if (ip == g_config.mos->netdev_table->ent[i]->ip_addr) { 51 return TRUE; 52 } 53 } 54 return FALSE; 55 } 56 } 57 58 return FALSE; 59 } 60 /*----------------------------------------------------------------------------*/ 61 static inline int 62 HandleActiveOpen(mtcp_manager_t mtcp, tcp_stream *cur_stream, 63 struct pkt_ctx *pctx) 64 { 65 const struct tcphdr* tcph = pctx->p.tcph; 66 67 cur_stream->rcvvar->irs = pctx->p.seq; 68 cur_stream->snd_nxt = pctx->p.ack_seq; 69 cur_stream->sndvar->peer_wnd = pctx->p.window; 70 cur_stream->rcvvar->snd_wl1 = cur_stream->rcvvar->irs - 1; 71 cur_stream->rcv_nxt = cur_stream->rcvvar->irs + 1; 72 cur_stream->rcvvar->last_ack_seq = pctx->p.ack_seq; 73 ParseTCPOptions(cur_stream, pctx->p.cur_ts, (uint8_t *)tcph + TCP_HEADER_LEN, 74 (tcph->doff << 2) - TCP_HEADER_LEN); 75 cur_stream->sndvar->cwnd = ((cur_stream->sndvar->cwnd == 1)? 76 (cur_stream->sndvar->mss * 2): cur_stream->sndvar->mss); 77 cur_stream->sndvar->ssthresh = cur_stream->sndvar->mss * 10; 78 if (HAS_STREAM_TYPE(cur_stream, MOS_SOCK_STREAM)) 79 UpdateRetransmissionTimer(mtcp, cur_stream, pctx->p.cur_ts); 80 81 return TRUE; 82 } 83 /*----------------------------------------------------------------------------*/ 84 /* ValidateSequence: validates sequence number of the segment */ 85 /* Return: TRUE if acceptable, FALSE if not acceptable */ 86 /*----------------------------------------------------------------------------*/ 87 static inline int 88 ValidateSequence(mtcp_manager_t mtcp, tcp_stream *cur_stream, 89 struct pkt_ctx *pctx) 90 { 91 const struct tcphdr* tcph = pctx->p.tcph; 92 93 /* Protect Against Wrapped Sequence number (PAWS) */ 94 if (!tcph->rst && cur_stream->saw_timestamp) { 95 struct tcp_timestamp ts; 96 97 if (!ParseTCPTimestamp(cur_stream, &ts, 98 (uint8_t *)tcph + TCP_HEADER_LEN, 99 (tcph->doff << 2) - TCP_HEADER_LEN)) { 100 /* if there is no timestamp */ 101 /* TODO: implement here */ 102 TRACE_DBG("No timestamp found.\n"); 103 return FALSE; 104 } 105 106 /* RFC1323: if SEG.TSval < TS.Recent, drop and send ack */ 107 if (TCP_SEQ_LT(ts.ts_val, cur_stream->rcvvar->ts_recent)) { 108 /* TODO: ts_recent should be invalidated 109 before timestamp wraparound for long idle flow */ 110 TRACE_DBG("PAWS Detect wrong timestamp. " 111 "seq: %u, ts_val: %u, prev: %u\n", 112 pctx->p.seq, ts.ts_val, cur_stream->rcvvar->ts_recent); 113 cur_stream->actions |= MOS_ACT_SEND_ACK_NOW; 114 return FALSE; 115 } else { 116 /* valid timestamp */ 117 if (TCP_SEQ_GT(ts.ts_val, cur_stream->rcvvar->ts_recent)) { 118 TRACE_TSTAMP("Timestamp update. cur: %u, prior: %u " 119 "(time diff: %uus)\n", 120 ts.ts_val, cur_stream->rcvvar->ts_recent, 121 TS_TO_USEC(pctx->p.cur_ts - cur_stream->rcvvar->ts_last_ts_upd)); 122 cur_stream->rcvvar->ts_last_ts_upd = pctx->p.cur_ts; 123 } 124 125 cur_stream->rcvvar->ts_recent = ts.ts_val; 126 cur_stream->rcvvar->ts_lastack_rcvd = ts.ts_ref; 127 } 128 } 129 130 /* TCP sequence validation */ 131 if (!TCP_SEQ_BETWEEN(pctx->p.seq + pctx->p.payloadlen, cur_stream->rcv_nxt, 132 cur_stream->rcv_nxt + cur_stream->rcvvar->rcv_wnd)) { 133 134 /* if RST bit is set, ignore the segment */ 135 if (tcph->rst) 136 return FALSE; 137 138 if (cur_stream->state == TCP_ST_ESTABLISHED) { 139 /* check if it is to get window advertisement */ 140 if (pctx->p.seq + 1 == cur_stream->rcv_nxt) { 141 TRACE_DBG("Window update request. (seq: %u, rcv_wnd: %u)\n", 142 seq, cur_stream->rcvvar->rcv_wnd); 143 cur_stream->actions |= MOS_ACT_SEND_ACK_AGG; 144 return FALSE; 145 146 } 147 148 if (TCP_SEQ_LEQ(pctx->p.seq, cur_stream->rcv_nxt)) { 149 cur_stream->actions |= MOS_ACT_SEND_ACK_AGG; 150 } else { 151 cur_stream->actions |= MOS_ACT_SEND_ACK_NOW; 152 } 153 } else { 154 if (cur_stream->state == TCP_ST_TIME_WAIT) { 155 TRACE_DBG("Stream %d: tw expire update to %u\n", 156 cur_stream->id, cur_stream->rcvvar->ts_tw_expire); 157 AddtoTimewaitList(mtcp, cur_stream, pctx->p.cur_ts); 158 } 159 cur_stream->actions |= MOS_ACT_SEND_CONTROL; 160 } 161 return FALSE; 162 } 163 164 return TRUE; 165 } 166 /*----------------------------------------------------------------------------*/ 167 static inline void 168 NotifyConnectionReset(mtcp_manager_t mtcp, tcp_stream *cur_stream) 169 { 170 TRACE_DBG("Stream %d: Notifying connection reset.\n", cur_stream->id); 171 /* TODO: implement this function */ 172 /* signal to user "connection reset" */ 173 } 174 /*----------------------------------------------------------------------------*/ 175 static inline int 176 ProcessRST(mtcp_manager_t mtcp, tcp_stream *cur_stream, 177 struct pkt_ctx *pctx) 178 { 179 /* TODO: we need reset validation logic */ 180 /* the sequence number of a RST should be inside window */ 181 /* (in SYN_SENT state, it should ack the previous SYN */ 182 183 TRACE_DBG("Stream %d: TCP RESET (%s)\n", 184 cur_stream->id, TCPStateToString(cur_stream)); 185 #if DUMP_STREAM 186 DumpStream(mtcp, cur_stream); 187 #endif 188 189 if (cur_stream->state <= TCP_ST_SYN_SENT) { 190 /* not handled here */ 191 return FALSE; 192 } 193 194 if (cur_stream->state == TCP_ST_SYN_RCVD) { 195 /* ACK number of last sent ACK packet == rcv_nxt + 1*/ 196 if (pctx->p.seq == 0 || 197 #ifdef BE_RESILIENT_TO_PACKET_DROP 198 pctx->p.seq == cur_stream->rcv_nxt + 1 || 199 #endif 200 pctx->p.ack_seq == cur_stream->rcv_nxt + 1) 201 { 202 cur_stream->state = TCP_ST_CLOSED_RSVD; 203 cur_stream->cb_events |= MOS_ON_TCP_STATE_CHANGE; 204 cur_stream->close_reason = TCP_RESET; 205 cur_stream->actions |= MOS_ACT_DESTROY; 206 } else { 207 RAISE_DEBUG_EVENT(mtcp, cur_stream, 208 "(SYN_RCVD): Ignore invalid RST. " 209 "ack_seq expected: %u, ack_seq rcvd: %u\n", 210 cur_stream->rcv_nxt + 1, pctx->p.ack_seq); 211 } 212 return TRUE; 213 } 214 215 /* if the application is already closed the connection, 216 just destroy the it */ 217 if (cur_stream->state == TCP_ST_FIN_WAIT_1 || 218 cur_stream->state == TCP_ST_FIN_WAIT_2 || 219 cur_stream->state == TCP_ST_LAST_ACK || 220 cur_stream->state == TCP_ST_CLOSING || 221 cur_stream->state == TCP_ST_TIME_WAIT) { 222 cur_stream->state = TCP_ST_CLOSED_RSVD; 223 cur_stream->close_reason = TCP_ACTIVE_CLOSE; 224 cur_stream->cb_events |= MOS_ON_TCP_STATE_CHANGE; 225 cur_stream->actions |= MOS_ACT_DESTROY; 226 return TRUE; 227 } 228 229 if (cur_stream->state >= TCP_ST_ESTABLISHED && 230 cur_stream->state <= TCP_ST_CLOSE_WAIT) { 231 /* ESTABLISHED, FIN_WAIT_1, FIN_WAIT_2, CLOSE_WAIT */ 232 /* TODO: flush all the segment queues */ 233 //NotifyConnectionReset(mtcp, cur_stream); 234 } 235 236 if (!(cur_stream->sndvar->on_closeq || cur_stream->sndvar->on_closeq_int || 237 cur_stream->sndvar->on_resetq || cur_stream->sndvar->on_resetq_int)) { 238 //cur_stream->state = TCP_ST_CLOSED_RSVD; 239 //cur_stream->actions |= MOS_ACT_DESTROY; 240 cur_stream->state = TCP_ST_CLOSED_RSVD; 241 cur_stream->cb_events |= MOS_ON_TCP_STATE_CHANGE; 242 cur_stream->close_reason = TCP_RESET; 243 if (HAS_STREAM_TYPE(cur_stream, MOS_SOCK_STREAM)) 244 RaiseCloseEvent(mtcp, cur_stream); 245 } 246 247 return TRUE; 248 } 249 /*----------------------------------------------------------------------------*/ 250 inline void 251 EstimateRTT(mtcp_manager_t mtcp, tcp_stream *cur_stream, uint32_t mrtt) 252 { 253 /* This function should be called for not retransmitted packets */ 254 /* TODO: determine tcp_rto_min */ 255 #define TCP_RTO_MIN 0 256 long m = mrtt; 257 uint32_t tcp_rto_min = TCP_RTO_MIN; 258 struct tcp_recv_vars *rcvvar = cur_stream->rcvvar; 259 260 if (m == 0) { 261 m = 1; 262 } 263 if (rcvvar->srtt != 0) { 264 /* rtt = 7/8 rtt + 1/8 new */ 265 m -= (rcvvar->srtt >> 3); 266 rcvvar->srtt += m; 267 if (m < 0) { 268 m = -m; 269 m -= (rcvvar->mdev >> 2); 270 if (m > 0) { 271 m >>= 3; 272 } 273 } else { 274 m -= (rcvvar->mdev >> 2); 275 } 276 rcvvar->mdev += m; 277 if (rcvvar->mdev > rcvvar->mdev_max) { 278 rcvvar->mdev_max = rcvvar->mdev; 279 if (rcvvar->mdev_max > rcvvar->rttvar) { 280 rcvvar->rttvar = rcvvar->mdev_max; 281 } 282 } 283 if (TCP_SEQ_GT(cur_stream->sndvar->snd_una, rcvvar->rtt_seq)) { 284 if (rcvvar->mdev_max < rcvvar->rttvar) { 285 rcvvar->rttvar -= (rcvvar->rttvar - rcvvar->mdev_max) >> 2; 286 } 287 rcvvar->rtt_seq = cur_stream->snd_nxt; 288 rcvvar->mdev_max = tcp_rto_min; 289 } 290 } else { 291 /* fresh measurement */ 292 rcvvar->srtt = m << 3; 293 rcvvar->mdev = m << 1; 294 rcvvar->mdev_max = rcvvar->rttvar = MAX(rcvvar->mdev, tcp_rto_min); 295 rcvvar->rtt_seq = cur_stream->snd_nxt; 296 } 297 298 TRACE_RTT("mrtt: %u (%uus), srtt: %u (%ums), mdev: %u, mdev_max: %u, " 299 "rttvar: %u, rtt_seq: %u\n", mrtt, mrtt * TIME_TICK, 300 rcvvar->srtt, TS_TO_MSEC((rcvvar->srtt) >> 3), rcvvar->mdev, 301 rcvvar->mdev_max, rcvvar->rttvar, rcvvar->rtt_seq); 302 } 303 /*----------------------------------------------------------------------------*/ 304 static inline void 305 ProcessACK(mtcp_manager_t mtcp, tcp_stream *cur_stream, 306 struct pkt_ctx *pctx) 307 { 308 const struct tcphdr* tcph = pctx->p.tcph; 309 uint32_t seq = pctx->p.seq; 310 uint32_t ack_seq = pctx->p.ack_seq; 311 struct tcp_send_vars *sndvar = cur_stream->sndvar; 312 uint32_t cwindow, cwindow_prev; 313 uint32_t rmlen; 314 uint32_t snd_wnd_prev; 315 uint32_t right_wnd_edge; 316 uint8_t dup; 317 318 cwindow = pctx->p.window; 319 if (!tcph->syn) { 320 cwindow = cwindow << sndvar->wscale_peer; 321 } 322 right_wnd_edge = sndvar->peer_wnd + cur_stream->rcvvar->snd_wl2; 323 324 /* If ack overs the sending buffer, return */ 325 if (cur_stream->state == TCP_ST_FIN_WAIT_1 || 326 cur_stream->state == TCP_ST_FIN_WAIT_2 || 327 cur_stream->state == TCP_ST_CLOSING || 328 cur_stream->state == TCP_ST_CLOSE_WAIT || 329 cur_stream->state == TCP_ST_LAST_ACK) { 330 if (sndvar->is_fin_sent && ack_seq == sndvar->fss + 1) { 331 ack_seq--; 332 } 333 } 334 335 if (TCP_SEQ_GT(ack_seq, sndvar->sndbuf->head_seq + sndvar->sndbuf->len)) { 336 TRACE_DBG("Stream %d (%s): invalid acknologement. " 337 "ack_seq: %u, possible max_ack_seq: %u\n", cur_stream->id, 338 TCPStateToString(cur_stream), ack_seq, 339 sndvar->sndbuf->head_seq + sndvar->sndbuf->len); 340 return; 341 } 342 343 #ifdef BE_RESILIENT_TO_PACKET_DROP 344 if (TCP_SEQ_GT(seq + pctx->p.payloadlen, cur_stream->rcv_nxt)) 345 cur_stream->rcv_nxt = seq + pctx->p.payloadlen; 346 #endif 347 348 /* Update window */ 349 if (TCP_SEQ_LT(cur_stream->rcvvar->snd_wl1, seq) || 350 (cur_stream->rcvvar->snd_wl1 == seq && 351 TCP_SEQ_LT(cur_stream->rcvvar->snd_wl2, ack_seq)) || 352 (cur_stream->rcvvar->snd_wl2 == ack_seq && 353 cwindow > sndvar->peer_wnd)) { 354 cwindow_prev = sndvar->peer_wnd; 355 sndvar->peer_wnd = cwindow; 356 cur_stream->rcvvar->snd_wl1 = seq; 357 cur_stream->rcvvar->snd_wl2 = ack_seq; 358 TRACE_CLWND("Window update. " 359 "ack: %u, peer_wnd: %u, snd_nxt-snd_una: %u\n", 360 ack_seq, cwindow, cur_stream->snd_nxt - sndvar->snd_una); 361 if (cwindow_prev < cur_stream->snd_nxt - sndvar->snd_una && 362 sndvar->peer_wnd >= cur_stream->snd_nxt - sndvar->snd_una) { 363 TRACE_CLWND("%u Broadcasting client window update! " 364 "ack_seq: %u, peer_wnd: %u (before: %u), " 365 "(snd_nxt - snd_una: %u)\n", 366 cur_stream->id, ack_seq, sndvar->peer_wnd, cwindow_prev, 367 cur_stream->snd_nxt - sndvar->snd_una); 368 if (HAS_STREAM_TYPE(cur_stream, MOS_SOCK_STREAM)) 369 RaiseWriteEvent(mtcp, cur_stream); 370 } 371 } 372 373 /* Check duplicated ack count */ 374 /* Duplicated ack if 375 1) ack_seq is old 376 2) payload length is 0. 377 3) advertised window not changed. 378 4) there is outstanding unacknowledged data 379 5) ack_seq == snd_una 380 */ 381 382 dup = FALSE; 383 if (TCP_SEQ_LT(ack_seq, cur_stream->snd_nxt)) { 384 if (ack_seq == cur_stream->rcvvar->last_ack_seq && pctx->p.payloadlen == 0) { 385 if (cur_stream->rcvvar->snd_wl2 + sndvar->peer_wnd == right_wnd_edge) { 386 if (cur_stream->rcvvar->dup_acks + 1 > cur_stream->rcvvar->dup_acks) { 387 cur_stream->rcvvar->dup_acks++; 388 } 389 dup = TRUE; 390 } 391 } 392 } 393 if (!dup) { 394 cur_stream->rcvvar->dup_acks = 0; 395 cur_stream->rcvvar->last_ack_seq = ack_seq; 396 } 397 398 /* Fast retransmission */ 399 if (dup && cur_stream->rcvvar->dup_acks == 3) { 400 TRACE_LOSS("Triple duplicated ACKs!! ack_seq: %u\n", ack_seq); 401 if (TCP_SEQ_LT(ack_seq, cur_stream->snd_nxt)) { 402 TRACE_LOSS("Reducing snd_nxt from %u to %u\n", 403 cur_stream->snd_nxt, ack_seq); 404 #if RTM_STAT 405 sndvar->rstat.tdp_ack_cnt++; 406 sndvar->rstat.tdp_ack_bytes += (cur_stream->snd_nxt - ack_seq); 407 #endif 408 if (ack_seq != sndvar->snd_una) { 409 TRACE_DBG("ack_seq and snd_una mismatch on tdp ack. " 410 "ack_seq: %u, snd_una: %u\n", 411 ack_seq, sndvar->snd_una); 412 } 413 cur_stream->snd_nxt = ack_seq; 414 } 415 416 /* update congestion control variables */ 417 /* ssthresh to half of min of cwnd and peer wnd */ 418 sndvar->ssthresh = MIN(sndvar->cwnd, sndvar->peer_wnd) / 2; 419 if (sndvar->ssthresh < 2 * sndvar->mss) { 420 sndvar->ssthresh = 2 * sndvar->mss; 421 } 422 sndvar->cwnd = sndvar->ssthresh + 3 * sndvar->mss; 423 TRACE_CONG("Fast retransmission. cwnd: %u, ssthresh: %u\n", 424 sndvar->cwnd, sndvar->ssthresh); 425 426 /* count number of retransmissions */ 427 if (sndvar->nrtx < TCP_MAX_RTX) { 428 sndvar->nrtx++; 429 } else { 430 TRACE_DBG("Exceed MAX_RTX.\n"); 431 } 432 433 cur_stream->actions |= MOS_ACT_SEND_DATA; 434 435 } else if (cur_stream->rcvvar->dup_acks > 3) { 436 /* Inflate congestion window until before overflow */ 437 if ((uint32_t)(sndvar->cwnd + sndvar->mss) > sndvar->cwnd) { 438 sndvar->cwnd += sndvar->mss; 439 TRACE_CONG("Dupack cwnd inflate. cwnd: %u, ssthresh: %u\n", 440 sndvar->cwnd, sndvar->ssthresh); 441 } 442 } 443 444 #if TCP_OPT_SACK_ENABLED 445 ParseSACKOption(cur_stream, ack_seq, (uint8_t *)tcph + TCP_HEADER_LEN, 446 (tcph->doff << 2) - TCP_HEADER_LEN); 447 #endif /* TCP_OPT_SACK_ENABLED */ 448 449 #if RECOVERY_AFTER_LOSS 450 /* updating snd_nxt (when recovered from loss) */ 451 if (TCP_SEQ_GT(ack_seq, cur_stream->snd_nxt)) { 452 #if RTM_STAT 453 sndvar->rstat.ack_upd_cnt++; 454 sndvar->rstat.ack_upd_bytes += (ack_seq - cur_stream->snd_nxt); 455 #endif 456 TRACE_LOSS("Updating snd_nxt from %u to %u\n", 457 cur_stream->snd_nxt, ack_seq); 458 cur_stream->snd_nxt = ack_seq; 459 if (sndvar->sndbuf->len == 0) { 460 if (HAS_STREAM_TYPE(cur_stream, MOS_SOCK_STREAM)) 461 RemoveFromSendList(mtcp, cur_stream); 462 } 463 } 464 #endif 465 466 /* If ack_seq is previously acked, return */ 467 if (TCP_SEQ_GEQ(sndvar->sndbuf->head_seq, ack_seq)) { 468 return; 469 } 470 471 /* Remove acked sequence from send buffer */ 472 rmlen = ack_seq - sndvar->sndbuf->head_seq; 473 if (rmlen > 0) { 474 /* Routine goes here only if there is new payload (not retransmitted) */ 475 uint16_t packets; 476 477 /* If acks new data */ 478 packets = rmlen / sndvar->eff_mss; 479 if ((rmlen / sndvar->eff_mss) * sndvar->eff_mss > rmlen) { 480 packets++; 481 } 482 483 /* Estimate RTT and calculate rto */ 484 if (cur_stream->saw_timestamp) { 485 EstimateRTT(mtcp, cur_stream, 486 pctx->p.cur_ts - cur_stream->rcvvar->ts_lastack_rcvd); 487 sndvar->rto = (cur_stream->rcvvar->srtt >> 3) + cur_stream->rcvvar->rttvar; 488 assert(sndvar->rto > 0); 489 } else { 490 //TODO: Need to implement timestamp estimation without timestamp 491 TRACE_RTT("NOT IMPLEMENTED.\n"); 492 } 493 494 /* Update congestion control variables */ 495 if (cur_stream->state >= TCP_ST_ESTABLISHED) { 496 if (sndvar->cwnd < sndvar->ssthresh) { 497 if ((sndvar->cwnd + sndvar->mss) > sndvar->cwnd) { 498 sndvar->cwnd += (sndvar->mss * packets); 499 } 500 TRACE_CONG("slow start cwnd: %u, ssthresh: %u\n", 501 sndvar->cwnd, sndvar->ssthresh); 502 } else { 503 uint32_t new_cwnd = sndvar->cwnd + 504 packets * sndvar->mss * sndvar->mss / 505 sndvar->cwnd; 506 if (new_cwnd > sndvar->cwnd) { 507 sndvar->cwnd = new_cwnd; 508 } 509 //TRACE_CONG("congestion avoidance cwnd: %u, ssthresh: %u\n", 510 // sndvar->cwnd, sndvar->ssthresh); 511 } 512 } 513 514 if (SBUF_LOCK(&sndvar->write_lock)) { 515 if (errno == EDEADLK) 516 perror("ProcessACK: write_lock blocked\n"); 517 assert(0); 518 } 519 if (HAS_STREAM_TYPE(cur_stream, MOS_SOCK_STREAM)) 520 SBRemove(mtcp->rbm_snd, sndvar->sndbuf, rmlen); 521 sndvar->snd_una = ack_seq; 522 snd_wnd_prev = sndvar->snd_wnd; 523 sndvar->snd_wnd = sndvar->sndbuf->size - sndvar->sndbuf->len; 524 525 /* If there was no available sending window */ 526 /* notify the newly available window to application */ 527 #if SELECTIVE_WRITE_EVENT_NOTIFY 528 if (snd_wnd_prev <= 0) { 529 #endif /* SELECTIVE_WRITE_EVENT_NOTIFY */ 530 if (HAS_STREAM_TYPE(cur_stream, MOS_SOCK_STREAM)) 531 RaiseWriteEvent(mtcp, cur_stream); 532 #if SELECTIVE_WRITE_EVENT_NOTIFY 533 } 534 #endif /* SELECTIVE_WRITE_EVENT_NOTIFY */ 535 536 SBUF_UNLOCK(&sndvar->write_lock); 537 if (HAS_STREAM_TYPE(cur_stream, MOS_SOCK_STREAM)) 538 UpdateRetransmissionTimer(mtcp, cur_stream, pctx->p.cur_ts); 539 } 540 } 541 /*----------------------------------------------------------------------------*/ 542 /* ProcessTCPPayload: merges TCP payload using receive ring buffer */ 543 /* Return: TRUE (1) in normal case, FALSE (0) if immediate ACK is required */ 544 /* CAUTION: should only be called at ESTABLISHED, FIN_WAIT_1, FIN_WAIT_2 */ 545 /*----------------------------------------------------------------------------*/ 546 static inline int 547 ProcessTCPPayload(mtcp_manager_t mtcp, tcp_stream *cur_stream, 548 struct pkt_ctx *pctx) 549 { 550 struct tcp_recv_vars *rcvvar = cur_stream->rcvvar; 551 uint32_t prev_rcv_nxt; 552 int ret = -1; 553 bool read_lock; 554 struct socket_map *walk; 555 556 if (!cur_stream->buffer_mgmt) 557 return FALSE; 558 559 /* if seq and segment length is lower than rcv_nxt, ignore and send ack */ 560 if (TCP_SEQ_LT(pctx->p.seq + pctx->p.payloadlen, cur_stream->rcv_nxt)) { 561 SOCKQ_FOREACH_START(walk, &cur_stream->msocks) { 562 HandleCallback(mtcp, MOS_NULL, walk, cur_stream->side, 563 pctx, MOS_ON_ERROR); 564 } SOCKQ_FOREACH_END; 565 return FALSE; 566 } 567 /* if payload exceeds receiving buffer, drop and send ack */ 568 if (TCP_SEQ_GT(pctx->p.seq + pctx->p.payloadlen, cur_stream->rcv_nxt + rcvvar->rcv_wnd)) { 569 SOCKQ_FOREACH_START(walk, &cur_stream->msocks) { 570 HandleCallback(mtcp, MOS_NULL, walk, cur_stream->side, 571 pctx, MOS_ON_ERROR); 572 } SOCKQ_FOREACH_END; 573 return FALSE; 574 } 575 576 /* allocate receive buffer if not exist */ 577 if (!rcvvar->rcvbuf) { 578 #ifdef NEWRB 579 rcvvar->rcvbuf = tcprb_new(mtcp->bufseg_pool, g_config.mos->rmem_size, cur_stream->buffer_mgmt); 580 #else 581 rcvvar->rcvbuf = RBInit(mtcp->rbm_rcv, rcvvar->irs + 1, cur_stream->buffer_mgmt); 582 #endif 583 if (!rcvvar->rcvbuf) { 584 TRACE_ERROR("Stream %d: Failed to allocate receive buffer.\n", 585 cur_stream->id); 586 cur_stream->state = TCP_ST_CLOSED_RSVD; 587 cur_stream->close_reason = TCP_NO_MEM; 588 cur_stream->cb_events |= MOS_ON_TCP_STATE_CHANGE; 589 if (HAS_STREAM_TYPE(cur_stream, MOS_SOCK_STREAM)) 590 RaiseErrorEvent(mtcp, cur_stream); 591 592 return ERROR; 593 } 594 } 595 596 read_lock = HAS_STREAM_TYPE(cur_stream, MOS_SOCK_STREAM); 597 598 if (read_lock && SBUF_LOCK(&rcvvar->read_lock)) { 599 if (errno == EDEADLK) 600 perror("ProcessTCPPayload: read_lock blocked\n"); 601 assert(0); 602 } 603 604 prev_rcv_nxt = cur_stream->rcv_nxt; 605 606 #ifdef NEWRB 607 tcprb_t *rb = rcvvar->rcvbuf; 608 loff_t off = seq2loff(rb, pctx->p.seq, (rcvvar->irs + 1)); 609 if (off >= 0) { 610 if (!HAS_STREAM_TYPE(cur_stream, MOS_SOCK_STREAM) && 611 HAS_STREAM_TYPE(cur_stream, MOS_SOCK_MONITOR_STREAM_ACTIVE)) 612 tcprb_setpile(rb, rb->pile + tcprb_cflen(rb)); 613 ret = tcprb_pwrite(rb, pctx->p.payload, pctx->p.payloadlen, off); 614 if (ret < 0) { 615 /* We try again after warning this result to the user. */ 616 SOCKQ_FOREACH_START(walk, &cur_stream->msocks) { 617 HandleCallback(mtcp, MOS_NULL, walk, cur_stream->side, 618 pctx, MOS_ON_ERROR); 619 } SOCKQ_FOREACH_END; 620 ret = tcprb_pwrite(rb, pctx->p.payload, pctx->p.payloadlen, off); 621 } 622 } 623 /* TODO: update monitor vars */ 624 #else 625 ret = RBPut(mtcp->rbm_rcv, cur_stream, pctx->p.payload, 626 (uint32_t)pctx->p.payloadlen, pctx->p.seq); 627 #endif 628 /* 629 * error can pop up due to disabled buffered management 630 * (only in monitor mode). In that case, ignore the warning 631 * message. 632 */ 633 if (ret < 0 && cur_stream->buffer_mgmt && mtcp->num_msp == 0) 634 TRACE_ERROR("Cannot merge payload. reason: %d\n", ret); 635 636 /* discard the buffer if the state is FIN_WAIT_1 or FIN_WAIT_2, 637 meaning that the connection is already closed by the application */ 638 #ifdef NEWRB 639 loff_t cftail = rb->pile + tcprb_cflen(rb); 640 if (cur_stream->state == TCP_ST_FIN_WAIT_1 || 641 cur_stream->state == TCP_ST_FIN_WAIT_2) { 642 /* XXX: Do we really need to update recv vars? */ 643 tcprb_setpile(rb, cftail); 644 } 645 if (cftail > 0 && (rcvvar->irs + 1) + cftail > cur_stream->rcv_nxt) { 646 RAISE_DEBUG_EVENT(mtcp, cur_stream, 647 "Move rcv_nxt from %u to %u.\n", 648 cur_stream->rcv_nxt, (rcvvar->irs + 1) + cftail); 649 cur_stream->rcv_nxt = (rcvvar->irs + 1) + cftail; 650 } 651 assert(cftail - rb->pile >= 0); 652 rcvvar->rcv_wnd = rb->len - (cftail - rb->pile); 653 #else 654 if (cur_stream->state == TCP_ST_FIN_WAIT_1 || 655 cur_stream->state == TCP_ST_FIN_WAIT_2) { 656 RBRemove(mtcp->rbm_rcv, 657 rcvvar->rcvbuf, rcvvar->rcvbuf->merged_len, AT_MTCP, 658 cur_stream->buffer_mgmt); 659 } 660 cur_stream->rcv_nxt = rcvvar->rcvbuf->head_seq + rcvvar->rcvbuf->merged_len; 661 rcvvar->rcv_wnd = rcvvar->rcvbuf->size - rcvvar->rcvbuf->merged_len; 662 #endif 663 //printf("rcv_wnd: %d\n", rcvvar->rcv_wnd); 664 665 if (read_lock) 666 SBUF_UNLOCK(&rcvvar->read_lock); 667 668 669 if (TCP_SEQ_LEQ(cur_stream->rcv_nxt, prev_rcv_nxt)) { 670 /* There are some lost packets */ 671 return FALSE; 672 } 673 674 TRACE_EPOLL("Stream %d data arrived. " 675 "len: %d, ET: %u, IN: %u, OUT: %u\n", 676 cur_stream->id, payloadlen, 677 cur_stream->socket? cur_stream->socket->epoll & MOS_EPOLLET : 0, 678 cur_stream->socket? cur_stream->socket->epoll & MOS_EPOLLIN : 0, 679 cur_stream->socket? cur_stream->socket->epoll & MOS_EPOLLOUT : 0); 680 681 if (cur_stream->state == TCP_ST_ESTABLISHED) 682 RaiseReadEvent(mtcp, cur_stream); 683 684 return TRUE; 685 } 686 /*----------------------------------------------------------------------------*/ 687 static inline void 688 Handle_TCP_ST_LISTEN (mtcp_manager_t mtcp, tcp_stream* cur_stream, 689 struct pkt_ctx *pctx) 690 { 691 692 const struct tcphdr* tcph = pctx->p.tcph; 693 694 if (tcph->syn) { 695 if (cur_stream->state == TCP_ST_LISTEN) 696 cur_stream->rcv_nxt++; 697 cur_stream->state = TCP_ST_SYN_RCVD; 698 cur_stream->cb_events |= MOS_ON_TCP_STATE_CHANGE | MOS_ON_CONN_START; 699 TRACE_STATE("Stream %d: TCP_ST_SYN_RCVD\n", cur_stream->id); 700 cur_stream->actions |= MOS_ACT_SEND_CONTROL; 701 if (IS_STREAM_TYPE(cur_stream, MOS_SOCK_MONITOR_STREAM_ACTIVE)) { 702 /** 703 * Passive stream context needs to initialize irs and rcv_nxt 704 * as it is not set neither during createserverstream or monitor 705 * creation. 706 */ 707 cur_stream->rcvvar->irs = 708 cur_stream->rcv_nxt = pctx->p.seq; 709 } 710 } else { 711 CTRACE_ERROR("Stream %d (TCP_ST_LISTEN): " 712 "Packet without SYN.\n", cur_stream->id); 713 } 714 715 } 716 /*----------------------------------------------------------------------------*/ 717 static inline void 718 Handle_TCP_ST_SYN_SENT (mtcp_manager_t mtcp, tcp_stream* cur_stream, 719 struct pkt_ctx *pctx) 720 { 721 const struct tcphdr* tcph = pctx->p.tcph; 722 723 /* when active open */ 724 if (tcph->ack) { 725 /* filter the unacceptable acks */ 726 if (TCP_SEQ_LEQ(pctx->p.ack_seq, cur_stream->sndvar->iss) 727 #ifndef BE_RESILIENT_TO_PACKET_DROP 728 || TCP_SEQ_GT(pctx->p.ack_seq, cur_stream->snd_nxt) 729 #endif 730 ) { 731 if (!tcph->rst) { 732 cur_stream->actions |= MOS_ACT_SEND_RST; 733 } 734 return; 735 } 736 /* accept the ack */ 737 cur_stream->sndvar->snd_una++; 738 } 739 740 if (tcph->rst) { 741 if (tcph->ack) { 742 cur_stream->state = TCP_ST_CLOSED_RSVD; 743 cur_stream->close_reason = TCP_RESET; 744 cur_stream->cb_events |= MOS_ON_TCP_STATE_CHANGE; 745 if (cur_stream->socket) { 746 if (HAS_STREAM_TYPE(cur_stream, MOS_SOCK_STREAM)) 747 RaiseErrorEvent(mtcp, cur_stream); 748 } else { 749 cur_stream->actions |= MOS_ACT_DESTROY; 750 } 751 } 752 return; 753 } 754 755 if (tcph->ack 756 #ifndef BE_RESILIENT_TO_PACKET_DROP 757 /* If we already lost SYNACK, let the ACK packet do the SYNACK's role */ 758 && tcph->syn 759 #endif 760 ) { 761 int ret = HandleActiveOpen(mtcp, cur_stream, pctx); 762 if (!ret) { 763 return; 764 } 765 766 #ifdef BE_RESILIENT_TO_PACKET_DROP 767 if (!tcph->syn) { 768 RAISE_DEBUG_EVENT(mtcp, cur_stream, 769 "We missed SYNACK. Replace it with an ACK packet.\n"); 770 /* correct some variables */ 771 cur_stream->rcv_nxt = pctx->p.seq; 772 } 773 #endif 774 775 cur_stream->sndvar->nrtx = 0; 776 if (HAS_STREAM_TYPE(cur_stream, MOS_SOCK_STREAM)) 777 RemoveFromRTOList(mtcp, cur_stream); 778 cur_stream->state = TCP_ST_ESTABLISHED; 779 cur_stream->cb_events |= /*MOS_ON_CONN_SETUP |*/ MOS_ON_TCP_STATE_CHANGE; 780 TRACE_STATE("Stream %d: TCP_ST_ESTABLISHED\n", cur_stream->id); 781 782 if (cur_stream->socket) { 783 if (HAS_STREAM_TYPE(cur_stream, MOS_SOCK_STREAM)) 784 RaiseWriteEvent(mtcp, cur_stream); 785 } else { 786 TRACE_STATE("Stream %d: ESTABLISHED, but no socket\n", cur_stream->id); 787 cur_stream->close_reason = TCP_ACTIVE_CLOSE; 788 cur_stream->actions |= MOS_ACT_SEND_RST; 789 cur_stream->actions |= MOS_ACT_DESTROY; 790 } 791 cur_stream->actions |= MOS_ACT_SEND_CONTROL; 792 if (g_config.mos->tcp_timeout > 0) 793 if (HAS_STREAM_TYPE(cur_stream, MOS_SOCK_STREAM)) 794 AddtoTimeoutList(mtcp, cur_stream); 795 796 #ifdef BE_RESILIENT_TO_PACKET_DROP 797 /* Handle this ack packet */ 798 if (!tcph->syn) 799 Handle_TCP_ST_ESTABLISHED(mtcp, cur_stream, pctx); 800 #endif 801 802 } else if (tcph->syn) { 803 cur_stream->state = TCP_ST_SYN_RCVD; 804 cur_stream->cb_events |= MOS_ON_TCP_STATE_CHANGE; 805 TRACE_STATE("Stream %d: TCP_ST_SYN_RCVD\n", cur_stream->id); 806 cur_stream->snd_nxt = cur_stream->sndvar->iss; 807 cur_stream->actions |= MOS_ACT_SEND_CONTROL; 808 } 809 } 810 /*----------------------------------------------------------------------------*/ 811 static inline void 812 Handle_TCP_ST_SYN_RCVD (mtcp_manager_t mtcp, tcp_stream* cur_stream, 813 struct pkt_ctx *pctx) 814 { 815 const struct tcphdr* tcph = pctx->p.tcph; 816 struct tcp_send_vars *sndvar = cur_stream->sndvar; 817 int ret; 818 819 if (tcph->ack) { 820 uint32_t prior_cwnd; 821 /* NOTE: We do not validate the ack number because first few packets 822 * can also come out of order */ 823 824 sndvar->snd_una++; 825 cur_stream->snd_nxt = pctx->p.ack_seq; 826 prior_cwnd = sndvar->cwnd; 827 sndvar->cwnd = ((prior_cwnd == 1)? 828 (sndvar->mss * 2): sndvar->mss); 829 830 //UpdateRetransmissionTimer(mtcp, cur_stream, cur_ts); 831 sndvar->nrtx = 0; 832 cur_stream->rcv_nxt = cur_stream->rcvvar->irs + 1; 833 if (HAS_STREAM_TYPE(cur_stream, MOS_SOCK_STREAM)) 834 RemoveFromRTOList(mtcp, cur_stream); 835 836 cur_stream->state = TCP_ST_ESTABLISHED; 837 cur_stream->cb_events |= /*MOS_ON_CONN_SETUP |*/ MOS_ON_TCP_STATE_CHANGE; 838 TRACE_STATE("Stream %d: TCP_ST_ESTABLISHED\n", cur_stream->id); 839 840 #ifdef BE_RESILIENT_TO_PACKET_DROP 841 if (pctx->p.ack_seq != sndvar->iss + 1) 842 Handle_TCP_ST_ESTABLISHED(mtcp, cur_stream, pctx); 843 #endif 844 845 /* update listening socket */ 846 if (HAS_STREAM_TYPE(cur_stream, MOS_SOCK_STREAM)) { 847 848 struct tcp_listener *listener = mtcp->listener; 849 850 ret = StreamEnqueue(listener->acceptq, cur_stream); 851 if (ret < 0) { 852 TRACE_ERROR("Stream %d: Failed to enqueue to " 853 "the listen backlog!\n", cur_stream->id); 854 cur_stream->close_reason = TCP_NOT_ACCEPTED; 855 cur_stream->state = TCP_ST_CLOSED_RSVD; 856 cur_stream->cb_events |= MOS_ON_TCP_STATE_CHANGE; 857 TRACE_STATE("Stream %d: TCP_ST_CLOSED_RSVD\n", cur_stream->id); 858 cur_stream->actions |= MOS_ACT_SEND_CONTROL; 859 } 860 861 /* raise an event to the listening socket */ 862 if (listener->socket && (listener->socket->epoll & MOS_EPOLLIN)) { 863 AddEpollEvent(mtcp->ep, 864 MOS_EVENT_QUEUE, listener->socket, MOS_EPOLLIN); 865 } 866 } 867 868 //TRACE_DBG("Stream %d inserted into acceptq.\n", cur_stream->id); 869 if (g_config.mos->tcp_timeout > 0) 870 if (HAS_STREAM_TYPE(cur_stream, MOS_SOCK_STREAM)) 871 AddtoTimeoutList(mtcp, cur_stream); 872 873 } else { 874 /* Handle retransmitted SYN packet */ 875 if (HAS_STREAM_TYPE(cur_stream, MOS_SOCK_MONITOR_STREAM_ACTIVE) && 876 tcph->syn) { 877 if (pctx->p.seq == cur_stream->pair_stream->sndvar->iss) { 878 TRACE_DBG("syn retransmit! (p.seq = %u / iss = %u)\n", 879 pctx->p.seq, cur_stream->pair_stream->sndvar->iss); 880 cur_stream->cb_events |= MOS_ON_REXMIT; 881 } 882 } 883 884 TRACE_DBG("Stream %d (TCP_ST_SYN_RCVD): No ACK.\n", 885 cur_stream->id); 886 /* retransmit SYN/ACK */ 887 cur_stream->snd_nxt = sndvar->iss; 888 cur_stream->actions |= MOS_ACT_SEND_CONTROL; 889 } 890 } 891 /*----------------------------------------------------------------------------*/ 892 static inline void 893 Handle_TCP_ST_ESTABLISHED (mtcp_manager_t mtcp, tcp_stream* cur_stream, 894 struct pkt_ctx *pctx) 895 { 896 const struct tcphdr* tcph = pctx->p.tcph; 897 898 if (tcph->syn) { 899 /* Handle retransmitted SYNACK packet */ 900 if (HAS_STREAM_TYPE(cur_stream, MOS_SOCK_MONITOR_STREAM_ACTIVE) && 901 tcph->ack) { 902 if (pctx->p.seq == cur_stream->pair_stream->sndvar->iss) { 903 TRACE_DBG("syn/ack retransmit! (p.seq = %u / iss = %u)\n", 904 pctx->p.seq, cur_stream->pair_stream->sndvar->iss); 905 cur_stream->cb_events |= MOS_ON_REXMIT; 906 } 907 } 908 909 TRACE_DBG("Stream %d (TCP_ST_ESTABLISHED): weird SYN. " 910 "seq: %u, expected: %u, ack_seq: %u, expected: %u\n", 911 cur_stream->id, pctx->p.seq, cur_stream->rcv_nxt, 912 pctx->p.ack_seq, cur_stream->snd_nxt); 913 cur_stream->snd_nxt = pctx->p.ack_seq; 914 cur_stream->actions |= MOS_ACT_SEND_CONTROL; 915 return; 916 } 917 918 if (pctx->p.payloadlen > 0) { 919 if (ProcessTCPPayload(mtcp, cur_stream, pctx)) { 920 /* if return is TRUE, send ACK */ 921 cur_stream->actions |= MOS_ACT_SEND_ACK_AGG; 922 } else { 923 cur_stream->actions |= MOS_ACT_SEND_ACK_NOW; 924 } 925 } 926 927 if (tcph->ack) { 928 if (cur_stream->sndvar->sndbuf) { 929 ProcessACK(mtcp, cur_stream, pctx); 930 } 931 } 932 933 if (tcph->fin) { 934 /* process the FIN only if the sequence is valid */ 935 /* FIN packet is allowed to push payload (should we check for PSH flag)? */ 936 if (!cur_stream->buffer_mgmt || 937 #ifdef BE_RESILIENT_TO_PACKET_DROP 938 TCP_SEQ_GEQ(pctx->p.seq + pctx->p.payloadlen, cur_stream->rcv_nxt) 939 #else 940 pctx->p.seq + pctx->p.payloadlen == cur_stream->rcv_nxt 941 #endif 942 ) { 943 #ifdef BE_RESILIENT_TO_PACKET_DROP 944 cur_stream->rcv_nxt = pctx->p.seq + pctx->p.payloadlen; 945 #endif 946 cur_stream->state = TCP_ST_CLOSE_WAIT; 947 cur_stream->cb_events |= MOS_ON_TCP_STATE_CHANGE; 948 TRACE_STATE("Stream %d: TCP_ST_CLOSE_WAIT\n", cur_stream->id); 949 cur_stream->rcv_nxt++; 950 cur_stream->actions |= MOS_ACT_SEND_CONTROL; 951 952 /* notify FIN to application */ 953 RaiseReadEvent(mtcp, cur_stream); 954 } else { 955 RAISE_DEBUG_EVENT(mtcp, cur_stream, 956 "Expected %u, but received %u (= %u + %u)\n", 957 cur_stream->rcv_nxt, pctx->p.seq + pctx->p.payloadlen, 958 pctx->p.seq, pctx->p.payloadlen); 959 960 cur_stream->actions |= MOS_ACT_SEND_ACK_NOW; 961 return; 962 } 963 } 964 } 965 /*----------------------------------------------------------------------------*/ 966 static inline void 967 Handle_TCP_ST_CLOSE_WAIT (mtcp_manager_t mtcp, tcp_stream* cur_stream, 968 struct pkt_ctx *pctx) 969 { 970 if (TCP_SEQ_LT(pctx->p.seq, cur_stream->rcv_nxt)) { 971 TRACE_DBG("Stream %d (TCP_ST_CLOSE_WAIT): " 972 "weird seq: %u, expected: %u\n", 973 cur_stream->id, pctx->p.seq, cur_stream->rcv_nxt); 974 cur_stream->actions |= MOS_ACT_SEND_CONTROL; 975 return; 976 } 977 978 if (cur_stream->sndvar->sndbuf) { 979 ProcessACK(mtcp, cur_stream, pctx); 980 } 981 } 982 /*----------------------------------------------------------------------------*/ 983 static inline void 984 Handle_TCP_ST_LAST_ACK (mtcp_manager_t mtcp, tcp_stream* cur_stream, 985 struct pkt_ctx *pctx) 986 { 987 const struct tcphdr* tcph = pctx->p.tcph; 988 989 if (TCP_SEQ_LT(pctx->p.seq, cur_stream->rcv_nxt)) { 990 TRACE_DBG("Stream %d (TCP_ST_LAST_ACK): " 991 "weird seq: %u, expected: %u\n", 992 cur_stream->id, pctx->p.seq, cur_stream->rcv_nxt); 993 return; 994 } 995 996 if (tcph->ack) { 997 if (cur_stream->sndvar->sndbuf) { 998 ProcessACK(mtcp, cur_stream, pctx); 999 } 1000 1001 if (!cur_stream->sndvar->is_fin_sent) { 1002 /* the case that FIN is not sent yet */ 1003 /* this is not ack for FIN, ignore */ 1004 TRACE_DBG("Stream %d (TCP_ST_LAST_ACK): " 1005 "No FIN sent yet.\n", cur_stream->id); 1006 #ifdef DBGMSG 1007 DumpIPPacket(mtcp, pctx->p.iph, pctx->p.ip_len); 1008 #endif 1009 #if DUMP_STREAM 1010 DumpStream(mtcp, cur_stream); 1011 DumpControlList(mtcp, mtcp->n_sender[0]); 1012 #endif 1013 return; 1014 } 1015 1016 /* check if ACK of FIN */ 1017 if (pctx->p.ack_seq == cur_stream->sndvar->fss + 1) { 1018 cur_stream->sndvar->snd_una++; 1019 if (HAS_STREAM_TYPE(cur_stream, MOS_SOCK_STREAM)) 1020 UpdateRetransmissionTimer(mtcp, cur_stream, pctx->p.cur_ts); 1021 cur_stream->state = TCP_ST_CLOSED_RSVD; 1022 cur_stream->close_reason = TCP_PASSIVE_CLOSE; 1023 cur_stream->cb_events |= MOS_ON_TCP_STATE_CHANGE; 1024 TRACE_STATE("Stream %d: TCP_ST_CLOSED_RSVD\n", 1025 cur_stream->id); 1026 cur_stream->actions |= MOS_ACT_DESTROY; 1027 } else { 1028 TRACE_DBG("Stream %d (TCP_ST_LAST_ACK): Not ACK of FIN. " 1029 "ack_seq: %u, expected: %u\n", 1030 cur_stream->id, pctx->p.ack_seq, cur_stream->sndvar->fss + 1); 1031 //cur_stream->snd_nxt = cur_stream->sndvar->fss; 1032 cur_stream->actions |= MOS_ACT_SEND_CONTROL; 1033 } 1034 } else { 1035 TRACE_DBG("Stream %d (TCP_ST_LAST_ACK): No ACK\n", 1036 cur_stream->id); 1037 //cur_stream->snd_nxt = cur_stream->sndvar->fss; 1038 cur_stream->actions |= MOS_ACT_SEND_CONTROL; 1039 } 1040 } 1041 /*----------------------------------------------------------------------------*/ 1042 static inline void 1043 Handle_TCP_ST_FIN_WAIT_1 (mtcp_manager_t mtcp, tcp_stream* cur_stream, 1044 struct pkt_ctx *pctx) 1045 { 1046 const struct tcphdr* tcph = pctx->p.tcph; 1047 1048 if (TCP_SEQ_LT(pctx->p.seq, cur_stream->rcv_nxt)) { 1049 RAISE_DEBUG_EVENT(mtcp, cur_stream, 1050 "Stream %d (FIN_WAIT_1): " 1051 "weird seq: %u, expected: %u\n", 1052 cur_stream->id, pctx->p.seq, cur_stream->rcv_nxt); 1053 cur_stream->actions |= MOS_ACT_SEND_CONTROL; 1054 return; 1055 } 1056 1057 if (tcph->ack) { 1058 if (cur_stream->sndvar->sndbuf) { 1059 ProcessACK(mtcp, cur_stream, pctx); 1060 } 1061 1062 if (cur_stream->sndvar->is_fin_sent && 1063 ((!HAS_STREAM_TYPE(cur_stream, MOS_SOCK_STREAM) && 1064 pctx->p.ack_seq == cur_stream->sndvar->fss) || 1065 pctx->p.ack_seq == cur_stream->sndvar->fss + 1)) { 1066 cur_stream->sndvar->snd_una = pctx->p.ack_seq; 1067 if (TCP_SEQ_GT(pctx->p.ack_seq, cur_stream->snd_nxt)) { 1068 TRACE_DBG("Stream %d: update snd_nxt to %u\n", 1069 cur_stream->id, pctx->p.ack_seq); 1070 cur_stream->snd_nxt = pctx->p.ack_seq; 1071 } 1072 //cur_stream->sndvar->snd_una++; 1073 //UpdateRetransmissionTimer(mtcp, cur_stream, cur_ts); 1074 cur_stream->sndvar->nrtx = 0; 1075 if (HAS_STREAM_TYPE(cur_stream, MOS_SOCK_STREAM)) 1076 RemoveFromRTOList(mtcp, cur_stream); 1077 cur_stream->state = TCP_ST_FIN_WAIT_2; 1078 cur_stream->cb_events |= MOS_ON_TCP_STATE_CHANGE; 1079 TRACE_STATE("Stream %d: TCP_ST_FIN_WAIT_2\n", 1080 cur_stream->id); 1081 } else { 1082 RAISE_DEBUG_EVENT(mtcp, cur_stream, 1083 "Failed to transit to FIN_WAIT_2, " 1084 "is_fin_sent: %s, ack_seq: %u, sndvar->fss: %u\n", 1085 cur_stream->sndvar->is_fin_sent ? "true" : "false", 1086 pctx->p.ack_seq, cur_stream->sndvar->fss); 1087 } 1088 1089 } else { 1090 RAISE_DEBUG_EVENT(mtcp, cur_stream, 1091 "Failed to transit to FIN_WAIT_2, " 1092 "We got a %s%s%s%s%s%s packet.", 1093 pctx->p.tcph->syn ? "S" : "", 1094 pctx->p.tcph->fin ? "F" : "", 1095 pctx->p.tcph->rst ? "R" : "", 1096 pctx->p.tcph->psh ? "P" : "", 1097 pctx->p.tcph->urg ? "U" : "", 1098 pctx->p.tcph->ack ? "A" : ""); 1099 1100 TRACE_DBG("Stream %d: does not contain an ack!\n", 1101 cur_stream->id); 1102 return; 1103 } 1104 1105 if (pctx->p.payloadlen > 0) { 1106 if (ProcessTCPPayload(mtcp, cur_stream, pctx)) { 1107 /* if return is TRUE, send ACK */ 1108 cur_stream->actions |= MOS_ACT_SEND_ACK_AGG; 1109 } else { 1110 cur_stream->actions |= MOS_ACT_SEND_ACK_NOW; 1111 } 1112 } 1113 1114 if (tcph->fin) { 1115 /* process the FIN only if the sequence is valid */ 1116 /* FIN packet is allowed to push payload (should we check for PSH flag)? */ 1117 if (!cur_stream->buffer_mgmt || 1118 #ifdef BE_RESILIENT_TO_PACKET_DROP 1119 TCP_SEQ_GEQ(pctx->p.seq + pctx->p.payloadlen, cur_stream->rcv_nxt) 1120 #else 1121 pctx->p.seq + pctx->p.payloadlen == cur_stream->rcv_nxt 1122 #endif 1123 ) { 1124 cur_stream->rcv_nxt++; 1125 1126 if (cur_stream->state == TCP_ST_FIN_WAIT_1) { 1127 cur_stream->state = TCP_ST_CLOSING; 1128 cur_stream->cb_events |= MOS_ON_TCP_STATE_CHANGE; 1129 TRACE_STATE("Stream %d: TCP_ST_CLOSING\n", cur_stream->id); 1130 1131 } else if (cur_stream->state == TCP_ST_FIN_WAIT_2) { 1132 cur_stream->state = TCP_ST_TIME_WAIT; 1133 cur_stream->cb_events |= MOS_ON_TCP_STATE_CHANGE; 1134 TRACE_STATE("Stream %d: TCP_ST_TIME_WAIT\n", cur_stream->id); 1135 AddtoTimewaitList(mtcp, cur_stream, pctx->p.cur_ts); 1136 } 1137 cur_stream->actions |= MOS_ACT_SEND_CONTROL; 1138 } else { 1139 RAISE_DEBUG_EVENT(mtcp, cur_stream, 1140 "Expected %u, but received %u (= %u + %u)\n", 1141 cur_stream->rcv_nxt, pctx->p.seq + pctx->p.payloadlen, 1142 pctx->p.seq, pctx->p.payloadlen); 1143 1144 cur_stream->actions |= MOS_ACT_SEND_ACK_NOW; 1145 return; 1146 } 1147 } 1148 } 1149 /*----------------------------------------------------------------------------*/ 1150 static inline void 1151 Handle_TCP_ST_FIN_WAIT_2 (mtcp_manager_t mtcp, tcp_stream* cur_stream, 1152 struct pkt_ctx *pctx) 1153 { 1154 const struct tcphdr* tcph = pctx->p.tcph; 1155 1156 if (tcph->ack) { 1157 if (cur_stream->sndvar->sndbuf) { 1158 ProcessACK(mtcp, cur_stream, pctx); 1159 } 1160 } else { 1161 TRACE_DBG("Stream %d: does not contain an ack!\n", 1162 cur_stream->id); 1163 return; 1164 } 1165 1166 if (pctx->p.payloadlen > 0) { 1167 if (ProcessTCPPayload(mtcp, cur_stream, pctx)) { 1168 /* if return is TRUE, send ACK */ 1169 cur_stream->actions |= MOS_ACT_SEND_ACK_AGG; 1170 } else { 1171 cur_stream->actions |= MOS_ACT_SEND_ACK_NOW; 1172 } 1173 } 1174 1175 if (tcph->fin) { 1176 /* process the FIN only if the sequence is valid */ 1177 /* FIN packet is allowed to push payload (should we check for PSH flag)? */ 1178 if (!cur_stream->buffer_mgmt || 1179 #ifdef BE_RESILIENT_TO_PACKET_DROP 1180 TCP_SEQ_GEQ(pctx->p.seq + pctx->p.payloadlen, cur_stream->rcv_nxt) 1181 #else 1182 pctx->p.seq + pctx->p.payloadlen == cur_stream->rcv_nxt 1183 #endif 1184 ) { 1185 cur_stream->state = TCP_ST_TIME_WAIT; 1186 cur_stream->cb_events |= MOS_ON_TCP_STATE_CHANGE; 1187 cur_stream->rcv_nxt++; 1188 TRACE_STATE("Stream %d: TCP_ST_TIME_WAIT\n", cur_stream->id); 1189 1190 AddtoTimewaitList(mtcp, cur_stream, pctx->p.cur_ts); 1191 cur_stream->actions |= MOS_ACT_SEND_CONTROL; 1192 } 1193 } else { 1194 TRACE_DBG("Stream %d (TCP_ST_FIN_WAIT_2): No FIN. " 1195 "seq: %u, ack_seq: %u, snd_nxt: %u, snd_una: %u\n", 1196 cur_stream->id, seq, ack_seq, 1197 cur_stream->snd_nxt, cur_stream->sndvar->snd_una); 1198 #if DBGMSG 1199 DumpIPPacket(mtcp, iph, ip_len); 1200 #endif 1201 } 1202 1203 } 1204 /*----------------------------------------------------------------------------*/ 1205 static inline void 1206 Handle_TCP_ST_CLOSING (mtcp_manager_t mtcp, tcp_stream* cur_stream, 1207 struct pkt_ctx *pctx) 1208 { 1209 const struct tcphdr* tcph = pctx->p.tcph; 1210 1211 if (tcph->ack) { 1212 if (cur_stream->sndvar->sndbuf) { 1213 ProcessACK(mtcp, cur_stream, pctx); 1214 } 1215 1216 if (!cur_stream->sndvar->is_fin_sent) { 1217 TRACE_DBG("Stream %d (TCP_ST_CLOSING): " 1218 "No FIN sent yet.\n", cur_stream->id); 1219 return; 1220 } 1221 1222 // check if ACK of FIN 1223 if (pctx->p.ack_seq != cur_stream->sndvar->fss + 1) { 1224 #if DBGMSG 1225 TRACE_DBG("Stream %d (TCP_ST_CLOSING): Not ACK of FIN. " 1226 "ack_seq: %u, snd_nxt: %u, snd_una: %u, fss: %u\n", 1227 cur_stream->id, ack_seq, cur_stream->snd_nxt, 1228 cur_stream->sndvar->snd_una, cur_stream->sndvar->fss); 1229 DumpIPPacketToFile(stderr, iph, ip_len); 1230 DumpStream(mtcp, cur_stream); 1231 #endif 1232 //assert(0); 1233 /* if the packet is not the ACK of FIN, ignore */ 1234 return; 1235 } 1236 1237 cur_stream->sndvar->snd_una = pctx->p.ack_seq; 1238 if (HAS_STREAM_TYPE(cur_stream, MOS_SOCK_STREAM)) 1239 UpdateRetransmissionTimer(mtcp, cur_stream, pctx->p.cur_ts); 1240 cur_stream->state = TCP_ST_TIME_WAIT; 1241 cur_stream->cb_events |= MOS_ON_TCP_STATE_CHANGE; 1242 TRACE_STATE("Stream %d: TCP_ST_TIME_WAIT\n", cur_stream->id); 1243 1244 AddtoTimewaitList(mtcp, cur_stream, pctx->p.cur_ts); 1245 1246 } else { 1247 TRACE_DBG("Stream %d (TCP_ST_CLOSING): Not ACK\n", 1248 cur_stream->id); 1249 return; 1250 } 1251 } 1252 /*----------------------------------------------------------------------------*/ 1253 void 1254 UpdateRecvTCPContext(mtcp_manager_t mtcp, struct tcp_stream *cur_stream, 1255 struct pkt_ctx *pctx) 1256 { 1257 struct tcphdr* tcph = pctx->p.tcph; 1258 int ret; 1259 1260 assert(cur_stream); 1261 1262 /* Validate sequence. if not valid, ignore the packet */ 1263 if (cur_stream->state > TCP_ST_SYN_RCVD) { 1264 1265 ret = ValidateSequence(mtcp, cur_stream, pctx); 1266 if (!ret) { 1267 TRACE_DBG("Stream %d: Unexpected sequence: %u, expected: %u\n", 1268 cur_stream->id, pctx->p.seq, cur_stream->rcv_nxt); 1269 #ifdef DBGMSG 1270 DumpIPPacket(mtcp, pctx->p.iph, pctx->p.ip_len); 1271 #endif 1272 #if DUMP_STREAM 1273 DumpStream(mtcp, cur_stream); 1274 #endif 1275 /* cur_stream->cb_events |= MOS_ON_ERROR; */ 1276 } 1277 } 1278 /* Update receive window size */ 1279 if (tcph->syn) { 1280 cur_stream->sndvar->peer_wnd = pctx->p.window; 1281 } else { 1282 cur_stream->sndvar->peer_wnd = 1283 (uint32_t)pctx->p.window << cur_stream->sndvar->wscale_peer; 1284 } 1285 1286 cur_stream->last_active_ts = pctx->p.cur_ts; 1287 if (HAS_STREAM_TYPE(cur_stream, MOS_SOCK_STREAM)) 1288 UpdateTimeoutList(mtcp, cur_stream); 1289 1290 /* Process RST: process here only if state > TCP_ST_SYN_SENT */ 1291 if (tcph->rst) { 1292 cur_stream->have_reset = TRUE; 1293 if (cur_stream->state > TCP_ST_SYN_SENT) { 1294 if (ProcessRST(mtcp, cur_stream, pctx)) { 1295 return; 1296 } 1297 } 1298 } 1299 1300 if (HAS_STREAM_TYPE(cur_stream, MOS_SOCK_MONITOR_STREAM_ACTIVE) && 1301 pctx->p.tcph->fin) { 1302 if (cur_stream->state == TCP_ST_CLOSE_WAIT || 1303 cur_stream->state == TCP_ST_LAST_ACK || 1304 cur_stream->state == TCP_ST_CLOSING || 1305 cur_stream->state == TCP_ST_TIME_WAIT) { 1306 /* Handle retransmitted FIN packet */ 1307 if (pctx->p.seq == cur_stream->pair_stream->sndvar->fss) { 1308 TRACE_DBG("FIN retransmit! (seq = %u / fss = %u)\n", 1309 pctx->p.seq, cur_stream->pair_stream->sndvar->fss, 1310 cur_stream->sndvar->fss); 1311 cur_stream->cb_events |= MOS_ON_REXMIT; 1312 } 1313 } 1314 } 1315 1316 switch (cur_stream->state) { 1317 case TCP_ST_LISTEN: 1318 Handle_TCP_ST_LISTEN(mtcp, cur_stream, pctx); 1319 break; 1320 1321 case TCP_ST_SYN_SENT: 1322 Handle_TCP_ST_SYN_SENT(mtcp, cur_stream, pctx); 1323 break; 1324 1325 case TCP_ST_SYN_RCVD: 1326 /* SYN retransmit implies our SYN/ACK was lost. Resend */ 1327 if (tcph->syn && pctx->p.seq == cur_stream->rcvvar->irs) 1328 Handle_TCP_ST_LISTEN(mtcp, cur_stream, pctx); 1329 else 1330 Handle_TCP_ST_SYN_RCVD(mtcp, cur_stream, pctx); 1331 break; 1332 1333 case TCP_ST_ESTABLISHED: 1334 Handle_TCP_ST_ESTABLISHED(mtcp, cur_stream, pctx); 1335 break; 1336 1337 case TCP_ST_CLOSE_WAIT: 1338 Handle_TCP_ST_CLOSE_WAIT(mtcp, cur_stream, pctx); 1339 break; 1340 1341 case TCP_ST_LAST_ACK: 1342 Handle_TCP_ST_LAST_ACK(mtcp, cur_stream, pctx); 1343 break; 1344 1345 case TCP_ST_FIN_WAIT_1: 1346 Handle_TCP_ST_FIN_WAIT_1(mtcp, cur_stream, pctx); 1347 break; 1348 1349 case TCP_ST_FIN_WAIT_2: 1350 Handle_TCP_ST_FIN_WAIT_2(mtcp, cur_stream, pctx); 1351 break; 1352 1353 case TCP_ST_CLOSING: 1354 Handle_TCP_ST_CLOSING(mtcp, cur_stream, pctx); 1355 break; 1356 1357 case TCP_ST_TIME_WAIT: 1358 /* the only thing that can arrive in this state is a retransmission 1359 of the remote FIN. Acknowledge it, and restart the 2 MSL timeout */ 1360 if (cur_stream->on_timewait_list) { 1361 RemoveFromTimewaitList(mtcp, cur_stream); 1362 AddtoTimewaitList(mtcp, cur_stream, pctx->p.cur_ts); 1363 } 1364 cur_stream->actions |= MOS_ACT_SEND_CONTROL; 1365 break; 1366 1367 case TCP_ST_CLOSED: 1368 case TCP_ST_CLOSED_RSVD: 1369 break; 1370 1371 default: 1372 break; 1373 } 1374 1375 TRACE_STATE("Stream %d: Events: %0lx, Action: %0x\n", 1376 cur_stream->id, cur_stream->cb_events, cur_stream->actions); 1377 return; 1378 } 1379 /*----------------------------------------------------------------------------*/ 1380 void 1381 DoActionEndTCPPacket(mtcp_manager_t mtcp, struct tcp_stream *cur_stream, 1382 struct pkt_ctx *pctx) 1383 { 1384 int i; 1385 1386 for (i = 1; i < MOS_ACT_CNT; i = i << 1) { 1387 1388 if (cur_stream->actions & i) { 1389 switch(i) { 1390 case MOS_ACT_SEND_DATA: 1391 AddtoSendList(mtcp, cur_stream); 1392 break; 1393 case MOS_ACT_SEND_ACK_NOW: 1394 EnqueueACK(mtcp, cur_stream, pctx->p.cur_ts, ACK_OPT_NOW); 1395 break; 1396 case MOS_ACT_SEND_ACK_AGG: 1397 EnqueueACK(mtcp, cur_stream, pctx->p.cur_ts, ACK_OPT_AGGREGATE); 1398 break; 1399 case MOS_ACT_SEND_CONTROL: 1400 AddtoControlList(mtcp, cur_stream, pctx->p.cur_ts); 1401 break; 1402 case MOS_ACT_SEND_RST: 1403 if (cur_stream->state <= TCP_ST_SYN_SENT) 1404 SendTCPPacketStandalone(mtcp, 1405 pctx->p.iph->daddr, pctx->p.tcph->dest, 1406 pctx->p.iph->saddr, pctx->p.tcph->source, 1407 0, pctx->p.seq + 1, 0, TCP_FLAG_RST | TCP_FLAG_ACK, 1408 NULL, 0, pctx->p.cur_ts, 0); 1409 else 1410 SendTCPPacketStandalone(mtcp, 1411 pctx->p.iph->daddr, pctx->p.tcph->dest, 1412 pctx->p.iph->saddr, pctx->p.tcph->source, 1413 pctx->p.ack_seq, 0, 0, TCP_FLAG_RST | TCP_FLAG_ACK, 1414 NULL, 0, pctx->p.cur_ts, 0); 1415 break; 1416 case MOS_ACT_DESTROY: 1417 DestroyTCPStream(mtcp, cur_stream); 1418 break; 1419 default: 1420 assert(1); 1421 break; 1422 } 1423 } 1424 } 1425 1426 cur_stream->actions = 0; 1427 } 1428 /*----------------------------------------------------------------------------*/ 1429 /** 1430 * Called (when monitoring mode is enabled).. for every outgoing packet to the 1431 * NIC. 1432 */ 1433 inline void 1434 UpdatePassiveRecvTCPContext(mtcp_manager_t mtcp, struct tcp_stream *cur_stream, 1435 struct pkt_ctx *pctx) 1436 { 1437 UpdateRecvTCPContext(mtcp, cur_stream, pctx); 1438 } 1439 /*----------------------------------------------------------------------------*/ 1440 /* NOTE TODO: This event prediction is additional overhaed of POST_RCV hook. 1441 * We can transparently optimize this by disabling prediction of events which 1442 * are not monitored by anyone. */ 1443 inline void 1444 PreRecvTCPEventPrediction(mtcp_manager_t mtcp, struct pkt_ctx *pctx, 1445 struct tcp_stream *recvside_stream) 1446 { 1447 #ifdef NEWRB 1448 #define DOESOVERLAP(a1, a2, b1, b2) \ 1449 ((a1 != b2) && (a2 != b1) && ((a1 > b2) != (a2 > b1))) 1450 1451 /* Check whether this packet is retransmitted or not. */ 1452 tcprb_t *rb; 1453 if (pctx->p.payloadlen > 0 && recvside_stream->rcvvar != NULL 1454 && (rb = recvside_stream->rcvvar->rcvbuf) != NULL) { 1455 struct _tcpfrag_t *f; 1456 loff_t off = seq2loff(rb, pctx->p.seq, recvside_stream->rcvvar->irs + 1); 1457 TAILQ_FOREACH(f, &rb->frags, link) 1458 if (DOESOVERLAP(f->head, f->tail, off, off + pctx->p.payloadlen)) { 1459 recvside_stream->cb_events |= MOS_ON_REXMIT; 1460 TRACE_DBG("RETX!\n"); 1461 break; 1462 } 1463 } 1464 #else 1465 /* Check whether this packet is retransmitted or not. */ 1466 struct tcp_ring_buffer *rcvbuf; 1467 if (pctx->p.payloadlen > 0 && recvside_stream->rcvvar != NULL 1468 && (rcvbuf = recvside_stream->rcvvar->rcvbuf) != NULL 1469 && RBDoesOverlap(mtcp->rbm_rcv, rcvbuf, pctx->p.seq, pctx->p.payloadlen)) { 1470 recvside_stream->cb_events |= MOS_ON_REXMIT; 1471 TRACE_DBG("RETX!\n"); 1472 } 1473 #endif 1474 } 1475 /*----------------------------------------------------------------------------*/ 1476