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 ctrlplane_ksync_sock_h
7 : #define ctrlplane_ksync_sock_h
8 :
9 : #include <atomic>
10 : #include <vector>
11 : #include <mutex>
12 :
13 : #include <boost/asio.hpp>
14 : #include <boost/asio/buffer.hpp>
15 :
16 : #include <boost/asio/netlink_protocol.hpp>
17 : #include <boost/asio/netlink_endpoint.hpp>
18 :
19 : #include <base/queue_task.h>
20 : #include <sandesh/common/vns_constants.h>
21 : #include <sandesh/common/vns_types.h>
22 : #include <io/tcp_session.h>
23 : #include <vr_types.h>
24 : #include <nl_util.h>
25 : #include "ksync_entry.h"
26 : #include "ksync_tx_queue.h"
27 :
28 : #define KSYNC_DEFAULT_MSG_SIZE 4096
29 : #define KSYNC_DEFAULT_Q_ID_SEQ 0x00000001
30 : #define KSYNC_ACK_WAIT_THRESHOLD 200
31 : #define KSYNC_SOCK_RECV_BUFF_SIZE (256 * 1024)
32 : #define KSYNC_BMC_ARR_SIZE 1024
33 :
34 : class KSyncEntry;
35 : class KSyncIoContext;
36 : class KSyncSockTcpSession;
37 : struct nl_client;
38 : class KSyncBulkSandeshContext;
39 :
40 : typedef std::vector<boost::asio::mutable_buffers_1> KSyncBufferList;
41 :
42 : uint32_t GetNetlinkSeqno(char *data);
43 : bool NetlinkMsgDone(char *data);
44 : bool ValidateNetlink(char *data);
45 : void GetNetlinkPayload(char *data, char **buf, uint32_t *buf_len);
46 : void InitNetlink(nl_client *client);
47 : void ResetNetlink(nl_client *client);
48 : void UpdateNetlink(nl_client *client, uint32_t len, uint32_t seq_no);
49 : void DecodeSandeshMessages(char *buf, uint32_t buf_len, SandeshContext *sandesh_context,
50 : uint32_t alignment);
51 :
52 : /* Base class to hold sandesh context information which is passed to
53 : * Sandesh decode
54 : */
55 : class AgentSandeshContext : public SandeshContext {
56 : public:
57 22460 : AgentSandeshContext() : errno_(0), ksync_io_ctx_(NULL) { }
58 95242 : virtual ~AgentSandeshContext() { }
59 :
60 : virtual void IfMsgHandler(vr_interface_req *req) = 0;
61 : virtual void NHMsgHandler(vr_nexthop_req *req) = 0;
62 : virtual void RouteMsgHandler(vr_route_req *req) = 0;
63 : virtual void MplsMsgHandler(vr_mpls_req *req) = 0;
64 : virtual int VrResponseMsgHandler(vr_response *resp) = 0;
65 : virtual void MirrorMsgHandler(vr_mirror_req *req) = 0;
66 : virtual void FlowMsgHandler(vr_flow_req *req) = 0;
67 0 : virtual void FlowResponseHandler(vr_flow_response *req) { assert(0); }
68 0 : virtual void FlowTableInfoHandler(vr_flow_table_data *r) { assert(0); }
69 0 : virtual void BridgeTableInfoHandler(vr_bridge_table_data *r) { assert(0);}
70 : virtual void VrfAssignMsgHandler(vr_vrf_assign_req *req) = 0;
71 : virtual void VrfMsgHandler(vr_vrf_req *req) = 0;
72 : virtual void VrfStatsMsgHandler(vr_vrf_stats_req *req) = 0;
73 : virtual void DropStatsMsgHandler(vr_drop_stats_req *req) = 0;
74 : virtual void VxLanMsgHandler(vr_vxlan_req *req) = 0;
75 0 : virtual void VrouterHugePageHandler(vr_hugepage_config *req) {}
76 : virtual void VrouterOpsMsgHandler(vrouter_ops *req) = 0;
77 : virtual void QosConfigMsgHandler(vr_qos_map_req *req) = 0;
78 : virtual void ForwardingClassMsgHandler(vr_fc_map_req *req) = 0;
79 53083 : virtual void SetErrno(int err) {errno_ = err;}
80 :
81 11629 : int GetErrno() const {return errno_;}
82 22498 : void set_ksync_io_ctx(const KSyncIoContext *ioc) {ksync_io_ctx_ = ioc;}
83 378 : const KSyncIoContext *ksync_io_ctx() const {return ksync_io_ctx_;}
84 : private:
85 : int errno_;
86 : const KSyncIoContext *ksync_io_ctx_;
87 : };
88 :
89 :
90 : /* Base class for context management. Used while sending and
91 : * receiving data via ksync socket
92 : */
93 : class IoContext {
94 : public:
95 : // Type of IoContext. The work-queue used for processing response is based
96 : // on this
97 : // IOC_UVE : Used for UVEs
98 : // IOC_KSYNC : Used for KSync objects
99 : enum Type {
100 : IOC_UVE,
101 : IOC_KSYNC,
102 : MAX_WORK_QUEUES // This should always be last
103 : };
104 : // Work-queue neames used for the ksync receive work-queues
105 : static const char *io_wq_names[MAX_WORK_QUEUES];
106 :
107 : IoContext() :
108 : sandesh_context_(NULL), msg_(NULL), msg_len_(0), seqno_(0),
109 : type_(IOC_KSYNC), index_(0), rx_buffer1_(NULL), rx_buffer2_(NULL) {
110 : }
111 0 : IoContext(char *msg, uint32_t len, uint32_t seq, AgentSandeshContext *ctx,
112 0 : Type type) :
113 0 : sandesh_context_(ctx), msg_(msg), msg_len_(len), seqno_(seq),
114 0 : type_(type), index_(0), rx_buffer1_(NULL), rx_buffer2_(NULL) {
115 0 : }
116 11249 : IoContext(char *msg, uint32_t len, uint32_t seq, AgentSandeshContext *ctx,
117 11249 : Type type, uint32_t index) :
118 11249 : sandesh_context_(ctx), msg_(msg), msg_len_(len), seqno_(seq),
119 11249 : type_(type), index_(index), rx_buffer1_(NULL), rx_buffer2_(NULL) {
120 11249 : }
121 11249 : virtual ~IoContext() {
122 11249 : if (msg_ != NULL)
123 11249 : free(msg_);
124 11249 : assert(rx_buffer1_ == NULL);
125 11249 : assert(rx_buffer2_ == NULL);
126 11249 : }
127 :
128 : bool operator<(const IoContext &rhs) const {
129 : return seqno_ < rhs.seqno_;
130 : }
131 :
132 0 : virtual void Handler() {}
133 0 : virtual void ErrorHandler(int err) {}
134 :
135 45374 : AgentSandeshContext *GetSandeshContext() { return sandesh_context_; }
136 33748 : Type type() { return type_; }
137 :
138 11249 : void SetSeqno(uint32_t seqno) {seqno_ = seqno;}
139 11250 : uint32_t GetSeqno() const {return seqno_;}
140 11249 : char *GetMsg() const { return msg_; }
141 22498 : uint32_t GetMsgLen() const { return msg_len_; }
142 11627 : char *rx_buffer1() const { return rx_buffer1_; }
143 378 : void reset_rx_buffer1() { rx_buffer1_ = NULL; }
144 11627 : char *rx_buffer2() { return rx_buffer2_; }
145 378 : void reset_rx_buffer2() { rx_buffer2_ = NULL; }
146 33748 : uint32_t index() const { return index_; }
147 :
148 : boost::intrusive::list_member_hook<> node_;
149 :
150 : protected:
151 : AgentSandeshContext *sandesh_context_;
152 :
153 : private:
154 : char *msg_;
155 : uint32_t msg_len_;
156 : uint32_t seqno_;
157 : Type type_;
158 : uint32_t index_;
159 : // Buffers allocated to read the ksync responses for this IoContext.
160 : // As an optimization, KSync Tx Queue will use these buffers to minimize
161 : // computation in KSync Tx Queue context.
162 : char *rx_buffer1_;
163 : char *rx_buffer2_;
164 :
165 : friend class KSyncSock;
166 : };
167 :
168 : /* IoContext tied to KSyncEntry */
169 : class KSyncIoContext : public IoContext {
170 : public:
171 : KSyncIoContext(KSyncSock *sock, KSyncEntry *sync_entry, int msg_len,
172 : char *msg, KSyncEntry::KSyncEvent event);
173 22498 : virtual ~KSyncIoContext() {
174 22498 : }
175 :
176 : virtual void Handler();
177 :
178 : void ErrorHandler(int err);
179 378 : KSyncEntry *GetKSyncEntry() const {return entry_;}
180 378 : KSyncEntry::KSyncEvent event() const {return event_;}
181 : private:
182 : KSyncEntry *entry_;
183 : KSyncEntry::KSyncEvent event_;
184 : AgentSandeshContext *agent_sandesh_ctx_;
185 : KSyncSock *sock_;
186 : };
187 :
188 : /*
189 : * KSync implementation of bunching messages.
190 : *
191 : * Message bunching has two parts,
192 : * Encoding messages
193 : * KSyncTxQueue is responsible to bunch KSync requests into single message
194 : *
195 : * KSyncTxQueue uses KSyncBulkMsgContext to bunch KSync events.
196 : * The IoContext for bunched KSync events are stored as list inside
197 : * KSyncBulkMessageContext
198 : *
199 : * KSync Events are bunched if pass following constraints,
200 : * - Number of KSync events is less than max_bulk_msg_count_
201 : * - Total size of buffer for events is less than max_bulk_buf_size_
202 : * - Type of IoContext is same.
203 : * When type of IoContext is same, it also ensures that KSync Responses
204 : * are processed in same WorkQueue.
205 : * - Each IoContext type has multiple work-queues. Flows are sprayed across
206 : * the work-queues based on flow-table index. Entries are bunched only if
207 : * they belong to same work-queue.
208 : * This also ensures that flows from a flow-table partition are processed
209 : * in single KSync Response Work-Queue
210 : * - The input queue for KSyncTxQueue is not empty.
211 : *
212 : * Decoding messages
213 : * KSync response are processed in context of KSyncSock::receive_work_queue_.
214 : * Each IoContext type has its own set of receive work-queue.
215 : *
216 : * Bulk decoder works on following assumption,
217 : * - Expects KSync Responses for each IoContext in KSyncBulkMsgContext
218 : * - Response for each IoContext can itself be more than one Sandesh Response.
219 : * - The sequence of response is same as sequence of IoContext entries
220 : * - Response for each IoContext starts with VrResponseMsg followed
221 : * optionally by other response. Hence, decoder moves to next IoContext
222 : * after getting VrResponse message
223 : *
224 : * class KSyncBulkSandeshContext is used to decode the Sandesh Responses
225 : * and move the IoContext on getting VrResponse
226 : */
227 : typedef boost::intrusive::member_hook<IoContext,
228 : boost::intrusive::list_member_hook<>,
229 : &IoContext::node_> KSyncSockNode;
230 : typedef boost::intrusive::list<IoContext, KSyncSockNode> IoContextList;
231 :
232 : class KSyncBulkMsgContext {
233 : public:
234 : const static unsigned kMaxRxBufferCount = 64;
235 : KSyncBulkMsgContext(IoContext::Type type, uint32_t index);
236 : KSyncBulkMsgContext(const KSyncBulkMsgContext &rhs);
237 : ~KSyncBulkMsgContext();
238 :
239 : void Insert(IoContext *ioc);
240 : void Data(KSyncBufferList *iovec);
241 22561 : IoContext::Type io_context_type() const {
242 22561 : return io_context_type_;
243 : }
244 : void AddReceiveBuffer(char *buff);
245 : char *GetReceiveBuffer();
246 22561 : uint32_t work_queue_index() const { return work_queue_index_; }
247 0 : void set_seqno(uint32_t seq) { seqno_ = seq; }
248 0 : uint32_t seqno() { return seqno_; }
249 : private:
250 : friend class KSyncBulkSandeshContext;
251 : // List of IoContext to be processed in this context
252 : IoContextList io_context_list_;
253 : // Type of message
254 : IoContext::Type io_context_type_;
255 : // Index of work-queue
256 : uint32_t work_queue_index_;
257 : // List of rx-buffers
258 : // The buffers are taken from IoContext added to the list above
259 : // If IoContext does not have buffer, then it dyamically allocates
260 : // buffer
261 : char *rx_buffers_[kMaxRxBufferCount];
262 : // Index of next buffer to process
263 : uint32_t rx_buffer_index_;
264 :
265 : ///////////////////////////////////////////////////////////////////////
266 : // Following fields are used for decode processing
267 : ///////////////////////////////////////////////////////////////////////
268 : // Number of responses already processed
269 : uint32_t vr_response_count_;
270 : // Iterator to IoContext being processed
271 : IoContextList::iterator io_context_list_it_;
272 : uint32_t seqno_;
273 : };
274 :
275 : class KSyncBulkSandeshContext : public AgentSandeshContext {
276 : public:
277 : KSyncBulkSandeshContext();
278 : virtual ~KSyncBulkSandeshContext();
279 :
280 : void IfMsgHandler(vr_interface_req *req);
281 : void NHMsgHandler(vr_nexthop_req *req);
282 : void RouteMsgHandler(vr_route_req *req);
283 : void MplsMsgHandler(vr_mpls_req *req);
284 : int VrResponseMsgHandler(vr_response *resp);
285 : void MirrorMsgHandler(vr_mirror_req *req);
286 : void FlowMsgHandler(vr_flow_req *req);
287 : void FlowResponseHandler(vr_flow_response *req);
288 : void VrfAssignMsgHandler(vr_vrf_assign_req *req);
289 : void VrfMsgHandler(vr_vrf_req *req);
290 : void VrfStatsMsgHandler(vr_vrf_stats_req *req);
291 : void DropStatsMsgHandler(vr_drop_stats_req *req);
292 : void VxLanMsgHandler(vr_vxlan_req *req);
293 : void VrouterOpsMsgHandler(vrouter_ops *req);
294 : void QosConfigMsgHandler(vr_qos_map_req *req);
295 : void ForwardingClassMsgHandler(vr_fc_map_req *req);
296 : void SetErrno(int err);
297 :
298 : bool Decoder(char *buff, uint32_t buff_len, uint32_t alignment, bool more);
299 : AgentSandeshContext *GetSandeshContext();
300 11312 : void set_bulk_message_context(KSyncBulkMsgContext *bulk_context) {
301 11312 : bulk_msg_context_ = bulk_context;
302 11312 : }
303 :
304 : void IoContextStart();
305 : void IoContextDone();
306 :
307 : private:
308 : KSyncBulkMsgContext *bulk_msg_context_;
309 : DISALLOW_COPY_AND_ASSIGN(KSyncBulkSandeshContext);
310 : };
311 :
312 : class KSyncSock {
313 : public:
314 : // Number of flow receive queues
315 : const static int kRxWorkQueueCount = 2;
316 : const static int kMsgGrowSize = 16;
317 : const static unsigned kBufLen = (4*1024);
318 :
319 : // Number of messages that can be bunched together
320 : const static unsigned kMaxBulkMsgCount = 16;
321 : // Sequence number to denote invalid builk-context
322 : const static unsigned kInvalidBulkSeqNo = 0xFFFFFFFF;
323 :
324 : typedef std::map<uint32_t, KSyncBulkMsgContext> WaitTree;
325 : typedef std::pair<uint32_t, KSyncBulkMsgContext> WaitTreePair;
326 : typedef boost::function<void(const boost::system::error_code &, size_t)>
327 : HandlerCb;
328 :
329 : // Request structure in the KSync Response Queue
330 : struct KSyncRxData {
331 : // buffer having KSync response
332 : char *buff_;
333 : // bulk context for decoding response
334 : KSyncBulkMsgContext *bulk_msg_context_;
335 :
336 2963 : KSyncRxData() : buff_(NULL), bulk_msg_context_(NULL) { }
337 45247 : KSyncRxData(const KSyncRxData &rhs) :
338 45247 : buff_(rhs.buff_), bulk_msg_context_(rhs.bulk_msg_context_) {
339 45247 : }
340 11312 : KSyncRxData(char *buff, KSyncBulkMsgContext *ctxt) :
341 11312 : buff_(buff), bulk_msg_context_(ctxt) {
342 11312 : }
343 : };
344 : typedef WorkQueue<KSyncRxData> KSyncReceiveQueue;
345 : // structure for ksyncrprocess Rx queue
346 : struct KSyncRxQueueData {
347 : KSyncEntry *entry_;
348 : KSyncEntry::KSyncEvent event_;
349 2963 : KSyncRxQueueData():entry_(NULL),event_(KSyncEntry::INVALID) {}
350 11249 : KSyncRxQueueData(KSyncEntry *entry, KSyncEntry::KSyncEvent event) :
351 11249 : entry_(entry), event_(event) {
352 11249 : }
353 : };
354 : typedef WorkQueue<KSyncRxQueueData> KSyncRxWorkQueue;
355 :
356 : KSyncSock();
357 : virtual ~KSyncSock();
358 :
359 : // Virtual methods
360 : virtual bool BulkDecoder(char *data, KSyncBulkSandeshContext *ctxt) = 0;
361 : virtual bool Decoder(char *data, AgentSandeshContext *ctxt) = 0;
362 :
363 : // Write a KSyncEntry to kernel
364 : void SendAsync(KSyncEntry *entry, int msg_len, char *msg,
365 : KSyncEntry::KSyncEvent event);
366 : std::size_t BlockingSend(char *msg, int msg_len);
367 : bool BlockingRecv();
368 : void GenericSend(IoContext *ctx);
369 : uint32_t AllocSeqNo(IoContext::Type type);
370 : uint32_t AllocSeqNo(IoContext::Type type, uint32_t instance);
371 : KSyncReceiveQueue *GetReceiveQueue(IoContext::Type type, uint32_t instance);
372 : KSyncReceiveQueue *GetReceiveQueue(uint32_t seqno);
373 :
374 : // Bulk Messaging methods
375 : KSyncBulkMsgContext *LocateBulkContext(uint32_t seqno,
376 : IoContext::Type io_context_type,
377 : uint32_t work_queue_index);
378 : int SendBulkMessage(KSyncBulkMsgContext *bulk_context, uint32_t seqno);
379 : bool TryAddToBulk(KSyncBulkMsgContext *bulk_context, IoContext *ioc);
380 : void OnEmptyQueue(bool done);
381 2 : int tx_count() const { return tx_count_; }
382 :
383 : // Start Ksync Asio operations
384 : static void Start(bool read_inline);
385 : static void Shutdown();
386 :
387 : // Partition to KSyncSock mapping
388 : static KSyncSock *Get(DBTablePartBase *partition);
389 : static KSyncSock *Get(int partition_id);
390 :
391 0 : static uint32_t GetPid() {return pid_;};
392 22542 : static int GetNetlinkFamilyId() {return vnsw_netlink_family_id_;}
393 : static void SetNetlinkFamilyId(int id);
394 :
395 11310 : static AgentSandeshContext *GetAgentSandeshContext(uint32_t type) {
396 11310 : return agent_sandesh_ctx_[type % kRxWorkQueueCount];
397 : }
398 24 : static void SetAgentSandeshContext(AgentSandeshContext *ctx, uint32_t idx) {
399 24 : agent_sandesh_ctx_[idx] = ctx;
400 24 : }
401 :
402 6 : const KSyncTxQueue *send_queue() const { return &send_queue_; }
403 10 : const KSyncReceiveQueue *get_receive_work_queue(uint16_t index) const {
404 10 : return ksync_rx_queue[index];
405 : }
406 : // Allocate a recieve work-queue
407 :
408 : KSyncReceiveQueue *AllocQueue(KSyncBulkSandeshContext ctxt[],
409 : uint32_t task_id, uint32_t instance,
410 : const char *name);
411 :
412 : uint32_t WaitTreeSize() const;
413 : void SetSeqno(uint32_t seq);
414 : void SetMeasureQueueDelay(bool val);
415 0 : void reset_use_wait_tree() { use_wait_tree_ = false; }
416 0 : void set_process_data_inline() { process_data_inline_ = true; }
417 : // API to enqueue ksync events to rx process work queue
418 : void EnqueueRxProcessData(KSyncEntry *entry, KSyncEntry::KSyncEvent event);
419 : protected:
420 : static void Init(bool use_work_queue, const std::string &cpu_pin_policy);
421 : static void SetSockTableEntry(KSyncSock *sock);
422 : bool ValidateAndEnqueue(char *data, KSyncBulkMsgContext *context);
423 : KSyncBulkSandeshContext *GetBulkSandeshContext(uint32_t seqno);
424 : void ProcessDataInline(char *data);
425 :
426 : std::mutex mutex_;
427 : nl_client *nl_client_;
428 : // Tree of all IoContext pending ksync response
429 : WaitTree wait_tree_;
430 : KSyncTxQueue send_queue_;
431 : KSyncReceiveQueue *uve_rx_queue[kRxWorkQueueCount];
432 : KSyncReceiveQueue *ksync_rx_queue[kRxWorkQueueCount];
433 :
434 : // Information maintained for bulk processing
435 :
436 : // Max messages in one bulk context
437 : uint32_t max_bulk_msg_count_;
438 : // Max buffer size in one bulk context
439 : uint32_t max_bulk_buf_size_;
440 :
441 : // Sequence number of first message in bulk context. Entry in WaitTree is
442 : // added based on this sequence number
443 : uint32_t bulk_seq_no_;
444 : // Current buffer size in bulk context
445 : uint32_t bulk_buf_size_;
446 : // Current message count in bulk context
447 : uint32_t bulk_msg_count_;
448 :
449 : uint32_t bmca_prod_;
450 : uint32_t bmca_cons_;
451 : KSyncBulkMsgContext *bulk_mctx_arr_[KSYNC_BMC_ARR_SIZE];
452 :
453 : private:
454 : friend class KSyncTxQueue;
455 : virtual void AsyncReceive(boost::asio::mutable_buffers_1, HandlerCb) = 0;
456 : virtual void AsyncSendTo(KSyncBufferList *iovec, uint32_t seq_no,
457 : HandlerCb cb) = 0;
458 : virtual std::size_t SendTo(KSyncBufferList *iovec, uint32_t seq_no) = 0;
459 : virtual void Receive(boost::asio::mutable_buffers_1) = 0;
460 : virtual uint32_t GetSeqno(char *data) = 0;
461 : virtual bool IsMoreData(char *data) = 0;
462 : virtual bool Validate(char *data) = 0;
463 :
464 : // Read handler registered with boost::asio. Demux done based on seqno_
465 : void ReadHandler(const boost::system::error_code& error,
466 : size_t bytes_transferred);
467 :
468 : // Write handler registered with boost::asio. Demux done based on seqno_
469 : void WriteHandler(const boost::system::error_code& error,
470 : size_t bytes_transferred);
471 :
472 : bool ProcessKernelData(KSyncBulkSandeshContext *ksync_context,
473 : const KSyncRxData &data);
474 : bool ProcessRxData(KSyncRxQueueData data);
475 : bool SendAsyncImpl(IoContext *ioc);
476 : bool SendAsyncStart() {
477 : std::scoped_lock lock(mutex_);
478 : return (wait_tree_.size() <= KSYNC_ACK_WAIT_THRESHOLD);
479 : }
480 :
481 : private:
482 : char *rx_buff_;
483 : std::atomic<uint32_t> seqno_;
484 : std::atomic<uint32_t> uve_seqno_;
485 : // Read ksync responses inline
486 : // The IoContext WaitTree is not used when response is read-inline
487 : bool read_inline_;
488 : KSyncBulkMsgContext *bulk_msg_context_;
489 : bool use_wait_tree_;
490 : bool process_data_inline_;
491 : KSyncBulkSandeshContext ksync_bulk_sandesh_context_[kRxWorkQueueCount];
492 : KSyncBulkSandeshContext uve_bulk_sandesh_context_[kRxWorkQueueCount];
493 :
494 : // Debug stats
495 : int tx_count_;
496 : int ack_count_;
497 : int err_count_;
498 :
499 : // IO context can defer ksync event processing
500 : // by defering them to this work queue, this queue gets
501 : // processed in Agent::KSync context
502 : KSyncRxWorkQueue rx_process_queue_;
503 : static std::unique_ptr<KSyncSock> sock_;
504 : static pid_t pid_;
505 : static int vnsw_netlink_family_id_;
506 : // AgentSandeshContext used for KSync response handling
507 : // AgentSandeshContext used for decode is picked based on work-queue index
508 : // Picking AgentSandeshContext based on work-queue index also makes it
509 : // thread safe
510 : static AgentSandeshContext *agent_sandesh_ctx_[kRxWorkQueueCount];
511 : static std::atomic<bool> shutdown_;
512 :
513 : DISALLOW_COPY_AND_ASSIGN(KSyncSock);
514 : };
515 :
516 : //netlink socket class for interacting with kernel
517 : class KSyncSockNetlink : public KSyncSock {
518 : public:
519 : KSyncSockNetlink(boost::asio::io_context &ios, int protocol);
520 : virtual ~KSyncSockNetlink();
521 :
522 : virtual uint32_t GetSeqno(char *data);
523 : virtual bool IsMoreData(char *data);
524 : virtual bool BulkDecoder(char *data, KSyncBulkSandeshContext *ctxt);
525 : virtual bool Decoder(char *data, AgentSandeshContext *ctxt);
526 : virtual bool Validate(char *data);
527 : virtual void AsyncReceive(boost::asio::mutable_buffers_1, HandlerCb);
528 : virtual void AsyncSendTo(KSyncBufferList *iovec, uint32_t seq_no,
529 : HandlerCb cb);
530 : virtual std::size_t SendTo(KSyncBufferList *iovec, uint32_t seq_no);
531 : virtual void Receive(boost::asio::mutable_buffers_1);
532 :
533 : static void NetlinkDecoder(char *data, SandeshContext *ctxt);
534 : static void NetlinkBulkDecoder(char *data, SandeshContext *ctxt, bool more);
535 : static void Init(boost::asio::io_context &ios, int protocol, bool use_work_queue,
536 : const std::string &cpu_pin_policy);
537 : private:
538 : boost::asio::netlink::raw::socket sock_;
539 : };
540 :
541 : //udp socket class for interacting with user vrouter
542 : class KSyncSockUdp : public KSyncSock {
543 : public:
544 : KSyncSockUdp(boost::asio::io_context &ios, int port);
545 0 : virtual ~KSyncSockUdp() { }
546 :
547 : virtual uint32_t GetSeqno(char *data);
548 : virtual bool IsMoreData(char *data);
549 : virtual bool BulkDecoder(char *data, KSyncBulkSandeshContext *ctxt);
550 : virtual bool Decoder(char *data, AgentSandeshContext *ctxt);
551 : virtual bool Validate(char *data);
552 : virtual void AsyncReceive(boost::asio::mutable_buffers_1, HandlerCb);
553 : virtual void AsyncSendTo(KSyncBufferList *iovec, uint32_t seq_no,
554 : HandlerCb cb);
555 : virtual std::size_t SendTo(KSyncBufferList *iovec, uint32_t seq_no);
556 : virtual void Receive(boost::asio::mutable_buffers_1);
557 :
558 : static void Init(boost::asio::io_context &ios, int port,
559 : const std::string &cpu_pin_policy);
560 : private:
561 : boost::asio::ip::udp::socket sock_;
562 : boost::asio::ip::udp::endpoint server_ep_;
563 : };
564 :
565 : //Unix domain socket class for interacting with user vrouter
566 : #define KSYNC_AGENT_VROUTER_SOCK_PATH "/var/run/vrouter/dpdk_netlink"
567 : class KSyncSockUds : public KSyncSock {
568 : public:
569 : KSyncSockUds(boost::asio::io_context &ios);
570 0 : virtual ~KSyncSockUds() {
571 0 : delete rx_buff_;
572 0 : delete rx_buff_q_;
573 0 : }
574 :
575 : virtual uint32_t GetSeqno(char *data);
576 : virtual bool IsMoreData(char *data);
577 : virtual bool BulkDecoder(char *data, KSyncBulkSandeshContext *ctxt);
578 : virtual bool Decoder(char *data, AgentSandeshContext *ctxt);
579 : virtual bool Validate(char *data);
580 : virtual void AsyncReceive(boost::asio::mutable_buffers_1, HandlerCb);
581 : virtual void AsyncSendTo(KSyncBufferList *iovec, uint32_t seq_no,
582 : HandlerCb cb);
583 : virtual std::size_t SendTo(KSyncBufferList *iovec, uint32_t seq_no);
584 : virtual void Receive(boost::asio::mutable_buffers_1);
585 : virtual bool Run(void);
586 :
587 : static void Init(boost::asio::io_context &ios,
588 : const std::string &cpu_pin_policy,
589 : const std::string &sockpathvr="");
590 : private:
591 : boost::asio::local::stream_protocol::socket sock_;
592 : boost::asio::local::stream_protocol::endpoint server_ep_;
593 : char *rx_buff_;
594 : char *rx_buff_q_;
595 : size_t remain_;
596 : int socket_;
597 : int connected_;
598 : static string sockpath_;
599 : };
600 :
601 : class KSyncSockTcpSessionReader : public TcpMessageReader {
602 : public:
603 : KSyncSockTcpSessionReader(TcpSession *session, ReceiveCallback callback);
604 0 : virtual ~KSyncSockTcpSessionReader() { }
605 :
606 : protected:
607 : virtual int MsgLength(Buffer buffer, int offset);
608 :
609 0 : virtual const int GetHeaderLenSize() {
610 0 : return sizeof(struct nlmsghdr);
611 : }
612 :
613 0 : virtual const int GetMaxMessageSize() {
614 0 : return kMaxMessageSize;
615 : }
616 :
617 : private:
618 : static const int kMaxMessageSize = 4096;
619 : };
620 :
621 : class KSyncSockTcpSession : public TcpSession {
622 : public:
623 : KSyncSockTcpSession(TcpServer *server, Socket *sock,
624 : bool async_ready = false);
625 : virtual ~KSyncSockTcpSession();
626 : protected:
627 : virtual void OnRead(Buffer buffer);
628 : private:
629 : KSyncSockTcpSessionReader *reader_;
630 : };
631 :
632 : class KSyncSockTcp : public KSyncSock, public TcpServer {
633 : public:
634 : KSyncSockTcp(EventManager *evm, boost::asio::ip::address ip_addr,
635 : int port);
636 0 : virtual ~KSyncSockTcp() { }
637 :
638 : virtual uint32_t GetSeqno(char *data);
639 : virtual bool IsMoreData(char *data);
640 : virtual bool BulkDecoder(char *data, KSyncBulkSandeshContext *ctxt);
641 : virtual bool Decoder(char *data, AgentSandeshContext *ctxt);
642 : virtual bool Validate(char *data);
643 : virtual void AsyncReceive(boost::asio::mutable_buffers_1, HandlerCb);
644 : virtual void AsyncSendTo(KSyncBufferList *iovec, uint32_t seq_no,
645 : HandlerCb cb);
646 : virtual std::size_t SendTo(KSyncBufferList *iovec, uint32_t seq_no);
647 : virtual void Receive(boost::asio::mutable_buffers_1);
648 : virtual TcpSession *AllocSession(Socket *socket);
649 : virtual bool Run(void);
650 :
651 : bool ReceiveMsg(const u_int8_t *msg, size_t size);
652 : void OnSessionEvent(TcpSession *session, TcpSession::Event event);
653 3 : bool connect_complete() const {
654 3 : return connect_complete_;
655 : }
656 : void AsyncReadStart();
657 :
658 : static void Init(EventManager *evm,
659 : boost::asio::ip::address ip_addr, int port,
660 : const std::string &cpu_pin_policy);
661 : private:
662 : EventManager *evm_;
663 : TcpSession *session_;
664 : boost::asio::ip::tcp::endpoint server_ep_;
665 : boost::asio::ip::tcp::socket *tcp_socket_;
666 : bool connect_complete_;
667 : char *rx_buff_;
668 : char *rx_buff_rem_;
669 : size_t remain_;
670 : };
671 : #endif // ctrlplane_ksync_sock_h
|
