Some database-related changes

Author: darrylmasson

DAQController.cc

@@ -184,6 +184,7 @@ int DAQController::Stop(){
   fStatus = DAXHelpers::Idle;
 
   fLog->SetRunId(-1);
+  fOptions.reset();
   std::cout<<"Finished end"<<std::endl;
   fStatus = DAXHelpers::Idle;
   return 0;

Options.cc

@@ -4,22 +4,18 @@
 
 #include <cmath>
 
-#include <mongocxx/uri.hpp>
-#include <mongocxx/database.hpp>
 #include <bsoncxx/array/view.hpp>
 #include <bsoncxx/types.hpp>
 #include <bsoncxx/json.hpp>
 #include <bsoncxx/builder/stream/document.hpp>
 #include <bsoncxx/exception/exception.hpp>
 
 Options::Options(std::shared_ptr<MongoLog>& log, std::string options_name, std::string hostname,
-          std::string suri, std::string dbname, std::string override_opts) : 
-    fLog(log), fDBname(dbname), fHostname(hostname) {
+          mongocxx::collection* opts_collection, mongocxx::collection* dac_collection,
+          mongocxx::collection* bm_collection, std::string override_opts) : 
+    fLog(log), fHostname(hostname), fDAC_collection(dac_collection),
+    fBM_collection(bm_collection) {
   bson_value = NULL;
-  mongocxx::uri uri{suri};
-  fClient = mongocxx::client{uri};
-  fDAC_collection = fClient[dbname]["dac_calibration"];
-  mongocxx::collection opts_collection = fClient[dbname]["options"];
   if(Load(options_name, opts_collection, override_opts)!=0)
     throw std::runtime_error("Can't initialize options class");
 }
@@ -31,11 +27,11 @@ Options::~Options(){
   }
 }
 
-int Options::Load(std::string name, mongocxx::collection& opts_collection,
+int Options::Load(std::string name, mongocxx::collection* opts_collection,
 	std::string override_opts){
   // Try to pull doc from DB
   bsoncxx::stdx::optional<bsoncxx::document::value> trydoc;
-  trydoc = opts_collection.find_one(bsoncxx::builder::stream::document{}<<
+  trydoc = opts_collection->find_one(bsoncxx::builder::stream::document{}<<
 				    "name" << name.c_str() << bsoncxx::builder::stream::finalize);
   if(!trydoc){
     fLog->Entry(MongoLog::Warning, "Failed to find your options file '%s' in DB", name.c_str());
@@ -53,7 +49,7 @@ int Options::Load(std::string name, mongocxx::collection& opts_collection,
   try{
     bsoncxx::array::view include_array = (*trydoc).view()["includes"].get_array().value;
     for(bsoncxx::array::element ele : include_array){
-      auto sd = opts_collection.find_one(bsoncxx::builder::stream::document{} <<
+      auto sd = opts_collection->find_one(bsoncxx::builder::stream::document{} <<
 					 "name" << ele.get_utf8().value.to_string() <<
 					 bsoncxx::builder::stream::finalize);
       if(sd)
@@ -340,7 +336,7 @@ std::vector<uint16_t> Options::GetDAC(int bid, int num_chan, uint16_t default_va
   auto q = document{} << std::to_string(bid) << open_document << "$exists" << 1 << close_document << finalize;
   auto opts = mongocxx::options::find{};
   opts.sort(sort_order.view());
-  auto cursor = fDAC_collection.find(std::move(q), opts);
+  auto cursor = fDAC_collection->find(std::move(q), opts);
   auto doc = cursor.begin();
   if (doc == cursor.end() || doc->find(std::to_string(bid)) == doc->end()) {
     fLog->Entry(MongoLog::Local, "No baseline calibrations? You must be new");
@@ -374,14 +370,14 @@ void Options::UpdateDAC(std::map<int, std::vector<uint16_t>>& all_dacs){
   auto write_doc = update_doc<<finalize;
   mongocxx::options::update options;
   options.upsert(true);
-  fDAC_collection.update_one(std::move(search_doc), std::move(write_doc), options);
+  fDAC_collection->update_one(std::move(search_doc), std::move(write_doc), options);
   return;
 }
 
 void Options::SaveBenchmarks(std::map<std::string, std::map<int, long>>& counters,
     long bytes, std::string sid, std::map<std::string, double>& times) {
   using namespace bsoncxx::builder::stream;
-  int level = GetInt("benchmark_level", 1);
+  int level = GetInt("benchmark_level", 0);
   if (level == 0) return;
   int run_id = GetInt("number", -1);
   std::map<std::string, std::map<int, long>> _counters;
@@ -419,6 +415,6 @@ void Options::SaveBenchmarks(std::map<std::string, std::map<int, long>>& counter
   auto write_doc = update_doc << finalize;
   mongocxx::options::update options;
   options.upsert(true);
-  fClient[fDBname]["redax_benchmarks"].update_one(search_doc.view(), write_doc.view(), options);
+  fBM_collection->update_one(search_doc.view(), write_doc.view(), options);
   return;
 }

Options.hh

@@ -10,7 +10,6 @@
 #include <bsoncxx/document/view.hpp>
 #include <bsoncxx/document/value.hpp>
 #include <mongocxx/collection.hpp>
-#include <mongocxx/client.hpp>
 
 struct BoardType{
   int link;
@@ -67,7 +66,7 @@ class MongoLog;
 class Options{
 
 public:
-  Options(std::shared_ptr<MongoLog>&, std::string, std::string, std::string, std::string, std::string);
+  Options(std::shared_ptr<MongoLog>&, std::string, std::string, mongocxx::collection*, mongocxx::collection*, mongocxx::collection*, std::string);
   ~Options();
 
   int GetInt(std::string, int=-1);
@@ -91,14 +90,13 @@ public:
       std::map<std::string, double>&);
 
 private:
-  int Load(std::string, mongocxx::collection&, std::string);
+  int Load(std::string, mongocxx::collection*, std::string);
   int Override(bsoncxx::document::view);
-  mongocxx::client fClient;
   bsoncxx::document::view bson_options;
   bsoncxx::document::value *bson_value;
   std::shared_ptr<MongoLog> fLog;
   mongocxx::collection fDAC_collection;
-  std::string fDBname;
+  mongocxx::collection fBM_collection;
   std::string fHostname;
   std::string fDetector;
 };

dispatcher/MongoConnect.py

@@ -421,9 +421,12 @@ def SendCommand(self, command, hosts, user, detector, mode="", delay=0):
             if delay == 0:
                 docs = doc_base
                 docs['host'] = hosts[0]+hosts[1] if isinstance(hosts, tuple) else hosts
+                docs['acknowledged'] = {h:0 for h in docs['host']}
             else:
                 docs = [dict(doc_base.items()), dict(doc_base.items())]
                 docs[0]['host'], docs[1]['host'] = hosts
+                docs[0]['acknowledged'] = {h:0 for h in docs[0]['host']}
+                docs[1]['acknowledged'] = {h:0 for h in docs[1]['host']}
                 docs[1]['createdAt'] += datetime.timedelta(seconds=delay)
             self.collections['command_queue'].insert(docs)
         except Exception as e:

docs/databases.md

@@ -122,7 +122,7 @@ The control database is used to propagate commands from the dispatcher to the re
      "user" : "web",
      "host" : ["fdaq00_reader_0"],
      "acknowledged" : {
-         "fdaq00_reader_0" : 1601469970934
+         "fdaq00_reader_0" : 0
      },
      "command" : "arm",
      "createdAt": <date object>
@@ -135,7 +135,7 @@ The control database is used to propagate commands from the dispatcher to the re
 |mode	|Options file to use for this run. Corresponds to the 'name' field of the options doc. |
 |user	|Who started the run? Corresponds to the last person to change the detector_status doc during normal operation. Exceptional stop commands can be automatically issued by various subsystems as well in case of errors.
 |host	|List of all hosts to which this command is directed. Readers and crate controllers will only process commands addressed to them. |
-|acknowledged	|Before attempting to process a command all reader and crate controller processes will first acknowledge the command as received. This does not indicate that processing the command was successful! It just indicates the thing tried. The dispatcher has to watch for the appropriate state change of the slave nodes in order to determine if the command achieved its goal. This is a dictionary, with values set to the timestamp (in ms) of when the acknowledgement happened. |
+|acknowledged	|Before attempting to process a command all reader and crate controller processes will first acknowledge the command as received. This does not indicate that processing the command was successful! It just indicates the thing tried. The dispatcher has to watch for the appropriate state change of the slave nodes in order to determine if the command achieved its goal. This is a dictionary, with values set to the timestamp of when the acknowledgement happened. This dictionary must be populated with 0s before insertion. |
 |command	|This is the actual command. 'arm' gets the DAQ ready to start. 'start' and 'stop' do what they say on the tin. 'stop' can also be used as a general reset command for a given instance. |
 
 ### db.options

helpers/runcommand.py

@@ -12,6 +12,8 @@ def main(coll):
     parser.add_argument('--host', nargs='+', default=[os.uname()[1]], help="Hosts to issue to")
 
     args = parser.parse_args()
+    if not isinstance(args.host, (list, tuple)):
+        args.host = [args.host]
 
     doc = {
             "command": args.command,
@@ -20,13 +22,14 @@ def main(coll):
             "host": args.host,
             "user": os.getlogin(),
             "run_identifier": '%06i' % args.number,
-            "createdAt": datetime.datetime.utcnow()
+            "createdAt": datetime.datetime.utcnow(),
+            "acknowledged": {h:0 for h in args.host}
             }
     coll.insert_one(doc)
     return
 
 if __name__ == '__main__':
-    with MongoClient("mongodb://daq:%s@xenon1t-daq:27017/admin" % os.environ['MONGO_PASSWORD_DAQ']) as client:
+    with MongoClient("mongodb://daq:%s@gw:27020/admin" % os.environ['MONGO_PASSWORD_DAQ']) as client:
         try:
             main(client['daq']['control'])
         except Exception as e:

main.cc

@@ -25,14 +25,11 @@ void SignalHandler(int signum) {
     return;
 }
 
-void UpdateStatus(std::string suri, std::string dbname, std::unique_ptr<DAQController>& controller) {
-  mongocxx::uri uri(suri);
-  mongocxx::client c(uri);
-  mongocxx::collection status = c[dbname]["status"];
+void UpdateStatus(std::string suri, mongocxx::collection* collection, dbname, std::unique_ptr<DAQController>& controller) {
   using namespace std::chrono;
   while (b_run == true) {
     try{
-      controller->StatusUpdate(&status);
+      controller->StatusUpdate(collection);
     }catch(const std::exception &e){
       std::cout<<"Can't connect to DB to update."<<std::endl;
       std::cout<<e.what()<<std::endl;
@@ -43,14 +40,14 @@ void UpdateStatus(std::string suri, std::string dbname, std::unique_ptr<DAQContr
 }
 
 int PrintUsage() {
-  std::cout<<"Welcome to REDAX readout\nAccepted command-line arguments:\n"
+  std::cout<<"Welcome to REDAX\nAccepted command-line arguments:\n"
     << "--id <id number>: id number of this readout instance, required\n"
     << "--uri <mongo uri>: full MongoDB URI, required\n"
     << "--db <database name>: name of the database to use, default \"daq\"\n"
     << "--logdir <directory>: where to write the logs, default pwd\n"
     << "--reader: this instance is a reader\n"
     << "--cc: this instance is a crate controller\n"
-    << "--arm-delay <delay>: ms to wait between the ARM command and the arming sequence, default 15000\n"
+    << "--arm-delay <delay>: ms to wait between the ARM command and the arming sequence, default 5000\n"
     << "--log-retention <value>: how many days to keep logfiles, default 7\n"
     << "--help: print this message\n"
     << "\n";
@@ -126,8 +123,9 @@ int main(int argc, char** argv){
   mongocxx::database db = client[dbname];
   mongocxx::collection control = db["control"];
   mongocxx::collection status = db["status"];
-  mongocxx::collection options_collection = db["options"];
+  mongocxx::collection opts_collection = db["options"];
   mongocxx::collection dac_collection = db["dac_calibration"];
+  mongocxx::collection benchmark_collection = db["redax_benchmarks"];
 
   // Logging
   auto fLog = std::make_shared<MongoLog>(log_retention, log_dir, suri, dbname, "log", hostname);
@@ -142,16 +140,16 @@ int main(int argc, char** argv){
     controller = std::make_unique<CControl_Handler>(fLog, hostname);
   else
     controller = std::make_unique<DAQController>(fLog, hostname);
-  std::thread status_update(&UpdateStatus, suri, dbname, std::ref(controller));
+  std::thread status_update(&UpdateStatus, &status, std::ref(controller));
 
   using namespace bsoncxx::builder::stream;
   // Sort oldest to newest
   auto opts = mongocxx::options::find_one_and_update{};
   opts.sort(document{} << "_id" << 1 << finalize);
-  auto query = document{} << "host" << hostname << "acknowledged." + hostname <<
-    open_document << "$exists" << 0 << close_document << finalize;
+  std::string ack_host = "acknowledged." + hostname;
+  auto query = document{} << "host" << hostname << ack_host << 0 << finalize;
   auto update = document{} << "$currentDate" << open_document <<
-    "acknowledged."+hostname << true << close_document << finalize;
+    ack_host << true << close_document << finalize;
   using namespace std::chrono;
   // Main program loop. Scan the database and look for commands addressed
   // to this hostname. 
@@ -196,6 +194,7 @@ int main(int argc, char** argv){
           fLog->Entry(MongoLog::Local, "Ack to stop took %i us",
               duration_cast<microseconds>(now-ack_time).count());
           fLog->SetRunId(-1);
+          fOptions.reset();
 	} else if(command == "arm"){
 	  // Can only arm if we're idle
 	  if(controller->status() == 0){
@@ -212,8 +211,8 @@ int main(int argc, char** argv){
 	    // Mongocxx types confusing so passing json strings around
             std::string mode = doc["mode"].get_utf8().value.to_string();
             fLog->Entry(MongoLog::Local, "Getting options doc for mode %s", mode.c_str());
-	    fOptions = std::make_shared<Options>(fLog, mode,
-				   hostname, suri, dbname, override_json);
+	    fOptions = std::make_shared<Options>(fLog, mode, &opts_collection,
+			      &dac_collection, &benchmark_collection, override_json);
             int dt = duration_cast<milliseconds>(system_clock::now()-ack_time).count();
             fLog->SetRunId(fOptions->GetInt("number", -1));
             fLog->Entry(MongoLog::Local, "Took %i ms to load config", dt);