ceph/src/seastar/src/net/native-stack-impl.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 #include <seastar/core/reactor.hh>
  25 #include <seastar/net/stack.hh>
  26 #include <iostream>
  27
  28 namespace seastar {
  29
  30 namespace net {
  31
  32 using namespace seastar;
  33
  34 template <typename Protocol>
  35 class native_server_socket_impl;
  36
  37 template <typename Protocol>
  38 class native_connected_socket_impl;
  39
  40 class native_network_stack;
  41
  42 // native_server_socket_impl
  43 template <typename Protocol>
  44 class native_server_socket_impl : public server_socket_impl {
  45     typename Protocol::listener _listener;
  46 public:
  47     native_server_socket_impl(Protocol& proto, uint16_t port, listen_options opt);
  48     virtual future<connected_socket, socket_address> accept() override;
  49     virtual void abort_accept() override;
  50 };
  51
  52 template <typename Protocol>
  53 native_server_socket_impl<Protocol>::native_server_socket_impl(Protocol& proto, uint16_t port, listen_options opt)
  54     : _listener(proto.listen(port)) {
  55 }
  56
  57 template <typename Protocol>
  58 future<connected_socket, socket_address>
  59 native_server_socket_impl<Protocol>::accept() {
  60     return _listener.accept().then([] (typename Protocol::connection conn) {
  61         return make_ready_future<connected_socket, socket_address>(
  62                 connected_socket(std::make_unique<native_connected_socket_impl<Protocol>>(make_lw_shared(std::move(conn)))),
  63                 make_ipv4_address(conn.foreign_ip().ip, conn.foreign_port()));
  64     });
  65 }
  66
  67 template <typename Protocol>
  68 void
  69 native_server_socket_impl<Protocol>::abort_accept() {
  70     _listener.abort_accept();
  71 }
  72
  73 // native_connected_socket_impl
  74 template <typename Protocol>
  75 class native_connected_socket_impl : public connected_socket_impl {
  76     lw_shared_ptr<typename Protocol::connection> _conn;
  77     class native_data_source_impl;
  78     class native_data_sink_impl;
  79 public:
  80     explicit native_connected_socket_impl(lw_shared_ptr<typename Protocol::connection> conn)
  81         : _conn(std::move(conn)) {}
  82     virtual data_source source() override;
  83     virtual data_sink sink() override;
  84     virtual void shutdown_input() override;
  85     virtual void shutdown_output() override;
  86     virtual void set_nodelay(bool nodelay) override;
  87     virtual bool get_nodelay() const override;
  88     void set_keepalive(bool keepalive) override;
  89     bool get_keepalive() const override;
  90     void set_keepalive_parameters(const keepalive_params&) override;
  91     keepalive_params get_keepalive_parameters() const override;
  92 };
  93
  94 template <typename Protocol>
  95 class native_socket_impl final : public socket_impl {
  96     Protocol& _proto;
  97     lw_shared_ptr<typename Protocol::connection> _conn;
  98 public:
  99     explicit native_socket_impl(Protocol& proto)
 100         : _proto(proto), _conn(nullptr) { }
 101
 102     virtual future<connected_socket> connect(socket_address sa, socket_address local, transport proto = transport::TCP) override {
 103         //TODO: implement SCTP
 104         assert(proto == transport::TCP);
 105
 106         // FIXME: local is ignored since native stack does not support multiple IPs yet
 107         assert(sa.as_posix_sockaddr().sa_family == AF_INET);
 108
 109         _conn = make_lw_shared<typename Protocol::connection>(_proto.connect(sa));
 110         return _conn->connected().then([conn = _conn]() mutable {
 111             auto csi = std::make_unique<native_connected_socket_impl<Protocol>>(std::move(conn));
 112             return make_ready_future<connected_socket>(connected_socket(std::move(csi)));
 113         });
 114     }
 115
 116     virtual void shutdown() override {
 117         if (_conn) {
 118             _conn->shutdown_connect();
 119         }
 120     }
 121 };
 122
 123 template <typename Protocol>
 124 class native_connected_socket_impl<Protocol>::native_data_source_impl final
 125     : public data_source_impl {
 126     typedef typename Protocol::connection connection_type;
 127     lw_shared_ptr<connection_type> _conn;
 128     size_t _cur_frag = 0;
 129     bool _eof = false;
 130     packet _buf;
 131 public:
 132     explicit native_data_source_impl(lw_shared_ptr<connection_type> conn)
 133         : _conn(std::move(conn)) {}
 134     virtual future<temporary_buffer<char>> get() override {
 135         if (_eof) {
 136             return make_ready_future<temporary_buffer<char>>(temporary_buffer<char>(0));
 137         }
 138         if (_cur_frag != _buf.nr_frags()) {
 139             auto& f = _buf.fragments()[_cur_frag++];
 140             return make_ready_future<temporary_buffer<char>>(
 141                     temporary_buffer<char>(f.base, f.size,
 142                             make_deleter(deleter(), [p = _buf.share()] () mutable {})));
 143         }
 144         return _conn->wait_for_data().then([this] {
 145             _buf = _conn->read();
 146             _cur_frag = 0;
 147             _eof = !_buf.len();
 148             return get();
 149         });
 150     }
 151     future<> close() override {
 152         _conn->close_write();
 153         return make_ready_future<>();
 154     }
 155 };
 156
 157 template <typename Protocol>
 158 class native_connected_socket_impl<Protocol>::native_data_sink_impl final
 159     : public data_sink_impl {
 160     typedef typename Protocol::connection connection_type;
 161     lw_shared_ptr<connection_type> _conn;
 162 public:
 163     explicit native_data_sink_impl(lw_shared_ptr<connection_type> conn)
 164         : _conn(std::move(conn)) {}
 165     using data_sink_impl::put;
 166     virtual future<> put(packet p) override {
 167         return _conn->send(std::move(p));
 168     }
 169     virtual future<> close() override {
 170         _conn->close_write();
 171         return make_ready_future<>();
 172     }
 173 };
 174
 175 template <typename Protocol>
 176 data_source native_connected_socket_impl<Protocol>::source() {
 177     return data_source(std::make_unique<native_data_source_impl>(_conn));
 178 }
 179
 180 template <typename Protocol>
 181 data_sink native_connected_socket_impl<Protocol>::sink() {
 182     return data_sink(std::make_unique<native_data_sink_impl>(_conn));
 183 }
 184
 185 template <typename Protocol>
 186 void
 187 native_connected_socket_impl<Protocol>::shutdown_input() {
 188     _conn->close_read();
 189 }
 190
 191 template <typename Protocol>
 192 void
 193 native_connected_socket_impl<Protocol>::shutdown_output() {
 194     _conn->close_write();
 195 }
 196
 197 template <typename Protocol>
 198 void
 199 native_connected_socket_impl<Protocol>::set_nodelay(bool nodelay) {
 200     // FIXME: implement
 201 }
 202
 203 template <typename Protocol>
 204 bool
 205 native_connected_socket_impl<Protocol>::get_nodelay() const {
 206     // FIXME: implement
 207     return true;
 208 }
 209
 210 template <typename Protocol>
 211 void native_connected_socket_impl<Protocol>::set_keepalive(bool keepalive) {
 212     // FIXME: implement
 213     std::cerr << "Keepalive is not supported by native stack" << std::endl;
 214 }
 215 template <typename Protocol>
 216 bool native_connected_socket_impl<Protocol>::get_keepalive() const {
 217     // FIXME: implement
 218     return false;
 219 }
 220
 221 template <typename Protocol>
 222 void native_connected_socket_impl<Protocol>::set_keepalive_parameters(const keepalive_params&) {
 223     // FIXME: implement
 224     std::cerr << "Keepalive parameters are not supported by native stack" << std::endl;
 225 }
 226
 227 template <typename Protocol>
 228 keepalive_params native_connected_socket_impl<Protocol>::get_keepalive_parameters() const {
 229     // FIXME: implement
 230     return tcp_keepalive_params {std::chrono::seconds(0), std::chrono::seconds(0), 0};
 231 }
 232
 233 }
 234
 235 }