ceph/src/seastar/src/net/udp.cc

   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 #include <seastar/net/ip.hh>
  23 #include <seastar/net/stack.hh>
  24
  25 namespace seastar {
  26
  27 using namespace net;
  28
  29 namespace net {
  30 namespace ipv4_udp_impl {
  31
  32 static inline
  33 ipv4_addr
  34 to_ipv4_addr(ipv4_address a, uint16_t port) {
  35     return {a.ip, port};
  36 }
  37
  38 class native_datagram : public udp_datagram_impl {
  39 private:
  40     ipv4_addr _src;
  41     ipv4_addr _dst;
  42     packet _p;
  43 public:
  44     native_datagram(ipv4_address src, ipv4_address dst, packet p)
  45             : _p(std::move(p)) {
  46         udp_hdr* hdr = _p.get_header<udp_hdr>();
  47         auto h = ntoh(*hdr);
  48         _p.trim_front(sizeof(*hdr));
  49         _src = to_ipv4_addr(src, h.src_port);
  50         _dst = to_ipv4_addr(dst, h.dst_port);
  51     }
  52
  53     virtual ipv4_addr get_src() override {
  54         return _src;
  55     };
  56
  57     virtual ipv4_addr get_dst() override {
  58         return _dst;
  59     };
  60
  61     virtual uint16_t get_dst_port() override {
  62         return _dst.port;
  63     }
  64
  65     virtual packet& get_data() override {
  66         return _p;
  67     }
  68 };
  69
  70 class native_channel : public udp_channel_impl {
  71 private:
  72     ipv4_udp& _proto;
  73     ipv4_udp::registration _reg;
  74     bool _closed;
  75     lw_shared_ptr<udp_channel_state> _state;
  76
  77 public:
  78     native_channel(ipv4_udp &proto, ipv4_udp::registration reg, lw_shared_ptr<udp_channel_state> state)
  79             : _proto(proto)
  80             , _reg(reg)
  81             , _closed(false)
  82             , _state(state)
  83     {
  84     }
  85
  86     ~native_channel()
  87     {
  88         if (!_closed)
  89             close();
  90     }
  91
  92     virtual future<udp_datagram> receive() override {
  93         return _state->_queue.pop_eventually();
  94     }
  95
  96     virtual future<> send(ipv4_addr dst, const char* msg) override {
  97         return send(dst, packet::from_static_data(msg, strlen(msg)));
  98     }
  99
 100     virtual future<> send(ipv4_addr dst, packet p) override {
 101         auto len = p.len();
 102         return _state->wait_for_send_buffer(len).then([this, dst, p = std::move(p), len] () mutable {
 103             p = packet(std::move(p), make_deleter([s = _state, len] { s->complete_send(len); }));
 104             _proto.send(_reg.port(), dst, std::move(p));
 105         });
 106     }
 107
 108     virtual bool is_closed() const override {
 109         return _closed;
 110     }
 111
 112     virtual void shutdown_input() override {
 113         _state->_queue.abort(std::make_exception_ptr(std::system_error(EBADF, std::system_category())));
 114     }
 115
 116     virtual void shutdown_output() override {
 117         _state->_queue.abort(std::make_exception_ptr(std::system_error(EPIPE, std::system_category())));
 118     }
 119
 120     virtual void close() override {
 121         _reg.unregister();
 122         _closed = true;
 123     }
 124 };
 125
 126 } /* namespace ipv4_udp_impl */
 127
 128 using namespace net::ipv4_udp_impl;
 129
 130 const int ipv4_udp::default_queue_size = 1024;
 131
 132 ipv4_udp::ipv4_udp(ipv4& inet)
 133     : _inet(inet)
 134 {
 135     _inet.register_packet_provider([this] {
 136         compat::optional<ipv4_traits::l4packet> l4p;
 137         if (!_packetq.empty()) {
 138             l4p = std::move(_packetq.front());
 139             _packetq.pop_front();
 140         }
 141         return l4p;
 142     });
 143 }
 144
 145 bool ipv4_udp::forward(forward_hash& out_hash_data, packet& p, size_t off)
 146 {
 147     auto uh = p.get_header<udp_hdr>(off);
 148
 149     if (uh) {
 150         out_hash_data.push_back(uh->src_port);
 151         out_hash_data.push_back(uh->dst_port);
 152     }
 153     return true;
 154 }
 155
 156 void ipv4_udp::received(packet p, ipv4_address from, ipv4_address to)
 157 {
 158     udp_datagram dgram(std::make_unique<native_datagram>(from, to, std::move(p)));
 159
 160     auto chan_it = _channels.find(dgram.get_dst_port());
 161     if (chan_it != _channels.end()) {
 162         auto chan = chan_it->second;
 163         chan->_queue.push(std::move(dgram));
 164     }
 165 }
 166
 167 void ipv4_udp::send(uint16_t src_port, ipv4_addr dst, packet &&p)
 168 {
 169     auto src = _inet.host_address();
 170     auto hdr = p.prepend_header<udp_hdr>();
 171     hdr->src_port = src_port;
 172     hdr->dst_port = dst.port;
 173     hdr->len = p.len();
 174     *hdr = hton(*hdr);
 175
 176     offload_info oi;
 177     checksummer csum;
 178     ipv4_traits::udp_pseudo_header_checksum(csum, src, dst, p.len());
 179     bool needs_frag = ipv4::needs_frag(p, ip_protocol_num::udp, _inet.hw_features());
 180     if (_inet.hw_features().tx_csum_l4_offload && !needs_frag) {
 181         hdr->cksum = ~csum.get();
 182         oi.needs_csum = true;
 183     } else {
 184         csum.sum(p);
 185         hdr->cksum = csum.get();
 186         oi.needs_csum = false;
 187     }
 188     oi.protocol = ip_protocol_num::udp;
 189     p.set_offload_info(oi);
 190
 191     _inet.get_l2_dst_address(dst).then([this, dst, p = std::move(p)] (ethernet_address e_dst) mutable {
 192         _packetq.emplace_back(ipv4_traits::l4packet{dst, std::move(p), e_dst, ip_protocol_num::udp});
 193     });
 194 }
 195
 196 uint16_t ipv4_udp::next_port(uint16_t port) {
 197     return (port + 1) == 0 ? min_anonymous_port : port + 1;
 198 }
 199
 200 udp_channel
 201 ipv4_udp::make_channel(ipv4_addr addr) {
 202     if (!is_ip_unspecified(addr)) {
 203         throw std::runtime_error("Binding to specific IP not supported yet");
 204     }
 205
 206     uint16_t bind_port;
 207
 208     if (!is_port_unspecified(addr)) {
 209         if (_channels.count(addr.port)) {
 210             throw std::runtime_error("Address already in use");
 211         }
 212         bind_port = addr.port;
 213     } else {
 214         auto starting_port = _next_anonymous_port;
 215         while (_channels.count(_next_anonymous_port)) {
 216             _next_anonymous_port = next_port(_next_anonymous_port);
 217             if (starting_port == _next_anonymous_port) {
 218                 throw std::runtime_error("No free port");
 219             }
 220         }
 221
 222         bind_port = _next_anonymous_port;
 223         _next_anonymous_port = next_port(_next_anonymous_port);
 224     }
 225
 226     auto chan_state = make_lw_shared<udp_channel_state>(_queue_size);
 227     _channels[bind_port] = chan_state;
 228     return udp_channel(std::make_unique<native_channel>(*this, registration(*this, bind_port), chan_state));
 229 }
 230
 231 } /* namespace net */
 232
 233 }
 234