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