1+ /*
2+ * === This file is part of ALICE O² ===
3+ *
4+ * Copyright 2020 CERN and copyright holders of ALICE O².
5+ * Author: Teo Mrnjavac <teo.mrnjavac@cern.ch>
6+ *
7+ * This program is free software: you can redistribute it and/or modify
8+ * it under the terms of the GNU General Public License as published by
9+ * the Free Software Foundation, either version 3 of the License, or
10+ * (at your option) any later version.
11+ *
12+ * This program is distributed in the hope that it will be useful,
13+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
14+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15+ * GNU General Public License for more details.
16+ *
17+ * You should have received a copy of the GNU General Public License
18+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
19+ *
20+ * In applying this license CERN does not waive the privileges and
21+ * immunities granted to it by virtue of its status as an
22+ * Intergovernmental Organization or submit itself to any jurisdiction.
23+ */
24+
25+ #include " OccFMQCommon.h"
26+
27+ #include " util/Defer.h"
28+
29+ #include < boost/algorithm/string.hpp>
30+ #include < iomanip>
31+
32+ std::string generateSubscriptionId (const std::string& prefix)
33+ {
34+ std::string id;
35+ try {
36+ boost::uuids::random_generator gen;
37+ id = boost::uuids::to_string (gen ());
38+ } catch (const boost::uuids::entropy_error &err) {
39+ OLOG (WARNING) << " [generateSubscriptionId] boost::uuids::entropy_error: " what () << " falling back to std::time"
40+ id = std::to_string (std::time (nullptr ));
41+ }
42+ return " OCC_" size () ? (prefix + " _" " "
43+ }
44+
45+ bool isIntermediateFMQState (const std::string& state)
46+ {
47+ return state.find (" INITIALIZING TASK"
48+ state.find (" RESETTING"
49+ state.find (" BINDING"
50+ state.find (" CONNECTING"
51+ }
52+
53+ std::tuple<OccLite::nopb::TransitionResponse, ::grpc::Status> doTransition (fair::mq::PluginServices* m_pluginServices, const OccLite::nopb::TransitionRequest& request)
54+ {
55+ std::string srcState = request.srcState ;
56+ std::string event = request.transitionEvent ;
57+ auto arguments = request.arguments ;
58+
59+ std::string currentState = fair::mq::PluginServices::ToStr (m_pluginServices->GetCurrentDeviceState ());
60+ if (srcState != currentState) {
61+ return std::make_tuple (OccLite::nopb::TransitionResponse (), grpc::Status (grpc::INVALID_ARGUMENT,
62+ " transition not possible: state mismatch: source: " " current: "
63+ currentState));
64+ }
65+
66+ OLOG (DEBUG) << " transition src: "
67+ << " currentState: "
68+ << " event: "
69+
70+ std::vector<std::string> newStates;
71+ const std::string finalState = EXPECTED_FINAL_STATE.at (event);
72+
73+ std::condition_variable cv;
74+ std::mutex cv_mu;
75+
76+ auto onDeviceStateChange = [&](fair::mq::PluginServices::DeviceState reachedState) {
77+ // CONFIGURE arguments must be pushed during InitializingDevice
78+ if (reachedState == fair::mq::PluginServices::DeviceState::InitializingDevice) {
79+
80+ // FIXME: workaround which special cases a stoi for certain properties
81+ // which must be pushed as int.
82+ // This should be removed once SetPropertyAsString becomes available.
83+ const std::vector<std::string> intKeys = {
84+ " rateLogging"
85+ " rcvBufSize"
86+ " sndBufSize"
87+ " linger"
88+ " rcvKernelSize"
89+ " sndKernelSize"
90+ };
91+ for (auto it = arguments.cbegin (); it != arguments.cend (); ++it) {
92+ std::string key = it->first ;
93+ std::string value = it->second ;
94+ if (boost::starts_with (key, " chans."
95+ key.erase (0 , 6 );
96+ std::vector<std::string> split;
97+ boost::split (split, key, std::bind1st (std::equal_to<char >(), ' .'
98+ if (std::find (intKeys.begin (), intKeys.end (), split.back ()) != intKeys.end ()) {
99+ auto intValue = std::stoi (value);
100+ m_pluginServices->SetProperty (key, intValue);
101+ }
102+ else {
103+ m_pluginServices->SetProperty (key, value);
104+ }
105+ }
106+ else {
107+ m_pluginServices->SetProperty (key, value);
108+ }
109+ }
110+ }
111+
112+ std::unique_lock<std::mutex> lk (cv_mu);
113+ newStates.push_back (fair::mq::PluginServices::ToStr (reachedState));
114+ OLOG (DEBUG) << " transition newStates vector: " boost::algorithm::join (newStates, " , "
115+ cv.notify_one ();
116+ };
117+
118+ auto id = generateSubscriptionId (" Transition"
119+
120+ m_pluginServices->SubscribeToDeviceStateChange (id, onDeviceStateChange);
121+ DEFER ({
122+ m_pluginServices->UnsubscribeFromDeviceStateChange (id);
123+ });
124+
125+ try {
126+ auto evt = fair::mq::PluginServices::ToDeviceStateTransition (event);
127+
128+ // FIXME: big ugly workaround over here
129+ // Since FairMQ currently (11/2018) can't yet implicitly create channels when receiving
130+ // chans.* properties during INITIALIZING DEVICE, we must fake a --channel-config cli
131+ // parameter during INIT and before the INIT DEVICE event.
132+ // We extract channel related properties from the OCC transition arguments vector and we
133+ // build up a vector of strings which mimics stuff along the lines of
134+ // --channel-config name=data,type=push,method=bind,address=tcp://*:5555,rateLogging=0"
135+ // See https://github.com/FairRootGroup/FairMQ/pull/111
136+ // When the relevant FairMQ 1.4.x version implements implicit channel creation, this whole
137+ // block should be removed with no loss of functionality.
138+ if (evt == fair::mq::PluginServices::DeviceStateTransition::InitDevice) {
139+ std::unordered_map<std::string, std::unordered_map<std::string, std::string>> channels;
140+ for (auto it = arguments.cbegin (); it != arguments.cend (); ++it) {
141+ std::string key = it->first ;
142+ std::string value = it->second ;
143+ if (boost::starts_with (key, " chans."
144+ key.erase (0 , 6 );
145+ std::vector<std::string> split;
146+ boost::split (split, key, std::bind1st (std::equal_to<char >(), ' .'
147+ if (split.size () != 3 )
148+ continue ;
149+ auto name = split[0 ];
150+ auto propKey = split[2 ];
151+ if (channels.find (name) == channels.end ()) // if map for this chan doesn't exist yet
152+ channels[name] = std::unordered_map<std::string, std::string>();
153+ channels[name][propKey] = value;
154+ }
155+ }
156+
157+ std::vector<std::string> channelLines;
158+ for (auto it = channels.cbegin (); it != channels.cend (); ++it) {
159+ std::vector<std::string> line;
160+ line.push_back (" name=" first );
161+ for (auto jt = it->second .cbegin (); jt != it->second .cend (); ++jt) {
162+ line.push_back (jt->first + " =" second );
163+ }
164+ channelLines.push_back (boost::join (line, " ,"
165+ OLOG (DEBUG) << " transition pushing channel configuration " back ();
166+ }
167+ if (!channelLines.empty ()) {
168+ m_pluginServices->SetProperty (" channel-config"
169+ }
170+ }
171+ // Run number must be pushed immediately before RUN transition
172+ else if (evt == fair::mq::PluginServices::DeviceStateTransition::Run) {
173+ try {
174+ for (auto const & [key, value] : arguments) {
175+ m_pluginServices->SetProperty (key, value);
176+ }
177+ }
178+ catch (std::runtime_error &e) {
179+ OLOG (WARNING) << " transition cannot push RUN transition arguments, reason:" what ();
180+ }
181+ }
182+ m_pluginServices->ChangeDeviceState (FMQ_CONTROLLER_NAME, evt);
183+ }
184+ catch (fair::mq::PluginServices::DeviceControlError& e) {
185+ OLOG (ERROR) << " transition cannot request transition: " what ();
186+ return std::make_tuple (OccLite::nopb::TransitionResponse (), grpc::Status (grpc::INTERNAL, " cannot request transition, OCC plugin has no device control"
187+ }
188+ catch (std::out_of_range& e) {
189+ OLOG (ERROR) << " transition invalid event name: "
190+ return std::make_tuple (OccLite::nopb::TransitionResponse (), grpc::Status (grpc::INVALID_ARGUMENT, " argument " " is not a valid transition name"
191+ }
192+
193+ {
194+ std::unique_lock<std::mutex> lk (cv_mu);
195+
196+ // IF we have no states in list yet, OR
197+ // we have some states, and the last one is an intermediate state (for which an autotransition is presumably about to happen)
198+ if (newStates.empty () || isIntermediateFMQState (newStates.back ())) {
199+ // We need to block until the transitions are complete
200+ for (;;) {
201+ cv.wait (lk);
202+ if (newStates.empty ()) {
203+ OLOG (ERROR) << " [request Transition] notify condition met but no states written"
204+ break ;
205+ }
206+
207+ OLOG (DEBUG) << " transition notify condition met, reached state: " back ();
208+ if (isIntermediateFMQState (newStates.back ())) { // if it's an auto state
209+ continue ;
210+ } else {
211+ break ;
212+ }
213+ }
214+ }
215+ }
216+
217+ if (newStates.empty ()) {
218+ return std::make_tuple (OccLite::nopb::TransitionResponse (), grpc::Status (grpc::INTERNAL,
219+ " no transitions made, current state stays "
220+ }
221+
222+ if (srcState == " IDLE" back () == " DEVICE READY"
223+ // Debug: list of FairMQ property keys
224+ auto pk = m_pluginServices->GetPropertyKeys ();
225+ for (const auto &k : pk) {
226+ OLOG (DEBUG) << std::setw (30 ) << k << " = " GetPropertyAsString (k);
227+ }
228+ auto chi = m_pluginServices->GetChannelInfo ();
229+ OLOG (DEBUG) << " channel info:"
230+ for (const auto &k : chi) {
231+ OLOG (DEBUG) << k.first << " : " second ;
232+ }
233+ }
234+
235+ auto response = OccLite::nopb::TransitionResponse ();
236+ response.state = newStates.back ();
237+ response.transitionEvent = event;
238+ response.ok = (newStates.back () == finalState);
239+ if (newStates.back () == " ERROR" // ERROR state
240+ response.trigger = OccLite::nopb::DEVICE_ERROR;
241+ } else if (newStates.back () == finalState) { // correct destination state
242+ response.trigger = OccLite::nopb::EXECUTOR;
243+ } else { // some other state, for whatever reason - we assume DEVICE_INTENTIONAL
244+ response.trigger = OccLite::nopb::DEVICE_INTENTIONAL;
245+ }
246+
247+ OLOG (DEBUG) << " transition done, states visited: " boost::algorithm::join (newStates, " , "
248+ return std::make_tuple (response, grpc::Status::OK);
249+ }
0 commit comments