Merge branch 'develop' into revert-249-revert-243-eflumerf/RemoveUPS

Author: eflumerf

artdaq/Application/DispatcherCore.cc

@@ -66,6 +66,7 @@ bool artdaq::DispatcherCore::initialize(fhicl::ParameterSet const& pset)
 	}
 
 	broadcast_mode_ = agg_pset.get<bool>("broadcast_mode", true);
+	allow_label_overwrites_ = agg_pset.get<bool>("allow_label_overwrites", true);
 	if (broadcast_mode_ && !agg_pset.has_key("broadcast_mode"))
 	{
 		agg_pset.put<bool>("broadcast_mode", true);
@@ -107,19 +108,40 @@ std::string artdaq::DispatcherCore::register_monitor(fhicl::ParameterSet const&
 		TLOG(TLVL_DEBUG + 32) << "Getting unique_label from input ParameterSet";
 		auto const& label = pset.get<std::string>("unique_label");
 		TLOG(TLVL_DEBUG + 32) << "Unique label is " << label;
+		TLOG(TLVL_INFO) << "Registering monitor with unique_label \"" << label << "\"";
+
+		bool overwriting_label = false;
+
 		{
 			std::lock_guard<std::mutex> lock(dispatcher_transfers_mutex_);
 			if (registered_monitors_.count(label) != 0u)
 			{
-				throw cet::exception("DispatcherCore") << "Unique label already exists!";  // NOLINT(cert-err60-cpp)
+				if (allow_label_overwrites_)
+				{
+					TLOG(TLVL_WARNING) << "Restarting companion art process with label " << label << "!";
+					overwriting_label = true;
+				}
+				else
+				{
+					throw cet::exception("DispatcherCore") << "Unique label already exists!";  // NOLINT(cert-err60-cpp)
+				}
 			}
 
 			registered_monitors_[label] = pset;
 		}
 
-		// ELF, Jul 21, 2020: This can take a long time, and we don't want to block the XMLRPC thread
-		boost::thread thread([this, label] { start_art_process_(label); });
-		thread.detach();
+		if (overwriting_label)
+		{
+			// ELF, Jul 21, 2020: This can take a long time, and we don't want to block the XMLRPC thread
+			boost::thread thread([this, label] { restart_art_process_(label); });
+			thread.detach();
+		}
+		else
+		{
+			// ELF, Jul 21, 2020: This can take a long time, and we don't want to block the XMLRPC thread
+			boost::thread thread([this, label] { start_art_process_(label); });
+			thread.detach();
+		}
 	}
 	catch (const cet::exception& e)
 	{
@@ -159,7 +181,7 @@ std::string artdaq::DispatcherCore::register_monitor(fhicl::ParameterSet const&
 
 std::string artdaq::DispatcherCore::unregister_monitor(std::string const& label)
 {
-	TLOG(TLVL_DEBUG + 32) << "DispatcherCore::unregister_monitor called with argument \"" << label << "\"";
+	TLOG(TLVL_INFO) << "DispatcherCore::unregister_monitor called for unique_label \"" << label << "\"";
 	check_filters_();
 
 	try
@@ -425,6 +447,12 @@ void artdaq::DispatcherCore::check_filters_()
 	}
 }
 
+void artdaq::DispatcherCore::restart_art_process_(std::string const& label)
+{
+	stop_art_process_(label);
+	start_art_process_(label);
+}
+
 void artdaq::DispatcherCore::start_art_process_(std::string const& label)
 {
 	if (event_store_ptr_ != nullptr)

artdaq/Application/DispatcherCore.hh

@@ -51,26 +51,8 @@ public:
 	 * \param pset ParameterSet used to configure the DispatcherCore
 	 * \return Whether the initialize attempt succeeded
 	 *
-	 * \verbatim
-	 * DispatcherCore accepts the following Parameters:
-	 * "daq" (REQUIRED): FHiCL table containing DAQ configuration
-	 *   "Dispatcher" (REQUIRED): FHiCL table containing Dispatcher paramters
-	 *     "expected_events_per_bunch" (REQUIRED): Number of events to collect before sending them to art
-	 *     "enq_timeout" (Default: 5.0): Maximum amount of time to wait while enqueueing events to the ConcurrentQueue
-	 *     "is_data_logger": True if the Dispatcher is a Data Logger
-	 *     "is_online_monitor": True if the Dispatcher is an Online Monitor. is_data_logger takes precedence
-	 *     "is_dispatcher": True if the Dispatcher is a Dispatcher. is_data_logger and is_online_monitor take precedence
-	 *       NOTE: At least ONE of these three parameters must be specified.
-	 *     "inrun_recv_timeout_usec" (Default: 100000): Amount of time to wait for new events while running
-	 *     "endrun_recv_timeout_usec" (Default: 20000000): Amount of time to wait for additional events at EndOfRun
-	 *     "pause_recv_timeout_usec" (Default: 3000000): Amount of time to wait for additional events at PauseRun
-	 *     "onmon_event_prescale" (Default: 1): Only send 1/N events to art for online monitoring (requires is_data_logger: true)
-	 *     "verbose" (Default: true): Whether to print transition messages
-	 *   "metrics": FHiCL table containing configuration for MetricManager
-	 * "outputs" (REQUIRED): FHiCL table containing output parameters
-	 *   "normalOutput" (REQUIRED): FHiCL table containing default output parameters
-	 *     "fileName" (Default: ""): Name template of the output file. Used to determine output directory
-	 * \endverbatim
+	 *  Configuration Parameters unique to the Dispatcher:
+	 *  "allow_label_overwrites" (default: true): Allow a new process to start with the same unique_label as an old one, stopping the appropriate art process and restarting with the new configuration.
 	 *  Note that the "Dispatcher" ParameterSet is also used to configure the EventStore. See that class' documentation for more information.
 	 */
 	bool initialize(fhicl::ParameterSet const& pset) override;
@@ -98,12 +80,14 @@ private:
 
 	void start_art_process_(std::string const& label);
 	void stop_art_process_(std::string const& label);
+	void restart_art_process_(std::string const& label);
 
 	std::mutex dispatcher_transfers_mutex_;
 	std::unordered_map<std::string, fhicl::ParameterSet> registered_monitors_;
 	std::unordered_map<std::string, pid_t> registered_monitor_pids_;
 	fhicl::ParameterSet pset_;  // The ParameterSet initially passed to the Dispatcher (contains input info)
 	bool broadcast_mode_;
+	bool allow_label_overwrites_;
 };
 
 #endif

artdaq/ArtModules/ArtdaqInputHelper.hh

@@ -492,6 +492,10 @@ void art::ArtdaqInputHelper<U>::readAndConstructPrincipal(std::unique_ptr<TBuffe
 		    << "Unable to fetch a high-resolution time with clock_gettime for art::SubRun Timestamp. ";
 	}
 
+	TLOG(TLVL_DEBUG + 37, "ArtdaqInputHelper") << "inR: " << static_cast<void*>(inR) << " run " << (inR ? std::to_string(inR->run()) : "invalid")
+	                                           << ", inSR: " << static_cast<void*>(inSR) << " run " << (inSR ? std::to_string(inSR->run()) : "invalid")
+	                                           << ", subrun " << (inSR ? std::to_string(inSR->subRun()) : "invalid");
+
 	// Process Run Aux
 	TLOG(TLVL_DEBUG + 37, "ArtdaqInputHelper") << "readAndConstructPrincipal: "
 	                                           << "processing Run auxiliary ...";
@@ -515,7 +519,7 @@ void art::ArtdaqInputHelper<U>::readAndConstructPrincipal(std::unique_ptr<TBuffe
 	TLOG(TLVL_DEBUG + 39, "ArtdaqInputHelper") << "readAndConstructPrincipal: "
 	                                           << "finished processing Run auxiliary.";
 
-	if (msg_type_code != artdaq::NetMonHeader::MessageType::Run)
+	if (msg_type_code != artdaq::NetMonHeader::MessageType::Run)  // SubRun or Event
 	{
 		TLOG(TLVL_DEBUG + 38, "ArtdaqInputHelper") << "readAndConstructPrincipal: "
 		                                           << "processing SubRun auxiliary ...";
@@ -548,13 +552,6 @@ void art::ArtdaqInputHelper<U>::readAndConstructPrincipal(std::unique_ptr<TBuffe
 		TLOG(TLVL_DEBUG + 39, "ArtdaqInputHelper") << "readAndConstructPrincipal: "
 		                                           << "finished processing SubRun auxiliary.";
 	}
-	else if (inSR == nullptr || !inSR->subRunID().isValid())
-	{
-		TLOG(TLVL_DEBUG + 39, "ArtdaqInputHelper") << "readAndConstructPrincipal: Faking Subrun 1 because there was no input Subrun for Run message";
-
-		art::SubRunID subrun_guess(outR->runID(), 1);
-		outSR = pm_.makeSubRunPrincipal(subrun_guess, currentTime);
-	}
 
 	if (msg_type_code == artdaq::NetMonHeader::MessageType::Event)
 	{  // Event message.
@@ -584,23 +581,23 @@ void art::ArtdaqInputHelper<U>::readAndConstructPrincipal(std::unique_ptr<TBuffe
 		TLOG(TLVL_DEBUG + 39, "ArtdaqInputHelper") << "readAndConstructPrincipal: "
 		                                           << "finished processing Event auxiliary.";
 	}
-
-	if (!outR && !outSR && !outE)
+	else if (msg_type_code == artdaq::NetMonHeader::MessageType::Subrun)
 	{
-		TLOG(TLVL_DEBUG + 39, "ArtdaqInputHelper") << "No principals created, making Flush Event based on whether there was an existing SubRun";
-
-		if (!inSR || !inSR->subRunID().isValid())
+		if (outSR == nullptr)
 		{
-			TLOG(TLVL_DEBUG + 39, "ArtdaqInputHelper") << "readAndConstructPrincipal: Making run flush event";
-			art::EventID const flush_evid(art::EventID::flushEvent(inR->runID()));
-			outSR = pm_.makeSubRunPrincipal(flush_evid.subRunID(), currentTime);
-			outE = pm_.makeEventPrincipal(flush_evid, currentTime);
+			TLOG(TLVL_DEBUG + 43, "ArtdaqInputHelper") << "SubrunDataFragment for current Subrun received, returning Flush event";
+			art::EventID const evid(art::EventID::flushEvent(inSR->subRunID()));
+			outE = pm_.makeEventPrincipal(evid, currentTime);
 		}
-		else
+	}
+	else if (msg_type_code == artdaq::NetMonHeader::MessageType::Run)
+	{
+		if (outR == nullptr)
 		{
-			TLOG(TLVL_DEBUG + 39, "ArtdaqInputHelper") << "readAndConstructPrincipal: Making subrun flush event";
-			art::EventID const flush_evid(art::EventID::flushEvent(inSR->subRunID()));
-			outE = pm_.makeEventPrincipal(flush_evid, currentTime);
+			TLOG(TLVL_DEBUG + 43, "ArtdaqInputHelper") << "RunDataFragment for current Run received, returning Flush subrun/event";
+			art::EventID const evid(art::EventID::flushEvent(inR->runID()));
+			outSR = pm_.makeSubRunPrincipal(evid.subRunID(), currentTime);
+			outE = pm_.makeEventPrincipal(evid, currentTime);
 		}
 	}
 }
@@ -646,64 +643,22 @@ bool art::ArtdaqInputHelper<U>::constructPrincipal(std::shared_ptr<ArtdaqEvent>
 		    << "The art::Event Timestamp will be zero for event " << eventPtr->header->event_id;
 	}
 
-	if (eventPtr->FirstFragmentType() == artdaq::Fragment::EndOfRunFragmentType)
-	{
-		TLOG(TLVL_DEBUG + 43, "ArtdaqInputHelper") << "EndOfRunFragment received, returning Flush event";
-		art::EventID const evid(art::EventID::flushEvent());
-		outR = pm_.makeRunPrincipal(evid.runID(), currentTime);
-		outSR = pm_.makeSubRunPrincipal(evid.subRunID(), currentTime);
-		outE = pm_.makeEventPrincipal(evid, currentTime);
-		return true;
-	}
-
-	if (eventPtr->FirstFragmentType() == artdaq::Fragment::EndOfSubrunFragmentType)
-	{
-		TLOG(TLVL_DEBUG + 43, "ArtdaqInputHelper") << "EndOfSubrunFragment received, creating new Subrun Principal";
-		// Check if inR == 0 or is a new run
-		if (inR == nullptr || !inR->runID().isValid() || inR->run() != eventPtr->header->run_id)
-		{
-			TLOG(TLVL_DEBUG + 43, "ArtdaqInputHelper") << "Making subrun principal with subrun_id " << eventPtr->header->subrun_id;
-			if (outSR) delete outSR;
-			outSR = pm_.makeSubRunPrincipal(eventPtr->header->run_id, eventPtr->header->subrun_id, currentTime);
-			art::EventID const evid(art::EventID::flushEvent(outSR->subRunID()));
-			outE = pm_.makeEventPrincipal(evid, currentTime);
-		}
-		else
-		{
-			// If the previous subrun was neither 0 nor flush and was identical with the current
-			// subrun, then it must have been associated with a data event.  In that case, we need
-			// to generate a flush event with a valid run but flush subrun and event number in order
-			// to end the subrun.
-			if (inSR != nullptr && !inSR->subRunID().isFlush() && inSR->subRun() == eventPtr->header->subrun_id)
-			{
-				TLOG(TLVL_DEBUG + 43, "ArtdaqInputHelper") << "Flushing old run id " << inR->runID();
-				art::EventID const evid(art::EventID::flushEvent(inR->runID()));
-				outSR = pm_.makeSubRunPrincipal(evid.subRunID(), currentTime);
-				outE = pm_.makeEventPrincipal(evid, currentTime);
-				// If this is either a new or another empty subrun, then generate a flush event with
-				// valid run and subrun numbers but flush event number
-				//} else if(inSR==0 || inSR->id().isFlush()){
-			}
-			else
-			{
-				TLOG(TLVL_DEBUG + 43, "ArtdaqInputHelper") << "Making subrun principal with subrun_id " << eventPtr->header->subrun_id;
-				outSR = pm_.makeSubRunPrincipal(eventPtr->header->run_id, eventPtr->header->subrun_id, currentTime);
-				art::EventID const evid(art::EventID::flushEvent(outSR->subRunID()));
-				outE = pm_.makeEventPrincipal(evid, currentTime);
-				// Possible error condition
-				//} else {
-			}
-		}
-		return true;
-	}
-
 	// make new run if inR is 0 or if the run has changed
 	if (inR == nullptr || !inR->runID().isValid() || inR->run() != eventPtr->header->run_id)
 	{
 		TLOG(TLVL_DEBUG + 43, "ArtdaqInputHelper") << "Making run principal with run_id " << eventPtr->header->run_id;
 		outR = pm_.makeRunPrincipal(eventPtr->header->run_id, currentTime);
 	}
 
+	if (eventPtr->FirstFragmentType() == artdaq::Fragment::EndOfRunFragmentType)
+	{
+		TLOG(TLVL_DEBUG + 43, "ArtdaqInputHelper") << "EndOfRunFragment received, returning Flush subrun/event";
+		art::EventID const evid(art::EventID::flushEvent(outR != nullptr ? outR->runID() : inR->runID()));
+		outSR = pm_.makeSubRunPrincipal(evid.subRunID(), currentTime);
+		outE = pm_.makeEventPrincipal(evid, currentTime);
+		return true;
+	}
+
 	// make new subrun if inSR is 0 or if the subrun has changed
 	art::SubRunID subrun_check(eventPtr->header->run_id, eventPtr->header->subrun_id);
 	if (inSR == nullptr || !inSR->subRunID().isValid() || subrun_check != inSR->subRunID())
@@ -712,6 +667,14 @@ bool art::ArtdaqInputHelper<U>::constructPrincipal(std::shared_ptr<ArtdaqEvent>
 		outSR = pm_.makeSubRunPrincipal(eventPtr->header->run_id, eventPtr->header->subrun_id, currentTime);
 	}
 
+	if (eventPtr->FirstFragmentType() == artdaq::Fragment::EndOfSubrunFragmentType)
+	{
+		TLOG(TLVL_DEBUG + 43, "ArtdaqInputHelper") << "EndOfSubrunFragment received, returning Flush event";
+		art::EventID const evid(art::EventID::flushEvent(outSR != nullptr ? outSR->subRunID() : inSR->subRunID()));
+		outE = pm_.makeEventPrincipal(evid, currentTime);
+		return true;
+	}
+
 	TLOG(TLVL_DEBUG + 43, "ArtdaqInputHelper") << "Making event principal with event_id " << eventPtr->header->event_id;
 	outE = pm_.makeEventPrincipal(eventPtr->header->run_id, eventPtr->header->subrun_id, eventPtr->header->event_id, currentTime);
 	return true;
@@ -1002,28 +965,7 @@ bool art::ArtdaqInputHelper<U>::readNext(art::RunPrincipal* const inR, art::SubR
 		TLOG(TLVL_DEBUG + 32, "ArtdaqInputHelper") << "First Fragment type is " << static_cast<int>(firstFragmentType);
 		if (constructPrincipal(eventMap, inR, inSR, outR, outSR, outE))
 		{
-			auto rfret = readFragments(eventMap->fragments, outR ? outR : inR, outSR ? outSR : inSR, outE);
-
-			// No event data
-			if (!rfret.second)
-			{
-				delete outE;
-				outE = nullptr;
-
-				if (inSR != nullptr && outSR != nullptr)
-				{
-					// No products added to subrun
-					if (!rfret.first)
-					{
-						// New subrun is identical to old
-						if (inSR->subRunID() == outSR->subRunID())
-						{
-							delete outSR;
-							outSR = nullptr;
-						}
-					}
-				}
-			}
+			readFragments(eventMap->fragments, outR ? outR : inR, outSR ? outSR : inSR, outE);
 		}
 	}
 	else
@@ -1091,25 +1033,6 @@ bool art::ArtdaqInputHelper<U>::readNext(art::RunPrincipal* const inR, art::SubR
 		}
 		else if (msg_type_code == artdaq::NetMonHeader::MessageType::Subrun)
 		{
-			// EndSubRun message.
-			// From the code above, EndRun and EndSubRun messages cause
-			// the construction of principals that have:
-			//    Run:Subrun:Event=flush:flush:flush.
-			// This is a problem when you have two neighboring EndSubRuns
-			// which are both associated with empty subruns because art will
-			// complain that you a new subrun with a subrun number identical
-			// to that of the previous subrun.  So the solution is to not
-			// return new principals.
-			if (inR != nullptr && inSR != nullptr && outR != nullptr && outSR != nullptr)
-			{
-				if (inR->runID().isFlush() && inSR->subRunID().isFlush() && outR->runID().isFlush() &&
-				    outSR->subRunID().isFlush())
-				{
-					outR = nullptr;
-					outSR = nullptr;
-					return outE ? true : false;
-				}
-			}
 			// FIXME: We need to merge these into the input SubRunPrincipal.
 			readDataProducts(msgs, outSR ? outSR : inSR);
 		}
@@ -1122,8 +1045,6 @@ bool art::ArtdaqInputHelper<U>::readNext(art::RunPrincipal* const inR, art::SubR
 			{
 				readFragments(eventMap->fragments, outR ? outR : inR, outSR ? outSR : inSR, outE);
 			}
-
-			TLOG(TLVL_DEBUG + 47, "ArtdaqInputHelper") << "readNext: returning true on Event message.";
 		}
 		else
 		{

artdaq/ArtModules/ArtdaqOutput.hh

@@ -558,7 +558,7 @@ inline void art::ArtdaqOutput::write(EventPrincipal& ep)
 
 	// Subrun number starts at 1
 	TLOG(TLVL_WRITE, "ArtdaqOutput") << "ArtdaqOutput::write: Setting Output Fragment Header Fields";
-	auto seqID = (static_cast<uint64_t>(ep.eventID().subRun() - 1) << 32) + ep.eventID().event();
+	auto seqID = (static_cast<uint64_t>(ep.eventID().subRun()) << 32) + ep.eventID().event();
 
 	art::ProcessTag tag("", processName());
 	auto res = ep.getMany(art::ModuleContext::invalid(), art::WrappedTypeID::make<artdaq::detail::RawEventHeader>(), art::MatchAllSelector(), tag);
@@ -637,8 +637,10 @@ inline void art::ArtdaqOutput::writeSubRun(SubRunPrincipal& srp)
 	TLOG(TLVL_WRITESUBRUN, "ArtdaqOutput") << "Begin: ArtdaqOutput::writeSubRun(SubRunPrincipal& srp)";
 	if (!initMsgSent_)
 	{
-		send_init_message();
-		initMsgSent_ = true;
+		TLOG(TLVL_WARNING, "ArtdaqOutput") << "Not sending Subrun message before Event!";
+		return;
+		// send_init_message();
+		// initMsgSent_ = true;
 	}
 
 	//
@@ -659,7 +661,7 @@ inline void art::ArtdaqOutput::writeSubRun(SubRunPrincipal& srp)
 	//
 	//  Begin preparing message.
 	//
-	auto msg = prepareMessage(last_sequence_id_ + 1, srp.subRun() + 1, artdaq::Fragment::SubrunDataFragmentType);
+	auto msg = prepareMessage(static_cast<uint64_t>(srp.subRun()) << 32, srp.subRun() + 1, artdaq::Fragment::SubrunDataFragmentType);
 	//
 	//  Write message type code.
 	//
@@ -752,8 +754,10 @@ inline void art::ArtdaqOutput::writeRun(RunPrincipal& rp)
 	(void)rp;
 	if (!initMsgSent_)
 	{
-		send_init_message();
-		initMsgSent_ = true;
+		TLOG(TLVL_WARNING, "ArtdaqOutput") << "Not sending Run message before Event!";
+		return;
+		// send_init_message();
+		// initMsgSent_ = true;
 	}
 
 	//
@@ -765,7 +769,7 @@ inline void art::ArtdaqOutput::writeRun(RunPrincipal& rp)
 	//
 	//  Begin preparing message.
 	//
-	auto msg = prepareMessage(last_sequence_id_ + 1, rp.run() + 1, artdaq::Fragment::RunDataFragmentType);
+	auto msg = prepareMessage(static_cast<uint64_t>(rp.run()) << 32, rp.run() + 1, artdaq::Fragment::RunDataFragmentType);
 	//
 	//  Write message type code.
 	//