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             : //
       7             : // sandesh_session.h
       8             : //
       9             : // Sandesh Session
      10             : //
      11             : 
      12             : #ifndef __SANDESH_SESSION_H__
      13             : #define __SANDESH_SESSION_H__
      14             : 
      15             : #include <mutex>
      16             : 
      17             : #include <boost/system/error_code.hpp>
      18             : #include <boost/asio.hpp>
      19             : #include <boost/tuple/tuple.hpp>
      20             : 
      21             : #include <base/util.h>
      22             : #include <io/ssl_session.h>
      23             : #include <io/udp_server.h>
      24             : 
      25             : #include <sandesh/transport/TBufferTransports.h>
      26             : #include <sandesh/sandesh.h>
      27             : #include <sandesh/sandesh_util.h>
      28             : #include <sandesh/sandesh_uve_types.h>
      29             : #include <sandesh/stats_client.h>
      30             : 
      31             : using contrail::sandesh::transport::TMemoryBuffer;
      32             : class SandeshSession;
      33             : class Sandesh;
      34             : 
      35             : class SandeshWriter {
      36             : public:
      37             :     static const uint32_t kEncodeBufferSize = 2048;
      38             :     static const unsigned int kDefaultSendSize = 16384;
      39             : 
      40             :     SandeshWriter(SandeshSession *session);
      41             :     ~SandeshWriter();
      42             :     void SendMsg(Sandesh *sandesh, bool more);
      43           0 :     void SendBuffer(boost::shared_ptr<TMemoryBuffer> sbuffer,
      44             :             bool more = false) {
      45           0 :         SendInternal(sbuffer);
      46           0 :     }
      47             :     void WriteReady(const boost::system::error_code &ec);
      48       12192 :     bool SendReady() {
      49       12192 :         std::scoped_lock lock(send_mutex_);
      50       12192 :         return ready_to_send_;
      51       12192 :     }
      52             : 
      53             :     static const std::string sandesh_open_;
      54             :     static const std::string sandesh_open_attr_length_;
      55             :     static const std::string sandesh_close_;
      56             : 
      57             : protected:
      58             :     friend class SandeshSessionTest;
      59             : 
      60             :     // Inline routines invoked by SendMsg()
      61             :     void SendMsgMore(boost::shared_ptr<TMemoryBuffer>);
      62             :     void SendMsgAll(boost::shared_ptr<TMemoryBuffer>);
      63             : 
      64             : private:
      65             :     friend class SandeshSendMsgUnitTest;
      66             : 
      67             :     SandeshSession *session_;
      68             : 
      69             :     void SendInternal(boost::shared_ptr<TMemoryBuffer>);
      70             :     void ConnectTimerExpired(const boost::system::error_code &error);
      71       24616 :     size_t send_buf_offset() { return send_buf_offset_; }
      72       21846 :     uint8_t* send_buf() const { return send_buf_; }
      73        2230 :     void set_send_buf(uint8_t *buf, size_t len) {
      74        2230 :         assert(len && (len < kDefaultSendSize));
      75        2230 :         memcpy(send_buf(), buf, len);
      76        2230 :         send_buf_offset_ = len;
      77        2230 :     }
      78        4413 :     void append_send_buf(uint8_t *buf, size_t len) {
      79        4413 :         assert(len && ((len + send_buf_offset_) < kDefaultSendSize));
      80        4413 :         memcpy(send_buf() + send_buf_offset_, buf, len);
      81        4413 :         send_buf_offset_ += len;
      82        4413 :     }
      83        2230 :     void reset_send_buf() {
      84        2230 :         send_buf_offset_ = 0;
      85        2230 :     }
      86             : 
      87             :     std::mutex send_mutex_;
      88             :     bool ready_to_send_;
      89             :     // send_buf_ is used to store unsent data
      90             :     uint8_t *send_buf_;
      91             :     size_t send_buf_offset_;
      92             : 
      93             : #define sXML_SANDESH_OPEN_ATTR_LENGTH  "<sandesh length=\""
      94             : #define sXML_SANDESH_OPEN              "<sandesh length=\"0000000000\">"
      95             : #define sXML_SANDESH_CLOSE             "</sandesh>"
      96             : 
      97             :     DISALLOW_COPY_AND_ASSIGN(SandeshWriter);
      98             : };
      99             : 
     100             : typedef boost::function<bool(const std::string&, SandeshSession *)>
     101             :     SandeshReceiveMsgCb;
     102             : 
     103             : class SandeshReader {
     104             : public:
     105             :     typedef boost::asio::const_buffer Buffer;
     106             : 
     107             :     SandeshReader(SandeshSession *session);
     108             :     virtual ~SandeshReader();
     109             :     virtual void OnRead(Buffer buffer);
     110             :     void SetReceiveMsgCb(SandeshReceiveMsgCb cb);
     111             :     static int ExtractMsgHeader(const std::string& msg, SandeshHeader& header,
     112             :             std::string& msg_type, uint32_t& header_offset);
     113             : 
     114             : private:
     115       15189 :     bool MsgLengthKnown() { return msg_length_ != (size_t)-1; }
     116             : 
     117       38502 :     size_t msg_length() { return msg_length_; }
     118             : 
     119       25070 :     void set_msg_length(size_t length) { msg_length_ = length; }
     120             : 
     121       12535 :     void reset_msg_length() { set_msg_length(-1); }
     122             : 
     123             :     void SetBuf(const std::string &str);
     124             :     void ReplaceBuf(const std::string &str);
     125             :     bool LeftOver() const;
     126             :     int MatchString(const std::string& match, size_t &m_offset);
     127             :     bool ExtractMsgLength(size_t &msg_length, int *result);
     128             :     bool ExtractMsg(Buffer buffer, int *result, bool NewBuf);
     129             : 
     130             :     std::string buf_;
     131             :     size_t offset_;
     132             :     size_t msg_length_;
     133             :     SandeshSession *session_;
     134             :     std::mutex cb_mutex_;
     135             :     SandeshReceiveMsgCb cb_;
     136             : 
     137             :     static const int kDefaultRecvSize = SandeshWriter::kDefaultSendSize;
     138             : 
     139             :     DISALLOW_COPY_AND_ASSIGN(SandeshReader);
     140             : };
     141             : 
     142             : class SandeshConnection;
     143             : 
     144             : class SandeshSession : public SslSession {
     145             : public:
     146             :     SandeshSession(SslServer *client, SslSocket *socket, int task_instance,
     147             :         int writer_task_id, int reader_task_id);
     148             :     virtual ~SandeshSession();
     149             :     virtual void Shutdown();
     150             :     virtual void OnRead(Buffer buffer);
     151           0 :     virtual void WriteReady(const boost::system::error_code &ec) {
     152           0 :         writer_->WriteReady(ec);
     153           0 :     }
     154             :     virtual bool EnqueueBuffer(u_int8_t *buf, u_int32_t buf_len);
     155       13069 :     Sandesh::SandeshQueue *send_queue() {
     156       13069 :         return send_queue_.get();
     157             :     }
     158           0 :     Sandesh::SandeshBufferQueue *send_buffer_queue() {
     159           0 :         return send_buffer_queue_.get();
     160             :     }
     161             :     SandeshWriter* writer() {
     162             :         return writer_.get();
     163             :     }
     164           0 :     void SetConnection(SandeshConnection *connection) {
     165           0 :         std::scoped_lock lock(conn_mutex_);
     166           0 :         connection_ = connection;
     167           0 :     }
     168          37 :     SandeshConnection *connection() {
     169          37 :         std::scoped_lock lock(conn_mutex_);
     170          37 :         return connection_;
     171          37 :     }
     172           0 :     void SetReceiveMsgCb(SandeshReceiveMsgCb cb) {
     173           0 :         reader_->SetReceiveMsgCb(cb);
     174           0 :     }
     175        5887 :     virtual int GetSessionInstance() const {
     176        5887 :         return instance_;
     177             :     }
     178             :     virtual void EnqueueClose();
     179             :     virtual boost::system::error_code SetSocketOptions();
     180             :     virtual std::string ToString() const;
     181             :     void set_stats_client(StatsClient *stats_client) { stats_client_ = stats_client;}
     182             :     static Sandesh * DecodeCtrlSandesh(const std::string& msg, const SandeshHeader& header,
     183             :         const std::string& sandesh_name, const uint32_t& header_offset);
     184             :     // Session statistics
     185             :     inline void increment_recv_msg() {
     186             :         sstats_.num_recv_msg++;
     187             :     }
     188             :     inline void increment_recv_msg_fail() {
     189             :         sstats_.num_recv_msg_fail++;
     190             :     }
     191           0 :     inline void increment_recv_fail() {
     192           0 :         sstats_.num_recv_fail++;
     193           0 :     }
     194       12973 :     inline void increment_send_msg() {
     195       12973 :         sstats_.num_send_msg++;
     196       12973 :     }
     197           0 :     inline void increment_send_msg_fail() {
     198           0 :         sstats_.num_send_msg_fail++;
     199           0 :     }
     200           0 :     inline void increment_send_buffer_fail() {
     201           0 :         sstats_.num_send_buffer_fail++;
     202           0 :     }
     203             :     inline void increment_wait_msgq_enqueue() {
     204             :         sstats_.num_wait_msgq_enqueue++;
     205             :     }
     206             :     inline void increment_wait_msgq_dequeue() {
     207             :         sstats_.num_wait_msgq_dequeue++;
     208             :     }
     209           0 :     inline void increment_write_ready_cb_error() {
     210           0 :         sstats_.num_write_ready_cb_error++;
     211           0 :     }
     212             :     const SandeshSessionStats& GetStats() const {
     213             :         return sstats_;
     214             :     }
     215             :     void SetSendQueueWaterMark(Sandesh::QueueWaterMarkInfo &wm_info);
     216             :     void ResetSendQueueWaterMark();
     217             :     SandeshLevel::type SendingLevel() const;
     218             : 
     219             : protected:
     220        5850 :     virtual int reader_task_id() const {
     221        5850 :         return reader_task_id_;
     222             :     }
     223             : 
     224             : private:
     225             :     friend class SandeshSessionTest;
     226             : 
     227             :     // 60 seconds - 45s + (3*5)s
     228             :     static const int kSessionKeepaliveIdleTime = 15; // in seconds
     229             :     static const int kSessionKeepaliveInterval = 3; // in seconds
     230             :     static const int kSessionKeepaliveProbes = 5; // count
     231             :     static const int kSessionTcpUserTimeout = 30000; // ms
     232             :     static const int kQueueSize = 200 * 1024 * 1024; // 200 MB
     233             : 
     234             :     bool SendMsg(SandeshElement element);
     235             :     bool SendBuffer(boost::shared_ptr<TMemoryBuffer> sbuffer);
     236             :     bool SessionSendReady();
     237             :     void SetSendingLevel(size_t count, SandeshLevel::type level);
     238             : 
     239             :     int instance_;
     240             :     boost::scoped_ptr<SandeshWriter> writer_;
     241             :     boost::scoped_ptr<SandeshReader> reader_;
     242             :     boost::scoped_ptr<Sandesh::SandeshQueue> send_queue_;
     243             :     boost::scoped_ptr<Sandesh::SandeshBufferQueue> send_buffer_queue_;
     244             :     StatsClient *stats_client_;
     245             :     SandeshConnection *connection_;
     246             :     std::mutex conn_mutex_;
     247             :     std::mutex send_mutex_;
     248             :     int keepalive_idle_time_;
     249             :     int keepalive_interval_;
     250             :     int keepalive_probes_;
     251             :     int tcp_user_timeout_;
     252             :     int reader_task_id_;
     253             :     SandeshLevel::type sending_level_;
     254             : 
     255             :     // Session statistics
     256             :     SandeshSessionStats sstats_;
     257             : 
     258             :     DISALLOW_COPY_AND_ASSIGN(SandeshSession);
     259             : };
     260             : 
     261             : #endif // __SANDESH_SESSION_H__