DecrepitRelicInputSourceImplementation.h

Go to the documentation of this file.

#ifndef art_Framework_Core_DecrepitRelicInputSourceImplementation_h
#define art_Framework_Core_DecrepitRelicInputSourceImplementation_h

/*----------------------------------------------------------------------

DecrepitRelicInputSourceImplementation: this is the relic we
inherited from CMS, which is not an interface. The remaining comments
are left over from that.

InputSource: Abstract interface for all input sources. Input
sources are responsible for creating an EventPrincipal, using data
controlled by the source, and external to the EventPrincipal itself.

The InputSource is also responsible for dealing with the "process
name list" contained within the EventPrincipal. Each InputSource has
to know what "process" (HLT, PROD, USER, USER1, etc.) the program is
part of. The InputSource is repsonsible for pushing this process name
onto the end of the process name list.

For now, we specify this process name to the constructor of the
InputSource. This should be improved.

 Some questions about this remain:

   1. What should happen if we "rerun" a process? i.e., if "USER1" is
   already last in our input file, and we run again a job which claims
   to be "USER1", what should happen? For now, we just quietly add
   this to the history.

   2. Do we need to detect a problem with a history like:
         HLT PROD USER1 PROD
   or is it up to the user not to do something silly? Right now, there
   is no protection against such sillyness.

Some examples of InputSource subclasses may be:

 1) EmptyEvent: creates EventPrincipals which contain no EDProducts.
 2) RootInput: creates EventPrincipals which "contain" the data
    read from a root file. This source should provide for delayed loading
    of data, thus the quotation marks around contain.
 3) DAQInputSource: creats EventPrincipals which contain raw data, as
    delivered by the L1 trigger and event builder.

----------------------------------------------------------------------*/

#include "art/Framework/Core/Frameworkfwd.h"
#include "art/Framework/Core/InputSource.h"
#include "art/Framework/Principal/EventPrincipal.h"
#include "art/Framework/Principal/RunPrincipal.h"
#include "art/Framework/Principal/SubRunPrincipal.h"
#include "art/Framework/Principal/fwd.h"
#include "canvas/Persistency/Provenance/RunID.h"
#include "canvas/Persistency/Provenance/SubRunID.h"
#include "canvas/Persistency/Provenance/Timestamp.h"
#include "cetlib/exempt_ptr.h"
#include "fhiclcpp/types/Atom.h"
#include "fhiclcpp/types/TableFragment.h"

#include <memory>
#include <string>

// ----------------------------------------------------------------------

namespace art {
  class ActivityRegistry;

  class DecrepitRelicInputSourceImplementation : public InputSource {
  public:
    DecrepitRelicInputSourceImplementation(
      DecrepitRelicInputSourceImplementation const&) = delete;
    DecrepitRelicInputSourceImplementation& operator=(
      DecrepitRelicInputSourceImplementation const&) = delete;

    struct Config {

      static constexpr char const*
      defaultMode()
      {
        return "RunsSubRunsAndEvents";
      }

      fhicl::Atom<int> maxEvents{fhicl::Name("maxEvents"), -1};
      fhicl::Atom<int> maxSubRuns{fhicl::Name("maxSubRuns"), -1};
      fhicl::Atom<int> reportFrequency{fhicl::Name("reportFrequency"), 1};
      fhicl::Atom<bool> errorOnFailureToPut{fhicl::Name("errorOnFailureToPut"),
                                            false};
      fhicl::Atom<std::string> processingMode{fhicl::Name("processingMode"),
                                              defaultMode()};
    };

    DecrepitRelicInputSourceImplementation(fhicl::TableFragment<Config> const&,
                                           ModuleDescription const&);

    // Prevent concrete instances of this class even though we don't
    // have any (other) pure virtual functions.
    virtual ~DecrepitRelicInputSourceImplementation() noexcept = 0;

    // Note: this virtual function is implemented here. It is overridden
    // in RootInput, but may still be called under normal circumstances
    // by the overriding function.
    input::ItemType nextItemType() override;

    std::unique_ptr<EventPrincipal> readEvent(
      cet::exempt_ptr<SubRunPrincipal const> srp) override;

    std::unique_ptr<EventPrincipal> readEvent(EventID const&) override;

    std::unique_ptr<SubRunPrincipal> readSubRun(
      cet::exempt_ptr<RunPrincipal const> rp) override;

    std::unique_ptr<RunPrincipal> readRun() override;

    std::unique_ptr<FileBlock> readFile() override;

    void closeFile() override;

    void skipEvents(int offset) override;

    void
    rewind() override
    {
      repeat_();
      doneReadAhead_ = false;
      state_ = input::IsInvalid;
      rewind_();
    }

    void issueReports(EventID const& eventID);

    void
    repeat_()
    {
      remainingEvents_ = maxEvents_;
      remainingSubRuns_ = maxSubRuns_;
      doneReadAhead_ = false;
    }

    int
    maxEvents() const
    {
      return maxEvents_;
    }

    int
    remainingEvents() const
    {
      return remainingEvents_;
    }

    int
    maxSubRuns() const
    {
      return maxSubRuns_;
    }

    int
    remainingSubRuns() const
    {
      return remainingSubRuns_;
    }

    void doBeginJob() override;

    void doEndJob() override;

    Timestamp const&
    timestamp() const
    {
      return time_;
    }

    ProcessingMode
    processingMode() const
    {
      return processingMode_;
    }

  protected:
    void
    setTimestamp(Timestamp const& theTime)
    {
      time_ = theTime;
    }

    input::ItemType
    state() const
    {
      return state_;
    }

    // Inform subclasses that they're going to be told to close the file
    // for non-exceptional reasons (such as hitting the event limit).
    virtual void
    finish()
    {}

    cet::exempt_ptr<RunPrincipal>
    runPrincipalExemptPtr()
    {
      return cachedRunPrincipal_;
    }
    cet::exempt_ptr<SubRunPrincipal>
    subRunPrincipalExemptPtr()
    {
      return cachedSubRunPrincipal_;
    }

    std::unique_ptr<RunPrincipal>
    runPrincipal()
    {
      cachedRunPrincipal_ = runPrincipal_.get();
      return std::move(runPrincipal_);
    }

    std::unique_ptr<SubRunPrincipal>
    subRunPrincipal()
    {
      cachedSubRunPrincipal_ = subRunPrincipal_.get();
      return std::move(subRunPrincipal_);
    }

    std::unique_ptr<EventPrincipal>
    eventPrincipal()
    {
      return std::move(eventPrincipal_);
    }

    void setRunPrincipal(std::unique_ptr<RunPrincipal>&& rp);
    void setSubRunPrincipal(std::unique_ptr<SubRunPrincipal>&& srp);
    void setEventPrincipal(std::unique_ptr<EventPrincipal>&& ep);
    void
    resetRunPrincipal()
    {
      runPrincipal_.reset();
    }
    void
    resetSubRunPrincipal()
    {
      subRunPrincipal_.reset();
    }
    void
    resetEventPrincipal()
    {
      eventPrincipal_.reset();
    }
    void
    reset()
    {
      doneReadAhead_ = false;
      state_ = input::IsInvalid;
    }

  private:
    bool
    eventLimitReached() const
    {
      return remainingEvents_ == 0;
    }
    bool
    subRunLimitReached() const
    {
      return remainingSubRuns_ == 0;
    }
    bool
    limitReached() const
    {
      return eventLimitReached() || subRunLimitReached();
    }
    virtual input::ItemType getNextItemType() = 0;
    input::ItemType nextItemType_();
    virtual std::unique_ptr<RunPrincipal> readRun_() = 0;
    virtual std::unique_ptr<SubRunPrincipal> readSubRun_() = 0;
    virtual std::unique_ptr<EventPrincipal> readEvent_() = 0;
    virtual std::unique_ptr<FileBlock> readFile_();
    virtual void
    closeFile_()
    {}
    virtual void skip(int);
    virtual void rewind_();
    virtual void beginJob();
    virtual void endJob();

  private:
    int maxEvents_;
    int maxSubRuns_;
    int const reportFrequency_;

    int remainingEvents_{maxEvents_};
    int remainingSubRuns_{maxSubRuns_};
    int readCount_{};
    ProcessingMode processingMode_{RunsSubRunsAndEvents};
    Timestamp time_{Timestamp::invalidTimestamp()};
    bool doneReadAhead_{false};
    input::ItemType state_{input::IsInvalid};
    std::unique_ptr<RunPrincipal> runPrincipal_{nullptr};
    std::unique_ptr<SubRunPrincipal> subRunPrincipal_{nullptr};
    std::unique_ptr<EventPrincipal> eventPrincipal_{nullptr};
    cet::exempt_ptr<RunPrincipal> cachedRunPrincipal_{nullptr};
    cet::exempt_ptr<SubRunPrincipal> cachedSubRunPrincipal_{nullptr};
  }; // DecrepitRelicInputSourceImplementation

} // art

  // ======================================================================

#endif /* art_Framework_Core_DecrepitRelicInputSourceImplementation_h */

// Local Variables:
// mode: c++
// End: