/** * Copyright (c) 2021, Timothy Stack * * All rights reserved. * * Redistribution and use in source and binary forms, with or without * modification, are permitted provided that the following conditions are met: * * * Redistributions of source code must retain the above copyright notice, this * list of conditions and the following disclaimer. * * Redistributions in binary form must reproduce the above copyright notice, * this list of conditions and the following disclaimer in the documentation * and/or other materials provided with the distribution. * * Neither the name of Timothy Stack nor the names of its contributors * may be used to endorse or promote products derived from this software * without specific prior written permission. * * THIS SOFTWARE IS PROVIDED BY THE REGENTS AND CONTRIBUTORS ''AS IS'' AND ANY * EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE * DISCLAIMED. IN NO EVENT SHALL THE REGENTS OR CONTRIBUTORS BE LIABLE FOR ANY * DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND * ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. * * @file isc.hh */ #include #include #include #include #include #include #include #include "injector.hh" #include "safe/safe.h" #include "time_util.hh" #ifndef lnav_isc_hh # define lnav_isc_hh namespace isc { struct msg { std::function m_callback; }; inline msg empty_msg() { return {[]() {}}; } class msg_port { public: msg_port() = default; void send(msg&& m) { safe::WriteAccess writable_msgs( this->mp_messages); writable_msgs->emplace_back(m); this->sp_cond.notify_all(); } template void process_for(const std::chrono::duration& rel_time) { std::deque tmp_msgs; { safe::WriteAccess writable_msgs(this->mp_messages); if (writable_msgs->empty()) { this->sp_cond.template wait_for(writable_msgs.lock, rel_time); } tmp_msgs.swap(*writable_msgs); } while (!tmp_msgs.empty()) { auto& m = tmp_msgs.front(); m.m_callback(); tmp_msgs.pop_front(); } } private: using message_list = std::deque; using safe_message_list = safe::Safe; std::condition_variable sp_cond; safe_message_list mp_messages; }; class service_base; using service_list = std::vector>; struct supervisor { explicit supervisor(service_list servs = {}, service_base* parent = nullptr); ~supervisor(); bool empty() const { return this->s_service_list.empty(); } void add_child_service(std::shared_ptr new_service); void stop_children(); void cleanup_children(); protected: service_list s_service_list; service_base* s_parent; }; class service_base : public std::enable_shared_from_this { public: explicit service_base(std::string name) : s_name(std::move(name)), s_children({}, this) { } virtual ~service_base() = default; bool is_looping() const { return this->s_looping; } msg_port& get_port() { return this->s_port; } friend supervisor; private: void start(); void stop(); protected: virtual void* run(); virtual void loop_body(){}; virtual void child_finished(std::shared_ptr child){}; virtual void stopped(){}; virtual std::chrono::milliseconds compute_timeout( mstime_t current_time) const { using namespace std::literals::chrono_literals; return 1s; }; const std::string s_name; bool s_started{false}; std::thread s_thread; std::atomic s_looping{true}; msg_port s_port; supervisor s_children; }; template class service : public service_base { public: explicit service(std::string sub_name = "") : service_base(std::string(__PRETTY_FUNCTION__) + " " + sub_name) { } template void send(F msg) { this->s_port.send({[lifetime = this->shared_from_this(), this, msg]() { msg(*(static_cast(this))); }}); } template void send_and_wait(F msg, const std::chrono::duration& rel_time) { msg_port reply_port; this->s_port.send( {[lifetime = this->shared_from_this(), this, &reply_port, msg]() { msg(*(static_cast(this))); reply_port.send(empty_msg()); }}); reply_port.template process_for(rel_time); } }; template struct to { void send(std::function cb) { auto& service = injector::get(); service.send(cb); } template void send_and_wait(std::function cb, const std::chrono::duration& rel_time) { auto& service = injector::get(); service.send_and_wait(cb, rel_time); } void send_and_wait(std::function cb) { using namespace std::literals::chrono_literals; this->send_and_wait(cb, 48h); } }; } // namespace isc #endif