public:
- virtual ~CFileWriterBase(){};
- virtual bool Create(char * name) = 0;
+ virtual ~CFileWriterBase(){};
+ virtual bool Create(char * name) = 0;
virtual bool write_packet(CCapPktRaw * lpPacket)=0;
+ virtual void flush_to_disk() = 0;
};
return true;
}
-
+void CErfFileWriter::flush_to_disk() {
+ if (m_fd) {
+ fflush(m_fd);
+ }
+}
bool CPcapFileWriter::Create(char *file_name){
m_fd=CAP_FOPEN_64(file_name,"wb");
virtual bool Create(char *file_name);
void Delete();
virtual bool write_packet(CCapPktRaw * lpPacket);
+
+ /**
+ * flush all packets to disk
+ *
+ */
+ void flush_to_disk();
+
private:
FILE *m_fd;
int m_cnt;
return init();
}
+void LibPCapWriter::flush_to_disk() {
+ if (m_is_open) {
+ fflush(m_file_handler);
+ }
+}
+
/**
*
* Write the libpcap header.
*/
void Close();
+ /**
+ * flush all packets to disk
+ *
+ * @author imarom (11/24/2016)
+ */
+ void flush_to_disk();
+
private:
bool init();
m_rx_features_info.m_rx_capture_info.enable(pcap_filename, limit);
- TrexStatelessCpToRxMsgBase *msg = new TrexStatelessRxStartCapture(m_port_id, m_rx_features_info.m_rx_capture_info);
- send_message_to_rx(msg);
+ TrexStatelessRxStartCapture *msg = new TrexStatelessRxStartCapture(m_port_id, m_rx_features_info.m_rx_capture_info);
+ send_message_to_rx((TrexStatelessCpToRxMsgBase *)msg);
+
+ /* as below, must wait for ACK from RX core before returning ACK */
+ msg->wait_for_reply();
}
void
TrexStatelessPort::stop_rx_capture() {
TrexStatelessCpToRxMsgBase *msg = new TrexStatelessRxStopCapture(m_port_id);
send_message_to_rx(msg);
+
+ /* update our cached data */
m_rx_features_info.m_rx_capture_info.disable();
}
m_rx_features_info.m_rx_queue_info.enable(size);
- TrexStatelessCpToRxMsgBase *msg = new TrexStatelessRxStartQueue(m_port_id, m_rx_features_info.m_rx_queue_info);
- send_message_to_rx(msg);
+ TrexStatelessRxStartQueue *msg = new TrexStatelessRxStartQueue(m_port_id, m_rx_features_info.m_rx_queue_info);
+ send_message_to_rx( (TrexStatelessCpToRxMsgBase *)msg );
+
+ /* we cannot return ACK to the user until the RX core has approved
+ this might cause the user to lose some packets from the queue
+ */
+ msg->wait_for_reply();
}
void
TrexStatelessPort::stop_rx_queue() {
TrexStatelessCpToRxMsgBase *msg = new TrexStatelessRxStopQueue(m_port_id);
send_message_to_rx(msg);
+
+ /* update our cached data */
m_rx_features_info.m_rx_queue_info.disable();
}
return NULL;
}
- /* ask RX core for the pkt queue */
- TrexStatelessMsgReply<RXPacketBuffer *> msg_reply;
-
- TrexStatelessCpToRxMsgBase *msg = new TrexStatelessRxQueueGetPkts(m_port_id, msg_reply);
- send_message_to_rx(msg);
+ TrexStatelessRxQueueGetPkts *msg = new TrexStatelessRxQueueGetPkts(m_port_id);
+ send_message_to_rx( (TrexStatelessCpToRxMsgBase *)msg );
- RXPacketBuffer *pkt_buffer = msg_reply.wait_for_reply();
+ RXPacketBuffer *pkt_buffer = msg->wait_for_reply();
return pkt_buffer;
}
bool
TrexStatelessRxStartCapture::handle(CRxCoreStateless *rx_core) {
- rx_core->start_capture(m_port_id, m_pcap_filename, m_limit, m_shared_counter);
+ rx_core->start_recorder(m_port_id, m_pcap_filename, m_limit, m_shared_counter);
+ set_reply(true);
+
return true;
}
bool
TrexStatelessRxStopCapture::handle(CRxCoreStateless *rx_core) {
- rx_core->stop_capture(m_port_id);
+ rx_core->stop_recorder(m_port_id);
return true;
}
bool
TrexStatelessRxStartQueue::handle(CRxCoreStateless *rx_core) {
rx_core->start_queue(m_port_id, m_size, m_shared_counter);
-
+
+ /* mark as done */
+ set_reply(true);
+
return true;
}
bool TrexStatelessRxQueueGetPkts::handle(CRxCoreStateless *rx_core) {
RXPacketBuffer *pkt_buffer = rx_core->get_rx_queue_pkts(m_port_id);
assert(pkt_buffer);
- m_reply.set(pkt_buffer);
+
+ /* set the reply */
+ set_reply(pkt_buffer);
return true;
}
};
+/**
+ * defines the base class for CP to RX with reply
+ *
+ * @author imarom (11/27/2016)
+ */
+template<typename T> class TrexStatelessCpToRxMsgReply : public TrexStatelessCpToRxMsgBase {
+
+public:
+
+ TrexStatelessCpToRxMsgReply() {
+ m_pending = true;
+ }
+
+ bool is_pending() const {
+ return m_pending;
+ }
+
+ void set_reply(const T &reply) {
+ m_reply = reply;
+
+ /* before marking as done - memory fence */
+ asm volatile("mfence" ::: "memory");
+ m_pending = false;
+ }
+
+ T wait_for_reply(int timeout_ms = 100, int backoff_ms = 1) {
+ int guard = timeout_ms;
+
+ while (is_pending()) {
+ guard -= backoff_ms;
+ if (guard < 0) {
+ throw TrexException("timeout: RX core has failed to reply");
+ }
+
+ delay(backoff_ms);
+ }
+
+ return m_reply;
+
+ }
+
+protected:
+ volatile bool m_pending;
+ T m_reply;
+};
+
+
class TrexStatelessRxEnableLatency : public TrexStatelessCpToRxMsgBase {
bool handle (CRxCoreStateless *rx_core);
};
};
-class TrexStatelessRxStartCapture : public TrexStatelessCpToRxMsgBase {
+
+class TrexStatelessRxStartCapture : public TrexStatelessCpToRxMsgReply<bool> {
public:
TrexStatelessRxStartCapture(uint8_t port_id, RXCaptureInfo &rx_capture_info) {
m_port_id = port_id;
uint8_t m_port_id;
};
-class TrexStatelessRxStartQueue : public TrexStatelessCpToRxMsgBase {
+
+class TrexStatelessRxStartQueue : public TrexStatelessCpToRxMsgReply<bool> {
public:
TrexStatelessRxStartQueue(uint8_t port_id, RXQueueInfo &rx_queue_info) {
m_port_id = port_id;
uint64_t *m_shared_counter;
};
+
class TrexStatelessRxStopQueue : public TrexStatelessCpToRxMsgBase {
public:
TrexStatelessRxStopQueue(uint8_t port_id) {
};
-template<typename T> class TrexStatelessMsgReply {
-public:
- TrexStatelessMsgReply() {
- m_pending = true;
- }
-
- bool is_pending() const {
- return m_pending;
- }
-
- void set(T reply) {
- m_reply = reply;
-
- /* before marking as done - memory fence */
- asm volatile("mfence" ::: "memory");
- m_pending = false;
- }
-
- T wait_for_reply(int timeout_ms = 100, int backoff_ms = 1) {
- int guard = timeout_ms;
-
- while (is_pending()) {
- guard -= backoff_ms;
- if (guard < 0) {
- throw TrexException("timeout: RX core has failed to reply");
- }
-
- delay(backoff_ms);
-
- }
- return m_reply;
-
- }
-private:
- bool m_pending;
- T m_reply;
-};
-
-
-class TrexStatelessRxQueueGetPkts : public TrexStatelessCpToRxMsgBase {
+class TrexStatelessRxQueueGetPkts : public TrexStatelessCpToRxMsgReply<RXPacketBuffer *> {
public:
- TrexStatelessRxQueueGetPkts(uint8_t port_id, TrexStatelessMsgReply<RXPacketBuffer *> &reply) : m_reply(reply) {
+ TrexStatelessRxQueueGetPkts(uint8_t port_id) {
m_port_id = port_id;
}
virtual bool handle(CRxCoreStateless *rx_core);
private:
- uint8_t m_port_id;
- TrexStatelessMsgReply<RXPacketBuffer*> &m_reply;
+ uint8_t m_port_id;
};
m_cpu_cp_u.Create(&m_cpu_dp_u);
- /* init per port manager */
+ /* create per port manager */
for (int i = 0; i < m_max_ports; i++) {
m_rx_port_mngr[i].create(cfg.m_ports[i],
m_rfc2544,
}
}
+/**
+ * for each port give a tick (for flushing if needed)
+ *
+ */
+void CRxCoreStateless::port_manager_tick() {
+ for (int i = 0; i < m_max_ports; i++) {
+ m_rx_port_mngr[i].tick();
+ }
+}
+
void CRxCoreStateless::handle_work_stage(bool do_try_rx_queue) {
int i = 0;
-
+ int j = 0;
+
while (m_state == STATE_WORKING) {
if (do_try_rx_queue) {
process_all_pending_pkts();
+ /* TODO: with scheduler, this should be solved better */
i++;
if (i == 100000) { // approx 10msec
i = 0;
periodic_check_for_cp_messages(); // m_state might change in here
+
+ j++;
+ if (j == 100) { // approx 1 sec
+ j = 0;
+ port_manager_tick();
+ }
}
-
+
rte_pause();
}
}
void
-CRxCoreStateless::start_capture(uint8_t port_id, const std::string &pcap_filename, uint64_t limit, uint64_t *shared_counter) {
- m_rx_port_mngr[port_id].start_capture(pcap_filename, limit, shared_counter);
+CRxCoreStateless::start_recorder(uint8_t port_id, const std::string &pcap_filename, uint64_t limit, uint64_t *shared_counter) {
+ m_rx_port_mngr[port_id].start_recorder(pcap_filename, limit, shared_counter);
}
void
-CRxCoreStateless::stop_capture(uint8_t port_id) {
- m_rx_port_mngr[port_id].stop_capture();
+CRxCoreStateless::stop_recorder(uint8_t port_id) {
+ m_rx_port_mngr[port_id].stop_recorder();
}
void
* @param pcap_filename
* @param limit
*/
- void start_capture(uint8_t port_id, const std::string &pcap_filename, uint64_t limit, uint64_t *shared_counter);
- void stop_capture(uint8_t port_id);
+ void start_recorder(uint8_t port_id, const std::string &pcap_filename, uint64_t limit, uint64_t *shared_counter);
+ void stop_recorder(uint8_t port_id);
/**
* start RX queueing of packets
void capture_pkt(rte_mbuf_t *m);
void handle_rx_queue_msgs(uint8_t thread_id, CNodeRing * r);
void handle_work_stage(bool do_try_rx_queue);
-
+ void port_manager_tick();
+
int process_all_pending_pkts(bool flush_rx = false);
void flush_all_pending_pkts() {
}
void
-RXPacketBuffer::handle_pkt(const rte_mbuf_t *m) {
+RXPacketBuffer::push(const rte_mbuf_t *m) {
assert(m_is_enabled);
/* if full - pop the oldest */
return output;
}
-/****************************** packet recorder ****************************/
+
+void
+RXQueue::start(uint64_t size, uint64_t *shared_counter) {
+ if (m_pkt_buffer) {
+ delete m_pkt_buffer;
+ }
+ m_pkt_buffer = new RXPacketBuffer(size, shared_counter);
+}
+
+void
+RXQueue::stop() {
+ if (m_pkt_buffer) {
+ delete m_pkt_buffer;
+ m_pkt_buffer = NULL;
+ }
+}
+
+RXPacketBuffer *
+RXQueue::fetch() {
+
+ if (!m_pkt_buffer) {
+ return nullptr;
+ }
+
+ /* hold a pointer to the old one */
+ RXPacketBuffer *old_buffer = m_pkt_buffer;
+
+ /* replace the old one with a new one and freeze the old */
+ m_pkt_buffer = old_buffer->freeze_and_clone();
+
+ return old_buffer;
+}
+
+void
+RXQueue::handle_pkt(const rte_mbuf_t *m) {
+ m_pkt_buffer->push(m);
+}
+
+/**************************************
+ * RX feature recorder
+ *
+ *************************************/
RXPacketRecorder::RXPacketRecorder() {
m_writer = NULL;
m_epoch = -1;
}
-RXPacketRecorder::~RXPacketRecorder() {
- stop();
-}
-
void
RXPacketRecorder::start(const std::string &pcap, uint64_t limit, uint64_t *shared_counter) {
m_writer = CCapWriterFactory::CreateWriter(LIBPCAP, (char *)pcap.c_str());
}
}
+void
+RXPacketRecorder::flush_to_disk() {
+ if (m_writer) {
+ m_writer->flush_to_disk();
+ }
+}
+
void
RXPacketRecorder::handle_pkt(const rte_mbuf_t *m) {
if (!m_writer) {
}
+/**************************************
+ * Port manager
+ *
+ *************************************/
+
+RXPortManager::RXPortManager() {
+ clear_all_features();
+ m_io = NULL;
+ m_cpu_dp_u = NULL;
+}
+
+
+void
+RXPortManager::create(CPortLatencyHWBase *io,
+ CRFC2544Info *rfc2544,
+ CRxCoreErrCntrs *err_cntrs,
+ CCpuUtlDp *cpu_util) {
+ m_io = io;
+ m_cpu_dp_u = cpu_util;
+
+ /* init features */
+ m_latency.create(rfc2544, err_cntrs);
+}
+
void RXPortManager::handle_pkt(const rte_mbuf_t *m) {
/* handle features */
m_latency.handle_pkt(m);
}
- if (is_feature_set(CAPTURE)) {
+ if (is_feature_set(RECORDER)) {
m_recorder.handle_pkt(m);
}
if (is_feature_set(QUEUE)) {
- m_pkt_buffer->handle_pkt(m);
+ m_queue.handle_pkt(m);
}
}
return cnt_p;
}
+void
+RXPortManager::tick() {
+ if (is_feature_set(RECORDER)) {
+ m_recorder.flush_to_disk();
+ }
+}
~RXPacketBuffer();
/**
- * handle a new packet
+ * push a packet to the buffer
*
*/
- void handle_pkt(const rte_mbuf_t *m);
+ void push(const rte_mbuf_t *m);
/**
* freezes the queue and clones a new one
};
+class RXQueue {
+public:
+ RXQueue() {
+ m_pkt_buffer = nullptr;
+ }
+
+ ~RXQueue() {
+ stop();
+ }
+
+ /**
+ * start RX queue
+ *
+ */
+ void start(uint64_t size, uint64_t *shared_counter);
+
+ /**
+ * fetch the current buffer
+ *
+ */
+ RXPacketBuffer * fetch();
+
+ /**
+ * stop RX queue
+ *
+ */
+ void stop();
+
+ void handle_pkt(const rte_mbuf_t *m);
+
+private:
+ RXPacketBuffer *m_pkt_buffer;
+};
+
/**************************************
* RX feature PCAP recorder
*
class RXPacketRecorder {
public:
RXPacketRecorder();
- ~RXPacketRecorder();
+
+ ~RXPacketRecorder() {
+ stop();
+ }
void start(const std::string &pcap, uint64_t limit, uint64_t *shared_counter);
void stop();
void handle_pkt(const rte_mbuf_t *m);
+ /**
+ * flush any cached packets to disk
+ *
+ */
+ void flush_to_disk();
+
private:
CFileWriterBase *m_writer;
CCapPktRaw m_pkt;
*/
class RXPortManager {
public:
- enum features_t {
- LATENCY = 0x1,
- CAPTURE = 0x2,
- QUEUE = 0x4
+ enum feature_t {
+ NO_FEATURES = 0x0,
+ LATENCY = 0x1,
+ RECORDER = 0x2,
+ QUEUE = 0x4
};
- RXPortManager() {
- m_features = 0;
- m_pkt_buffer = NULL;
- m_io = NULL;
- m_cpu_dp_u = NULL;
- }
+ RXPortManager();
void create(CPortLatencyHWBase *io,
CRFC2544Info *rfc2544,
CRxCoreErrCntrs *err_cntrs,
- CCpuUtlDp *cpu_util) {
- m_io = io;
- m_cpu_dp_u = cpu_util;
- m_latency.create(rfc2544, err_cntrs);
- }
+ CCpuUtlDp *cpu_util);
void clear_stats() {
m_latency.reset_stats();
return m_latency;
}
+ /* latency */
void enable_latency() {
set_feature(LATENCY);
}
unset_feature(LATENCY);
}
- /**
- * capturing of RX packets
- *
- * @author imarom (11/2/2016)
- *
- * @param pcap
- * @param limit_pkts
- */
- void start_capture(const std::string &pcap, uint64_t limit_pkts, uint64_t *shared_counter) {
+ /* recorder */
+ void start_recorder(const std::string &pcap, uint64_t limit_pkts, uint64_t *shared_counter) {
m_recorder.start(pcap, limit_pkts, shared_counter);
- set_feature(CAPTURE);
+ set_feature(RECORDER);
}
- void stop_capture() {
+ void stop_recorder() {
m_recorder.stop();
- unset_feature(CAPTURE);
+ unset_feature(RECORDER);
}
- /**
- * queueing packets
- *
- */
+ /* queue */
void start_queue(uint32_t size, uint64_t *shared_counter) {
- if (m_pkt_buffer) {
- delete m_pkt_buffer;
- }
- m_pkt_buffer = new RXPacketBuffer(size, shared_counter);
+ m_queue.start(size, shared_counter);
set_feature(QUEUE);
}
void stop_queue() {
- if (m_pkt_buffer) {
- delete m_pkt_buffer;
- m_pkt_buffer = NULL;
- }
- unset_feature(QUEUE);
+ m_queue.stop();
+ unset_feature(QUEUE);
}
RXPacketBuffer *get_pkt_buffer() {
if (!is_feature_set(QUEUE)) {
- return NULL;
+ return nullptr;
}
-
- assert(m_pkt_buffer);
-
- /* hold a pointer to the old one */
- RXPacketBuffer *old_buffer = m_pkt_buffer;
-
- /* replace the old one with a new one and freeze the old */
- m_pkt_buffer = old_buffer->freeze_and_clone();
-
- return old_buffer;
+
+ return m_queue.fetch();
}
+
/**
* fetch and process all packets
*/
void handle_pkt(const rte_mbuf_t *m);
-
+ /**
+ * maintenance
+ *
+ * @author imarom (11/24/2016)
+ */
+ void tick();
+
bool has_features_set() {
- return (m_features != 0);
+ return (m_features != NO_FEATURES);
}
private:
+ void clear_all_features() {
+ m_features = NO_FEATURES;
+ }
- void set_feature(features_t feature) {
+ void set_feature(feature_t feature) {
m_features |= feature;
}
- void unset_feature(features_t feature) {
+ void unset_feature(feature_t feature) {
m_features &= (~feature);
}
- bool is_feature_set(features_t feature) {
+ bool is_feature_set(feature_t feature) const {
return ( (m_features & feature) == feature );
}
uint32_t m_features;
- RXPacketRecorder m_recorder;
+
RXLatency m_latency;
- RXPacketBuffer *m_pkt_buffer;
+ RXPacketRecorder m_recorder;
+ RXQueue m_queue;
+
+
CCpuUtlDp *m_cpu_dp_u;
CPortLatencyHWBase *m_io;
};
Code Review / trex.git / commitdiff