Line data    Source code

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