ngcbPP.hpp

Go to the documentation of this file.

#ifndef ngcbPP_H
#define ngcbPP_H
 
 
#ifndef __cplusplus
#error This is a C++ include file and cannot be used from plain C
#endif
 
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <unistd.h>
#include <errno.h>
#include <dlfcn.h>
#include <ngc/core/ngcb.h>
#include <ngc/core/ngcbSched.h>
 
/*
 * Processing pipeline result codes
 */
#define ngcbPP_SUCCESS      0x0000  //< success
#define ngcbPP_SKIP         0x0001  //< skip processing
#define ngcbPP_FAILURE      0x0100  //< processing failed
#define ngcbPP_ERR_FATAL    0x1100  //< fatal error (cannot continue)

#define ngcbPPLOG(...) do {char s[320]; snprintf(s, 320, __VA_ARGS__); Log(s);} while (0)

#define ngcbPP_MAX_LIB 32

struct ngcb_pp_descr {
  int result;

  int ftype;

  int fcnt; 

  int bitPix; 

  double bscale;

  double bzero; 

  int sx; 

  int sy; 

  int nx; 

  int ny;    

  double crpix1;

  double crpix2;

  ngcb_pp_descr() {
    result = ngcbPP_SUCCESS;
    ftype = 0; fcnt = 0;
    bitPix = 16; bscale = 1.0; bzero = 0.0;
    sx = 1; sy = 1; nx = 0; ny = 0;
    crpix1 = 0.0; crpix2 = 0.0;
  }

  size_t NumPix() {return ((size_t)nx * (size_t)ny);}

  size_t Size() {return (NumPix() * (size_t)(abs(bitPix) / 8));}
};

typedef void *(*ngcb_pp_thr)(void *arg);

class ngcb_pp_sync 
{
public:
  pthread_t thid;
  
  ngcb_pp_sync() {
    initialized_ = 0;
    erms_[0] = '\0';
    // cppcheck-suppress useInitializationList
    thid = ngcbThrId();
    if (ngcbSemInit(&semStart_, 0, ngcbSEM_FAST, NULL) != 0)
      {
      strcpy(ErrMsg(), "semaphore creation failed");
      return;
      }
    else if (ngcbSemInit(&semAck_, 0, ngcbSEM_FAST, NULL) != 0)
      {
      ngcbSemDestroy(&semStart_);
      strcpy(ErrMsg(), "semaphore creation failed");
      return;
      } 
    initialized_ = 1;
  }

  virtual ~ngcb_pp_sync() {
    if (initialized_)
      {
      ngcbSemDestroy(&semStart_);
      ngcbSemDestroy(&semAck_);
      }
  }

  int Initialized() {return (initialized_);}

  char *ErrMsg() {return (erms_);}

  void Start() {if (initialized_) ngcbSemPost(&semStart_);}

  void Join() {if (initialized_) ngcbSemWait(&semAck_);}

  void Wait() {if (initialized_) ngcbSemWait(&semStart_);}

  void Ack() {if (initialized_) ngcbSemPost(&semAck_);}
protected:
private:
  int initialized_;

  char erms_[256];

  ngcb_sema_t semStart_;

  ngcb_sema_t semAck_;
};

class ngcbPP
{
public:
  int verboseLevel;

  int logLevel;

  ngcbPP *next;

  ngcb_pp_sync *psync;

  ngcb_pp_thr proc;

  pthread_t thid;

  void *procBuf;
  
  ngcb_pp_descr procDescr;

  ngcbPP() {Init_(-1);}

  explicit ngcbPP(int id) {Init_(id); }

  virtual ~ngcbPP() {
    ngcbPPLOG("deleting processor %d\n", Id());
    if (next != NULL) 
      {
      if (join_)
    {
    /*
     * Wait for next thread
     */
    if (next->proc != NULL)
      {
      ngcbPPLOG("joining processor %d\n", next->Id());
      ngcbThrJoin(next->thid); // thread is terminated
      next->proc = NULL; 
      ngcbPPLOG("processor %d joined\n", next->Id());
      }
    else if (next->psync != NULL)
      {
      ngcbPPLOG("waiting for processor %d\n", next->Id());
      next->psync->Join(); // thread is now in waiting state
      ngcbPPLOG("cancelling processor %d\n", next->Id());
      if (ngcbThrCancel(next->psync->thid) == 0)
        {
        ngcbPPLOG("joining processor %d\n", next->Id());
        ngcbThrJoin(next->psync->thid); // thread is terminated
        delete next->psync; next->psync = NULL;
        ngcbPPLOG("processor %d joined\n", next->Id());
        }
      }
    join_ = 0;
    }
      delete next; next = NULL;
      }
    ngcbPPLOG("processor %d deleted\n", Id());
  }

  int Initialized() {return (initialized);}

  char *ErrMsg() {return (erms_);}

  void Id(int id) {id_ = id;}

  int Id() {return (id_);}

  int NumBuf() {return (numBuf);}

  int RdOnly() {return (rdOnly);}

  void *Execute(void *buffer, ngcb_pp_descr *descr) {
    void *in = buffer; 
    void *out;
 
    /*
     * Start our processor
     */
    ngcbPPLOG("processor: %d\n", Id());
    out = Process(in, descr);
    if (descr->result & ngcbPP_SKIP)
      {
      /*
       * Frame is not ready for further processing
       */
      return (out);
      }
 
    if (next != NULL)
      {
      if (out != NULL)
    {
    /*
     * We forward the result as input to the next instance
     */
    in = out;
    }
 
      if ((next->proc != NULL) || (next->psync != NULL))
    {
    if (join_)
      {
      join_ = 0;
 
      /*
       * Wait for result from next thread
       */
      ngcbPPLOG("waiting for processor %d\n", next->Id());
      if (next->proc != NULL)
        {
        if (ngcbThrJoin(next->thid) != 0)
          {
          descr->result |= ngcbPP_ERR_FATAL;
          sprintf(ErrMsg(), "processer %d thread synchronization failed",
              Id());
          return (NULL);
          }
        }
      else
        {
        next->psync->Join();
        }
      ngcbPPLOG("processor %d ready\n", next->Id());
 
      out = next->procBuf;
      tmpDescr_ = next->procDescr;
      next->procDescr = *descr;
      *descr = tmpDescr_;
 
      if (descr->result & ngcbPP_FAILURE)
        {
        strcpy(ErrMsg(), next->ErrMsg());
        if (descr->result & ngcbPP_ERR_FATAL)
          {
          return (NULL);
          }
        }
      }
    else
      {
      /*
       * We do not yet have a final result
       */
      out = NULL;
      next->procDescr = *descr;
      }
 
    /*
     * Start next processing thread
     */
    next->procBuf = in;
    next->procDescr.result = 0x0;
    if (next->proc != NULL)
      {
      ngcbPPLOG("creating thread for processor %d\n", next->Id());
      if (ngcbThrCreate(next->proc, (void *)next, 0, &next->thid) != 0)
        {
        descr->result |= ngcbPP_ERR_FATAL;
        sprintf(ErrMsg(), "processor %d thread creation failed", Id());
        return (NULL);
        }
      }
    else
      {
      ngcbPPLOG("starting thread for processor %d\n", next->Id());
      next->psync->Start();
      }
    join_ = 1;
    }
      else
    {
    /*
     * Execute next processor
     */
    out = next->Execute(in, descr);
    if (descr->result & ngcbPP_FAILURE) strcpy(ErrMsg(), next->ErrMsg());
    }
      }
    else
      {
      /*
       * End of processing pipeline
       */
      ngcbPPLOG("pipeline ready\n");
      }
 
    return (out);
  }
  
  virtual void *Process(void *buffer, ngcb_pp_descr *descr) {
    descr->result = ngcbPP_SUCCESS;
    return (buffer);
  }
  
protected:
  int initialized;

  int numBuf;
  
  int rdOnly;

  void Init_(int id) {
    initialized = 0;
    id_ = id; 
    verboseLevel = 0; logLevel = 0;
    next = NULL;
    psync = NULL;
    proc = NULL; 
    procBuf = NULL; 
    thid = ngcbThrId();
    join_ = 0;
 
    erms_[0] = '\0';
 
    numBuf = 0;
    rdOnly = 1; 
    initialized = 1;
  }

  void Log(const char *msg) {
    if (verboseLevel > 0) fprintf(stdout, "%s", msg);
  }

  void *Malloc(size_t size) {
    unsigned char *m = (unsigned char *)malloc(size);    
    if (m != NULL)
      {
      unsigned int i;
      int pageSize = (int)sysconf(_SC_PAGESIZE);
      if (pageSize < 1) pageSize = 4096; /* TBD */
      
      /*
       * Touch each page to get it mapped into RAM
       */
      for (i=0;i<size;i+=pageSize) m[i] = (unsigned char)0;
      }
    return ((void *)m);
  }
 
private:
  int id_;

  char erms_[256];

  int join_;

  ngcb_pp_descr tmpDescr_;
};

static void *ngcbPP_Proc(void *arg)
{
  ngcbPP *p = static_cast<ngcbPP *>(arg);
  sigset_t set;
 
  /*
   * We do not want to interfere with signals here
   */
  sigfillset(&set);
  pthread_sigmask(SIG_BLOCK, &set, (sigset_t *)NULL);
 
  /*
   * Execute processor
   */
  p->procBuf = p->Execute(p->procBuf, &p->procDescr);
 
  return (NULL);
}

static void *ngcpPP_ProcSync(void *arg)
{
  ngcbPP *p = static_cast<ngcbPP *>(arg);
  sigset_t set;
 
  /*
   * We do not want to interfere with signals here
   */
  sigfillset(&set);
  pthread_sigmask(SIG_BLOCK, &set, (sigset_t *)NULL);
 
  while (1)
    {
    /*
     * Wait for start signal
     */
    p->psync->Wait();
 
    /*
     * Execute processor
     */
    p->procBuf = p->Execute(p->procBuf, &p->procDescr);
 
    /*
     * Send acknowledge
     */
    p->psync->Ack();
    }
 
  return (NULL);
}

class ngcbPP_FACTORY
{
public:
  ngcbPP_FACTORY() {
    erms_[0] = '\0';
  }

  virtual ~ngcbPP_FACTORY() {}

  char *ErrMsg() {return (erms_);}

  ngcbPP *Create(int id) {
    ngcbPP *p = Make(id);
    if (p->Initialized())
      {
      /*
       * Processor initialization successful
       */
      return (p);
      }
    else
      {
      /*
       * Processor initialization failed
       */
      strcpy(ErrMsg(), p->ErrMsg());
      delete p;
      return (NULL);
      }
  }
 
protected:
  virtual ngcbPP *Make(int id) {return (new ngcbPP(id));}
 
private:
  char erms_[256];
};

#define NGC_REGISTER_PP_MOD(type, name) extern "C" { type name; \
  int _ngcb_pp_ ## name = 1 ; }

class ngcbPIPELINE
{
public:
  int verboseLevel;

  int logLevel;

  ngcbPIPELINE() {
    int i;
    verboseLevel = 0; logLevel = 0; erms_[0] = '\0';
    first_ = NULL;
    pp_ = first_;
    idx_ = 0;
    ndl_ = 0;
    for (i=0;i<ngcbPP_MAX_LIB;i++) dl_[i] = NULL;
  }

  virtual ~ngcbPIPELINE() {
    if (first_ != NULL)
      {
      /*
       * Delete whole pipeline
       */
      delete first_;
      }
 
    if (ndl_ > 0)
      {
      int i;
 
      ngcbPPLOG("unloading libraries\n");
      for (i=0;i<ndl_;i++)
    {
    if (dl_[i] != NULL)
      {
      if (dlclose(dl_[i]) != 0) ngcbPPLOG("failure - %s\n", dlerror());
      dl_[i] = NULL;
      }
    }
      }
  }

  char *ErrMsg() {return (erms_);}

  void Add(ngcbPP *newProc) {
    if (newProc == NULL)
      {
      return;
      }
 
    if (newProc->Id() < 0)
      {
      /*
       * The ID is no yet assigned
       */
      newProc->Id(idx_++);
      }
 
    newProc->verboseLevel = verboseLevel;
    newProc->logLevel = logLevel;
    if (first_ == NULL)
      {
      /*
       * This is our first processing step
       */
      first_ = newProc;
      pp_ = first_;
      }
    else
      {
      /*
       * Add processor to the end of the pipeline
       */
      pp_->next = newProc;
      pp_ = pp_->next;;
      }
  }

  int Lib(const char *name) {
    char libName[128];
    void *dl;
 
    if (ndl_ >= ngcbPP_MAX_LIB)
      {
      strcpy(ErrMsg(), "maximum number of external libraries reached");
      return (-1);
      }
 
    /*
     * Load library
     */
    snprintf(libName, sizeof(libName), "lib%s.so", name);
    ngcbPPLOG("loading library <%s>\n", libName);
    dl = dlopen(libName, RTLD_NOW|RTLD_LOCAL);
    if (dl == NULL)
      {
      strcpy(ErrMsg(), dlerror());
      return (-1);
      }
    dl_[ndl_++] = dl;
 
    return (0);
  }

  int Add(const char *module) {
    char cookie[128];
    void *dl = NULL;
    ngcbPP_FACTORY *f = NULL;
    ngcbPP *newProc = NULL;
 
    if (module == NULL)
      {
      /*
       * Add dummy processor instance just for test
       */
      newProc = new ngcbPP(idx_++);
      if (newProc->Initialized())
    {
    Add(newProc);
    return (0);
    }
      else
    {
    strcpy(ErrMsg(), newProc->ErrMsg());
    delete newProc;
    return(-1);
    }
      }
 
    /*
     * Identification string
     */
    snprintf(cookie, sizeof(cookie), "_ngcb_pp_%s", module);
 
    if (ndl_ > 0)
      {
      int i;
 
      /*
       * First search loaded libraries
       */
      ngcbPPLOG("browsing libraries\n");
      for (i=0;i<ndl_;i++)
    {
    if (dl_[i] != NULL)
      {
      /*
       * Identification
       */
      if (dlsym(dl_[i], cookie) != NULL)
        {
        ngcbPPLOG("module <%s> identified\n", module);
 
        /*
         * Lookup processor factory symbol
         */
        dlerror();
        f = static_cast<ngcbPP_FACTORY *>(dlsym(dl_[i], module));
        if (f == NULL)
          {
          char *err = dlerror();
          if (err != NULL)
        {
        ngcbPPLOG("%s\n", err);
        }
          else 
        {
        ngcbPPLOG("invalid factory symbol <%s>\n", module);
        }
          }
        else
          {
          ngcbPPLOG("module <%s> registered\n", module);
          break;
          }
        }
      }
    }
      }
    
    if (f == NULL)
      {
      char libName[128];
 
      /*
       * Load module by name
       */
      ngcbPPLOG("loading module <%s>\n", module);
      snprintf(libName, sizeof(libName), "lib%s.so", module);
      dl = dlopen(libName, RTLD_NOW|RTLD_LOCAL);
      if (dl == NULL)
    {
    strcpy(ErrMsg(), dlerror());
    return (-1);
    }
 
      /*
       * Identification
       */
      if (dlsym(dl, cookie) == NULL)
    {
    ngcbPPLOG("unloading module <%s>\n", module);
    if (dlclose(dl) != 0)
      {
      ngcbPPLOG("failure - %s\n", dlerror());
      }
    sprintf(ErrMsg(), "module <%s> rejected - identification failed",
        module);
    return (-1);
    }
 
      /*
       * Lookup processor factory symbol
       */
      dlerror();
      f = static_cast<ngcbPP_FACTORY *>(dlsym(dl, module));
      if (f == NULL)
    {
    char *err = dlerror();
    if (err != NULL) strcpy(ErrMsg(), err);
    else sprintf(ErrMsg(), "invalid factory symbol <%s>", module);
    
    /*
     * Unload the module again
     */
    ngcbPPLOG("unloading module <%s>\n", module);
    if (dlclose(dl) != 0)
      {
      ngcbPPLOG("failure - %s\n", dlerror());
      }
    return (-1);
    }
      }
   
    /*
     * Create processor
     */
    ngcbPPLOG("creating processor for module <%s>\n", module);
    newProc = f->Create(idx_++);
    if (newProc == NULL)
      {
      if (dl != NULL)
    {
    /*
     * Unload the module again
     */
    ngcbPPLOG("unloading module <%s>\n", module);
    if (dlclose(dl) != 0)
      {
      ngcbPPLOG("failure - %s\n", dlerror());
      }
    return (-1);
    }
      }
    
    /*
     * Add processor
     */
    Add(newProc);
 
    return (0);
  }

  void *Execute(void *buffer, ngcb_pp_descr *descr) {
    void *out;
 
    /*
     * We start with success
     */ 
    descr->result = ngcbPP_SUCCESS;
 
    /*
     * Enter the processing pipeline
     */
    out = first_->Execute(buffer, descr);
    if (descr->result & ngcbPP_FAILURE) strcpy(ErrMsg(), first_->ErrMsg());
 
    return (out);
  }

  void Parallelize(int fastSync=0) {
    ngcbPP *p;
 
    /*
     * Assign built-in threading
     */
    for (p = first_; p != NULL; p = p->next)
      {
      if (p->next != NULL) p->next->proc = ngcbPP_Proc;
      }
 
    if (!fastSync)
      {
      /*
       * This was enough
       */
      return;
      }
    
    for (p = first_; p != NULL; p = p->next)
      {
      if (p->next != NULL)
    {
    if ((p->NumBuf() > 1) || (p->RdOnly() && p->next->RdOnly()))
      {
      ngcb_pp_sync *psync;
 
      /*
       * Create fast synchronization primitive
       */
      psync = new ngcb_pp_sync();
      if (psync->Initialized())
        {
        /*
         * Create static processing thread
         */
        ngcbPPLOG("creating processor %d\n", p->next->Id());
        p->next->psync = psync;
        if (ngcbThrCreate(ngcpPP_ProcSync, (void *)p->next, 0, 
                  &psync->thid) == 0)
          {
          /*
           * Disable built-in thread (slow-sync)
           */
          p->next->proc = NULL; 
          }
        else
          {
          ngcbPPLOG("fast-sync not available - thread creation failed\n");
          p->next->psync = NULL;
          delete psync;
          break;
          }
        }
      else
        {
        ngcbPPLOG("fast-sync not available - %s\n", psync->ErrMsg());
        p->next->psync = NULL;
        delete psync;
        }
      }
    }
      }
  }

  void VerboseLevel(int level) {
    ngcbPP *p;
    verboseLevel = level;
    for (p = first_; p != NULL; p = p->next) p->verboseLevel = level;
  }

  void LogLevel(int level) {
    ngcbPP *p;
    logLevel = level;
    for (p = first_; p != NULL; p = p->next) p->logLevel = level;
  }
 
protected:
  void Log(const char *msg) {
    if (verboseLevel > 0) fprintf(stdout, "%s", msg);
  }
 
private:
  ngcbPP *first_;

  ngcbPP *pp_;

  int idx_;

  char erms_[256];

  void *dl_[ngcbPP_MAX_LIB];

  int ndl_;
};
 
#endif