--- /dev/null
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+/*
+ * This file is open source software, licensed to you under the terms
+ * of the Apache License, Version 2.0 (the "License"). See the NOTICE file
+ * distributed with this work for additional information regarding copyright
+ * ownership. You may not use this file except in compliance with the License.
+ *
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+/*
+ * Copyright (C) 2014 Cloudius Systems, Ltd.
+ */
+
+#ifndef CEPH_DPDK_TCP_H_
+#define CEPH_DPDK_TCP_H_
+
+#include <unordered_map>
+#include <map>
+#include <queue>
+#include <functional>
+#include <deque>
+#include <chrono>
+#include <random>
+#include <stdexcept>
+#include <system_error>
+
+#define CRYPTOPP_ENABLE_NAMESPACE_WEAK 1
+#include <cryptopp/md5.h>
+
+#include "msg/async/dpdk/EventDPDK.h"
+
+#include "include/utime.h"
+#include "common/Throttle.h"
+#include "common/ceph_time.h"
+#include "msg/async/Event.h"
+#include "IPChecksum.h"
+#include "IP.h"
+#include "const.h"
+#include "byteorder.h"
+#include "shared_ptr.h"
+#include "PacketUtil.h"
+
+struct tcp_hdr;
+
+enum class tcp_state : uint16_t {
+ CLOSED = (1 << 0),
+ LISTEN = (1 << 1),
+ SYN_SENT = (1 << 2),
+ SYN_RECEIVED = (1 << 3),
+ ESTABLISHED = (1 << 4),
+ FIN_WAIT_1 = (1 << 5),
+ FIN_WAIT_2 = (1 << 6),
+ CLOSE_WAIT = (1 << 7),
+ CLOSING = (1 << 8),
+ LAST_ACK = (1 << 9),
+ TIME_WAIT = (1 << 10)
+};
+
+inline tcp_state operator|(tcp_state s1, tcp_state s2) {
+ return tcp_state(uint16_t(s1) | uint16_t(s2));
+}
+
+inline std::ostream & operator<<(std::ostream & str, const tcp_state& s) {
+ switch (s) {
+ case tcp_state::CLOSED: return str << "CLOSED";
+ case tcp_state::LISTEN: return str << "LISTEN";
+ case tcp_state::SYN_SENT: return str << "SYN_SENT";
+ case tcp_state::SYN_RECEIVED: return str << "SYN_RECEIVED";
+ case tcp_state::ESTABLISHED: return str << "ESTABLISHED";
+ case tcp_state::FIN_WAIT_1: return str << "FIN_WAIT_1";
+ case tcp_state::FIN_WAIT_2: return str << "FIN_WAIT_2";
+ case tcp_state::CLOSE_WAIT: return str << "CLOSE_WAIT";
+ case tcp_state::CLOSING: return str << "CLOSING";
+ case tcp_state::LAST_ACK: return str << "LAST_ACK";
+ case tcp_state::TIME_WAIT: return str << "TIME_WAIT";
+ default: return str << "UNKNOWN";
+ }
+}
+
+struct tcp_option {
+ // The kind and len field are fixed and defined in TCP protocol
+ enum class option_kind: uint8_t { mss = 2, win_scale = 3, sack = 4, timestamps = 8, nop = 1, eol = 0 };
+ enum class option_len: uint8_t { mss = 4, win_scale = 3, sack = 2, timestamps = 10, nop = 1, eol = 1 };
+ struct mss {
+ option_kind kind = option_kind::mss;
+ option_len len = option_len::mss;
+ uint16_t mss;
+ struct mss hton() {
+ struct mss m = *this;
+ m.mss = ::hton(m.mss);
+ return m;
+ }
+ } __attribute__((packed));
+ struct win_scale {
+ option_kind kind = option_kind::win_scale;
+ option_len len = option_len::win_scale;
+ uint8_t shift;
+ } __attribute__((packed));
+ struct sack {
+ option_kind kind = option_kind::sack;
+ option_len len = option_len::sack;
+ } __attribute__((packed));
+ struct timestamps {
+ option_kind kind = option_kind::timestamps;
+ option_len len = option_len::timestamps;
+ uint32_t t1;
+ uint32_t t2;
+ } __attribute__((packed));
+ struct nop {
+ option_kind kind = option_kind::nop;
+ } __attribute__((packed));
+ struct eol {
+ option_kind kind = option_kind::eol;
+ } __attribute__((packed));
+ static const uint8_t align = 4;
+
+ void parse(uint8_t* beg, uint8_t* end);
+ uint8_t fill(tcp_hdr* th, uint8_t option_size);
+ uint8_t get_size(bool syn_on, bool ack_on);
+
+ // For option negotiattion
+ bool _mss_received = false;
+ bool _win_scale_received = false;
+ bool _timestamps_received = false;
+ bool _sack_received = false;
+
+ // Option data
+ uint16_t _remote_mss = 536;
+ uint16_t _local_mss;
+ uint8_t _remote_win_scale = 0;
+ uint8_t _local_win_scale = 0;
+};
+inline uint8_t*& operator+=(uint8_t*& x, tcp_option::option_len len) { x += uint8_t(len); return x; }
+inline uint8_t& operator+=(uint8_t& x, tcp_option::option_len len) { x += uint8_t(len); return x; }
+
+struct tcp_sequence {
+ uint32_t raw;
+};
+
+tcp_sequence ntoh(tcp_sequence ts) {
+ return tcp_sequence { ::ntoh(ts.raw) };
+}
+
+tcp_sequence hton(tcp_sequence ts) {
+ return tcp_sequence { ::hton(ts.raw) };
+}
+
+inline std::ostream& operator<<(std::ostream& os, const tcp_sequence& s) {
+ return os << s.raw;
+}
+
+inline tcp_sequence make_seq(uint32_t raw) { return tcp_sequence{raw}; }
+inline tcp_sequence& operator+=(tcp_sequence& s, int32_t n) { s.raw += n; return s; }
+inline tcp_sequence& operator-=(tcp_sequence& s, int32_t n) { s.raw -= n; return s; }
+inline tcp_sequence operator+(tcp_sequence s, int32_t n) { return s += n; }
+inline tcp_sequence operator-(tcp_sequence s, int32_t n) { return s -= n; }
+inline int32_t operator-(tcp_sequence s, tcp_sequence q) { return s.raw - q.raw; }
+inline bool operator==(tcp_sequence s, tcp_sequence q) { return s.raw == q.raw; }
+inline bool operator!=(tcp_sequence s, tcp_sequence q) { return !(s == q); }
+inline bool operator<(tcp_sequence s, tcp_sequence q) { return s - q < 0; }
+inline bool operator>(tcp_sequence s, tcp_sequence q) { return q < s; }
+inline bool operator<=(tcp_sequence s, tcp_sequence q) { return !(s > q); }
+inline bool operator>=(tcp_sequence s, tcp_sequence q) { return !(s < q); }
+
+struct tcp_hdr {
+ uint16_t src_port;
+ uint16_t dst_port;
+ tcp_sequence seq;
+ tcp_sequence ack;
+ uint8_t rsvd1 : 4;
+ uint8_t data_offset : 4;
+ uint8_t f_fin : 1;
+ uint8_t f_syn : 1;
+ uint8_t f_rst : 1;
+ uint8_t f_psh : 1;
+ uint8_t f_ack : 1;
+ uint8_t f_urg : 1;
+ uint8_t rsvd2 : 2;
+ uint16_t window;
+ uint16_t checksum;
+ uint16_t urgent;
+
+ tcp_hdr hton() {
+ tcp_hdr hdr = *this;
+ hdr.src_port = ::hton(src_port);
+ hdr.dst_port = ::hton(dst_port);
+ hdr.seq = ::hton(seq);
+ hdr.ack = ::hton(ack);
+ hdr.window = ::hton(window);
+ hdr.checksum = ::hton(checksum);
+ hdr.urgent = ::hton(urgent);
+ return hdr;
+ }
+
+ tcp_hdr ntoh() {
+ tcp_hdr hdr = *this;
+ hdr.src_port = ::ntoh(src_port);
+ hdr.dst_port = ::ntoh(dst_port);
+ hdr.seq = ::ntoh(seq);
+ hdr.ack = ::ntoh(ack);
+ hdr.window = ::ntoh(window);
+ hdr.checksum = ::ntoh(checksum);
+ hdr.urgent = ::ntoh(urgent);
+ return hdr;
+ }
+} __attribute__((packed));
+
+struct tcp_tag {};
+using tcp_packet_merger = packet_merger<tcp_sequence, tcp_tag>;
+
+template <typename InetTraits>
+class tcp {
+ public:
+ using ipaddr = typename InetTraits::address_type;
+ using inet_type = typename InetTraits::inet_type;
+ using connid = l4connid<InetTraits>;
+ using connid_hash = typename connid::connid_hash;
+ class connection;
+ class listener;
+ private:
+ class tcb;
+
+ class C_handle_delayed_ack : public EventCallback {
+ tcb *tc;
+
+ public:
+ C_handle_delayed_ack(tcb *t): tc(t) { }
+ void do_request(int r) {
+ tc->_nr_full_seg_received = 0;
+ tc->output();
+ }
+ };
+
+ class C_handle_retransmit : public EventCallback {
+ tcb *tc;
+
+ public:
+ C_handle_retransmit(tcb *t): tc(t) { }
+ void do_request(int r) {
+ tc->retransmit();
+ }
+ };
+
+ class C_handle_persist : public EventCallback {
+ tcb *tc;
+
+ public:
+ C_handle_persist(tcb *t): tc(t) { }
+ void do_request(int r) {
+ tc->persist();
+ }
+ };
+
+ class C_all_data_acked : public EventCallback {
+ tcb *tc;
+
+ public:
+ C_all_data_acked(tcb *t): tc(t) {}
+ void do_request(int fd_or_id) {
+ tc->close_final_cleanup();
+ }
+ };
+
+ class C_actual_remove_tcb : public EventCallback {
+ lw_shared_ptr<tcb> tc;
+ public:
+ C_actual_remove_tcb(tcb *t): tc(t->shared_from_this()) {}
+ void do_request(int r) {
+ delete this;
+ }
+ };
+
+ class tcb : public enable_lw_shared_from_this<tcb> {
+ using clock_type = ceph::coarse_real_clock;
+ static constexpr tcp_state CLOSED = tcp_state::CLOSED;
+ static constexpr tcp_state LISTEN = tcp_state::LISTEN;
+ static constexpr tcp_state SYN_SENT = tcp_state::SYN_SENT;
+ static constexpr tcp_state SYN_RECEIVED = tcp_state::SYN_RECEIVED;
+ static constexpr tcp_state ESTABLISHED = tcp_state::ESTABLISHED;
+ static constexpr tcp_state FIN_WAIT_1 = tcp_state::FIN_WAIT_1;
+ static constexpr tcp_state FIN_WAIT_2 = tcp_state::FIN_WAIT_2;
+ static constexpr tcp_state CLOSE_WAIT = tcp_state::CLOSE_WAIT;
+ static constexpr tcp_state CLOSING = tcp_state::CLOSING;
+ static constexpr tcp_state LAST_ACK = tcp_state::LAST_ACK;
+ static constexpr tcp_state TIME_WAIT = tcp_state::TIME_WAIT;
+ tcp_state _state = CLOSED;
+ tcp& _tcp;
+ UserspaceEventManager &manager;
+ connection* _conn = nullptr;
+ bool _connect_done = false;
+ ipaddr _local_ip;
+ ipaddr _foreign_ip;
+ uint16_t _local_port;
+ uint16_t _foreign_port;
+ struct unacked_segment {
+ Packet p;
+ uint16_t data_len;
+ unsigned nr_transmits;
+ clock_type::time_point tx_time;
+ };
+ struct send {
+ tcp_sequence unacknowledged;
+ tcp_sequence next;
+ uint32_t window;
+ uint8_t window_scale;
+ uint16_t mss;
+ tcp_sequence urgent;
+ tcp_sequence wl1;
+ tcp_sequence wl2;
+ tcp_sequence initial;
+ std::deque<unacked_segment> data;
+ std::deque<Packet> unsent;
+ uint32_t unsent_len = 0;
+ uint32_t queued_len = 0;
+ bool closed = false;
+ // Wait for all data are acked
+ int _all_data_acked_fd = -1;
+ // Limit number of data queued into send queue
+ Throttle user_queue_space;
+ // Round-trip time variation
+ std::chrono::microseconds rttvar;
+ // Smoothed round-trip time
+ std::chrono::microseconds srtt;
+ bool first_rto_sample = true;
+ clock_type::time_point syn_tx_time;
+ // Congestion window
+ uint32_t cwnd;
+ // Slow start threshold
+ uint32_t ssthresh;
+ // Duplicated ACKs
+ uint16_t dupacks = 0;
+ unsigned syn_retransmit = 0;
+ unsigned fin_retransmit = 0;
+ uint32_t limited_transfer = 0;
+ uint32_t partial_ack = 0;
+ tcp_sequence recover;
+ bool window_probe = false;
+ send(CephContext *c): user_queue_space(c, "DPDK::tcp::tcb::user_queue_space", 81920) {}
+ } _snd;
+ struct receive {
+ tcp_sequence next;
+ uint32_t window;
+ uint8_t window_scale;
+ uint16_t mss;
+ tcp_sequence urgent;
+ tcp_sequence initial;
+ std::deque<Packet> data;
+ tcp_packet_merger out_of_order;
+ } _rcv;
+ EventCenter *center;
+ int fd;
+ // positive means no errno, 0 means eof, nagetive means error
+ int16_t _errno = 1;
+ tcp_option _option;
+ EventCallbackRef delayed_ack_event;
+ Tub<uint64_t> _delayed_ack_fd;
+ // Retransmission timeout
+ std::chrono::microseconds _rto{1000*1000};
+ std::chrono::microseconds _persist_time_out{1000*1000};
+ static constexpr std::chrono::microseconds _rto_min{1000*1000};
+ static constexpr std::chrono::microseconds _rto_max{60000*1000};
+ // Clock granularity
+ static constexpr std::chrono::microseconds _rto_clk_granularity{1000};
+ static constexpr uint16_t _max_nr_retransmit{5};
+ EventCallbackRef retransmit_event;
+ Tub<uint64_t> retransmit_fd;
+ EventCallbackRef persist_event;
+ EventCallbackRef all_data_ack_event;
+ Tub<uint64_t> persist_fd;
+ uint16_t _nr_full_seg_received = 0;
+ struct isn_secret {
+ // 512 bits secretkey for ISN generating
+ uint32_t key[16];
+ isn_secret () {
+ std::random_device rd;
+ std::default_random_engine e(rd());
+ std::uniform_int_distribution<uint32_t> dist{};
+ for (auto& k : key) {
+ k = dist(e);
+ }
+ }
+ };
+ static isn_secret _isn_secret;
+ tcp_sequence get_isn();
+ circular_buffer<typename InetTraits::l4packet> _packetq;
+ bool _poll_active = false;
+ public:
+ // callback
+ void close_final_cleanup();
+ ostream& _prefix(std::ostream *_dout);
+
+ public:
+ tcb(tcp& t, connid id);
+ ~tcb();
+ void input_handle_listen_state(tcp_hdr* th, Packet p);
+ void input_handle_syn_sent_state(tcp_hdr* th, Packet p);
+ void input_handle_other_state(tcp_hdr* th, Packet p);
+ void output_one(bool data_retransmit = false);
+ bool is_all_data_acked();
+ int send(Packet p);
+ void connect();
+ Tub<Packet> read();
+ void close();
+ void remove_from_tcbs() {
+ auto id = connid{_local_ip, _foreign_ip, _local_port, _foreign_port};
+ _tcp._tcbs.erase(id);
+ }
+ Tub<typename InetTraits::l4packet> get_packet();
+ void output() {
+ if (!_poll_active) {
+ _poll_active = true;
+
+ auto tcb = this->shared_from_this();
+ _tcp._inet.wait_l2_dst_address(_foreign_ip, Packet(), [tcb] (const ethernet_address &dst, Packet p, int r) {
+ if (r == 0) {
+ tcb->_tcp.poll_tcb(dst, std::move(tcb));
+ } else if (r == -ETIMEDOUT) {
+ // in other states connection should time out
+ if (tcb->in_state(SYN_SENT)) {
+ tcb->_errno = -ETIMEDOUT;
+ tcb->cleanup();
+ }
+ } else if (r == -EBUSY) {
+ // retry later
+ tcb->_poll_active = false;
+ tcb->start_retransmit_timer();
+ }
+ });
+ }
+ }
+
+ int16_t get_errno() const {
+ return _errno;
+ }
+
+ tcp_state& state() {
+ return _state;
+ }
+
+ uint64_t peek_sent_available() {
+ if (!in_state(ESTABLISHED))
+ return 0;
+ uint64_t left = _snd.user_queue_space.get_max() - _snd.user_queue_space.get_current();
+ return left;
+ }
+
+ int is_connected() const {
+ if (_errno <= 0)
+ return _errno;
+ return _connect_done;
+ }
+
+ private:
+ void respond_with_reset(tcp_hdr* th);
+ bool merge_out_of_order();
+ void insert_out_of_order(tcp_sequence seq, Packet p);
+ void trim_receive_data_after_window();
+ bool should_send_ack(uint16_t seg_len);
+ void clear_delayed_ack();
+ Packet get_transmit_packet();
+ void retransmit_one() {
+ bool data_retransmit = true;
+ output_one(data_retransmit);
+ }
+ void start_retransmit_timer() {
+ if (retransmit_fd)
+ center->delete_time_event(*retransmit_fd);
+ retransmit_fd.construct(center->create_time_event(_rto.count(), retransmit_event));
+ };
+ void stop_retransmit_timer() {
+ if (retransmit_fd) {
+ center->delete_time_event(*retransmit_fd);
+ retransmit_fd.destroy();
+ }
+ };
+ void start_persist_timer() {
+ if (persist_fd)
+ center->delete_time_event(*persist_fd);
+ persist_fd.construct(center->create_time_event(_persist_time_out.count(), persist_event));
+ };
+ void stop_persist_timer() {
+ if (persist_fd) {
+ center->delete_time_event(*persist_fd);
+ persist_fd.destroy();
+ }
+ };
+ void persist();
+ void retransmit();
+ void fast_retransmit();
+ void update_rto(clock_type::time_point tx_time);
+ void update_cwnd(uint32_t acked_bytes);
+ void cleanup();
+ uint32_t can_send() {
+ if (_snd.window_probe) {
+ return 1;
+ }
+ // Can not send more than advertised window allows
+ auto x = std::min(uint32_t(_snd.unacknowledged + _snd.window - _snd.next), _snd.unsent_len);
+ // Can not send more than congestion window allows
+ x = std::min(_snd.cwnd, x);
+ if (_snd.dupacks == 1 || _snd.dupacks == 2) {
+ // RFC5681 Step 3.1
+ // Send cwnd + 2 * smss per RFC3042
+ auto flight = flight_size();
+ auto max = _snd.cwnd + 2 * _snd.mss;
+ x = flight <= max ? std::min(x, max - flight) : 0;
+ _snd.limited_transfer += x;
+ } else if (_snd.dupacks >= 3) {
+ // RFC5681 Step 3.5
+ // Sent 1 full-sized segment at most
+ x = std::min(uint32_t(_snd.mss), x);
+ }
+ return x;
+ }
+ uint32_t flight_size() {
+ uint32_t size = 0;
+ std::for_each(_snd.data.begin(), _snd.data.end(),
+ [&] (unacked_segment& seg) { size += seg.p.len(); });
+ return size;
+ }
+ uint16_t local_mss() {
+ return _tcp.get_hw_features().mtu - tcp_hdr_len_min - InetTraits::ip_hdr_len_min;
+ }
+ void queue_packet(Packet p) {
+ _packetq.emplace_back(
+ typename InetTraits::l4packet{_foreign_ip, std::move(p)});
+ }
+ void signal_data_received() {
+ manager.notify(fd, EVENT_READABLE);
+ }
+ void signal_all_data_acked() {
+ if (_snd._all_data_acked_fd >= 0 && _snd.unsent_len == 0 && _snd.queued_len == 0)
+ manager.notify(_snd._all_data_acked_fd, EVENT_READABLE);
+ }
+ void do_syn_sent() {
+ _state = SYN_SENT;
+ _snd.syn_tx_time = clock_type::now();
+ // Send <SYN> to remote
+ output();
+ }
+ void do_syn_received() {
+ _state = SYN_RECEIVED;
+ _snd.syn_tx_time = clock_type::now();
+ // Send <SYN,ACK> to remote
+ output();
+ }
+ void do_established() {
+ _state = ESTABLISHED;
+ update_rto(_snd.syn_tx_time);
+ _connect_done = true;
+ manager.notify(fd, EVENT_READABLE|EVENT_WRITABLE);
+ }
+ void do_reset() {
+ _state = CLOSED;
+ // Free packets to be sent which are waiting for user_queue_space
+ _snd.user_queue_space.reset();
+ cleanup();
+ _errno = -ECONNRESET;
+ manager.notify(fd, EVENT_READABLE);
+
+ if (_snd._all_data_acked_fd >= 0)
+ manager.notify(_snd._all_data_acked_fd, EVENT_READABLE);
+ }
+ void do_time_wait() {
+ // FIXME: Implement TIME_WAIT state timer
+ _state = TIME_WAIT;
+ cleanup();
+ }
+ void do_closed() {
+ _state = CLOSED;
+ cleanup();
+ }
+ void do_setup_isn() {
+ _snd.initial = get_isn();
+ _snd.unacknowledged = _snd.initial;
+ _snd.next = _snd.initial + 1;
+ _snd.recover = _snd.initial;
+ }
+ void do_local_fin_acked() {
+ _snd.unacknowledged += 1;
+ _snd.next += 1;
+ }
+ bool syn_needs_on() {
+ return in_state(SYN_SENT | SYN_RECEIVED);
+ }
+ bool fin_needs_on() {
+ return in_state(FIN_WAIT_1 | CLOSING | LAST_ACK) && _snd.closed &&
+ _snd.unsent_len == 0 && _snd.queued_len == 0;
+ }
+ bool ack_needs_on() {
+ return !in_state(CLOSED | LISTEN | SYN_SENT);
+ }
+ bool foreign_will_not_send() {
+ return in_state(CLOSING | TIME_WAIT | CLOSE_WAIT | LAST_ACK | CLOSED);
+ }
+ bool in_state(tcp_state state) {
+ return uint16_t(_state) & uint16_t(state);
+ }
+ void exit_fast_recovery() {
+ _snd.dupacks = 0;
+ _snd.limited_transfer = 0;
+ _snd.partial_ack = 0;
+ }
+ uint32_t data_segment_acked(tcp_sequence seg_ack);
+ bool segment_acceptable(tcp_sequence seg_seq, unsigned seg_len);
+ void init_from_options(tcp_hdr* th, uint8_t* opt_start, uint8_t* opt_end);
+ friend class connection;
+
+ friend class C_handle_delayed_ack;
+ friend class C_handle_retransmit;
+ friend class C_handle_persist;
+ friend class C_all_data_acked;
+ };
+
+ CephContext *cct;
+ // ipv4_l4<ip_protocol_num::tcp>
+ inet_type& _inet;
+ EventCenter *center;
+ UserspaceEventManager &manager;
+ std::unordered_map<connid, lw_shared_ptr<tcb>, connid_hash> _tcbs;
+ std::unordered_map<uint16_t, listener*> _listening;
+ std::random_device _rd;
+ std::default_random_engine _e;
+ std::uniform_int_distribution<uint16_t> _port_dist{41952, 65535};
+ circular_buffer<std::pair<lw_shared_ptr<tcb>, ethernet_address>> _poll_tcbs;
+ // queue for packets that do not belong to any tcb
+ circular_buffer<ipv4_traits::l4packet> _packetq;
+ Throttle _queue_space;
+ // Limit number of data queued into send queue
+ public:
+ class connection {
+ lw_shared_ptr<tcb> _tcb;
+ public:
+ explicit connection(lw_shared_ptr<tcb> tcbp) : _tcb(std::move(tcbp)) { _tcb->_conn = this; }
+ connection(const connection&) = delete;
+ connection(connection&& x) noexcept : _tcb(std::move(x._tcb)) {
+ _tcb->_conn = this;
+ }
+ ~connection();
+ void operator=(const connection&) = delete;
+ connection& operator=(connection&& x) {
+ if (this != &x) {
+ this->~connection();
+ new (this) connection(std::move(x));
+ }
+ return *this;
+ }
+ int fd() const {
+ return _tcb->fd;
+ }
+ int send(Packet p) {
+ return _tcb->send(std::move(p));
+ }
+ Tub<Packet> read() {
+ return _tcb->read();
+ }
+ int16_t get_errno() const {
+ return _tcb->get_errno();
+ }
+ void close_read();
+ void close_write();
+ entity_addr_t remote_addr() const {
+ entity_addr_t addr;
+ auto net_ip = _tcb->_foreign_ip.hton();
+ memcpy((void*)&addr.in4_addr().sin_addr.s_addr,
+ &net_ip, sizeof(addr.in4_addr().sin_addr.s_addr));
+ addr.set_family(AF_INET);
+ return addr;
+ }
+ uint64_t peek_sent_available() {
+ return _tcb->peek_sent_available();
+ }
+ int is_connected() const { return _tcb->is_connected(); }
+ };
+ class listener {
+ tcp& _tcp;
+ uint16_t _port;
+ int _fd = -1;
+ int16_t _errno;
+ queue<connection> _q;
+ size_t _q_max_length;
+
+ private:
+ listener(tcp& t, uint16_t port, size_t queue_length)
+ : _tcp(t), _port(port), _errno(0), _q(), _q_max_length(queue_length) {
+ }
+ public:
+ listener(const listener&) = delete;
+ void operator=(const listener&) = delete;
+ listener(listener&& x)
+ : _tcp(x._tcp), _port(x._port), _fd(std::move(x._fd)), _errno(x._errno),
+ _q(std::move(x._q)) {
+ if (_fd >= 0)
+ _tcp._listening[_port] = this;
+ }
+ ~listener() {
+ abort_accept();
+ }
+ int listen() {
+ if (_tcp._listening.find(_port) != _tcp._listening.end())
+ return -EADDRINUSE;
+ _tcp._listening.emplace(_port, this);
+ _fd = _tcp.manager.get_eventfd();
+ return 0;
+ }
+ Tub<connection> accept() {
+ Tub<connection> c;
+ if (!_q.empty()) {
+ c = std::move(_q.front());
+ _q.pop();
+ }
+ return c;
+ }
+ void abort_accept() {
+ while (!_q.empty())
+ _q.pop();
+ if (_fd >= 0) {
+ _tcp._listening.erase(_port);
+ _tcp.manager.close(_fd);
+ _fd = -1;
+ }
+ }
+ int16_t get_errno() const {
+ return _errno;
+ }
+ bool full() const {
+ return _q.size() == _q_max_length;
+ }
+ int fd() const {
+ return _fd;
+ }
+ friend class tcp;
+ };
+ public:
+ explicit tcp(CephContext *c, inet_type& inet, EventCenter *cen);
+ void received(Packet p, ipaddr from, ipaddr to);
+ bool forward(forward_hash& out_hash_data, Packet& p, size_t off);
+ listener listen(uint16_t port, size_t queue_length = 100);
+ connection connect(const entity_addr_t &addr);
+ const hw_features& get_hw_features() const { return _inet._inet.get_hw_features(); }
+ void poll_tcb(const ethernet_address &dst, lw_shared_ptr<tcb> tcb) {
+ _poll_tcbs.emplace_back(std::move(tcb), dst);
+ }
+ bool push_listen_queue(uint16_t port, tcb *t) {
+ auto listener = _listening.find(port);
+ if (listener == _listening.end() || listener->second->full()) {
+ return false;
+ }
+ listener->second->_q.push(connection(t->shared_from_this()));
+ manager.notify(listener->second->_fd, EVENT_READABLE);
+ return true;
+ }
+
+ private:
+ void send_packet_without_tcb(ipaddr from, ipaddr to, Packet p);
+ void respond_with_reset(tcp_hdr* rth, ipaddr local_ip, ipaddr foreign_ip);
+ friend class listener;
+};
+
+template <typename InetTraits>
+tcp<InetTraits>::tcp(CephContext *c, inet_type& inet, EventCenter *cen)
+ : cct(c), _inet(inet), center(cen),
+ manager(static_cast<DPDKDriver*>(cen->get_driver())->manager),
+ _e(_rd()), _queue_space(cct, "DPDK::tcp::queue_space", 81920) {
+ int tcb_polled = 0u;
+ _inet.register_packet_provider([this, tcb_polled] () mutable {
+ Tub<typename InetTraits::l4packet> l4p;
+ auto c = _poll_tcbs.size();
+ if (!_packetq.empty() && (!(tcb_polled % 128) || c == 0)) {
+ l4p = std::move(_packetq.front());
+ _packetq.pop_front();
+ _queue_space.put(l4p->p.len());
+ } else {
+ while (c--) {
+ tcb_polled++;
+ lw_shared_ptr<tcb> tcb;
+ ethernet_address dst;
+ std::tie(tcb, dst) = std::move(_poll_tcbs.front());
+ _poll_tcbs.pop_front();
+ l4p = std::move(tcb->get_packet());
+ if (l4p) {
+ l4p->e_dst = dst;
+ break;
+ }
+ }
+ }
+ return l4p;
+ });
+}
+
+template <typename InetTraits>
+auto tcp<InetTraits>::listen(uint16_t port, size_t queue_length) -> listener {
+ return listener(*this, port, queue_length);
+}
+
+template <typename InetTraits>
+typename tcp<InetTraits>::connection tcp<InetTraits>::connect(const entity_addr_t &addr) {
+ uint16_t src_port;
+ connid id;
+ auto src_ip = _inet._inet.host_address();
+ auto dst_ip = ipv4_address(addr);
+ auto dst_port = addr.get_port();
+
+ do {
+ src_port = _port_dist(_e);
+ id = connid{src_ip, dst_ip, src_port, (uint16_t)dst_port};
+ if (_tcbs.find(id) == _tcbs.end()) {
+ if (_inet._inet.netif()->hw_queues_count() == 1 ||
+ _inet._inet.netif()->hash2cpu(
+ id.hash(_inet._inet.netif()->rss_key())) == center->get_id())
+ break;
+ }
+ } while (true);
+
+ auto tcbp = make_lw_shared<tcb>(*this, id);
+ _tcbs.insert({id, tcbp});
+ tcbp->connect();
+ return connection(tcbp);
+}
+
+template <typename InetTraits>
+bool tcp<InetTraits>::forward(forward_hash& out_hash_data, Packet& p, size_t off) {
+ auto th = p.get_header<tcp_hdr>(off);
+ if (th) {
+ out_hash_data.push_back(th->src_port);
+ out_hash_data.push_back(th->dst_port);
+ }
+ return true;
+}
+
+template <typename InetTraits>
+void tcp<InetTraits>::received(Packet p, ipaddr from, ipaddr to) {
+ auto th = p.get_header<tcp_hdr>(0);
+ if (!th) {
+ return;
+ }
+ // th->data_offset is correct even before ntoh()
+ if (unsigned(th->data_offset * 4) < sizeof(*th)) {
+ return;
+ }
+
+ if (!get_hw_features().rx_csum_offload) {
+ checksummer csum;
+ InetTraits::tcp_pseudo_header_checksum(csum, from, to, p.len());
+ csum.sum(p);
+ if (csum.get() != 0) {
+ return;
+ }
+ }
+ auto h = th->ntoh();
+ auto id = connid{to, from, h.dst_port, h.src_port};
+ auto tcbi = _tcbs.find(id);
+ lw_shared_ptr<tcb> tcbp;
+ if (tcbi == _tcbs.end()) {
+ auto listener = _listening.find(id.local_port);
+ if (listener == _listening.end() || listener->second->full()) {
+ // 1) In CLOSE state
+ // 1.1 all data in the incoming segment is discarded. An incoming
+ // segment containing a RST is discarded. An incoming segment not
+ // containing a RST causes a RST to be sent in response.
+ // FIXME:
+ // if ACK off: <SEQ=0><ACK=SEG.SEQ+SEG.LEN><CTL=RST,ACK>
+ // if ACK on: <SEQ=SEG.ACK><CTL=RST>
+ return respond_with_reset(&h, id.local_ip, id.foreign_ip);
+ } else {
+ // 2) In LISTEN state
+ // 2.1 first check for an RST
+ if (h.f_rst) {
+ // An incoming RST should be ignored
+ return;
+ }
+ // 2.2 second check for an ACK
+ if (h.f_ack) {
+ // Any acknowledgment is bad if it arrives on a connection
+ // still in the LISTEN state.
+ // <SEQ=SEG.ACK><CTL=RST>
+ return respond_with_reset(&h, id.local_ip, id.foreign_ip);
+ }
+ // 2.3 third check for a SYN
+ if (h.f_syn) {
+ // check the security
+ // NOTE: Ignored for now
+ tcbp = make_lw_shared<tcb>(*this, id);
+ _tcbs.insert({id, tcbp});
+ return tcbp->input_handle_listen_state(&h, std::move(p));
+ }
+ // 2.4 fourth other text or control
+ // So you are unlikely to get here, but if you do, drop the
+ // segment, and return.
+ return;
+ }
+ } else {
+ tcbp = tcbi->second;
+ if (tcbp->state() == tcp_state::SYN_SENT) {
+ // 3) In SYN_SENT State
+ return tcbp->input_handle_syn_sent_state(&h, std::move(p));
+ } else {
+ // 4) In other state, can be one of the following:
+ // SYN_RECEIVED, ESTABLISHED, FIN_WAIT_1, FIN_WAIT_2
+ // CLOSE_WAIT, CLOSING, LAST_ACK, TIME_WAIT
+ return tcbp->input_handle_other_state(&h, std::move(p));
+ }
+ }
+}
+
+// Send packet does not belong to any tcb
+template <typename InetTraits>
+void tcp<InetTraits>::send_packet_without_tcb(ipaddr from, ipaddr to, Packet p) {
+ if (_queue_space.get_or_fail(p.len())) { // drop packets that do not fit the queue
+ _inet.wait_l2_dst_address(to, std::move(p), [this, to] (const ethernet_address &e_dst, Packet p, int r) mutable {
+ if (r == 0)
+ _packetq.emplace_back(ipv4_traits::l4packet{to, std::move(p), e_dst, ip_protocol_num::tcp});
+ });
+ }
+}
+
+template <typename InetTraits>
+tcp<InetTraits>::connection::~connection() {
+ if (_tcb) {
+ _tcb->_conn = nullptr;
+ close_read();
+ close_write();
+ }
+}
+
+template <typename InetTraits>
+tcp<InetTraits>::tcb::tcb(tcp& t, connid id)
+ : _tcp(t), manager(t.manager), _local_ip(id.local_ip) , _foreign_ip(id.foreign_ip),
+ _local_port(id.local_port), _foreign_port(id.foreign_port),
+ _snd(_tcp.cct),
+ center(t.center),
+ fd(t.manager.get_eventfd()),
+ delayed_ack_event(new tcp<InetTraits>::C_handle_delayed_ack(this)),
+ retransmit_event(new tcp<InetTraits>::C_handle_retransmit(this)),
+ persist_event(new tcp<InetTraits>::C_handle_persist(this)),
+ all_data_ack_event(new tcp<InetTraits>::C_all_data_acked(this)) {}
+
+template <typename InetTraits>
+tcp<InetTraits>::tcb::~tcb()
+{
+ if (_delayed_ack_fd)
+ center->delete_time_event(*_delayed_ack_fd);
+ if (retransmit_fd)
+ center->delete_time_event(*retransmit_fd);
+ if (persist_fd)
+ center->delete_time_event(*persist_fd);
+ delete delayed_ack_event;
+ delete retransmit_event;
+ delete persist_event;
+ delete all_data_ack_event;
+ manager.close(fd);
+ fd = -1;
+}
+
+template <typename InetTraits>
+void tcp<InetTraits>::tcb::respond_with_reset(tcp_hdr* rth)
+{
+ _tcp.respond_with_reset(rth, _local_ip, _foreign_ip);
+}
+
+template <typename InetTraits>
+uint32_t tcp<InetTraits>::tcb::data_segment_acked(tcp_sequence seg_ack) {
+ uint32_t total_acked_bytes = 0;
+ // Full ACK of segment
+ while (!_snd.data.empty()
+ && (_snd.unacknowledged + _snd.data.front().p.len() <= seg_ack)) {
+ auto acked_bytes = _snd.data.front().p.len();
+ _snd.unacknowledged += acked_bytes;
+ // Ignore retransmitted segments when setting the RTO
+ if (_snd.data.front().nr_transmits == 0) {
+ update_rto(_snd.data.front().tx_time);
+ }
+ update_cwnd(acked_bytes);
+ total_acked_bytes += acked_bytes;
+ _snd.user_queue_space.put(_snd.data.front().data_len);
+ manager.notify(fd, EVENT_WRITABLE);
+ _snd.data.pop_front();
+ }
+ // Partial ACK of segment
+ if (_snd.unacknowledged < seg_ack) {
+ auto acked_bytes = seg_ack - _snd.unacknowledged;
+ if (!_snd.data.empty()) {
+ auto& unacked_seg = _snd.data.front();
+ unacked_seg.p.trim_front(acked_bytes);
+ }
+ _snd.unacknowledged = seg_ack;
+ update_cwnd(acked_bytes);
+ total_acked_bytes += acked_bytes;
+ }
+ return total_acked_bytes;
+}
+
+template <typename InetTraits>
+bool tcp<InetTraits>::tcb::segment_acceptable(tcp_sequence seg_seq, unsigned seg_len) {
+ if (seg_len == 0 && _rcv.window == 0) {
+ // SEG.SEQ = RCV.NXT
+ return seg_seq == _rcv.next;
+ } else if (seg_len == 0 && _rcv.window > 0) {
+ // RCV.NXT =< SEG.SEQ < RCV.NXT+RCV.WND
+ return (_rcv.next <= seg_seq) && (seg_seq < _rcv.next + _rcv.window);
+ } else if (seg_len > 0 && _rcv.window > 0) {
+ // RCV.NXT =< SEG.SEQ < RCV.NXT+RCV.WND
+ // or
+ // RCV.NXT =< SEG.SEQ+SEG.LEN-1 < RCV.NXT+RCV.WND
+ bool x = (_rcv.next <= seg_seq) && seg_seq < (_rcv.next + _rcv.window);
+ bool y = (_rcv.next <= seg_seq + seg_len - 1) && (seg_seq + seg_len - 1 < _rcv.next + _rcv.window);
+ return x || y;
+ } else {
+ // SEG.LEN > 0 RCV.WND = 0, not acceptable
+ return false;
+ }
+}
+
+template <typename InetTraits>
+void tcp<InetTraits>::tcb::init_from_options(tcp_hdr* th, uint8_t* opt_start, uint8_t* opt_end) {
+ // Handle tcp options
+ _option.parse(opt_start, opt_end);
+
+ // Remote receive window scale factor
+ _snd.window_scale = _option._remote_win_scale;
+ // Local receive window scale factor
+ _rcv.window_scale = _option._local_win_scale;
+
+ // Maximum segment size remote can receive
+ _snd.mss = _option._remote_mss;
+ // Maximum segment size local can receive
+ _rcv.mss = _option._local_mss = local_mss();
+
+ // Linux's default window size
+ _rcv.window = 29200 << _rcv.window_scale;
+ _snd.window = th->window << _snd.window_scale;
+
+ // Segment sequence number used for last window update
+ _snd.wl1 = th->seq;
+ // Segment acknowledgment number used for last window update
+ _snd.wl2 = th->ack;
+
+ // Setup initial congestion window
+ if (2190 < _snd.mss) {
+ _snd.cwnd = 2 * _snd.mss;
+ } else if (1095 < _snd.mss && _snd.mss <= 2190) {
+ _snd.cwnd = 3 * _snd.mss;
+ } else {
+ _snd.cwnd = 4 * _snd.mss;
+ }
+
+ // Setup initial slow start threshold
+ _snd.ssthresh = th->window << _snd.window_scale;
+}
+
+template <typename InetTraits>
+Packet tcp<InetTraits>::tcb::get_transmit_packet() {
+ // easy case: empty queue
+ if (_snd.unsent.empty()) {
+ return Packet();
+ }
+ auto can_send = this->can_send();
+ // Max number of TCP payloads we can pass to NIC
+ uint32_t len;
+ if (_tcp.get_hw_features().tx_tso) {
+ // FIXME: Info tap device the size of the splitted packet
+ len = _tcp.get_hw_features().max_packet_len - tcp_hdr_len_min - InetTraits::ip_hdr_len_min;
+ } else {
+ len = std::min(uint16_t(_tcp.get_hw_features().mtu - tcp_hdr_len_min - InetTraits::ip_hdr_len_min), _snd.mss);
+ }
+ can_send = std::min(can_send, len);
+ // easy case: one small packet
+ if (_snd.unsent.front().len() <= can_send) {
+ auto p = std::move(_snd.unsent.front());
+ _snd.unsent.pop_front();
+ _snd.unsent_len -= p.len();
+ return p;
+ }
+ // moderate case: need to split one packet
+ if (_snd.unsent.front().len() > can_send) {
+ auto p = _snd.unsent.front().share(0, can_send);
+ _snd.unsent.front().trim_front(can_send);
+ _snd.unsent_len -= p.len();
+ return p;
+ }
+ // hard case: merge some packets, possibly split last
+ auto p = std::move(_snd.unsent.front());
+ _snd.unsent.pop_front();
+ can_send -= p.len();
+ while (!_snd.unsent.empty()
+ && _snd.unsent.front().len() <= can_send) {
+ can_send -= _snd.unsent.front().len();
+ p.append(std::move(_snd.unsent.front()));
+ _snd.unsent.pop_front();
+ }
+ // FIXME: this will result in calling "deleter" of packet which free managed objects
+ // will used later
+ // if (!_snd.unsent.empty() && can_send) {
+ // auto& q = _snd.unsent.front();
+ // p.append(q.share(0, can_send));
+ // q.trim_front(can_send);
+ // }
+ _snd.unsent_len -= p.len();
+ return p;
+}
+
+template <typename InetTraits>
+void tcp<InetTraits>::tcb::output_one(bool data_retransmit) {
+ if (in_state(CLOSED)) {
+ return;
+ }
+
+ Packet p = data_retransmit ? _snd.data.front().p.share() : get_transmit_packet();
+ Packet clone = p.share(); // early clone to prevent share() from calling packet::unuse_internal_data() on header.
+ uint16_t len = p.len();
+ bool syn_on = syn_needs_on();
+ bool ack_on = ack_needs_on();
+
+ auto options_size = _option.get_size(syn_on, ack_on);
+ auto th = p.prepend_header<tcp_hdr>(options_size);
+
+ th->src_port = _local_port;
+ th->dst_port = _foreign_port;
+
+ th->f_syn = syn_on;
+ th->f_ack = ack_on;
+ if (ack_on) {
+ clear_delayed_ack();
+ }
+ th->f_urg = false;
+ th->f_psh = false;
+
+ tcp_sequence seq;
+ if (data_retransmit) {
+ seq = _snd.unacknowledged;
+ } else {
+ seq = syn_on ? _snd.initial : _snd.next;
+ _snd.next += len;
+ }
+ th->seq = seq;
+ th->ack = _rcv.next;
+ th->data_offset = (sizeof(*th) + options_size) / 4;
+ th->window = _rcv.window >> _rcv.window_scale;
+ th->checksum = 0;
+
+ // FIXME: does the FIN have to fit in the window?
+ bool fin_on = fin_needs_on();
+ th->f_fin = fin_on;
+
+ // Add tcp options
+ _option.fill(th, options_size);
+ *th = th->hton();
+
+ offload_info oi;
+ checksummer csum;
+ uint16_t pseudo_hdr_seg_len = 0;
+
+ oi.tcp_hdr_len = sizeof(tcp_hdr) + options_size;
+
+ if (_tcp.get_hw_features().tx_csum_l4_offload) {
+ oi.needs_csum = true;
+
+ //
+ // tx checksum offloading: both virtio-net's VIRTIO_NET_F_CSUM dpdk's
+ // PKT_TX_TCP_CKSUM - requires th->checksum to be initialized to ones'
+ // complement sum of the pseudo header.
+ //
+ // For TSO the csum should be calculated for a pseudo header with
+ // segment length set to 0. All the rest is the same as for a TCP Tx
+ // CSUM offload case.
+ //
+ if (_tcp.get_hw_features().tx_tso && len > _snd.mss) {
+ oi.tso_seg_size = _snd.mss;
+ } else {
+ pseudo_hdr_seg_len = sizeof(*th) + options_size + len;
+ }
+ } else {
+ pseudo_hdr_seg_len = sizeof(*th) + options_size + len;
+ oi.needs_csum = false;
+ }
+
+ InetTraits::tcp_pseudo_header_checksum(csum, _local_ip, _foreign_ip,
+ pseudo_hdr_seg_len);
+
+ if (_tcp.get_hw_features().tx_csum_l4_offload) {
+ th->checksum = ~csum.get();
+ } else {
+ csum.sum(p);
+ th->checksum = csum.get();
+ }
+
+ oi.protocol = ip_protocol_num::tcp;
+
+ p.set_offload_info(oi);
+
+ if (!data_retransmit && (len || syn_on || fin_on)) {
+ auto now = clock_type::now();
+ if (len) {
+ unsigned nr_transmits = 0;
+ _snd.data.emplace_back(unacked_segment{std::move(clone),
+ len, nr_transmits, now});
+ }
+ if (!retransmit_fd) {
+ start_retransmit_timer();
+ }
+ }
+
+ queue_packet(std::move(p));
+}
+
+template <typename InetTraits>
+bool tcp<InetTraits>::tcb::is_all_data_acked() {
+ if (_snd.data.empty() && _snd.unsent_len == 0 && _snd.queued_len == 0) {
+ return true;
+ }
+ return false;
+}
+
+template <typename InetTraits>
+Tub<Packet> tcp<InetTraits>::tcb::read() {
+ Tub<Packet> p;
+ if (_rcv.data.empty())
+ return p;
+
+ p.construct();
+ for (auto&& q : _rcv.data) {
+ p->append(std::move(q));
+ }
+ _rcv.data.clear();
+ return p;
+}
+
+template <typename InetTraits>
+int tcp<InetTraits>::tcb::send(Packet p) {
+ // We can not send after the connection is closed
+ assert(!_snd.closed);
+
+ if (in_state(CLOSED))
+ return -ECONNRESET;
+
+ auto len = p.len();
+ if (!_snd.user_queue_space.get_or_fail(len)) {
+ // note: caller must ensure enough queue space to send
+ ceph_abort();
+ }
+ // TODO: Handle p.len() > max user_queue_space case
+ _snd.queued_len += len;
+ _snd.unsent_len += len;
+ _snd.queued_len -= len;
+ _snd.unsent.push_back(std::move(p));
+ if (can_send() > 0) {
+ output();
+ }
+ return len;
+}
+
+template <typename InetTraits>
+void tcp<InetTraits>::tcb::close() {
+ if (in_state(CLOSED) || _snd.closed) {
+ return ;
+ }
+ // TODO: We should make this asynchronous
+
+ _errno = -EPIPE;
+ center->delete_file_event(fd, EVENT_READABLE|EVENT_WRITABLE);
+ bool acked = is_all_data_acked();
+ if (!acked) {
+ _snd._all_data_acked_fd = manager.get_eventfd();
+ center->create_file_event(_snd._all_data_acked_fd, EVENT_READABLE, all_data_ack_event);
+ } else {
+ close_final_cleanup();
+ }
+}
+
+template <typename InetTraits>
+bool tcp<InetTraits>::tcb::should_send_ack(uint16_t seg_len) {
+ // We've received a TSO packet, do ack immediately
+ if (seg_len > _rcv.mss) {
+ _nr_full_seg_received = 0;
+ if (_delayed_ack_fd) {
+ center->delete_time_event(*_delayed_ack_fd);
+ _delayed_ack_fd.destroy();
+ }
+ return true;
+ }
+
+ // We've received a full sized segment, ack for every second full sized segment
+ if (seg_len == _rcv.mss) {
+ if (_nr_full_seg_received++ >= 1) {
+ _nr_full_seg_received = 0;
+ if (_delayed_ack_fd) {
+ center->delete_time_event(*_delayed_ack_fd);
+ _delayed_ack_fd.destroy();
+ }
+ return true;
+ }
+ }
+
+ // If the timer is armed and its callback hasn't been run.
+ if (_delayed_ack_fd) {
+ return false;
+ }
+
+ // If the timer is not armed, schedule a delayed ACK.
+ // The maximum delayed ack timer allowed by RFC1122 is 500ms, most
+ // implementations use 200ms.
+ _delayed_ack_fd.construct(center->create_time_event(200*1000, delayed_ack_event));
+ return false;
+}
+
+template <typename InetTraits>
+void tcp<InetTraits>::tcb::clear_delayed_ack() {
+ if (_delayed_ack_fd) {
+ center->delete_time_event(*_delayed_ack_fd);
+ _delayed_ack_fd.destroy();
+ }
+}
+
+template <typename InetTraits>
+bool tcp<InetTraits>::tcb::merge_out_of_order() {
+ bool merged = false;
+ if (_rcv.out_of_order.map.empty()) {
+ return merged;
+ }
+ for (auto it = _rcv.out_of_order.map.begin(); it != _rcv.out_of_order.map.end();) {
+ auto& p = it->second;
+ auto seg_beg = it->first;
+ auto seg_len = p.len();
+ auto seg_end = seg_beg + seg_len;
+ if (seg_beg <= _rcv.next && seg_end > _rcv.next) {
+ // This segment has been received out of order and its previous
+ // segment has been received now
+ auto trim = _rcv.next - seg_beg;
+ if (trim) {
+ p.trim_front(trim);
+ seg_len -= trim;
+ }
+ _rcv.next += seg_len;
+ _rcv.data.push_back(std::move(p));
+ // Since c++11, erase() always returns the value of the following element
+ it = _rcv.out_of_order.map.erase(it);
+ merged = true;
+ } else if (_rcv.next >= seg_end) {
+ // This segment has been receive already, drop it
+ it = _rcv.out_of_order.map.erase(it);
+ } else {
+ // seg_beg > _rcv.need, can not merge. Note, seg_beg can grow only,
+ // so we can stop looking here.
+ it++;
+ break;
+ }
+ }
+ return merged;
+}
+
+template <typename InetTraits>
+void tcp<InetTraits>::tcb::insert_out_of_order(tcp_sequence seg, Packet p) {
+ _rcv.out_of_order.merge(seg, std::move(p));
+}
+
+template <typename InetTraits>
+void tcp<InetTraits>::tcb::trim_receive_data_after_window() {
+ abort();
+}
+
+template <typename InetTraits>
+void tcp<InetTraits>::tcb::fast_retransmit() {
+ if (!_snd.data.empty()) {
+ auto& unacked_seg = _snd.data.front();
+ unacked_seg.nr_transmits++;
+ retransmit_one();
+ output();
+ }
+}
+
+template <typename InetTraits>
+void tcp<InetTraits>::tcb::update_rto(clock_type::time_point tx_time) {
+ // Update RTO according to RFC6298
+ auto R = std::chrono::duration_cast<std::chrono::microseconds>(clock_type::now() - tx_time);
+ if (_snd.first_rto_sample) {
+ _snd.first_rto_sample = false;
+ // RTTVAR <- R/2
+ // SRTT <- R
+ _snd.rttvar = R / 2;
+ _snd.srtt = R;
+ } else {
+ // RTTVAR <- (1 - beta) * RTTVAR + beta * |SRTT - R'|
+ // SRTT <- (1 - alpha) * SRTT + alpha * R'
+ // where alpha = 1/8 and beta = 1/4
+ auto delta = _snd.srtt > R ? (_snd.srtt - R) : (R - _snd.srtt);
+ _snd.rttvar = _snd.rttvar * 3 / 4 + delta / 4;
+ _snd.srtt = _snd.srtt * 7 / 8 + R / 8;
+ }
+ // RTO <- SRTT + max(G, K * RTTVAR)
+ _rto = _snd.srtt + std::max(_rto_clk_granularity, 4 * _snd.rttvar);
+
+ // Make sure 1 sec << _rto << 60 sec
+ _rto = std::max(_rto, _rto_min);
+ _rto = std::min(_rto, _rto_max);
+}
+
+template <typename InetTraits>
+void tcp<InetTraits>::tcb::update_cwnd(uint32_t acked_bytes) {
+ uint32_t smss = _snd.mss;
+ if (_snd.cwnd < _snd.ssthresh) {
+ // In slow start phase
+ _snd.cwnd += std::min(acked_bytes, smss);
+ } else {
+ // In congestion avoidance phase
+ uint32_t round_up = 1;
+ _snd.cwnd += std::max(round_up, smss * smss / _snd.cwnd);
+ }
+}
+
+
+template <typename InetTraits>
+void tcp<InetTraits>::tcb::cleanup() {
+ manager.notify(fd, EVENT_READABLE);
+ _snd.closed = true;
+ _snd.unsent.clear();
+ _snd.data.clear();
+ _rcv.out_of_order.map.clear();
+ _rcv.data.clear();
+ stop_retransmit_timer();
+ clear_delayed_ack();
+ center->dispatch_event_external(new tcp<InetTraits>::C_actual_remove_tcb(this));
+ remove_from_tcbs();
+}
+
+template <typename InetTraits>
+tcp_sequence tcp<InetTraits>::tcb::get_isn() {
+ // Per RFC6528, TCP SHOULD generate its Initial Sequence Numbers
+ // with the expression:
+ // ISN = M + F(localip, localport, remoteip, remoteport, secretkey)
+ // M is the 4 microsecond timer
+ using namespace std::chrono;
+ uint32_t hash[4];
+ hash[0] = _local_ip.ip;
+ hash[1] = _foreign_ip.ip;
+ hash[2] = (_local_port << 16) + _foreign_port;
+ hash[3] = _isn_secret.key[15];
+ CryptoPP::Weak::MD5::Transform(hash, _isn_secret.key);
+ auto seq = hash[0];
+ auto m = duration_cast<microseconds>(clock_type::now().time_since_epoch());
+ seq += m.count() / 4;
+ return make_seq(seq);
+}
+
+template <typename InetTraits>
+Tub<typename InetTraits::l4packet> tcp<InetTraits>::tcb::get_packet() {
+ _poll_active = false;
+ if (_packetq.empty()) {
+ output_one();
+ }
+
+ Tub<typename InetTraits::l4packet> p;
+ if (in_state(CLOSED)) {
+ return p;
+ }
+
+ assert(!_packetq.empty());
+
+ p = std::move(_packetq.front());
+ _packetq.pop_front();
+ if (!_packetq.empty() || (_snd.dupacks < 3 && can_send() > 0)) {
+ // If there are packets to send in the queue or tcb is allowed to send
+ // more add tcp back to polling set to keep sending. In addition, dupacks >= 3
+ // is an indication that an segment is lost, stop sending more in this case.
+ output();
+ }
+ return p;
+}
+
+template <typename InetTraits>
+void tcp<InetTraits>::connection::close_read() {
+ // do nothing
+ // _tcb->manager.notify(_tcb->fd, EVENT_READABLE);
+}
+
+template <typename InetTraits>
+void tcp<InetTraits>::connection::close_write() {
+ _tcb->close();
+}
+
+template <typename InetTraits>
+constexpr uint16_t tcp<InetTraits>::tcb::_max_nr_retransmit;
+
+template <typename InetTraits>
+constexpr std::chrono::microseconds tcp<InetTraits>::tcb::_rto_min;
+
+template <typename InetTraits>
+constexpr std::chrono::microseconds tcp<InetTraits>::tcb::_rto_max;
+
+template <typename InetTraits>
+constexpr std::chrono::microseconds tcp<InetTraits>::tcb::_rto_clk_granularity;
+
+template <typename InetTraits>
+typename tcp<InetTraits>::tcb::isn_secret tcp<InetTraits>::tcb::_isn_secret;
+
+
+#endif /* TCP_HH_ */
Code Review / stor4nfv.git / blobdiff