From f5820b347ea6920764023d6cf71f7a254bd7106d Mon Sep 17 00:00:00 2001 From: "Matthias P. Braendli" Date: Tue, 16 Jan 2018 10:55:15 +0100 Subject: Do some InputReader cleanup --- src/InputReader.h | 150 +++++++++++++++++++++++------------------------------- 1 file changed, 64 insertions(+), 86 deletions(-) (limited to 'src/InputReader.h') diff --git a/src/InputReader.h b/src/InputReader.h index 7d6b373..c897c2d 100644 --- a/src/InputReader.h +++ b/src/InputReader.h @@ -3,7 +3,7 @@ Her Majesty the Queen in Right of Canada (Communications Research Center Canada) - Copyright (C) 2016 + Copyright (C) 2018 Matthias P. Braendli, matthias.braendli@mpb.li http://www.opendigitalradio.org @@ -33,6 +33,7 @@ #include #include +#include #include #if defined(HAVE_ZEROMQ) # include "zmq.hpp" @@ -40,47 +41,8 @@ #endif #include "porting.h" #include "Log.h" -#include "lib/UdpSocket.h" -#include -#include #include -#include -#include -#define SOCKET int #define INVALID_SOCKET -1 -#define SOCKET_ERROR -1 - -/* Known types of input streams. Description taken from the CRC mmbTools forum. - - All numbers are little-endian. - - Framed format is used for file recording. It is the default format. The - padding can be removed from data. Format: - uint32_t nbFrames - for each frame - uint16_t frameSize - uint8_t data[frameSize] - - Streamed format is used for streamed applications. As the total number of - frames is unknown before end of transmission, the corresponding field is - removed. The padding can be removed from data. Format: - for each frame - uint16_t frameSize - uint8_t data[frameSize] - - Raw format is a bit-by-bit (but byte aligned on sync) recording of a G.703 - data stream. The padding is always present. Format: - for each frame - uint8_t data[6144] - - Please note that our raw format can also be referred to as ETI(NI, G.703) or ETI(NI). -*/ -enum EtiStreamType { - ETI_STREAM_TYPE_NONE = 0, - ETI_STREAM_TYPE_RAW, - ETI_STREAM_TYPE_STREAMED, - ETI_STREAM_TYPE_FRAMED, -}; class InputReader { @@ -91,43 +53,25 @@ class InputReader virtual int GetNextFrame(void* buffer) = 0; // Print some information - virtual void PrintInfo() = 0; + virtual void PrintInfo() const = 0; }; class InputFileReader : public InputReader { public: - InputFileReader() : - streamtype_(ETI_STREAM_TYPE_NONE), - inputfile_(NULL) { } - - ~InputFileReader() - { - if (inputfile_ != NULL) { - fprintf(stderr, "\nClosing input file...\n"); - - fclose(inputfile_); - } - } + InputFileReader() = default; + InputFileReader(const InputFileReader& other) = delete; + InputFileReader& operator=(const InputFileReader& other) = delete; // open file and determine stream type // When loop=1, GetNextFrame will never return 0 int Open(std::string filename, bool loop); // Print information about the file opened - void PrintInfo(); - + void PrintInfo() const; int GetNextFrame(void* buffer); - EtiStreamType GetStreamType() - { - return streamtype_; - } - private: - InputFileReader(const InputFileReader& other) = delete; - InputFileReader& operator=(const InputFileReader& other) = delete; - int IdentifyType(); // Rewind the file, and replay anew @@ -136,19 +80,60 @@ class InputFileReader : public InputReader bool loop_; // if shall we loop the file over and over std::string filename_; - EtiStreamType streamtype_; - FILE* inputfile_; - size_t inputfilelength_; - uint64_t nbframes_; // 64-bit because 32-bit overflow is - // after 2**32 * 24ms ~= 3.3 years + /* Known types of input streams. Description taken from the CRC + * mmbTools forum. All values are are little-endian. */ + enum class EtiStreamType { + /* Not yet identified */ + None, + + /* Raw format is a bit-by-bit (but byte aligned on sync) recording + * of a G.703 data stream. The padding is always present. + * The raw format can also be referred to as ETI(NI, G.703) or ETI(NI). + * Format: + for each frame: + uint8_t data[6144] + */ + Raw, + + /* Streamed format is used for streamed applications. As the total + * number of frames is unknown before end of transmission, the + * corresponding field is removed. The padding can be removed from + * data. + * Format: + for each frame: + uint16_t frameSize + uint8_t data[frameSize] + */ + Streamed, + + /* Framed format is used for file recording. It is the default format. + * The padding can be removed from data. + * Format: + uint32_t nbFrames + for each frame: + uint16_t frameSize + uint8_t data[frameSize] + */ + Framed, + }; + + EtiStreamType streamtype_ = EtiStreamType::None; + struct FILEDeleter{ void operator()(FILE* fd){ if(fd) fclose(fd);}}; + std::unique_ptr inputfile_; + + size_t inputfilelength_ = 0; + uint64_t nbframes_ = 0; // 64-bit because 32-bit overflow is + // after 2**32 * 24ms ~= 3.3 years }; class InputTcpReader : public InputReader { public: InputTcpReader(); - ~InputTcpReader(); + InputTcpReader(const InputTcpReader& other) = delete; + InputTcpReader& operator=(const InputTcpReader& other) = delete; + virtual ~InputTcpReader(); // Endpoint is either host:port or tcp://host:port void Open(const std::string& endpoint); @@ -159,21 +144,19 @@ class InputTcpReader : public InputReader virtual int GetNextFrame(void* buffer); // Print some information - virtual void PrintInfo(); + virtual void PrintInfo() const; private: - InputTcpReader(const InputTcpReader& other) = delete; - InputTcpReader& operator=(const InputTcpReader& other) = delete; - SOCKET m_sock; + int m_sock = INVALID_SOCKET; std::string m_uri; }; struct zmq_input_overflow : public std::exception { - const char* what () const throw () - { - return "InputZMQ buffer overflow"; - } + const char* what () const throw () + { + return "InputZMQ buffer overflow"; + } }; #if defined(HAVE_ZEROMQ) @@ -189,17 +172,12 @@ struct InputZeroMQThreadData class InputZeroMQWorker { public: - InputZeroMQWorker() : - running(false), - zmqcontext(1), - m_to_drop(0) { } - void Start(struct InputZeroMQThreadData* workerdata); void Stop(); + bool is_running(void) const { return running; } - bool is_running(void) { return running; } private: - bool running; + std::atomic running = ATOMIC_VAR_INIT(false); void RecvProcess(struct InputZeroMQThreadData* workerdata); @@ -212,7 +190,7 @@ class InputZeroMQWorker * * Here we keep track of how many ETI frames we must drop */ - int m_to_drop; + int m_to_drop = 0; }; class InputZeroMQReader : public InputReader @@ -232,7 +210,7 @@ class InputZeroMQReader : public InputReader int GetNextFrame(void* buffer); - void PrintInfo(); + void PrintInfo() const; private: InputZeroMQReader(const InputZeroMQReader& other) = delete; -- cgit v1.2.3 From 7db444f95310419382146d7f072670f2df855a5f Mon Sep 17 00:00:00 2001 From: "Matthias P. Braendli" Date: Tue, 16 Jan 2018 11:19:26 +0100 Subject: Simplify InputZeroMQReader, remove worker class --- src/InputReader.h | 62 ++++++++++------------------------ src/InputZeroMQReader.cpp | 86 ++++++++++++++++++++++------------------------- 2 files changed, 58 insertions(+), 90 deletions(-) (limited to 'src/InputReader.h') diff --git a/src/InputReader.h b/src/InputReader.h index c897c2d..07326cf 100644 --- a/src/InputReader.h +++ b/src/InputReader.h @@ -162,65 +162,37 @@ struct zmq_input_overflow : public std::exception #if defined(HAVE_ZEROMQ) /* A ZeroMQ input. See www.zeromq.org for more info */ -struct InputZeroMQThreadData -{ - ThreadsafeQueue > > *in_messages; - std::string uri; - size_t max_queued_frames; -}; - -class InputZeroMQWorker +class InputZeroMQReader : public InputReader { public: - void Start(struct InputZeroMQThreadData* workerdata); - void Stop(); - bool is_running(void) const { return running; } + InputZeroMQReader() = default; + InputZeroMQReader(const InputZeroMQReader& other) = delete; + InputZeroMQReader& operator=(const InputZeroMQReader& other) = delete; + ~InputZeroMQReader(); + + int Open(const std::string& uri, size_t max_queued_frames); + int GetNextFrame(void* buffer); + void PrintInfo() const; private: - std::atomic running = ATOMIC_VAR_INIT(false); + std::atomic m_running = ATOMIC_VAR_INIT(false); + std::string m_uri; + size_t m_max_queued_frames = 0; + ThreadsafeQueue > > m_in_messages; - void RecvProcess(struct InputZeroMQThreadData* workerdata); + void RecvProcess(void); - zmq::context_t zmqcontext; // is thread-safe - boost::thread recv_thread; + zmq::context_t m_zmqcontext; // is thread-safe + boost::thread m_recv_thread; /* We must be careful to keep frame phase consistent. If we * drop a single ETI frame, we will break the transmission * frame vs. ETI frame phase. * - * Here we keep track of how many ETI frames we must drop + * Here we keep track of how many ETI frames we must drop. */ int m_to_drop = 0; }; -class InputZeroMQReader : public InputReader -{ - public: - InputZeroMQReader() - { - workerdata_.in_messages = &in_messages_; - } - - ~InputZeroMQReader() - { - worker_.Stop(); - } - - int Open(const std::string& uri, size_t max_queued_frames); - - int GetNextFrame(void* buffer); - - void PrintInfo() const; - - private: - InputZeroMQReader(const InputZeroMQReader& other) = delete; - InputZeroMQReader& operator=(const InputZeroMQReader& other) = delete; - std::string uri_; - - InputZeroMQWorker worker_; - ThreadsafeQueue > > in_messages_; - struct InputZeroMQThreadData workerdata_; -}; - #endif diff --git a/src/InputZeroMQReader.cpp b/src/InputZeroMQReader.cpp index 5d0e513..aa342d5 100644 --- a/src/InputZeroMQReader.cpp +++ b/src/InputZeroMQReader.cpp @@ -3,7 +3,7 @@ Her Majesty the Queen in Right of Canada (Communications Research Center Canada) - Copyright (C) 2017 + Copyright (C) 2018 Matthias P. Braendli, matthias.braendli@mpb.li http://opendigitalradio.org @@ -68,20 +68,28 @@ struct zmq_dab_message_t #define ZMQ_DAB_MESSAGE_T_HEADERSIZE \ (sizeof(uint32_t) + NUM_FRAMES_PER_ZMQ_MESSAGE*sizeof(uint16_t)) +InputZeroMQReader::~InputZeroMQReader() +{ + m_running = false; + m_zmqcontext.close(); + if (m_recv_thread.joinable()) { + m_recv_thread.join(); + } +} + int InputZeroMQReader::Open(const string& uri, size_t max_queued_frames) { // The URL might start with zmq+tcp:// if (uri.substr(0, 4) == "zmq+") { - uri_ = uri.substr(4); + m_uri = uri.substr(4); } else { - uri_ = uri; + m_uri = uri; } - workerdata_.uri = uri_; - workerdata_.max_queued_frames = max_queued_frames; - // launch receiver thread - worker_.Start(&workerdata_); + m_max_queued_frames = max_queued_frames; + + m_recv_thread = boost::thread(&InputZeroMQReader::RecvProcess, this); return 0; } @@ -90,7 +98,7 @@ int InputZeroMQReader::GetNextFrame(void* buffer) { const size_t framesize = 6144; - if (not worker_.is_running()) { + if (not m_running) { return 0; } @@ -100,18 +108,18 @@ int InputZeroMQReader::GetNextFrame(void* buffer) * (4 ETI frames in TM1) and we should make sure that * we can serve the data required for a full transmission frame. */ - if (in_messages_.size() < 4) { + if (m_in_messages.size() < 4) { const size_t prebuffering = 10; etiLog.log(trace, "ZMQ,wait1"); - in_messages_.wait_and_pop(incoming, prebuffering); + m_in_messages.wait_and_pop(incoming, prebuffering); } else { etiLog.log(trace, "ZMQ,wait2"); - in_messages_.wait_and_pop(incoming); + m_in_messages.wait_and_pop(incoming); } etiLog.log(trace, "ZMQ,pop"); - if (not worker_.is_running()) { + if (not m_running) { throw zmq_input_overflow(); } @@ -123,54 +131,55 @@ int InputZeroMQReader::GetNextFrame(void* buffer) void InputZeroMQReader::PrintInfo() const { fprintf(stderr, "Input ZeroMQ:\n"); - fprintf(stderr, " Receiving from %s\n\n", uri_.c_str()); + fprintf(stderr, " Receiving from %s\n\n", m_uri.c_str()); } -// ------------- Worker functions - -void InputZeroMQWorker::RecvProcess(struct InputZeroMQThreadData* workerdata) +void InputZeroMQReader::RecvProcess() { set_thread_name("zmqinput"); - size_t queue_size = 0; + m_running = true; + size_t queue_size = 0; bool buffer_full = false; - zmq::socket_t subscriber(zmqcontext, ZMQ_SUB); + zmq::socket_t subscriber(m_zmqcontext, ZMQ_SUB); // zmq sockets are not thread safe. That's why // we create it here, and not at object creation. bool success = true; try { - subscriber.connect(workerdata->uri.c_str()); + subscriber.connect(m_uri.c_str()); } catch (zmq::error_t& err) { - etiLog.level(error) << "Failed to connect ZeroMQ socket to '" << workerdata->uri << "': '" << err.what() << "'"; + etiLog.level(error) << "Failed to connect ZeroMQ socket to '" << + m_uri << "': '" << err.what() << "'"; success = false; } if (success) try { - subscriber.setsockopt(ZMQ_SUBSCRIBE, NULL, 0); // subscribe to all messages + // subscribe to all messages + subscriber.setsockopt(ZMQ_SUBSCRIBE, NULL, 0); } catch (zmq::error_t& err) { - etiLog.level(error) << "Failed to subscribe ZeroMQ socket to messages: '" << err.what() << "'"; + etiLog.level(error) << "Failed to subscribe ZeroMQ socket to messages: '" << + err.what() << "'"; success = false; } if (success) try { - while (running) - { + while (m_running) { zmq::message_t incoming; subscriber.recv(&incoming); if (m_to_drop) { - queue_size = workerdata->in_messages->size(); + queue_size = m_in_messages.size(); if (queue_size > 4) { - workerdata->in_messages->notify(); + m_in_messages.notify(); } m_to_drop--; } - else if (queue_size < workerdata->max_queued_frames) { + else if (queue_size < m_max_queued_frames) { if (buffer_full) { etiLog.level(info) << "ZeroMQ buffer recovered: " << queue_size << " elements"; @@ -214,14 +223,14 @@ void InputZeroMQWorker::RecvProcess(struct InputZeroMQThreadData* workerdata) offset += framesize; - queue_size = workerdata->in_messages->push(buf); + queue_size = m_in_messages.push(buf); etiLog.log(trace, "ZMQ,push %zu", queue_size); } } } } else { - workerdata->in_messages->notify(); + m_in_messages.notify(); if (!buffer_full) { etiLog.level(warn) << "ZeroMQ buffer overfull !"; @@ -230,7 +239,7 @@ void InputZeroMQWorker::RecvProcess(struct InputZeroMQThreadData* workerdata) throw runtime_error("ZMQ input full"); } - queue_size = workerdata->in_messages->size(); + queue_size = m_in_messages.size(); /* Drop three more incoming ETI frames before * we start accepting them again, to guarantee @@ -256,21 +265,8 @@ void InputZeroMQWorker::RecvProcess(struct InputZeroMQThreadData* workerdata) subscriber.close(); - running = false; - workerdata->in_messages->notify(); -} - -void InputZeroMQWorker::Start(struct InputZeroMQThreadData* workerdata) -{ - running = true; - recv_thread = boost::thread(&InputZeroMQWorker::RecvProcess, this, workerdata); -} - -void InputZeroMQWorker::Stop() -{ - running = false; - zmqcontext.close(); - recv_thread.join(); + m_running = false; + m_in_messages.notify(); } #endif -- cgit v1.2.3