X-Git-Url: https://gerrit.automotivelinux.org/gerrit/gitweb?a=blobdiff_plain;f=CAN-binder%2Flow-can-binding%2Fcan%2Fcan-bus.cpp;h=6f9e78d958f7e71682dfb458d2de56e725e6ed36;hb=e00c07be2835dd082862730189c87baacbfbdbeb;hp=61730a17afc2810790b51c4b91fbad2adcfd1abf;hpb=ebd4cbb403fb05ed21eaf5150244dfd91bd44fb2;p=apps%2Fagl-service-can-low-level.git diff --git a/CAN-binder/low-can-binding/can/can-bus.cpp b/CAN-binder/low-can-binding/can/can-bus.cpp index 61730a17..6f9e78d9 100644 --- a/CAN-binder/low-can-binding/can/can-bus.cpp +++ b/CAN-binder/low-can-binding/can/can-bus.cpp @@ -15,14 +15,15 @@ * limitations under the License. */ -#include -#include -#include -#include #include #include #include #include +#include +#include +#include +#include +#include #include "can-bus.hpp" @@ -46,6 +47,49 @@ can_bus_t::can_bus_t(utils::config_parser_t conf_file) std::map> can_bus_t::can_devices_; +/// @brief Listen for all device sockets and fill can_messages_queue with them. +/// Reading blocks until message arrive on listened sockets. +/// +/// @return 0 if ok -1 if not +int can_bus_t::can_reader() +{ + fd_set rfds; + int sock_max = INVALID_SOCKET; + + FD_ZERO(&rfds); + + for(auto& can_dev : can_devices_) + { + FD_SET(can_dev.second->get_socket().socket(), &rfds); + if (sock_max < can_dev.second->get_socket().socket()) + sock_max = can_dev.second->get_socket().socket(); + } + + int ret; + while(is_reading_) + { + ret = select(sock_max + 1, &rfds, nullptr, nullptr, nullptr); + + if(ret == -1) + perror("select()"); + else if(ret > 0) + { + for(const auto& s: can_devices_) + { + if(FD_ISSET(s.second->get_socket().socket(), &rfds)) + { + can_message_t msg; + s.second->get_socket() >> msg; + push_new_can_message(msg); + } + } + } + else + printf("Timeout\n"); + } + return 0; +} + /// @brief Will make the decoding operation on a classic CAN message. It will not /// handle CAN commands nor diagnostic messages that have their own method to get /// this happens. @@ -59,19 +103,21 @@ std::map> can_bus_t::can_devices_; int can_bus_t::process_can_signals(can_message_t& can_message) { int processed_signals = 0; - std::vector signals; + struct utils::signals_found signals; openxc_DynamicField search_key, decoded_message; openxc_VehicleMessage vehicle_message; + configuration_t& conf = configuration_t::instance(); + utils::signals_manager_t& sm = utils::signals_manager_t::instance(); // First we have to found which can_signal_t it is search_key = build_DynamicField((double)can_message.get_id()); - configuration_t::instance().find_can_signals(search_key, signals); + signals = sm.find_signals(search_key); // Decoding the message ! Don't kill the messenger ! - for(auto& sig : signals) + for(auto& sig : signals.can_signals) { - std::lock_guard subscribed_signals_lock(get_subscribed_signals_mutex()); - std::map& s = get_subscribed_signals(); + std::lock_guard subscribed_signals_lock(sm.get_subscribed_signals_mutex()); + std::map& s = sm.get_subscribed_signals(); // DEBUG message to make easier debugger STL containers... //DEBUG(binder_interface, "Operator[] key char: %s, event valid? %d", sig.generic_name, afb_event_is_valid(s[sig.generic_name])); @@ -80,7 +126,7 @@ int can_bus_t::process_can_signals(can_message_t& can_message) //DEBUG(binder_interface, "Nb elt matched string: %d", (int)s.count(std::string(sig.generic_name)); if( s.find(sig->get_name()) != s.end() && afb_event_is_valid(s[sig->get_name()])) { - decoded_message = decoder_t::translateSignal(*sig, can_message, configuration_t::instance().get_can_signals()); + decoded_message = decoder_t::translateSignal(*sig, can_message, conf.get_can_signals()); openxc_SimpleMessage s_message = build_SimpleMessage(sig->get_name(), decoded_message); vehicle_message = build_VehicleMessage(s_message); @@ -91,7 +137,7 @@ int can_bus_t::process_can_signals(can_message_t& can_message) } } - DEBUG(binder_interface, "process_can_signals: %d/%d CAN signals processed.", processed_signals, (int)signals.size()); + DEBUG(binder_interface, "%s: %d/%d CAN signals processed.", __FUNCTION__, processed_signals, (int)signals.can_signals.size()); return processed_signals; } @@ -107,8 +153,10 @@ int can_bus_t::process_diagnostic_signals(diagnostic_manager_t& manager, const c { int processed_signals = 0; - std::lock_guard subscribed_signals_lock(get_subscribed_signals_mutex()); - std::map& s = get_subscribed_signals(); + utils::signals_manager_t& sm = utils::signals_manager_t::instance(); + + std::lock_guard subscribed_signals_lock(sm.get_subscribed_signals_mutex()); + std::map& s = sm.get_subscribed_signals(); openxc_VehicleMessage vehicle_message = manager.find_and_decode_adr(can_message); if( (vehicle_message.has_simple_message && vehicle_message.simple_message.has_name) && @@ -164,6 +212,7 @@ void can_bus_t::can_event_push() openxc_VehicleMessage v_message; openxc_SimpleMessage s_message; json_object* jo; + utils::signals_manager_t& sm = utils::signals_manager_t::instance(); while(is_pushing_) { @@ -175,8 +224,8 @@ void can_bus_t::can_event_push() s_message = get_simple_message(v_message); { - std::lock_guard subscribed_signals_lock(get_subscribed_signals_mutex()); - std::map& s = get_subscribed_signals(); + std::lock_guard subscribed_signals_lock(sm.get_subscribed_signals_mutex()); + std::map& s = sm.get_subscribed_signals(); if(s.find(std::string(s_message.name)) != s.end() && afb_event_is_valid(s[std::string(s_message.name)])) { jo = json_object_new_object(); @@ -193,6 +242,11 @@ void can_bus_t::can_event_push() /// and push subscribed events. void can_bus_t::start_threads() { + is_reading_ = true; + th_reading_ = std::thread(&can_bus_t::can_reader, this); + if(!th_reading_.joinable()) + is_reading_ = false; + is_decoding_ = true; th_decoding_ = std::thread(&can_bus_t::can_decode_message, this); if(!th_decoding_.joinable()) @@ -209,6 +263,7 @@ void can_bus_t::start_threads() /// they'll finish their job. void can_bus_t::stop_threads() { + is_reading_ = false; is_decoding_ = false; is_pushing_ = false; } @@ -237,26 +292,27 @@ int can_bus_t::init_can_dev() for(const auto& device : devices_name) { can_bus_t::can_devices_[device] = std::make_shared(device, i); - if (can_bus_t::can_devices_[device]->open() == 0) + if (can_bus_t::can_devices_[device]->open() >= 0) { - DEBUG(binder_interface, "Start reading thread"); - NOTICE(binder_interface, "%s device opened and reading", device.c_str()); - can_bus_t::can_devices_[device]->start_reading(*this); + can_bus_t::can_devices_[device]->configure(); + DEBUG(binder_interface, "%s: Start reading thread", __FUNCTION__); + NOTICE(binder_interface, "%s: %s device opened and reading", __FUNCTION__, device.c_str()); + //can_bus_t::can_devices_[device]->start_reading(*this); i++; } else { - ERROR(binder_interface, "Can't open device %s", device.c_str()); + ERROR(binder_interface, "%s: Can't open device %s", __FUNCTION__, device.c_str()); return 1; } } - NOTICE(binder_interface, "Initialized %d/%d can bus device(s)", i, (int)t); + NOTICE(binder_interface, "%s: Initialized %d/%d can bus device(s)", __FUNCTION__, i, (int)t); return 0; } - ERROR(binder_interface, "init_can_dev: Error at CAN device initialization. No devices read from configuration file"); + ERROR(binder_interface, "%s: Error at CAN device initialization. No devices read from configuration file", __FUNCTION__); return 1; } - ERROR(binder_interface, "init_can_dev: Can't read INI configuration file"); + ERROR(binder_interface, "%s: Can't read INI configuration file", __FUNCTION__); return 2; } @@ -287,7 +343,7 @@ can_message_t can_bus_t::next_can_message() { can_msg = can_message_q_.front(); can_message_q_.pop(); - DEBUG(binder_interface, "next_can_message: Here is the next can message : id %X, length %X, data %02X%02X%02X%02X%02X%02X%02X%02X", can_msg.get_id(), can_msg.get_length(), + DEBUG(binder_interface, "%s: Here is the next can message : id %X, length %X, data %02X%02X%02X%02X%02X%02X%02X%02X", __FUNCTION__, can_msg.get_id(), can_msg.get_length(), can_msg.get_data()[0], can_msg.get_data()[1], can_msg.get_data()[2], can_msg.get_data()[3], can_msg.get_data()[4], can_msg.get_data()[5], can_msg.get_data()[6], can_msg.get_data()[7]); return can_msg; } @@ -314,7 +370,7 @@ openxc_VehicleMessage can_bus_t::next_vehicle_message() { v_msg = vehicle_message_q_.front(); vehicle_message_q_.pop(); - DEBUG(binder_interface, "next_vehicle_message: next vehicle message poped"); + DEBUG(binder_interface, "%s: next vehicle message poped", __FUNCTION__); return v_msg; } @@ -329,6 +385,15 @@ void can_bus_t::push_new_vehicle_message(const openxc_VehicleMessage& v_msg) vehicle_message_q_.push(v_msg); } +/// @brief Create a RX_SETUP receive job for the BCM socket of a CAN signal. +/// +/// @return 0 if ok -1 if not. + int can_bus_t::create_rx_filter(const can_signal_t& s) + { + const std::string& bus = s.get_message().get_bus_name(); + return can_bus_t::can_devices_[bus]->create_rx_filter(s); + } + /// @brief Return a map with the can_bus_dev_t initialized /// /// @return map can_bus_dev_m_ map