Seastar
High performance C++ framework for concurrent servers
All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Friends Modules Pages
tcp.hh
1/*
2 * This file is open source software, licensed to you under the terms
3 * of the Apache License, Version 2.0 (the "License"). See the NOTICE file
4 * distributed with this work for additional information regarding copyright
5 * ownership. You may not use this file except in compliance with the License.
6 *
7 * You may obtain a copy of the License at
8 *
9 * http://www.apache.org/licenses/LICENSE-2.0
10 *
11 * Unless required by applicable law or agreed to in writing,
12 * software distributed under the License is distributed on an
13 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14 * KIND, either express or implied. See the License for the
15 * specific language governing permissions and limitations
16 * under the License.
17 */
18/*
19 * Copyright (C) 2014 Cloudius Systems, Ltd.
20 */
21
22#pragma once
23
24#ifndef SEASTAR_MODULE
25#include <unordered_map>
26#include <map>
27#include <functional>
28#include <deque>
29#include <chrono>
30#include <random>
31#include <stdexcept>
32#include <system_error>
33#include <gnutls/crypto.h>
34#endif
35#include <seastar/core/shared_ptr.hh>
36#include <seastar/core/queue.hh>
37#include <seastar/core/semaphore.hh>
38#include <seastar/core/byteorder.hh>
40#include <seastar/net/net.hh>
41#include <seastar/net/ip_checksum.hh>
42#include <seastar/net/ip.hh>
43#include <seastar/net/const.hh>
44#include <seastar/net/packet-util.hh>
45#include <seastar/util/std-compat.hh>
46
47namespace seastar {
48
49using namespace std::chrono_literals;
50
51namespace net {
52
53struct tcp_hdr;
54
55inline auto tcp_error(int err) {
56 return std::system_error(err, std::system_category());
57}
58
59inline auto tcp_reset_error() {
60 return tcp_error(ECONNRESET);
61};
62
63inline auto tcp_connect_error() {
64 return tcp_error(ECONNABORTED);
65}
66
67inline auto tcp_refused_error() {
68 return tcp_error(ECONNREFUSED);
69};
70
71enum class tcp_state : uint16_t {
72 CLOSED = (1 << 0),
73 LISTEN = (1 << 1),
74 SYN_SENT = (1 << 2),
75 SYN_RECEIVED = (1 << 3),
76 ESTABLISHED = (1 << 4),
77 FIN_WAIT_1 = (1 << 5),
78 FIN_WAIT_2 = (1 << 6),
79 CLOSE_WAIT = (1 << 7),
80 CLOSING = (1 << 8),
81 LAST_ACK = (1 << 9),
82 TIME_WAIT = (1 << 10)
83};
84
85inline tcp_state operator|(tcp_state s1, tcp_state s2) {
86 return tcp_state(uint16_t(s1) | uint16_t(s2));
87}
88
89template <typename... Args>
90void tcp_debug(const char* fmt, Args&&... args) {
91#if TCP_DEBUG
92 print(fmt, std::forward<Args>(args)...);
93#endif
94}
95
96struct tcp_option {
97 // The kind and len field are fixed and defined in TCP protocol
98 enum class option_kind: uint8_t { mss = 2, win_scale = 3, sack = 4, timestamps = 8, nop = 1, eol = 0 };
99 enum class option_len: uint8_t { mss = 4, win_scale = 3, sack = 2, timestamps = 10, nop = 1, eol = 1 };
100 static void write(char* p, option_kind kind, option_len len) {
101 p[0] = static_cast<uint8_t>(kind);
102 if (static_cast<uint8_t>(len) > 1) {
103 p[1] = static_cast<uint8_t>(len);
104 }
105 }
106 struct mss {
107 static constexpr option_kind kind = option_kind::mss;
108 static constexpr option_len len = option_len::mss;
109 uint16_t mss;
110 static tcp_option::mss read(const char* p) {
112 x.mss = read_be<uint16_t>(p + 2);
113 return x;
114 }
115 void write(char* p) const {
116 tcp_option::write(p, kind, len);
117 write_be<uint16_t>(p + 2, mss);
118 }
119 };
120 struct win_scale {
121 static constexpr option_kind kind = option_kind::win_scale;
122 static constexpr option_len len = option_len::win_scale;
123 uint8_t shift;
124 static tcp_option::win_scale read(const char* p) {
126 x.shift = p[2];
127 return x;
128 }
129 void write(char* p) const {
130 tcp_option::write(p, kind, len);
131 p[2] = shift;
132 }
133 };
134 struct sack {
135 static constexpr option_kind kind = option_kind::sack;
136 static constexpr option_len len = option_len::sack;
137 static tcp_option::sack read(const char* p) {
138 return {};
139 }
140 void write(char* p) const {
141 tcp_option::write(p, kind, len);
142 }
143 };
144 struct timestamps {
145 static constexpr option_kind kind = option_kind::timestamps;
146 static constexpr option_len len = option_len::timestamps;
147 uint32_t t1;
148 uint32_t t2;
149 static tcp_option::timestamps read(const char* p) {
151 ts.t1 = read_be<uint32_t>(p + 2);
152 ts.t2 = read_be<uint32_t>(p + 6);
153 return ts;
154 }
155 void write(char* p) const {
156 tcp_option::write(p, kind, len);
157 write_be<uint32_t>(p + 2, t1);
158 write_be<uint32_t>(p + 6, t2);
159 }
160 };
161 struct nop {
162 static constexpr option_kind kind = option_kind::nop;
163 static constexpr option_len len = option_len::nop;
164 void write(char* p) const {
165 tcp_option::write(p, kind, len);
166 }
167 };
168 struct eol {
169 static constexpr option_kind kind = option_kind::eol;
170 static constexpr option_len len = option_len::eol;
171 void write(char* p) const {
172 tcp_option::write(p, kind, len);
173 }
174 };
175 static const uint8_t align = 4;
176
177 void parse(uint8_t* beg, uint8_t* end);
178 uint8_t fill(void* h, const tcp_hdr* th, uint8_t option_size);
179 uint8_t get_size(bool syn_on, bool ack_on);
180
181 // For option negotiattion
182 bool _mss_received = false;
183 bool _win_scale_received = false;
184 bool _timestamps_received = false;
185 bool _sack_received = false;
186
187 // Option data
188 uint16_t _remote_mss = 536;
189 uint16_t _local_mss;
190 uint8_t _remote_win_scale = 0;
191 uint8_t _local_win_scale = 0;
192};
193inline char*& operator+=(char*& x, tcp_option::option_len len) { x += uint8_t(len); return x; }
194inline const char*& operator+=(const char*& x, tcp_option::option_len len) { x += uint8_t(len); return x; }
195inline uint8_t& operator+=(uint8_t& x, tcp_option::option_len len) { x += uint8_t(len); return x; }
196
197struct tcp_seq {
198 uint32_t raw;
199};
200
201inline tcp_seq ntoh(tcp_seq s) {
202 return tcp_seq { ntoh(s.raw) };
203}
204
205inline tcp_seq hton(tcp_seq s) {
206 return tcp_seq { hton(s.raw) };
207}
208
209inline
210std::ostream& operator<<(std::ostream& os, tcp_seq s) {
211 return os << s.raw;
212}
213
214inline tcp_seq make_seq(uint32_t raw) { return tcp_seq{raw}; }
215inline tcp_seq& operator+=(tcp_seq& s, int32_t n) { s.raw += n; return s; }
216inline tcp_seq& operator-=(tcp_seq& s, int32_t n) { s.raw -= n; return s; }
217inline tcp_seq operator+(tcp_seq s, int32_t n) { return s += n; }
218inline tcp_seq operator-(tcp_seq s, int32_t n) { return s -= n; }
219inline int32_t operator-(tcp_seq s, tcp_seq q) { return s.raw - q.raw; }
220inline bool operator==(tcp_seq s, tcp_seq q) { return s.raw == q.raw; }
221inline bool operator!=(tcp_seq s, tcp_seq q) { return !(s == q); }
222inline bool operator<(tcp_seq s, tcp_seq q) { return s - q < 0; }
223inline bool operator>(tcp_seq s, tcp_seq q) { return q < s; }
224inline bool operator<=(tcp_seq s, tcp_seq q) { return !(s > q); }
225inline bool operator>=(tcp_seq s, tcp_seq q) { return !(s < q); }
226
227struct tcp_hdr {
228 static constexpr size_t len = 20;
229 uint16_t src_port;
230 uint16_t dst_port;
231 tcp_seq seq;
232 tcp_seq ack;
233 uint8_t rsvd1 : 4;
234 uint8_t data_offset : 4;
235 uint8_t f_fin : 1;
236 uint8_t f_syn : 1;
237 uint8_t f_rst : 1;
238 uint8_t f_psh : 1;
239 uint8_t f_ack : 1;
240 uint8_t f_urg : 1;
241 uint8_t rsvd2 : 2;
242 uint16_t window;
243 uint16_t checksum;
244 uint16_t urgent;
245 static tcp_hdr read(const char* p) {
246 tcp_hdr h;
247 h.src_port = read_be<uint16_t>(p + 0);
248 h.dst_port = read_be<uint16_t>(p + 2);
249 h.seq = tcp_seq{read_be<uint32_t>(p + 4)};
250 h.ack = tcp_seq{read_be<uint32_t>(p + 8)};
251 h.rsvd1 = p[12] & 15;
252 h.data_offset = uint8_t(p[12]) >> 4;
253 h.f_fin = (uint8_t(p[13]) >> 0) & 1;
254 h.f_syn = (uint8_t(p[13]) >> 1) & 1;
255 h.f_rst = (uint8_t(p[13]) >> 2) & 1;
256 h.f_psh = (uint8_t(p[13]) >> 3) & 1;
257 h.f_ack = (uint8_t(p[13]) >> 4) & 1;
258 h.f_urg = (uint8_t(p[13]) >> 5) & 1;
259 h.rsvd2 = (uint8_t(p[13]) >> 6) & 3;
260 h.window = read_be<uint16_t>(p + 14);
261 h.checksum = read_be<uint16_t>(p + 16);
262 h.urgent = read_be<uint16_t>(p + 18);
263 return h;
264 }
265 void write(char* p) const {
266 write_be<uint16_t>(p + 0, src_port);
267 write_be<uint16_t>(p + 2, dst_port);
268 write_be<uint32_t>(p + 4, seq.raw);
269 write_be<uint32_t>(p + 8, ack.raw);
270 p[12] = rsvd1 | (data_offset << 4);
271 p[13] = (f_fin << 0)
272 | (f_syn << 1)
273 | (f_rst << 2)
274 | (f_psh << 3)
275 | (f_ack << 4)
276 | (f_urg << 5)
277 | (rsvd2 << 6);
278 write_be<uint16_t>(p + 14, window);
279 write_be<uint16_t>(p + 16, checksum);
280 write_be<uint16_t>(p + 18, urgent);
281 }
282 static void write_nbo_checksum(char* p, uint16_t checksum_in_network_byte_order) {
283 std::copy_n(reinterpret_cast<const char*>(&checksum_in_network_byte_order), 2, p + 16);
284 }
285};
286
287struct tcp_tag {};
289
290template <typename InetTraits>
291class tcp {
292public:
293 using ipaddr = typename InetTraits::address_type;
294 using inet_type = typename InetTraits::inet_type;
296 using connid_hash = typename connid::connid_hash;
297 class connection;
298 class listener;
299private:
300 class tcb;
301
302 class tcb : public enable_lw_shared_from_this<tcb> {
303 using clock_type = lowres_clock;
304 static constexpr tcp_state CLOSED = tcp_state::CLOSED;
305 static constexpr tcp_state LISTEN = tcp_state::LISTEN;
306 static constexpr tcp_state SYN_SENT = tcp_state::SYN_SENT;
307 static constexpr tcp_state SYN_RECEIVED = tcp_state::SYN_RECEIVED;
308 static constexpr tcp_state ESTABLISHED = tcp_state::ESTABLISHED;
309 static constexpr tcp_state FIN_WAIT_1 = tcp_state::FIN_WAIT_1;
310 static constexpr tcp_state FIN_WAIT_2 = tcp_state::FIN_WAIT_2;
311 static constexpr tcp_state CLOSE_WAIT = tcp_state::CLOSE_WAIT;
312 static constexpr tcp_state CLOSING = tcp_state::CLOSING;
313 static constexpr tcp_state LAST_ACK = tcp_state::LAST_ACK;
314 static constexpr tcp_state TIME_WAIT = tcp_state::TIME_WAIT;
315 tcp_state _state = CLOSED;
316 tcp& _tcp;
317 connection* _conn = nullptr;
318 promise<> _connect_done;
319 std::optional<promise<>> _fin_recvd_promise = promise<>();
320 ipaddr _local_ip;
321 ipaddr _foreign_ip;
322 uint16_t _local_port;
323 uint16_t _foreign_port;
324 struct unacked_segment {
325 packet p;
326 uint16_t data_len;
327 unsigned nr_transmits;
328 clock_type::time_point tx_time;
329 };
330 struct send {
331 tcp_seq unacknowledged;
332 tcp_seq next;
333 uint32_t window;
334 uint8_t window_scale;
335 uint16_t mss;
336 tcp_seq urgent;
337 tcp_seq wl1;
338 tcp_seq wl2;
339 tcp_seq initial;
340 std::deque<unacked_segment> data;
341 std::deque<packet> unsent;
342 uint32_t unsent_len = 0;
343 bool closed = false;
344 promise<> _window_opened;
345 // Wait for all data are acked
346 std::optional<promise<>> _all_data_acked_promise;
347 // Limit number of data queued into send queue
348 size_t max_queue_space = 212992;
349 size_t current_queue_space = 0;
350 // wait for there is at least one byte available in the queue
351 std::optional<promise<>> _send_available_promise;
352 // Round-trip time variation
353 std::chrono::milliseconds rttvar;
354 // Smoothed round-trip time
355 std::chrono::milliseconds srtt;
356 bool first_rto_sample = true;
357 clock_type::time_point syn_tx_time;
358 // Congestion window
359 uint32_t cwnd;
360 // Slow start threshold
361 uint32_t ssthresh;
362 // Duplicated ACKs
363 uint16_t dupacks = 0;
364 unsigned syn_retransmit = 0;
365 unsigned fin_retransmit = 0;
366 uint32_t limited_transfer = 0;
367 uint32_t partial_ack = 0;
368 tcp_seq recover;
369 bool window_probe = false;
370 uint8_t zero_window_probing_out = 0;
371 } _snd;
372 struct receive {
373 tcp_seq next;
374 uint32_t window;
375 uint8_t window_scale;
376 uint16_t mss;
377 tcp_seq urgent;
378 tcp_seq initial;
379 std::deque<packet> data;
380 // The total size of data stored in std::deque<packet> data
381 size_t data_size = 0;
382 tcp_packet_merger out_of_order;
383 std::optional<promise<>> _data_received_promise;
384 // The maximun memory buffer size allowed for receiving
385 // Currently, it is the same as default receive window size when window scaling is enabled
386 size_t max_receive_buf_size = 3737600;
387 } _rcv;
388 tcp_option _option;
389 timer<lowres_clock> _delayed_ack;
390 // Retransmission timeout
391 std::chrono::milliseconds _rto{1000};
392 std::chrono::milliseconds _persist_time_out{1000};
393 static constexpr std::chrono::milliseconds _rto_min{1000};
394 static constexpr std::chrono::milliseconds _rto_max{60000};
395 // Clock granularity
396 static constexpr std::chrono::milliseconds _rto_clk_granularity{1};
397 static constexpr uint16_t _max_nr_retransmit{5};
398 timer<lowres_clock> _retransmit;
399 timer<lowres_clock> _persist;
400 uint16_t _nr_full_seg_received = 0;
401 struct isn_secret {
402 // 512 bits secretkey for ISN generating
403 uint32_t key[16];
404 isn_secret () {
405 std::random_device rd;
406 std::default_random_engine e(rd());
407 std::uniform_int_distribution<uint32_t> dist{};
408 for (auto& k : key) {
409 k = dist(e);
410 }
411 }
412 };
413 static isn_secret _isn_secret;
414 tcp_seq get_isn();
416 bool _poll_active = false;
417 uint32_t get_default_receive_window_size() {
418 // Linux's default window size
419 constexpr uint32_t size = 29200;
420 return size << _rcv.window_scale;
421 }
422 // Returns the current receive window according to available receiving buffer size
423 uint32_t get_modified_receive_window_size() {
424 uint32_t left = _rcv.data_size > _rcv.max_receive_buf_size ? 0 : _rcv.max_receive_buf_size - _rcv.data_size;
425 return std::min(left, get_default_receive_window_size());
426 }
427 public:
428 tcb(tcp& t, connid id);
429 void input_handle_listen_state(tcp_hdr* th, packet p);
430 void input_handle_syn_sent_state(tcp_hdr* th, packet p);
431 void input_handle_other_state(tcp_hdr* th, packet p);
432 void output_one(bool data_retransmit = false);
433 future<> wait_for_data();
434 future<> wait_input_shutdown();
435 void abort_reader() noexcept;
436 future<> wait_for_all_data_acked();
437 future<> wait_send_available();
438 future<> send(packet p);
439 void connect();
440 packet read();
441 void close() noexcept;
442 void remove_from_tcbs() {
443 auto id = connid{_local_ip, _foreign_ip, _local_port, _foreign_port};
444 _tcp._tcbs.erase(id);
445 }
446 std::optional<typename InetTraits::l4packet> get_packet();
447 void output() {
448 if (!_poll_active) {
449 _poll_active = true;
450 // FIXME: future is discarded
451 (void)_tcp.poll_tcb(_foreign_ip, this->shared_from_this()).then_wrapped([this] (auto&& f) {
452 try {
453 f.get();
454 } catch(arp_queue_full_error& ex) {
455 // retry later
456 _poll_active = false;
457 this->start_retransmit_timer();
458 } catch(arp_timeout_error& ex) {
459 if (this->in_state(SYN_SENT)) {
460 _connect_done.set_exception(ex);
461 this->cleanup();
462 }
463 // in other states connection should time out
464 }
465 });
466 }
467 }
468 future<> connect_done() {
469 return _connect_done.get_future();
470 }
471 tcp_state& state() {
472 return _state;
473 }
474 private:
475 void respond_with_reset(tcp_hdr* th);
476 bool merge_out_of_order();
477 void insert_out_of_order(tcp_seq seq, packet p);
478 void trim_receive_data_after_window();
479 bool should_send_ack(uint16_t seg_len);
480 void clear_delayed_ack() noexcept;
481 packet get_transmit_packet();
482 void retransmit_one() {
483 bool data_retransmit = true;
484 output_one(data_retransmit);
485 }
486 void start_retransmit_timer() {
487 auto now = clock_type::now();
488 start_retransmit_timer(now);
489 };
490 void start_retransmit_timer(clock_type::time_point now) {
491 auto tp = now + _rto;
492 _retransmit.rearm(tp);
493 };
494 void stop_retransmit_timer() noexcept {
495 _retransmit.cancel();
496 };
497 void start_persist_timer() {
498 auto now = clock_type::now();
499 start_persist_timer(now);
500 };
501 void start_persist_timer(clock_type::time_point now) {
502 auto tp = now + _persist_time_out;
503 _persist.rearm(tp);
504 };
505 void stop_persist_timer() {
506 _persist.cancel();
507 };
508 void persist();
509 void retransmit();
510 void fast_retransmit();
511 void update_rto(clock_type::time_point tx_time);
512 void update_cwnd(uint32_t acked_bytes);
513 void cleanup();
514 uint32_t can_send() {
515 if (_snd.window_probe) {
516 return 1;
517 }
518
519 // Can not send if send window is zero
520 if (_snd.window == 0) {
521 return 0;
522 }
523
524 // Can not send if send window is less than unacknowledged data size
525 auto window_used = uint32_t(_snd.next - _snd.unacknowledged);
526 if (window_used > _snd.window) {
527 return 0;
528 }
529
530 // Can not send more than advertised window allows or unsent data size
531 auto x = std::min(_snd.window - window_used, _snd.unsent_len);
532
533 // Can not send more than congestion window allows
534 x = std::min(_snd.cwnd, x);
535 if (_snd.dupacks == 1 || _snd.dupacks == 2) {
536 // RFC5681 Step 3.1
537 // Send cwnd + 2 * smss per RFC3042
538 auto flight = flight_size();
539 auto max = _snd.cwnd + 2 * _snd.mss;
540 x = flight <= max ? std::min(x, max - flight) : 0;
541 _snd.limited_transfer += x;
542 } else if (_snd.dupacks >= 3) {
543 // RFC5681 Step 3.5
544 // Sent 1 full-sized segment at most
545 x = std::min(uint32_t(_snd.mss), x);
546 }
547 return x;
548 }
549 uint32_t flight_size() {
550 uint32_t size = 0;
551 std::for_each(_snd.data.begin(), _snd.data.end(), [&] (unacked_segment& seg) { size += seg.p.len(); });
552 return size;
553 }
554 uint16_t local_mss() {
555 return _tcp.hw_features().mtu - net::tcp_hdr_len_min - InetTraits::ip_hdr_len_min;
556 }
557 void queue_packet(packet p) {
558 _packetq.emplace_back(typename InetTraits::l4packet{_foreign_ip, std::move(p)});
559 }
560 void signal_data_received() {
561 if (_rcv._data_received_promise) {
562 _rcv._data_received_promise->set_value();
563 _rcv._data_received_promise = {};
564 }
565 }
566 void signal_all_data_acked() {
567 if (_snd._all_data_acked_promise && _snd.unsent_len == 0) {
568 _snd._all_data_acked_promise->set_value();
569 _snd._all_data_acked_promise = {};
570 }
571 }
572 void signal_send_available() {
573 if (_snd._send_available_promise && _snd.max_queue_space > _snd.current_queue_space) {
574 _snd._send_available_promise->set_value();
575 _snd._send_available_promise = {};
576 }
577 }
578 void do_syn_sent() {
579 _state = SYN_SENT;
580 _snd.syn_tx_time = clock_type::now();
581 // Send <SYN> to remote
582 output();
583 }
584 void do_syn_received() {
585 _state = SYN_RECEIVED;
586 _snd.syn_tx_time = clock_type::now();
587 // Send <SYN,ACK> to remote
588 output();
589 }
590 void do_established() {
591 _state = ESTABLISHED;
592 update_rto(_snd.syn_tx_time);
593 _connect_done.set_value();
594 }
595 void do_reset() {
596 _state = CLOSED;
597 cleanup();
598 if (_rcv._data_received_promise) {
599 _rcv._data_received_promise->set_exception(tcp_reset_error());
600 _rcv._data_received_promise = std::nullopt;
601 }
602 if (_snd._all_data_acked_promise) {
603 _snd._all_data_acked_promise->set_exception(tcp_reset_error());
604 _snd._all_data_acked_promise = std::nullopt;
605 }
606 if (_snd._send_available_promise) {
607 _snd._send_available_promise->set_exception(tcp_reset_error());
608 _snd._send_available_promise = std::nullopt;
609 }
610 }
611 void do_time_wait() {
612 // FIXME: Implement TIME_WAIT state timer
613 _state = TIME_WAIT;
614 cleanup();
615 }
616 void do_closed() {
617 _state = CLOSED;
618 cleanup();
619 }
620 void do_setup_isn() {
621 _snd.initial = get_isn();
622 _snd.unacknowledged = _snd.initial;
623 _snd.next = _snd.initial + 1;
624 _snd.recover = _snd.initial;
625 }
626 void do_local_fin_acked() {
627 _snd.unacknowledged += 1;
628 _snd.next += 1;
629 }
630 bool syn_needs_on() const noexcept {
631 return in_state(SYN_SENT | SYN_RECEIVED);
632 }
633 bool fin_needs_on() const noexcept {
634 return in_state(FIN_WAIT_1 | CLOSING | LAST_ACK) && _snd.closed &&
635 _snd.unsent_len == 0;
636 }
637 bool ack_needs_on() const noexcept {
638 return !in_state(CLOSED | LISTEN | SYN_SENT);
639 }
640 bool foreign_will_not_send() const noexcept {
641 return in_state(CLOSING | TIME_WAIT | CLOSE_WAIT | LAST_ACK | CLOSED);
642 }
643 bool in_state(tcp_state state) const noexcept {
644 return uint16_t(_state) & uint16_t(state);
645 }
646 void exit_fast_recovery() {
647 _snd.dupacks = 0;
648 _snd.limited_transfer = 0;
649 _snd.partial_ack = 0;
650 }
651 uint32_t data_segment_acked(tcp_seq seg_ack);
652 bool segment_acceptable(tcp_seq seg_seq, unsigned seg_len);
653 void init_from_options(tcp_hdr* th, uint8_t* opt_start, uint8_t* opt_end);
654 friend class connection;
655 };
656 inet_type& _inet;
657 std::unordered_map<connid, lw_shared_ptr<tcb>, connid_hash> _tcbs;
658 std::unordered_map<uint16_t, listener*> _listening;
659 std::random_device _rd;
660 std::default_random_engine _e;
661 std::uniform_int_distribution<uint16_t> _port_dist{41952, 65535};
663 // queue for packets that do not belong to any tcb
665 semaphore _queue_space = {212992};
666 metrics::metric_groups _metrics;
667public:
668 const inet_type& inet() const {
669 return _inet;
670 }
673 public:
674 explicit connection(lw_shared_ptr<tcb> tcbp) : _tcb(std::move(tcbp)) { _tcb->_conn = this; }
675 connection(const connection&) = delete;
676 connection(connection&& x) noexcept : _tcb(std::move(x._tcb)) {
677 _tcb->_conn = this;
678 }
679 ~connection();
680 void operator=(const connection&) = delete;
681 connection& operator=(connection&& x) {
682 if (this != &x) {
683 this->~connection();
684 new (this) connection(std::move(x));
685 }
686 return *this;
687 }
688 future<> connected() {
689 return _tcb->connect_done();
690 }
691 future<> send(packet p) {
692 return _tcb->send(std::move(p));
693 }
694 future<> wait_for_data() {
695 return _tcb->wait_for_data();
696 }
697 future<> wait_input_shutdown() {
698 return _tcb->wait_input_shutdown();
699 }
700 packet read() {
701 return _tcb->read();
702 }
703 ipaddr foreign_ip() {
704 return _tcb->_foreign_ip;
705 }
706 uint16_t foreign_port() {
707 return _tcb->_foreign_port;
708 }
709 ipaddr local_ip() {
710 return _tcb->_local_ip;
711 }
712 uint16_t local_port() {
713 return _tcb->_local_port;
714 }
715 void shutdown_connect();
716 void close_read() noexcept;
717 void close_write() noexcept;
718 };
719 class listener {
720 tcp& _tcp;
721 uint16_t _port;
723 size_t _pending = 0;
724 private:
725 listener(tcp& t, uint16_t port, size_t queue_length)
726 : _tcp(t), _port(port), _q(queue_length) {
727 _tcp._listening.emplace(_port, this);
728 }
729 public:
730 listener(listener&& x)
731 : _tcp(x._tcp), _port(x._port), _q(std::move(x._q)) {
732 _tcp._listening[_port] = this;
733 x._port = 0;
734 }
735 ~listener() {
736 if (_port) {
737 _tcp._listening.erase(_port);
738 }
739 }
740 future<connection> accept() {
741 return _q.pop_eventually();
742 }
743 void abort_accept() {
744 _q.abort(std::make_exception_ptr(std::system_error(ECONNABORTED, std::system_category())));
745 }
746 bool full() { return _pending + _q.size() >= _q.max_size(); }
747 void inc_pending() { _pending++; }
748 void dec_pending() { _pending--; }
749
750 const tcp& get_tcp() const {
751 return _tcp;
752 }
753 uint16_t port() const {
754 return _port;
755 }
756 friend class tcp;
757 };
758public:
759 explicit tcp(inet_type& inet);
760 void received(packet p, ipaddr from, ipaddr to);
761 bool forward(forward_hash& out_hash_data, packet& p, size_t off);
762 listener listen(uint16_t port, size_t queue_length = 100);
764 const net::hw_features& hw_features() const { return _inet._inet.hw_features(); }
765 future<> poll_tcb(ipaddr to, lw_shared_ptr<tcb> tcb);
766 void add_connected_tcb(lw_shared_ptr<tcb> tcbp, uint16_t local_port) {
767 auto it = _listening.find(local_port);
768 if (it != _listening.end()) {
769 it->second->_q.push(connection(tcbp));
770 it->second->dec_pending();
771 }
772 }
773private:
774 void send_packet_without_tcb(ipaddr from, ipaddr to, packet p);
775 void respond_with_reset(tcp_hdr* rth, ipaddr local_ip, ipaddr foreign_ip);
776 friend class listener;
777};
778
779template <typename InetTraits>
780tcp<InetTraits>::tcp(inet_type& inet)
781 : _inet(inet)
782 , _e(_rd()) {
783 namespace sm = metrics;
784
785 _metrics.add_group("tcp", {
786 sm::make_counter("linearizations", [] { return tcp_packet_merger::linearizations(); },
787 sm::description("Counts a number of times a buffer linearization was invoked during the buffers merge process. "
788 "Divide it by a total TCP receive packet rate to get an everage number of lineraizations per TCP packet."))
789 });
790
791 _inet.register_packet_provider([this, tcb_polled = 0u] () mutable {
792 std::optional<typename InetTraits::l4packet> l4p;
793 auto c = _poll_tcbs.size();
794 if (!_packetq.empty() && (!(tcb_polled % 128) || c == 0)) {
795 l4p = std::move(_packetq.front());
796 _packetq.pop_front();
797 _queue_space.signal(l4p.value().p.len());
798 } else {
799 while (c--) {
800 tcb_polled++;
801 lw_shared_ptr<tcb> tcb;
802 ethernet_address dst;
803 std::tie(tcb, dst) = std::move(_poll_tcbs.front());
804 _poll_tcbs.pop_front();
805 l4p = tcb->get_packet();
806 if (l4p) {
807 l4p.value().e_dst = dst;
808 break;
809 }
810 }
811 }
812 return l4p;
813 });
814}
815
816template <typename InetTraits>
817future<> tcp<InetTraits>::poll_tcb(ipaddr to, lw_shared_ptr<tcb> tcb) {
818 return _inet.get_l2_dst_address(to).then([this, tcb = std::move(tcb)] (ethernet_address dst) {
819 _poll_tcbs.emplace_back(std::move(tcb), dst);
820 });
821}
822
823template <typename InetTraits>
824auto tcp<InetTraits>::listen(uint16_t port, size_t queue_length) -> listener {
825 return listener(*this, port, queue_length);
826}
827
828template <typename InetTraits>
829auto tcp<InetTraits>::connect(socket_address sa) -> connection {
830 connid id;
831 auto src_ip = _inet._inet.host_address();
832 auto dst_ip = ipv4_address(sa);
833 auto dst_port = net::ntoh(sa.u.in.sin_port);
834
835 if (smp::count > 1) {
836 do {
837 id = connid{src_ip, dst_ip, _port_dist(_e), dst_port};
838 } while (_inet._inet.netif()->hash2cpu(id.hash(_inet._inet.netif()->rss_key())) != this_shard_id()
839 || _tcbs.find(id) != _tcbs.end());
840 } else {
841 id = connid{src_ip, dst_ip, _port_dist(_e), dst_port};
842 }
843
844 auto tcbp = make_lw_shared<tcb>(*this, id);
845 _tcbs.insert({id, tcbp});
846 tcbp->connect();
847 return connection(tcbp);
848}
849
850template <typename InetTraits>
851bool tcp<InetTraits>::forward(forward_hash& out_hash_data, packet& p, size_t off) {
852 auto th = p.get_header(off, tcp_hdr::len);
853 if (th) {
854 // src_port, dst_port in network byte order
855 out_hash_data.push_back(uint8_t(th[0]));
856 out_hash_data.push_back(uint8_t(th[1]));
857 out_hash_data.push_back(uint8_t(th[2]));
858 out_hash_data.push_back(uint8_t(th[3]));
859 }
860 return true;
861}
862
863template <typename InetTraits>
864void tcp<InetTraits>::received(packet p, ipaddr from, ipaddr to) {
865 auto th = p.get_header(0, tcp_hdr::len);
866 if (!th) {
867 return;
868 }
869 // data_offset is correct even before ntoh()
870 auto data_offset = uint8_t(th[12]) >> 4;
871 if (size_t(data_offset * 4) < tcp_hdr::len) {
872 return;
873 }
874
875 if (!hw_features().rx_csum_offload) {
876 checksummer csum;
877 InetTraits::tcp_pseudo_header_checksum(csum, from, to, p.len());
878 csum.sum(p);
879 if (csum.get() != 0) {
880 return;
881 }
882 }
883 auto h = tcp_hdr::read(th);
884 auto id = connid{to, from, h.dst_port, h.src_port};
885 auto tcbi = _tcbs.find(id);
886 lw_shared_ptr<tcb> tcbp;
887 if (tcbi == _tcbs.end()) {
888 auto listener = _listening.find(id.local_port);
889 if (listener == _listening.end() || listener->second->full()) {
890 // 1) In CLOSE state
891 // 1.1 all data in the incoming segment is discarded. An incoming
892 // segment containing a RST is discarded. An incoming segment not
893 // containing a RST causes a RST to be sent in response.
894 // FIXME:
895 // if ACK off: <SEQ=0><ACK=SEG.SEQ+SEG.LEN><CTL=RST,ACK>
896 // if ACK on: <SEQ=SEG.ACK><CTL=RST>
897 return respond_with_reset(&h, id.local_ip, id.foreign_ip);
898 } else {
899 // 2) In LISTEN state
900 // 2.1 first check for an RST
901 if (h.f_rst) {
902 // An incoming RST should be ignored
903 return;
904 }
905 // 2.2 second check for an ACK
906 if (h.f_ack) {
907 // Any acknowledgment is bad if it arrives on a connection
908 // still in the LISTEN state.
909 // <SEQ=SEG.ACK><CTL=RST>
910 return respond_with_reset(&h, id.local_ip, id.foreign_ip);
911 }
912 // 2.3 third check for a SYN
913 if (h.f_syn) {
914 // check the security
915 // NOTE: Ignored for now
916 tcbp = make_lw_shared<tcb>(*this, id);
917 _tcbs.insert({id, tcbp});
918 // TODO: we need to remove the tcb and decrease the pending if
919 // it stays SYN_RECEIVED state forever.
920 listener->second->inc_pending();
921
922 return tcbp->input_handle_listen_state(&h, std::move(p));
923 }
924 // 2.4 fourth other text or control
925 // So you are unlikely to get here, but if you do, drop the
926 // segment, and return.
927 return;
928 }
929 } else {
930 tcbp = tcbi->second;
931 if (tcbp->state() == tcp_state::SYN_SENT) {
932 // 3) In SYN_SENT State
933 return tcbp->input_handle_syn_sent_state(&h, std::move(p));
934 } else {
935 // 4) In other state, can be one of the following:
936 // SYN_RECEIVED, ESTABLISHED, FIN_WAIT_1, FIN_WAIT_2
937 // CLOSE_WAIT, CLOSING, LAST_ACK, TIME_WAIT
938 return tcbp->input_handle_other_state(&h, std::move(p));
939 }
940 }
941}
942
943// Send packet does not belong to any tcb
944template <typename InetTraits>
945void tcp<InetTraits>::send_packet_without_tcb(ipaddr from, ipaddr to, packet p) {
946 if (_queue_space.try_wait(p.len())) { // drop packets that do not fit the queue
947 // FIXME: future is discarded
948 (void)_inet.get_l2_dst_address(to).then([this, to, p = std::move(p)] (ethernet_address e_dst) mutable {
949 _packetq.emplace_back(ipv4_traits::l4packet{to, std::move(p), e_dst, ip_protocol_num::tcp});
950 });
951 }
952}
953
954template <typename InetTraits>
955tcp<InetTraits>::connection::~connection() {
956 if (_tcb) {
957 _tcb->_conn = nullptr;
958 close_read();
959 close_write();
960 }
961}
962
963template <typename InetTraits>
964tcp<InetTraits>::tcb::tcb(tcp& t, connid id)
965 : _tcp(t)
966 , _local_ip(id.local_ip)
967 , _foreign_ip(id.foreign_ip)
968 , _local_port(id.local_port)
969 , _foreign_port(id.foreign_port)
970 , _delayed_ack([this] { _nr_full_seg_received = 0; output(); })
971 , _retransmit([this] { retransmit(); })
972 , _persist([this] { persist(); }) {
973}
974
975template <typename InetTraits>
976void tcp<InetTraits>::tcb::respond_with_reset(tcp_hdr* rth) {
977 _tcp.respond_with_reset(rth, _local_ip, _foreign_ip);
978}
979
980template <typename InetTraits>
981void tcp<InetTraits>::respond_with_reset(tcp_hdr* rth, ipaddr local_ip, ipaddr foreign_ip) {
982 if (rth->f_rst) {
983 return;
984 }
985 packet p;
986 auto th = p.prepend_uninitialized_header(tcp_hdr::len);
987 auto h = tcp_hdr{};
988 h.src_port = rth->dst_port;
989 h.dst_port = rth->src_port;
990 if (rth->f_ack) {
991 h.seq = rth->ack;
992 }
993 // If this RST packet is in response to a SYN packet. We ACK the ISN.
994 if (rth->f_syn) {
995 h.ack = rth->seq + 1;
996 h.f_ack = true;
997 }
998 h.f_rst = true;
999 h.data_offset = tcp_hdr::len / 4;
1000 h.checksum = 0;
1001 h.write(th);
1002
1003 checksummer csum;
1004 offload_info oi;
1005 InetTraits::tcp_pseudo_header_checksum(csum, local_ip, foreign_ip, tcp_hdr::len);
1006 uint16_t checksum;
1007 if (hw_features().tx_csum_l4_offload) {
1008 checksum = ~csum.get();
1009 oi.needs_csum = true;
1010 } else {
1011 csum.sum(p);
1012 checksum = csum.get();
1013 oi.needs_csum = false;
1014 }
1015 tcp_hdr::write_nbo_checksum(th, checksum);
1016
1017 oi.protocol = ip_protocol_num::tcp;
1018 oi.tcp_hdr_len = tcp_hdr::len;
1019 p.set_offload_info(oi);
1020
1021 send_packet_without_tcb(local_ip, foreign_ip, std::move(p));
1022}
1023
1024template <typename InetTraits>
1025uint32_t tcp<InetTraits>::tcb::data_segment_acked(tcp_seq seg_ack) {
1026 uint32_t total_acked_bytes = 0;
1027 // Full ACK of segment
1028 while (!_snd.data.empty()
1029 && (_snd.unacknowledged + _snd.data.front().p.len() <= seg_ack)) {
1030 auto acked_bytes = _snd.data.front().p.len();
1031 _snd.unacknowledged += acked_bytes;
1032 // Ignore retransmitted segments when setting the RTO
1033 if (_snd.data.front().nr_transmits == 0) {
1034 update_rto(_snd.data.front().tx_time);
1035 }
1036 update_cwnd(acked_bytes);
1037 total_acked_bytes += acked_bytes;
1038 _snd.current_queue_space -= _snd.data.front().data_len;
1039 signal_send_available();
1040 _snd.data.pop_front();
1041 }
1042 // Partial ACK of segment
1043 if (_snd.unacknowledged < seg_ack) {
1044 auto acked_bytes = seg_ack - _snd.unacknowledged;
1045 if (!_snd.data.empty()) {
1046 auto& unacked_seg = _snd.data.front();
1047 unacked_seg.p.trim_front(acked_bytes);
1048 }
1049 _snd.unacknowledged = seg_ack;
1050 update_cwnd(acked_bytes);
1051 total_acked_bytes += acked_bytes;
1052 }
1053 return total_acked_bytes;
1054}
1055
1056template <typename InetTraits>
1057bool tcp<InetTraits>::tcb::segment_acceptable(tcp_seq seg_seq, unsigned seg_len) {
1058 if (seg_len == 0 && _rcv.window == 0) {
1059 // SEG.SEQ = RCV.NXT
1060 return seg_seq == _rcv.next;
1061 } else if (seg_len == 0 && _rcv.window > 0) {
1062 // RCV.NXT =< SEG.SEQ < RCV.NXT+RCV.WND
1063 return (_rcv.next <= seg_seq) && (seg_seq < _rcv.next + _rcv.window);
1064 } else if (seg_len > 0 && _rcv.window > 0) {
1065 // RCV.NXT =< SEG.SEQ < RCV.NXT+RCV.WND
1066 // or
1067 // RCV.NXT =< SEG.SEQ+SEG.LEN-1 < RCV.NXT+RCV.WND
1068 bool x = (_rcv.next <= seg_seq) && seg_seq < (_rcv.next + _rcv.window);
1069 bool y = (_rcv.next <= seg_seq + seg_len - 1) && (seg_seq + seg_len - 1 < _rcv.next + _rcv.window);
1070 return x || y;
1071 } else {
1072 // SEG.LEN > 0 RCV.WND = 0, not acceptable
1073 return false;
1074 }
1075}
1076
1077template <typename InetTraits>
1078void tcp<InetTraits>::tcb::init_from_options(tcp_hdr* th, uint8_t* opt_start, uint8_t* opt_end) {
1079 // Handle tcp options
1080 _option.parse(opt_start, opt_end);
1081
1082 // Remote receive window scale factor
1083 _snd.window_scale = _option._remote_win_scale;
1084 // Local receive window scale factor
1085 _rcv.window_scale = _option._local_win_scale;
1086
1087 // Maximum segment size remote can receive
1088 _snd.mss = _option._remote_mss;
1089 // Maximum segment size local can receive
1090 _rcv.mss = _option._local_mss = local_mss();
1091
1092 _rcv.window = get_default_receive_window_size();
1093 _snd.window = th->window << _snd.window_scale;
1094
1095 // Segment sequence number used for last window update
1096 _snd.wl1 = th->seq;
1097 // Segment acknowledgment number used for last window update
1098 _snd.wl2 = th->ack;
1099
1100 // Setup initial congestion window
1101 if (2190 < _snd.mss) {
1102 _snd.cwnd = 2 * _snd.mss;
1103 } else if (1095 < _snd.mss && _snd.mss <= 2190) {
1104 _snd.cwnd = 3 * _snd.mss;
1105 } else {
1106 _snd.cwnd = 4 * _snd.mss;
1107 }
1108
1109 // Setup initial slow start threshold
1110 _snd.ssthresh = th->window << _snd.window_scale;
1111}
1112
1113template <typename InetTraits>
1114void tcp<InetTraits>::tcb::input_handle_listen_state(tcp_hdr* th, packet p) {
1115 auto opt_len = th->data_offset * 4 - tcp_hdr::len;
1116 auto opt_start = reinterpret_cast<uint8_t*>(p.get_header(0, th->data_offset * 4)) + tcp_hdr::len;
1117 auto opt_end = opt_start + opt_len;
1118 p.trim_front(th->data_offset * 4);
1119 tcp_seq seg_seq = th->seq;
1120
1121 // Set RCV.NXT to SEG.SEQ+1, IRS is set to SEG.SEQ
1122 _rcv.next = seg_seq + 1;
1123 _rcv.initial = seg_seq;
1124
1125 // ISS should be selected and a SYN segment sent of the form:
1126 // <SEQ=ISS><ACK=RCV.NXT><CTL=SYN,ACK>
1127 // SND.NXT is set to ISS+1 and SND.UNA to ISS
1128 // NOTE: In previous code, _snd.next is set to ISS + 1 only when SYN is
1129 // ACKed. Now, we set _snd.next to ISS + 1 here, so in output_one(): we
1130 // have
1131 // th->seq = syn_on ? _snd.initial : _snd.next
1132 // to make sure retransmitted SYN has correct SEQ number.
1133 do_setup_isn();
1134
1135 _rcv.urgent = _rcv.next;
1136
1137 tcp_debug("listen: LISTEN -> SYN_RECEIVED\n");
1138 init_from_options(th, opt_start, opt_end);
1139 do_syn_received();
1140}
1141
1142template <typename InetTraits>
1143void tcp<InetTraits>::tcb::input_handle_syn_sent_state(tcp_hdr* th, packet p) {
1144 auto opt_len = th->data_offset * 4 - tcp_hdr::len;
1145 auto opt_start = reinterpret_cast<uint8_t*>(p.get_header(0, th->data_offset * 4)) + tcp_hdr::len;
1146 auto opt_end = opt_start + opt_len;
1147 p.trim_front(th->data_offset * 4);
1148 tcp_seq seg_seq = th->seq;
1149 auto seg_ack = th->ack;
1150
1151 bool acceptable = false;
1152 // 3.1 first check the ACK bit
1153 if (th->f_ack) {
1154 // If SEG.ACK =< ISS, or SEG.ACK > SND.NXT, send a reset (unless the
1155 // RST bit is set, if so drop the segment and return)
1156 if (seg_ack <= _snd.initial || seg_ack > _snd.next) {
1157 return respond_with_reset(th);
1158 }
1159
1160 // If SND.UNA =< SEG.ACK =< SND.NXT then the ACK is acceptable.
1161 acceptable = _snd.unacknowledged <= seg_ack && seg_ack <= _snd.next;
1162 }
1163
1164 // 3.2 second check the RST bit
1165 if (th->f_rst) {
1166 // If the ACK was acceptable then signal the user "error: connection
1167 // reset", drop the segment, enter CLOSED state, delete TCB, and
1168 // return. Otherwise (no ACK) drop the segment and return.
1169 if (acceptable) {
1170 _connect_done.set_exception(tcp_refused_error());
1171 return do_reset();
1172 } else {
1173 return;
1174 }
1175 }
1176
1177 // 3.3 third check the security and precedence
1178 // NOTE: Ignored for now
1179
1180 // 3.4 fourth check the SYN bit
1181 if (th->f_syn) {
1182 // RCV.NXT is set to SEG.SEQ+1, IRS is set to SEG.SEQ. SND.UNA should
1183 // be advanced to equal SEG.ACK (if there is an ACK), and any segments
1184 // on the retransmission queue which are thereby acknowledged should be
1185 // removed.
1186 _rcv.next = seg_seq + 1;
1187 _rcv.initial = seg_seq;
1188 if (th->f_ack) {
1189 // TODO: clean retransmission queue
1190 _snd.unacknowledged = seg_ack;
1191 }
1192 if (_snd.unacknowledged > _snd.initial) {
1193 // If SND.UNA > ISS (our SYN has been ACKed), change the connection
1194 // state to ESTABLISHED, form an ACK segment
1195 // <SEQ=SND.NXT><ACK=RCV.NXT><CTL=ACK>
1196 tcp_debug("syn: SYN_SENT -> ESTABLISHED\n");
1197 init_from_options(th, opt_start, opt_end);
1198 do_established();
1199 output();
1200 } else {
1201 // Otherwise enter SYN_RECEIVED, form a SYN,ACK segment
1202 // <SEQ=ISS><ACK=RCV.NXT><CTL=SYN,ACK>
1203 tcp_debug("syn: SYN_SENT -> SYN_RECEIVED\n");
1204 do_syn_received();
1205 }
1206 }
1207
1208 // 3.5 fifth, if neither of the SYN or RST bits is set then drop the
1209 // segment and return.
1210 return;
1211}
1212
1213template <typename InetTraits>
1214void tcp<InetTraits>::tcb::input_handle_other_state(tcp_hdr* th, packet p) {
1215 p.trim_front(th->data_offset * 4);
1216 bool do_output = false;
1217 bool do_output_data = false;
1218 tcp_seq seg_seq = th->seq;
1219 auto seg_ack = th->ack;
1220 auto seg_len = p.len();
1221
1222 // 4.1 first check sequence number
1223 if (!segment_acceptable(seg_seq, seg_len)) {
1224 //<SEQ=SND.NXT><ACK=RCV.NXT><CTL=ACK>
1225 return output();
1226 }
1227
1228 // In the following it is assumed that the segment is the idealized
1229 // segment that begins at RCV.NXT and does not exceed the window.
1230 if (seg_seq < _rcv.next) {
1231 // ignore already acknowledged data
1232 auto dup = std::min(uint32_t(_rcv.next - seg_seq), seg_len);
1233 p.trim_front(dup);
1234 seg_len -= dup;
1235 seg_seq += dup;
1236 }
1237 // FIXME: We should trim data outside the right edge of the receive window as well
1238
1239 if (seg_seq != _rcv.next) {
1240 insert_out_of_order(seg_seq, std::move(p));
1241 // A TCP receiver SHOULD send an immediate duplicate ACK
1242 // when an out-of-order segment arrives.
1243 return output();
1244 }
1245
1246 // 4.2 second check the RST bit
1247 if (th->f_rst) {
1248 if (in_state(SYN_RECEIVED)) {
1249 // If this connection was initiated with a passive OPEN (i.e.,
1250 // came from the LISTEN state), then return this connection to
1251 // LISTEN state and return. The user need not be informed. If
1252 // this connection was initiated with an active OPEN (i.e., came
1253 // from SYN_SENT state) then the connection was refused, signal
1254 // the user "connection refused". In either case, all segments
1255 // on the retransmission queue should be removed. And in the
1256 // active OPEN case, enter the CLOSED state and delete the TCB,
1257 // and return.
1258 _connect_done.set_exception(tcp_refused_error());
1259 return do_reset();
1260 }
1261 if (in_state(ESTABLISHED | FIN_WAIT_1 | FIN_WAIT_2 | CLOSE_WAIT)) {
1262 // If the RST bit is set then, any outstanding RECEIVEs and SEND
1263 // should receive "reset" responses. All segment queues should be
1264 // flushed. Users should also receive an unsolicited general
1265 // "connection reset" signal. Enter the CLOSED state, delete the
1266 // TCB, and return.
1267 return do_reset();
1268 }
1269 if (in_state(CLOSING | LAST_ACK | TIME_WAIT)) {
1270 // If the RST bit is set then, enter the CLOSED state, delete the
1271 // TCB, and return.
1272 return do_closed();
1273 }
1274 }
1275
1276 // 4.3 third check security and precedence
1277 // NOTE: Ignored for now
1278
1279 // 4.4 fourth, check the SYN bit
1280 if (th->f_syn) {
1281 // SYN_RECEIVED, ESTABLISHED, FIN_WAIT_1, FIN_WAIT_2
1282 // CLOSE_WAIT, CLOSING, LAST_ACK, TIME_WAIT
1283
1284 // If the SYN is in the window it is an error, send a reset, any
1285 // outstanding RECEIVEs and SEND should receive "reset" responses,
1286 // all segment queues should be flushed, the user should also
1287 // receive an unsolicited general "connection reset" signal, enter
1288 // the CLOSED state, delete the TCB, and return.
1289 respond_with_reset(th);
1290 return do_reset();
1291
1292 // If the SYN is not in the window this step would not be reached
1293 // and an ack would have been sent in the first step (sequence
1294 // number check).
1295 }
1296
1297 // 4.5 fifth check the ACK field
1298 if (!th->f_ack) {
1299 // if the ACK bit is off drop the segment and return
1300 return;
1301 } else {
1302 // SYN_RECEIVED STATE
1303 if (in_state(SYN_RECEIVED)) {
1304 // If SND.UNA =< SEG.ACK =< SND.NXT then enter ESTABLISHED state
1305 // and continue processing.
1306 if (_snd.unacknowledged <= seg_ack && seg_ack <= _snd.next) {
1307 tcp_debug("SYN_RECEIVED -> ESTABLISHED\n");
1308 do_established();
1309 _tcp.add_connected_tcb(this->shared_from_this(), _local_port);
1310 } else {
1311 // <SEQ=SEG.ACK><CTL=RST>
1312 return respond_with_reset(th);
1313 }
1314 }
1315 auto update_window = [this, th, seg_seq, seg_ack] {
1316 tcp_debug("window update seg_seq=%d, seg_ack=%d, old window=%d new window=%d\n",
1317 seg_seq, seg_ack, _snd.window, th->window << _snd.window_scale);
1318 _snd.window = th->window << _snd.window_scale;
1319 _snd.wl1 = seg_seq;
1320 _snd.wl2 = seg_ack;
1321 _snd.zero_window_probing_out = 0;
1322 if (_snd.window == 0) {
1323 _persist_time_out = _rto;
1324 start_persist_timer();
1325 } else {
1326 stop_persist_timer();
1327 }
1328 };
1329 // ESTABLISHED STATE or
1330 // CLOSE_WAIT STATE: Do the same processing as for the ESTABLISHED state.
1331 if (in_state(ESTABLISHED | CLOSE_WAIT)){
1332 // When we are in zero window probing phase and packets_out = 0 we bypass "duplicated ack" check
1333 auto packets_out = _snd.next - _snd.unacknowledged - _snd.zero_window_probing_out;
1334 // If SND.UNA < SEG.ACK =< SND.NXT then, set SND.UNA <- SEG.ACK.
1335 if (_snd.unacknowledged < seg_ack && seg_ack <= _snd.next) {
1336 // Remote ACKed data we sent
1337 auto acked_bytes = data_segment_acked(seg_ack);
1338
1339 // If SND.UNA < SEG.ACK =< SND.NXT, the send window should be updated.
1340 if (_snd.wl1 < seg_seq || (_snd.wl1 == seg_seq && _snd.wl2 <= seg_ack)) {
1341 update_window();
1342 }
1343
1344 // some data is acked, try send more data
1345 do_output_data = true;
1346
1347 auto set_retransmit_timer = [this] {
1348 if (_snd.data.empty()) {
1349 // All outstanding segments are acked, turn off the timer.
1350 stop_retransmit_timer();
1351 // Signal the waiter of this event
1352 signal_all_data_acked();
1353 } else {
1354 // Restart the timer becasue new data is acked.
1355 start_retransmit_timer();
1356 }
1357 };
1358
1359 if (_snd.dupacks >= 3) {
1360 // We are in fast retransmit / fast recovery phase
1361 uint32_t smss = _snd.mss;
1362 if (seg_ack > _snd.recover) {
1363 tcp_debug("ack: full_ack\n");
1364 // Set cwnd to min (ssthresh, max(FlightSize, SMSS) + SMSS)
1365 _snd.cwnd = std::min(_snd.ssthresh, std::max(flight_size(), smss) + smss);
1366 // Exit the fast recovery procedure
1367 exit_fast_recovery();
1368 set_retransmit_timer();
1369 } else {
1370 tcp_debug("ack: partial_ack\n");
1371 // Retransmit the first unacknowledged segment
1372 fast_retransmit();
1373 // Deflate the congestion window by the amount of new data
1374 // acknowledged by the Cumulative Acknowledgment field
1375 _snd.cwnd -= acked_bytes;
1376 // If the partial ACK acknowledges at least one SMSS of new
1377 // data, then add back SMSS bytes to the congestion window
1378 if (acked_bytes >= smss) {
1379 _snd.cwnd += smss;
1380 }
1381 // Send a new segment if permitted by the new value of
1382 // cwnd. Do not exit the fast recovery procedure For
1383 // the first partial ACK that arrives during fast
1384 // recovery, also reset the retransmit timer.
1385 if (++_snd.partial_ack == 1) {
1386 start_retransmit_timer();
1387 }
1388 }
1389 } else {
1390 // RFC5681: The fast retransmit algorithm uses the arrival
1391 // of 3 duplicate ACKs (as defined in section 2, without
1392 // any intervening ACKs which move SND.UNA) as an
1393 // indication that a segment has been lost.
1394 //
1395 // So, here we reset dupacks to zero becasue this ACK moves
1396 // SND.UNA.
1397 exit_fast_recovery();
1398 set_retransmit_timer();
1399 }
1400 } else if ((packets_out > 0) && !_snd.data.empty() && seg_len == 0 &&
1401 th->f_fin == 0 && th->f_syn == 0 &&
1402 th->ack == _snd.unacknowledged &&
1403 uint32_t(th->window << _snd.window_scale) == _snd.window) {
1404 // Note:
1405 // RFC793 states:
1406 // If the ACK is a duplicate (SEG.ACK < SND.UNA), it can be ignored
1407 // RFC5681 states:
1408 // The TCP sender SHOULD use the "fast retransmit" algorithm to detect
1409 // and repair loss, based on incoming duplicate ACKs.
1410 // Here, We follow RFC5681.
1411 _snd.dupacks++;
1412 uint32_t smss = _snd.mss;
1413 // 3 duplicated ACKs trigger a fast retransmit
1414 if (_snd.dupacks == 1 || _snd.dupacks == 2) {
1415 // RFC5681 Step 3.1
1416 // Send cwnd + 2 * smss per RFC3042
1417 do_output_data = true;
1418 } else if (_snd.dupacks == 3) {
1419 // RFC6582 Step 3.2
1420 if (seg_ack - 1 > _snd.recover) {
1421 _snd.recover = _snd.next - 1;
1422 // RFC5681 Step 3.2
1423 _snd.ssthresh = std::max((flight_size() - _snd.limited_transfer) / 2, 2 * smss);
1424 fast_retransmit();
1425 } else {
1426 // Do not enter fast retransmit and do not reset ssthresh
1427 }
1428 // RFC5681 Step 3.3
1429 _snd.cwnd = _snd.ssthresh + 3 * smss;
1430 } else if (_snd.dupacks > 3) {
1431 // RFC5681 Step 3.4
1432 _snd.cwnd += smss;
1433 // RFC5681 Step 3.5
1434 do_output_data = true;
1435 }
1436 } else if (seg_ack > _snd.next) {
1437 // If the ACK acks something not yet sent (SEG.ACK > SND.NXT)
1438 // then send an ACK, drop the segment, and return
1439 return output();
1440 } else if (_snd.window == 0 && th->window > 0) {
1441 update_window();
1442 do_output_data = true;
1443 }
1444 }
1445 // FIN_WAIT_1 STATE
1446 if (in_state(FIN_WAIT_1)) {
1447 // In addition to the processing for the ESTABLISHED state, if
1448 // our FIN is now acknowledged then enter FIN-WAIT-2 and continue
1449 // processing in that state.
1450 if (seg_ack == _snd.next + 1) {
1451 tcp_debug("ack: FIN_WAIT_1 -> FIN_WAIT_2\n");
1452 _state = FIN_WAIT_2;
1453 do_local_fin_acked();
1454 }
1455 }
1456 // FIN_WAIT_2 STATE
1457 if (in_state(FIN_WAIT_2)) {
1458 // In addition to the processing for the ESTABLISHED state, if
1459 // the retransmission queue is empty, the user’s CLOSE can be
1460 // acknowledged ("ok") but do not delete the TCB.
1461 // TODO
1462 }
1463 // CLOSING STATE
1464 if (in_state(CLOSING)) {
1465 if (seg_ack == _snd.next + 1) {
1466 tcp_debug("ack: CLOSING -> TIME_WAIT\n");
1467 do_local_fin_acked();
1468 return do_time_wait();
1469 } else {
1470 return;
1471 }
1472 }
1473 // LAST_ACK STATE
1474 if (in_state(LAST_ACK)) {
1475 if (seg_ack == _snd.next + 1) {
1476 tcp_debug("ack: LAST_ACK -> CLOSED\n");
1477 do_local_fin_acked();
1478 return do_closed();
1479 }
1480 }
1481 // TIME_WAIT STATE
1482 if (in_state(TIME_WAIT)) {
1483 // The only thing that can arrive in this state is a
1484 // retransmission of the remote FIN. Acknowledge it, and restart
1485 // the 2 MSL timeout.
1486 // TODO
1487 }
1488 }
1489
1490 // 4.6 sixth, check the URG bit
1491 if (th->f_urg) {
1492 // TODO
1493 }
1494
1495 // 4.7 seventh, process the segment text
1496 if (in_state(ESTABLISHED | FIN_WAIT_1 | FIN_WAIT_2)) {
1497 if (p.len()) {
1498 // Once the TCP takes responsibility for the data it advances
1499 // RCV.NXT over the data accepted, and adjusts RCV.WND as
1500 // apporopriate to the current buffer availability. The total of
1501 // RCV.NXT and RCV.WND should not be reduced.
1502 _rcv.data_size += p.len();
1503 _rcv.data.push_back(std::move(p));
1504 _rcv.next += seg_len;
1505 auto merged = merge_out_of_order();
1506 _rcv.window = get_modified_receive_window_size();
1507 signal_data_received();
1508 // Send an acknowledgment of the form:
1509 // <SEQ=SND.NXT><ACK=RCV.NXT><CTL=ACK>
1510 // This acknowledgment should be piggybacked on a segment being
1511 // transmitted if possible without incurring undue delay.
1512 if (merged) {
1513 // TCP receiver SHOULD send an immediate ACK when the
1514 // incoming segment fills in all or part of a gap in the
1515 // sequence space.
1516 do_output = true;
1517 } else {
1518 do_output = should_send_ack(seg_len);
1519 }
1520 }
1521 } else if (in_state(CLOSE_WAIT | CLOSING | LAST_ACK | TIME_WAIT)) {
1522 // This should not occur, since a FIN has been received from the
1523 // remote side. Ignore the segment text.
1524 return;
1525 }
1526
1527 // 4.8 eighth, check the FIN bit
1528 if (th->f_fin) {
1529 if (_fin_recvd_promise) {
1530 _fin_recvd_promise->set_value();
1531 _fin_recvd_promise.reset();
1532 }
1533 if (in_state(CLOSED | LISTEN | SYN_SENT)) {
1534 // Do not process the FIN if the state is CLOSED, LISTEN or SYN-SENT
1535 // since the SEG.SEQ cannot be validated; drop the segment and return.
1536 return;
1537 }
1538 auto fin_seq = seg_seq + seg_len;
1539 if (fin_seq == _rcv.next) {
1540 _rcv.next = fin_seq + 1;
1541 signal_data_received();
1542
1543 // If this <FIN> packet contains data as well, we can ACK both data
1544 // and <FIN> in a single packet, so canncel the previous ACK.
1545 clear_delayed_ack();
1546 do_output = false;
1547 // Send ACK for the FIN!
1548 output();
1549
1550 if (in_state(SYN_RECEIVED | ESTABLISHED)) {
1551 tcp_debug("fin: SYN_RECEIVED or ESTABLISHED -> CLOSE_WAIT\n");
1552 _state = CLOSE_WAIT;
1553 }
1554 if (in_state(FIN_WAIT_1)) {
1555 // If our FIN has been ACKed (perhaps in this segment), then
1556 // enter TIME-WAIT, start the time-wait timer, turn off the other
1557 // timers; otherwise enter the CLOSING state.
1558 // Note: If our FIN has been ACKed, we should be in FIN_WAIT_2
1559 // not FIN_WAIT_1 if we reach here.
1560 tcp_debug("fin: FIN_WAIT_1 -> CLOSING\n");
1561 _state = CLOSING;
1562 }
1563 if (in_state(FIN_WAIT_2)) {
1564 tcp_debug("fin: FIN_WAIT_2 -> TIME_WAIT\n");
1565 return do_time_wait();
1566 }
1567 }
1568 }
1569 if (do_output || (do_output_data && can_send())) {
1570 // Since we will do output, we can canncel scheduled delayed ACK.
1571 clear_delayed_ack();
1572 output();
1573 }
1574}
1575
1576template <typename InetTraits>
1577packet tcp<InetTraits>::tcb::get_transmit_packet() {
1578 // easy case: empty queue
1579 if (_snd.unsent.empty()) {
1580 return packet();
1581 }
1582 auto can_send = this->can_send();
1583 // Max number of TCP payloads we can pass to NIC
1584 uint32_t len;
1585 if (_tcp.hw_features().tx_tso) {
1586 // FIXME: Info tap device the size of the splitted packet
1587 len = _tcp.hw_features().max_packet_len - net::tcp_hdr_len_min - InetTraits::ip_hdr_len_min;
1588 } else {
1589 len = std::min(uint16_t(_tcp.hw_features().mtu - net::tcp_hdr_len_min - InetTraits::ip_hdr_len_min), _snd.mss);
1590 }
1591 can_send = std::min(can_send, len);
1592 // easy case: one small packet
1593 if (_snd.unsent.size() == 1 && _snd.unsent.front().len() <= can_send) {
1594 auto p = std::move(_snd.unsent.front());
1595 _snd.unsent.pop_front();
1596 _snd.unsent_len -= p.len();
1597 return p;
1598 }
1599 // moderate case: need to split one packet
1600 if (_snd.unsent.front().len() > can_send) {
1601 auto p = _snd.unsent.front().share(0, can_send);
1602 _snd.unsent.front().trim_front(can_send);
1603 _snd.unsent_len -= p.len();
1604 return p;
1605 }
1606 // hard case: merge some packets, possibly split last
1607 auto p = std::move(_snd.unsent.front());
1608 _snd.unsent.pop_front();
1609 can_send -= p.len();
1610 while (!_snd.unsent.empty()
1611 && _snd.unsent.front().len() <= can_send) {
1612 can_send -= _snd.unsent.front().len();
1613 p.append(std::move(_snd.unsent.front()));
1614 _snd.unsent.pop_front();
1615 }
1616 if (!_snd.unsent.empty() && can_send) {
1617 auto& q = _snd.unsent.front();
1618 p.append(q.share(0, can_send));
1619 q.trim_front(can_send);
1620 }
1621 _snd.unsent_len -= p.len();
1622 return p;
1623}
1624
1625template <typename InetTraits>
1626void tcp<InetTraits>::tcb::output_one(bool data_retransmit) {
1627 if (in_state(CLOSED)) {
1628 return;
1629 }
1630
1631 packet p = data_retransmit ? _snd.data.front().p.share() : get_transmit_packet();
1632 packet clone = p.share(); // early clone to prevent share() from calling packet::unuse_internal_data() on header.
1633 uint16_t len = p.len();
1634 bool syn_on = syn_needs_on();
1635 bool ack_on = ack_needs_on();
1636
1637 auto options_size = _option.get_size(syn_on, ack_on);
1638 auto th = p.prepend_uninitialized_header(tcp_hdr::len + options_size);
1639 auto h = tcp_hdr{};
1640
1641 h.src_port = _local_port;
1642 h.dst_port = _foreign_port;
1643
1644 h.f_syn = syn_on;
1645 h.f_ack = ack_on;
1646 if (ack_on) {
1647 clear_delayed_ack();
1648 }
1649 h.f_urg = false;
1650 h.f_psh = false;
1651
1652 tcp_seq seq;
1653 if (data_retransmit) {
1654 seq = _snd.unacknowledged;
1655 } else {
1656 seq = syn_on ? _snd.initial : _snd.next;
1657 _snd.next += len;
1658 }
1659 h.seq = seq;
1660 h.ack = _rcv.next;
1661 h.data_offset = (tcp_hdr::len + options_size) / 4;
1662 h.window = _rcv.window >> _rcv.window_scale;
1663 h.checksum = 0;
1664
1665 // FIXME: does the FIN have to fit in the window?
1666 bool fin_on = fin_needs_on();
1667 h.f_fin = fin_on;
1668
1669 // Add tcp options
1670 _option.fill(th, &h, options_size);
1671 h.write(th);
1672
1673 offload_info oi;
1674 checksummer csum;
1675 uint16_t pseudo_hdr_seg_len = 0;
1676
1677 oi.tcp_hdr_len = tcp_hdr::len + options_size;
1678
1679 if (_tcp.hw_features().tx_csum_l4_offload) {
1680 oi.needs_csum = true;
1681
1682 //
1683 // tx checksum offloading: both virtio-net's VIRTIO_NET_F_CSUM dpdk's
1684 // PKT_TX_TCP_CKSUM - requires th->checksum to be initialized to ones'
1685 // complement sum of the pseudo header.
1686 //
1687 // For TSO the csum should be calculated for a pseudo header with
1688 // segment length set to 0. All the rest is the same as for a TCP Tx
1689 // CSUM offload case.
1690 //
1691 if (_tcp.hw_features().tx_tso && len > _snd.mss) {
1692 oi.tso_seg_size = _snd.mss;
1693 } else {
1694 pseudo_hdr_seg_len = tcp_hdr::len + options_size + len;
1695 }
1696 } else {
1697 pseudo_hdr_seg_len = tcp_hdr::len + options_size + len;
1698 oi.needs_csum = false;
1699 }
1700
1701 InetTraits::tcp_pseudo_header_checksum(csum, _local_ip, _foreign_ip,
1702 pseudo_hdr_seg_len);
1703
1704 uint16_t checksum;
1705 if (_tcp.hw_features().tx_csum_l4_offload) {
1706 checksum = ~csum.get();
1707 } else {
1708 csum.sum(p);
1709 checksum = csum.get();
1710 }
1711 tcp_hdr::write_nbo_checksum(th, checksum);
1712
1713 oi.protocol = ip_protocol_num::tcp;
1714
1715 p.set_offload_info(oi);
1716
1717 if (!data_retransmit && (len || syn_on || fin_on)) {
1718 auto now = clock_type::now();
1719 if (len) {
1720 unsigned nr_transmits = 0;
1721 _snd.data.emplace_back(unacked_segment{std::move(clone),
1722 len, nr_transmits, now});
1723 }
1724 if (!_retransmit.armed()) {
1725 start_retransmit_timer(now);
1726 }
1727 }
1728
1729
1730 // if advertised TCP receive window is 0 we may only transmit zero window probing segment.
1731 // Payload size of this segment is 1. Queueing anything bigger when _snd.window == 0 is bug
1732 // and violation of RFC
1733 assert((_snd.window > 0) || ((_snd.window == 0) && (len <= 1)));
1734 queue_packet(std::move(p));
1735}
1736
1737template <typename InetTraits>
1738future<> tcp<InetTraits>::tcb::wait_for_data() {
1739 if (!_rcv.data.empty() || foreign_will_not_send()) {
1740 return make_ready_future<>();
1741 }
1742 _rcv._data_received_promise = promise<>();
1743 return _rcv._data_received_promise->get_future();
1744}
1745
1746template <typename InetTraits>
1747future<> tcp<InetTraits>::tcb::wait_input_shutdown() {
1748 if (!_fin_recvd_promise) {
1749 return make_ready_future<>();
1750 }
1751 return _fin_recvd_promise->get_future();
1752}
1753
1754template <typename InetTraits>
1755void
1756tcp<InetTraits>::tcb::abort_reader() noexcept {
1757 if (_rcv._data_received_promise) {
1758 _rcv._data_received_promise->set_exception(
1759 std::make_exception_ptr(std::system_error(ECONNABORTED, std::system_category())));
1760 _rcv._data_received_promise = std::nullopt;
1761 }
1762 if (_fin_recvd_promise) {
1763 _fin_recvd_promise->set_value();
1764 _fin_recvd_promise.reset();
1765 }
1766}
1767
1768template <typename InetTraits>
1769future<> tcp<InetTraits>::tcb::wait_for_all_data_acked() {
1770 if (_snd.data.empty() && _snd.unsent_len == 0) {
1771 return make_ready_future<>();
1772 }
1773 _snd._all_data_acked_promise = promise<>();
1774 return _snd._all_data_acked_promise->get_future();
1775}
1776
1777template <typename InetTraits>
1779 // An initial send sequence number (ISS) is selected. A SYN segment of the
1780 // form <SEQ=ISS><CTL=SYN> is sent. Set SND.UNA to ISS, SND.NXT to ISS+1,
1781 // enter SYN-SENT state, and return.
1782 do_setup_isn();
1783
1784 // Local receive window scale factor
1785 _rcv.window_scale = _option._local_win_scale = 7;
1786 // Maximum segment size local can receive
1787 _rcv.mss = _option._local_mss = local_mss();
1788 _rcv.window = get_default_receive_window_size();
1789
1790 do_syn_sent();
1791}
1792
1793template <typename InetTraits>
1794packet tcp<InetTraits>::tcb::read() {
1795 packet p;
1796 for (auto&& q : _rcv.data) {
1797 p.append(std::move(q));
1798 }
1799 _rcv.data_size = 0;
1800 _rcv.data.clear();
1801 _rcv.window = get_default_receive_window_size();
1802 return p;
1803}
1804
1805template <typename InetTraits>
1806future<> tcp<InetTraits>::tcb::wait_send_available() {
1807 if (_snd.max_queue_space > _snd.current_queue_space) {
1808 return make_ready_future<>();
1809 }
1810 _snd._send_available_promise = promise<>();
1811 return _snd._send_available_promise->get_future();
1812}
1813
1814template <typename InetTraits>
1815future<> tcp<InetTraits>::tcb::send(packet p) {
1816 // We can not send after the connection is closed
1817 if (_snd.closed || in_state(CLOSED)) {
1818 return make_exception_future<>(tcp_reset_error());
1819 }
1820
1821 auto len = p.len();
1822 _snd.current_queue_space += len;
1823 _snd.unsent_len += len;
1824 _snd.unsent.push_back(std::move(p));
1825
1826 if (can_send() > 0) {
1827 output();
1828 }
1829
1830 return wait_send_available();
1831}
1832
1833template <typename InetTraits>
1834void tcp<InetTraits>::tcb::close() noexcept {
1835 if (in_state(CLOSED) || _snd.closed) {
1836 return;
1837 }
1838 // TODO: We should return a future to upper layer
1839 (void)wait_for_all_data_acked().then([this, zis = this->shared_from_this()] () mutable {
1840 _snd.closed = true;
1841 tcp_debug("close: unsent_len=%d\n", _snd.unsent_len);
1842 if (in_state(CLOSE_WAIT)) {
1843 tcp_debug("close: CLOSE_WAIT -> LAST_ACK\n");
1844 _state = LAST_ACK;
1845 } else if (in_state(ESTABLISHED)) {
1846 tcp_debug("close: ESTABLISHED -> FIN_WAIT_1\n");
1847 _state = FIN_WAIT_1;
1848 }
1849 // Send <FIN> to remote
1850 // Note: we call output_one to make sure a packet with FIN actually
1851 // sent out. If we only call output() and _packetq is not empty,
1852 // tcp::tcb::get_packet(), packet with FIN will not be generated.
1853 output_one();
1854 output();
1855 });
1856}
1857
1858template <typename InetTraits>
1859bool tcp<InetTraits>::tcb::should_send_ack(uint16_t seg_len) {
1860 // We've received a TSO packet, do ack immediately
1861 if (seg_len > _rcv.mss) {
1862 _nr_full_seg_received = 0;
1863 _delayed_ack.cancel();
1864 return true;
1865 }
1866
1867 // We've received a full sized segment, ack for every second full sized segment
1868 if (seg_len == _rcv.mss) {
1869 if (_nr_full_seg_received++ >= 1) {
1870 _nr_full_seg_received = 0;
1871 _delayed_ack.cancel();
1872 return true;
1873 }
1874 }
1875
1876 // If the timer is armed and its callback hasn't been run.
1877 if (_delayed_ack.armed()) {
1878 return false;
1879 }
1880
1881 // If the timer is not armed, schedule a delayed ACK.
1882 // The maximum delayed ack timer allowed by RFC1122 is 500ms, most
1883 // implementations use 200ms.
1884 _delayed_ack.arm(200ms);
1885 return false;
1886}
1887
1888template <typename InetTraits>
1889void tcp<InetTraits>::tcb::clear_delayed_ack() noexcept {
1890 _delayed_ack.cancel();
1891}
1892
1893template <typename InetTraits>
1894bool tcp<InetTraits>::tcb::merge_out_of_order() {
1895 bool merged = false;
1896 if (_rcv.out_of_order.map.empty()) {
1897 return merged;
1898 }
1899 for (auto it = _rcv.out_of_order.map.begin(); it != _rcv.out_of_order.map.end();) {
1900 auto& p = it->second;
1901 auto seg_beg = it->first;
1902 auto seg_len = p.len();
1903 auto seg_end = seg_beg + seg_len;
1904 if (seg_beg <= _rcv.next && _rcv.next < seg_end) {
1905 // This segment has been received out of order and its previous
1906 // segment has been received now
1907 auto trim = _rcv.next - seg_beg;
1908 if (trim) {
1909 p.trim_front(trim);
1910 seg_len -= trim;
1911 }
1912 _rcv.next += seg_len;
1913 _rcv.data_size += p.len();
1914 _rcv.data.push_back(std::move(p));
1915 // Since c++11, erase() always returns the value of the following element
1916 it = _rcv.out_of_order.map.erase(it);
1917 merged = true;
1918 } else if (_rcv.next >= seg_end) {
1919 // This segment has been receive already, drop it
1920 it = _rcv.out_of_order.map.erase(it);
1921 } else {
1922 // seg_beg > _rcv.need, can not merge. Note, seg_beg can grow only,
1923 // so we can stop looking here.
1924 it++;
1925 break;
1926 }
1927 }
1928 return merged;
1929}
1930
1931template <typename InetTraits>
1932void tcp<InetTraits>::tcb::insert_out_of_order(tcp_seq seg, packet p) {
1933 _rcv.out_of_order.merge(seg, std::move(p));
1934}
1935
1936template <typename InetTraits>
1937void tcp<InetTraits>::tcb::trim_receive_data_after_window() {
1938 abort();
1939}
1940
1941template <typename InetTraits>
1942void tcp<InetTraits>::tcb::persist() {
1943 tcp_debug("persist timer fired\n");
1944 // Send 1 byte packet to probe peer's window size
1945 _snd.window_probe = true;
1946 _snd.zero_window_probing_out++;
1947 output_one();
1948 _snd.window_probe = false;
1949
1950 output();
1951 // Perform binary exponential back-off per RFC1122
1952 _persist_time_out = std::min(_persist_time_out * 2, _rto_max);
1953 start_persist_timer();
1954}
1955
1956template <typename InetTraits>
1957void tcp<InetTraits>::tcb::retransmit() {
1958 auto output_update_rto = [this] {
1959 output();
1960 // According to RFC6298, Update RTO <- RTO * 2 to perform binary exponential back-off
1961 this->_rto = std::min(this->_rto * 2, this->_rto_max);
1962 start_retransmit_timer();
1963 };
1964
1965 // Retransmit SYN
1966 if (syn_needs_on()) {
1967 if (_snd.syn_retransmit++ < _max_nr_retransmit) {
1968 output_update_rto();
1969 } else {
1970 _connect_done.set_exception(tcp_connect_error());
1971 cleanup();
1972 return;
1973 }
1974 }
1975
1976 // Retransmit FIN
1977 if (fin_needs_on()) {
1978 if (_snd.fin_retransmit++ < _max_nr_retransmit) {
1979 output_update_rto();
1980 } else {
1981 cleanup();
1982 return;
1983 }
1984 }
1985
1986 // Retransmit Data
1987 if (_snd.data.empty()) {
1988 return;
1989 }
1990
1991 // If there are unacked data, retransmit the earliest segment
1992 auto& unacked_seg = _snd.data.front();
1993
1994 // According to RFC5681
1995 // Update ssthresh only for the first retransmit
1996 uint32_t smss = _snd.mss;
1997 if (unacked_seg.nr_transmits == 0) {
1998 _snd.ssthresh = std::max(flight_size() / 2, 2 * smss);
1999 }
2000 // RFC6582 Step 4
2001 _snd.recover = _snd.next - 1;
2002 // Start the slow start process
2003 _snd.cwnd = smss;
2004 // End fast recovery
2005 exit_fast_recovery();
2006
2007 if (unacked_seg.nr_transmits < _max_nr_retransmit) {
2008 unacked_seg.nr_transmits++;
2009 } else {
2010 // Delete connection when max num of retransmission is reached
2011 do_reset();
2012 return;
2013 }
2014 retransmit_one();
2015
2016 output_update_rto();
2017}
2018
2019template <typename InetTraits>
2020void tcp<InetTraits>::tcb::fast_retransmit() {
2021 if (!_snd.data.empty()) {
2022 auto& unacked_seg = _snd.data.front();
2023 unacked_seg.nr_transmits++;
2024 retransmit_one();
2025 output();
2026 }
2027}
2028
2029template <typename InetTraits>
2030void tcp<InetTraits>::tcb::update_rto(clock_type::time_point tx_time) {
2031 // Update RTO according to RFC6298
2032 auto R = std::chrono::duration_cast<std::chrono::milliseconds>(clock_type::now() - tx_time);
2033 if (_snd.first_rto_sample) {
2034 _snd.first_rto_sample = false;
2035 // RTTVAR <- R/2
2036 // SRTT <- R
2037 _snd.rttvar = R / 2;
2038 _snd.srtt = R;
2039 } else {
2040 // RTTVAR <- (1 - beta) * RTTVAR + beta * |SRTT - R'|
2041 // SRTT <- (1 - alpha) * SRTT + alpha * R'
2042 // where alpha = 1/8 and beta = 1/4
2043 auto delta = _snd.srtt > R ? (_snd.srtt - R) : (R - _snd.srtt);
2044 _snd.rttvar = _snd.rttvar * 3 / 4 + delta / 4;
2045 _snd.srtt = _snd.srtt * 7 / 8 + R / 8;
2046 }
2047 // RTO <- SRTT + max(G, K * RTTVAR)
2048 _rto = _snd.srtt + std::max(_rto_clk_granularity, 4 * _snd.rttvar);
2049
2050 // Make sure 1 sec << _rto << 60 sec
2051 _rto = std::max(_rto, _rto_min);
2052 _rto = std::min(_rto, _rto_max);
2053}
2054
2055template <typename InetTraits>
2056void tcp<InetTraits>::tcb::update_cwnd(uint32_t acked_bytes) {
2057 uint32_t smss = _snd.mss;
2058 if (_snd.cwnd < _snd.ssthresh) {
2059 // In slow start phase
2060 _snd.cwnd += std::min(acked_bytes, smss);
2061 } else {
2062 // In congestion avoidance phase
2063 uint32_t round_up = 1;
2064 _snd.cwnd += std::max(round_up, smss * smss / _snd.cwnd);
2065 }
2066}
2067
2068template <typename InetTraits>
2069void tcp<InetTraits>::tcb::cleanup() {
2070 _snd.unsent.clear();
2071 _snd.data.clear();
2072 _rcv.out_of_order.map.clear();
2073 _rcv.data_size = 0;
2074 _rcv.data.clear();
2075 stop_retransmit_timer();
2076 clear_delayed_ack();
2077 remove_from_tcbs();
2078}
2079
2080template <typename InetTraits>
2081tcp_seq tcp<InetTraits>::tcb::get_isn() {
2082 // Per RFC6528, TCP SHOULD generate its Initial Sequence Numbers
2083 // with the expression:
2084 // ISN = M + F(localip, localport, remoteip, remoteport, secretkey)
2085 // M is the 4 microsecond timer
2086 using namespace std::chrono;
2087 uint32_t hash[4];
2088 hash[0] = _local_ip.ip;
2089 hash[1] = _foreign_ip.ip;
2090 hash[2] = (_local_port << 16) + _foreign_port;
2091 gnutls_hash_hd_t md5_hash_handle;
2092 // GnuTLS digests do not init at all, so this should never fail.
2093 gnutls_hash_init(&md5_hash_handle, GNUTLS_DIG_MD5);
2094 gnutls_hash(md5_hash_handle, hash, 3 * sizeof(hash[0]));
2095 gnutls_hash(md5_hash_handle, _isn_secret.key, sizeof(_isn_secret.key));
2096 // reuse "hash" for the output of digest
2097 assert(sizeof(hash) == gnutls_hash_get_len(GNUTLS_DIG_MD5));
2098 gnutls_hash_deinit(md5_hash_handle, hash);
2099 auto seq = hash[0];
2100 auto m = duration_cast<microseconds>(clock_type::now().time_since_epoch());
2101 seq += m.count() / 4;
2102 return make_seq(seq);
2103}
2104
2105template <typename InetTraits>
2106std::optional<typename InetTraits::l4packet> tcp<InetTraits>::tcb::get_packet() {
2107 _poll_active = false;
2108 if (_packetq.empty()) {
2109 output_one();
2110 }
2111
2112 if (in_state(CLOSED)) {
2113 return std::optional<typename InetTraits::l4packet>();
2114 }
2115
2116 assert(!_packetq.empty());
2117
2118 auto p = std::move(_packetq.front());
2119 _packetq.pop_front();
2120 if (!_packetq.empty() || (_snd.dupacks < 3 && can_send() > 0 && (_snd.window > 0))) {
2121 // If there are packets to send in the queue or tcb is allowed to send
2122 // more add tcp back to polling set to keep sending. In addition, dupacks >= 3
2123 // is an indication that an segment is lost, stop sending more in this case.
2124 // Finally - we can't send more until window is opened again.
2125 output();
2126 }
2127 return p;
2128}
2129
2130template <typename InetTraits>
2131void tcp<InetTraits>::connection::close_read() noexcept {
2132 _tcb->abort_reader();
2133}
2134
2135template <typename InetTraits>
2136void tcp<InetTraits>::connection::close_write() noexcept {
2137 _tcb->close();
2138}
2139
2140template <typename InetTraits>
2141void tcp<InetTraits>::connection::shutdown_connect() {
2142 if (_tcb->syn_needs_on()) {
2143 _tcb->_connect_done.set_exception(tcp_refused_error());
2144 _tcb->cleanup();
2145 } else {
2146 close_read();
2147 close_write();
2148 }
2149}
2150
2151template <typename InetTraits>
2152typename tcp<InetTraits>::tcb::isn_secret tcp<InetTraits>::tcb::_isn_secret;
2153
2154}
2155
2156}
bool try_wait(size_t nr=1) noexcept
Definition: semaphore.hh:434
void signal(size_t nr=1) noexcept
Definition: semaphore.hh:396
Definition: shared_ptr.hh:148
A representation of a possibly not-yet-computed value.
Definition: future.hh:1240
Low-resolution and efficient steady clock.
Definition: lowres_clock.hh:56
static time_point now() noexcept
Definition: lowres_clock.hh:74
holds the metric definition.
Definition: metrics_registration.hh:94
metric_groups & add_group(const group_name_type &name, const std::initializer_list< metric_definition > &l)
Add metrics belonging to the same group.
Definition: arp.hh:205
Definition: net.hh:51
Definition: packet.hh:87
Definition: tcp.hh:671
Definition: tcp.hh:719
Definition: tcp.hh:291
void set_value(A &&... a) noexcept
Sets the promises value.
Definition: future.hh:990
void set_exception(std::exception_ptr &&ex) noexcept
Marks the promise as failed.
Definition: future.hh:998
Definition: queue.hh:44
size_t size() const noexcept
Returns the number of items currently in the queue.
Definition: queue.hh:109
future< T > pop_eventually() noexcept
Definition: queue.hh:225
size_t max_size() const noexcept
Definition: queue.hh:117
void abort(std::exception_ptr ex) noexcept
Definition: queue.hh:131
Definition: socket_defs.hh:47
bool cancel() noexcept
void rearm(time_point until, std::optional< duration > period={}) noexcept
Definition: timer.hh:167
future< T > get_future() noexcept
Gets the promise's associated future.
Definition: future.hh:1926
future now()
Returns a ready future.
Definition: later.hh:35
impl::metric_definition_impl make_counter(metric_name_type name, T &&val, description d=description(), std::vector< label_instance > labels={})
create a counter metric
Definition: metrics.hh:528
server_socket listen(socket_address sa)
future< connected_socket > connect(socket_address sa)
header for metrics creation.
Definition: net.hh:75
Definition: tcp.hh:197
Definition: tcp.hh:287
Seastar API namespace.
Definition: abort_on_ebadf.hh:26
shard_id this_shard_id() noexcept
Returns shard_id of the of the current shard.
Definition: shard_id.hh:52
Definition: ethernet.hh:37
Definition: ip.hh:108
Definition: tcp.hh:227
Definition: tcp.hh:168
Definition: tcp.hh:106
Definition: tcp.hh:161
Definition: tcp.hh:134
Definition: tcp.hh:96