LEMWorker_module.cc

Go to the documentation of this file.

////////////////////////////////////////////////////////////////////////
// \file   LEMWorker_module.cc
// \brief  This runs on a high-mem node and reconstitutes LEMSummary events
// \author Christopher Backhouse - bckhouse@caltech.edu
////////////////////////////////////////////////////////////////////////

// NOvASoft includes
#include "RecoBase/Cluster.h"
#include "Utilities/AssociationUtil.h"

// Framework includes
#include "art/Framework/Core/EDProducer.h"
#include "canvas/Persistency/Common/FindManyP.h"
#include "art/Framework/Core/ModuleMacros.h"
#include "art/Framework/Principal/Event.h"
#include "fhiclcpp/ParameterSet.h"

#include "LEM/LEMInput.h"
#include "LEM/LEMWebSettings.h"
#include "LEM/Util.h"
#include "LEM/WebUtils.h"

#include <iostream>

#include <curl/curl.h>

namespace
{
  struct LEMWorkerParams
  {
    template<class T> using Atom = fhicl::Atom<T>;
    template<class T> using Table = fhicl::Table<T>;
    using Comment = fhicl::Comment;
    using Name = fhicl::Name;

    Atom<std::string> releaseOverride{
      Name("ReleaseOverride"),
      Comment("Specify this release to the server. Empty string for current release.")
    };

    Table<lem::LEMWebSettings> web{Name("WebSettings")};
  };
}

namespace lem
{
  /// This runs on a high-mem node and reconstitutes LEMSummary events
  class LEMWorker: public art::EDProducer
  {
  public:
    // Allows 'nova --print-description' to work
    using Parameters = art::EDProducer::Table<LEMWorkerParams>;

    explicit LEMWorker(const Parameters& params);
    ~LEMWorker();

    virtual void produce(art::Event& evt);

  protected:
    std::string GetWork() const;

    LEMWorkerParams fParams;

    std::string fWorkerID;
  };

  //......................................................................
  LEMWorker::LEMWorker(const Parameters& params)
    : fParams(params())
  {
    produces<std::vector<LEMInput>>();

    // Randomize ports and so on properly
    srand(getpid());

    // Generate a unique ID so that the server can identify us, eg to detect
    // duplicate requests.
    char host[1024];
    gethostname(host, 1023);
    host[1023] = 0;
    const char* user = getenv("USER");
    if(!user) user = "(unknown)";
    fWorkerID = TString::Format("%s@%s.%d.%ld",
                                user, host, getpid(), time(0)).Data();

    curl_global_init(CURL_GLOBAL_ALL);
  }

  //......................................................................
  LEMWorker::~LEMWorker()
  {
    curl_global_cleanup();
  }

  struct GetWorkArgs
  {
    std::string params;
    LEMWebSettings web;
  };

  void* GetWorkThreadFunc(void* x)
  {
    const GetWorkArgs* args = (GetWorkArgs*)x;

    bool ok;
    const std::string work = get_query(args->web.host(),
                                       args->params,
                                       args->web.minPort(),
                                       args->web.maxPort(),
                                       args->web.queryTimeout(),
                                       args->web.retryTimeout(),
                                       ok);
    if(!ok){
      std::cerr << "No response from server after many retries. "
                << "Maybe there's no work for us. Aborting"
                << std::endl;
      abort();
    }

    delete args;
    return new std::string(work);
  }

  //......................................................................
  std::string LEMWorker::GetWork() const
  {
    // Instead of hanging around wasting valuable computation time waiting for
    // the server to give us the next event, fetch the next one in parallel
    // with ID'ing the previous event and hopefully have it ready when
    // produce() wants it.

    static bool once = true;
    static pthread_t thread;

    const LEMWebSettings& web = fParams.web();

    const std::string rel = fParams.releaseOverride().empty() ? get_release() : fParams.releaseOverride();

    const std::string query = TString::Format("get_work?rel=%s&id=%s&timeout=%d",
                                              rel.c_str(),
                                              fWorkerID.c_str(),
                                              web.queryTimeout()).Data();

    GetWorkArgs* args = new GetWorkArgs;

    args->params = query;
    args->web = web;

    void* ret;
    if(once){
      once = false;
      // First time out just make a synchronous call to get the ball rolling.
      // Clone the args since the thread function deletes them
      ret = GetWorkThreadFunc(new GetWorkArgs(*args));
    }
    else{
      // Otherwise, fetch the result from the thread, wait for it if necessary.
      pthread_join(thread, &ret);
    }

    // Get ahold of the result more safely.
    const std::string work = *((std::string*)ret);
    delete (std::string*)ret;

    // And fire up a new request, ready for next time.
    pthread_create(&thread, 0, GetWorkThreadFunc, args);

    return work;
  }

  //......................................................................
  void LEMWorker::produce(art::Event& evt)
  {
    const std::string work = GetWork();

    char workstr[work.size()+1];
    strcpy(workstr, work.c_str());

    LEMInput input;

    input.tag = &strtok(workstr, "\n")[4]; // skip over "tag="

    char* estr = strtok(0, "\n");
    assert(estr);
    int n = sscanf(estr, "E=%lf", &input.totalGeV);
    assert(n == 1);

    while(char* chanqstr = strtok(0, "\n")){
      LiteHit hit;
      n = sscanf(chanqstr, "%hu %f", &hit.cellIdx, &hit.pecorr);
      assert(n == 2);
      input.hits.push_back(hit);
    }

    std::unique_ptr<std::vector<LEMInput>> sumcol(new std::vector<LEMInput>(1, input));
    evt.put(std::move(sumcol));
  }

  DEFINE_ART_MODULE(LEMWorker)

} //namespace lem
////////////////////////////////////////////////////////////////////////