Line data    Source code

       1             : /*
       2             :  * Copyright (c) 2013 Juniper Networks, Inc. All rights reserved.
       3             :  * Copyright Contributors to the OpenSDN project
       4             :  */
       5             : 
       6             : #ifndef SRC_IO_TCP_SESSION_H_
       7             : #define SRC_IO_TCP_SESSION_H_
       8             : 
       9             : #include <deque>
      10             : #include <list>
      11             : #include <string>
      12             : #include <mutex>
      13             : #include <atomic>
      14             : 
      15             : #include <boost/asio/buffer.hpp>
      16             : #include <boost/asio/io_service.hpp>
      17             : #include <boost/asio/ip/tcp.hpp>
      18             : #include <boost/asio/strand.hpp>
      19             : #include <boost/intrusive_ptr.hpp>
      20             : #include <boost/function.hpp>
      21             : #include <boost/scoped_ptr.hpp>
      22             : 
      23             : #include "base/util.h"
      24             : #include "base/task.h"
      25             : #include "io/tcp_server.h"
      26             : 
      27             : #define SSL_SHORT_READ_ERROR 335544539
      28             : 
      29             : class EventManager;
      30             : class TcpServer;
      31             : class TcpSession;
      32             : class TcpMessageWriter;
      33             : 
      34             : // TcpSession
      35             : //
      36             : // Concurrency: the session is created by the event manager thread, which
      37             : // also invokes the AsyncHandlers. ReleaseBuffer and Send will typically be
      38             : // invoked by a different thread.
      39             : class TcpSession {
      40             : public:
      41             :     static const int kDefaultBufferSize = 16 * 1024;
      42             :     static const int kDefaultWriteBufferSize = 32 * 1024;
      43             : 
      44             :     enum Event {
      45             :         EVENT_NONE,
      46             :         ACCEPT,
      47             :         CONNECT_COMPLETE,
      48             :         CONNECT_FAILED,
      49             :         CLOSE
      50             :     };
      51             : 
      52             :     enum Direction {
      53             :         ACTIVE,
      54             :         PASSIVE
      55             :     };
      56             : 
      57             :     typedef boost::asio::ip::tcp::socket Socket;
      58             :     typedef boost::asio::ip::tcp::socket::native_handle_type NativeSocketType;
      59             :     typedef boost::asio::ip::tcp::endpoint Endpoint;
      60             :     typedef boost::function<void(TcpSession *, Event)> EventObserver;
      61             :     typedef boost::asio::const_buffer Buffer;
      62             : 
      63             :     // TcpSession constructor takes ownership of socket.
      64             :     TcpSession(TcpServer *server, Socket *socket,
      65             :                bool async_read_ready = true,
      66             :                size_t buffer_send_size = TcpSession::kDefaultWriteBufferSize);
      67             :     // Performs a non-blocking send operation.
      68             :     virtual bool Send(const uint8_t *data, size_t size, size_t *sent);
      69             : 
      70             :     // Called by TcpServer to trigger async read.
      71             :     virtual bool Connected(Endpoint remote);
      72             : 
      73             :     // Called by TcpServer to trigger async read.
      74             :     virtual void Accepted();
      75             : 
      76             :     void ConnectFailed();
      77             : 
      78             :     void Close();
      79             : 
      80      107313 :     virtual std::string ToString() const { return name_; }
      81             : 
      82             :     // Getters and setters
      83      795217 :     virtual Socket *socket() const { return socket_.get(); }
      84             :     NativeSocketType sock_descriptor() { return socket_->native_handle(); }
      85      211025 :     TcpServer *server() { return server_.get(); }
      86             :     int32_t local_port() const;
      87             :     int32_t remote_port() const;
      88             : 
      89             :     // Concurrency: changing the observer guarantees mutual exclusion with
      90             :     // the observer invocation. e.g. if the caller sets the observer to NULL
      91             :     // it is guaranteed that the observer will not get invoked after this
      92             :     // method returns.
      93             :     void set_observer(EventObserver observer);
      94             : 
      95             :     // Buffers must be freed in arrival order.
      96             :     virtual void ReleaseBuffer(Buffer buffer);
      97             : 
      98             :     // This function returns the instance to run SessionTask.
      99             :     // Returning Task::kTaskInstanceAny would allow multiple session tasks to
     100             :     // run in parallel.
     101             :     // Derived class may override implementation if it expects the all the
     102             :     // Tasks of the session to run in specific instance
     103             :     // Note: Two tasks of same task ID and task instance can't run
     104             :     //       at in parallel
     105             :     // E.g. BgpSession is created per BgpPeer and to ensure that
     106             :     //      there is one SessionTask per peer, PeerIndex is returned
     107             :     //      from this function
     108             :     virtual int GetSessionInstance() const;
     109             : 
     110     1524767 :     static const uint8_t *BufferData(const Buffer &buffer) {
     111     1524767 :         return boost::asio::buffer_cast<const uint8_t *>(buffer);
     112             :     }
     113     2626016 :     static size_t BufferSize(const Buffer &buffer) {
     114     2626016 :         return boost::asio::buffer_size(buffer);
     115             :     }
     116             : 
     117     1430665 :     bool IsEstablished() const {
     118     1430665 :         std::scoped_lock lock(mutex_);
     119     1430691 :         return established_;
     120     1430688 :     }
     121             : 
     122        9857 :     bool IsClosed() const {
     123        9857 :         std::scoped_lock lock(mutex_);
     124        9861 :         return closed_;
     125        9862 :     }
     126             : 
     127        2931 :     bool IsServerSession() {
     128        2931 :         if (direction_ == PASSIVE) return true;
     129        1476 :         return false;
     130             :     }
     131             : 
     132     8040612 :     Endpoint remote_endpoint() const {
     133     8040612 :         return remote_;
     134             :     }
     135             : 
     136     1623678 :     const std::string &remote_addr_string() const {
     137     1623678 :         return remote_addr_str_;
     138             :     }
     139             : 
     140             :     Endpoint local_endpoint() const;
     141             : 
     142             :     const boost::system::error_code &close_reason() const {
     143             :         return close_reason_;
     144             :     }
     145             : 
     146             :     virtual boost::system::error_code SetSocketOptions();
     147             :     static bool IsSocketErrorHard(const boost::system::error_code &ec);
     148       13209 :     void set_read_on_connect(bool read) { read_on_connect_ = read; }
     149             :     void SessionEstablished(Endpoint remote, Direction direction);
     150             : 
     151             :     virtual void AsyncReadStart();
     152             :     virtual void SetDeferReader(bool defer_reader);
     153             :     // Is the reader deferred ? If reader is deferred, SetDeferReader needs
     154             :     // to be called to undefer/restart reading.
     155     1275419 :     virtual bool IsReaderDeferred() const {
     156     1275419 :         return defer_reader_;
     157             :     }
     158             : 
     159        1058 :     const io::SocketStats &GetSocketStats() const { return stats_; }
     160             :     void GetRxSocketStats(SocketIOStats *socket_stats) const;
     161             :     void GetTxSocketStats(SocketIOStats *socket_stats) const;
     162             : 
     163           6 :     void GetRxSocketStats(SocketIOStats &socket_stats) const {
     164           6 :         GetRxSocketStats(&socket_stats);
     165           6 :     }
     166             : 
     167           6 :     void GetTxSocketStats(SocketIOStats &socket_stats) const {
     168           6 :         GetTxSocketStats(&socket_stats);
     169           6 :     }
     170             : 
     171             :     int SetMd5SocketOption(uint32_t peer_ip, const std::string &md5_password);
     172             :     int ClearMd5SocketOption(uint32_t peer_ip);
     173             :     int SetDscpSocketOption(uint8_t value);
     174             :     uint8_t GetDscpValue() const;
     175       23232 :     const std::string &ToUVEKey() const { return uve_key_str_; }
     176             :     boost::system::error_code SetTcpNoDelay();
     177             :     boost::system::error_code SetTcpSendBufSize(uint32_t size);
     178             :     boost::system::error_code SetTcpRecvBufSize(uint32_t size);
     179             : 
     180             : protected:
     181             :     typedef boost::intrusive_ptr<TcpSession> TcpSessionPtr;
     182             :     static void AsyncReadHandler(TcpSessionPtr session);
     183             :     static void AsyncWriteHandler(TcpSessionPtr session,
     184             :                                   const boost::system::error_code &error,
     185             :                                   std::size_t bytes_transferred);
     186             : 
     187             :     void AsyncReadStartInternal(TcpSessionPtr session);
     188             :     virtual Task* CreateReaderTask(boost::asio::mutable_buffer, size_t);
     189             : 
     190             :     virtual ~TcpSession();
     191             : 
     192             :     // Read handler. Called from a TBB task.
     193             :     virtual void OnRead(Buffer buffer) = 0;
     194             :     // Callback after socket is ready for write.
     195             :     virtual void WriteReady(const boost::system::error_code &error);
     196             : 
     197             :     void AsyncWriteInternal(TcpSessionPtr session);
     198             : 
     199             :     virtual void AsyncReadSome();
     200             :     virtual size_t GetReadBufferSize() const;
     201             :     virtual size_t ReadSome(boost::asio::mutable_buffer buffer,
     202             :                             boost::system::error_code *error);
     203             :     virtual void AsyncWrite(const uint8_t *data, std::size_t size);
     204             : 
     205     1271553 :     virtual int reader_task_id() const {
     206     1271553 :         return reader_task_id_;
     207             :     }
     208             : 
     209     1187468 :     bool established() const { return established_; }
     210             : 
     211         114 :     EventObserver observer() { return observer_; }
     212             :     boost::system::error_code SetSocketKeepaliveOptions(int keepalive_time,
     213             :             int keepalive_intvl, int keepalive_probes,
     214             :             int tcp_user_timeout_val = 0);
     215             : 
     216             :     void CloseInternal(const boost::system::error_code &ec,
     217             :                        bool call_observer, bool notify_server = true);
     218             : 
     219             :     void TriggerAsyncReadHandler();
     220             : 
     221             :     // Protects session state and buffer queue.
     222             :     mutable std::mutex mutex_;
     223             :     io::SocketStats stats_;
     224             : 
     225             : protected:
     226             :     typedef boost::asio::strand<boost::asio::io_context::executor_type> Strand;
     227             :     boost::scoped_ptr<Strand> io_strand_;
     228             : 
     229             : private:
     230             :     class Reader;
     231             :     friend class TcpServer;
     232             :     friend class TcpMessageWriter;
     233             :     friend void intrusive_ptr_add_ref(TcpSession *session);
     234             :     friend void intrusive_ptr_release(TcpSession *session);
     235             :     typedef std::list<boost::asio::mutable_buffer> BufferQueue;
     236             : 
     237             :     static void WriteReadyInternal(TcpSessionPtr session,
     238             :                                    const boost::system::error_code &error,
     239             :                                    uint64_t block_start_time);
     240             :     void ReleaseBufferLocked(Buffer buffer);
     241             :     void SetEstablished(Endpoint remote, Direction dir);
     242             : 
     243     3369316 :     bool IsClosedLocked() const {
     244     3369316 :         return closed_;
     245             :     }
     246             : 
     247     2096138 :     bool IsEstablishedLocked() const {
     248     2096138 :         return (established_ && !tcp_close_in_progress_);
     249             :     }
     250             : 
     251             :     void SetName();
     252             : 
     253             :     boost::asio::mutable_buffer AllocateBuffer(size_t buffer_size);
     254             :     void DeleteBuffer(boost::asio::mutable_buffer buffer);
     255             : 
     256             :     static int reader_task_id_;
     257             : 
     258             :     TcpServerPtr server_;
     259             :     boost::scoped_ptr<Socket> socket_;
     260             :     bool read_on_connect_;
     261             : 
     262             :     /**************** protected by mutex_ ****************/
     263             :     bool established_;             // In TCP ESTABLISHED state.
     264             :     bool closed_;                  // Close has been called.
     265             :     Endpoint remote_;              // Remote end-point
     266             :     std::string remote_addr_str_;  // Remote end-point address string
     267             :     Direction direction_;          // direction (active, passive)
     268             :     BufferQueue buffer_queue_;
     269             :     boost::system::error_code close_reason_;
     270             :     /**************** end protected by mutex_ ****************/
     271             : 
     272             :     // Protects observer manipulation and invocation. When this lock is
     273             :     // held the session mutex should not be held and vice-versa.
     274             :     std::mutex obs_mutex_;
     275             :     EventObserver observer_;
     276             : 
     277             :     boost::scoped_ptr<TcpMessageWriter> writer_;
     278             : 
     279             :     std::atomic<int> refcount_;
     280             :     std::string name_;
     281             :     std::atomic<bool> defer_reader_;
     282             :     std::string uve_key_str_;
     283             :     std::atomic<bool> write_blocked_;
     284             :     std::atomic<bool> tcp_close_in_progress_;
     285             : 
     286             :     DISALLOW_COPY_AND_ASSIGN(TcpSession);
     287             : };
     288             : 
     289    45113497 : inline void intrusive_ptr_add_ref(TcpSession *session) {
     290    45113497 :     session->refcount_.fetch_add(1);
     291    45113497 : }
     292             : 
     293    45096622 : inline void intrusive_ptr_release(TcpSession *session) {
     294    45096622 :     int prev = session->refcount_.fetch_sub(1);
     295    45096622 :     if (prev == 1) {
     296       28508 :         delete session;
     297             :     }
     298    45096621 : }
     299             : 
     300             : // TcpMessageReader
     301             : //
     302             : // Provides base implementation of OnRead() for TcpSession assuming
     303             : // fixed message header length
     304             : //
     305             : class TcpMessageReader {
     306             : public:
     307             :     typedef boost::asio::const_buffer Buffer;
     308             :     typedef boost::function<bool(const uint8_t *, size_t)> ReceiveCallback;
     309             : 
     310             :     TcpMessageReader(TcpSession *session, ReceiveCallback callback);
     311             :     virtual ~TcpMessageReader();
     312             :     virtual void OnRead(Buffer buffer);
     313             : 
     314             : protected:
     315             :     virtual int MsgLength(Buffer buffer, int offset) = 0;
     316             :     virtual const int GetHeaderLenSize() = 0;
     317             :     virtual const int GetMaxMessageSize() = 0;
     318             : 
     319             : private:
     320             :     typedef std::deque<Buffer> BufferQueue;
     321             : 
     322             :     // Copy the queue into one contiguous buffer.
     323             :     uint8_t *BufferConcat(uint8_t *data, Buffer buffer, int msglength);
     324             : 
     325             :     int QueueByteLength() const;
     326             : 
     327             :     Buffer PullUp(uint8_t *data, Buffer buffer, size_t size) const;
     328             : 
     329             :     int AllocBufferSize(int length);
     330             : 
     331             :     TcpSession *session_;
     332             :     ReceiveCallback callback_;
     333             :     BufferQueue queue_;
     334             :     int offset_;
     335             :     int remain_;
     336             : 
     337             :     DISALLOW_COPY_AND_ASSIGN(TcpMessageReader);
     338             : };
     339             : 
     340             : #endif  // SRC_IO_TCP_SESSION_H_