source: tags/1.2/disparm.cpp @ 5

#include <iostream>
#include <fstream>
#include <libgen.h>
#include <cstdlib>
#include <string.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <unistd.h>

#include "disparm.h"
#include "tofromstring.h"
#include "token.h"
#include "shellenv.h"
#include <errno.h>
#include <dirent.h>

const std::string disparm::keyname       = "DISPARM_KEY";
const std::string disparm::committedname = "DISPARM_COMMITTED";
const std::string disparm::valuename     = "DISPARM_VALUE";
const std::string disparm::rcname        = "DISPARM_RC";
const std::string disparm::linename      = "DISPARM_LINENO";

const std::string disparm::defpath_pool   = "pool.disparm";
const std::string disparm::definputfilename = "-";

//
//constructor disparm
//
disparm::disparm()
{
  char defpoolname_c[this->defpath_pool.length()+1];        // default path_pool
  strcpy(defpoolname_c,this->defpath_pool.c_str());         
  //
  // A note about basename and dirname: they need as argument a char*
  // NOT a const char*. Both functions may modify the argument,
  // so, in general, do not use the same argument more than
  // once.
  //
  char defpath_lockdir_c[this->defpath_lockdir.length()+1];  // default directory for the lockfile
  strcpy(defpath_lockdir_c,this->defpath_lockdir.c_str());
  this->setpath_pool(this->defpath_pool);
  this->defpath_lockdir = std::string(dirname(defpoolname_c));
  this->setpath_link("");
  this->setpath_lock("");     // the name of the lock folder
  this->setbaselinkfilename("");
  this->setinputfilename(this->definputfilename);  // default input file name
  this->setpath_lockdir(this->defpath_lockdir);      // default lock directory
  this->setmaxiter(1);                             // default maximum times a string can be given out
  this->pmarker="pp";                              // start of record with pointer to next string in pool
  this->pmarkerlen = this->pmarker.length();
  this->lockname="lock.disparm";

  std::string shell = getenv("SHELL");
  if (shell.find("csh") != std::string::npos) 
    this->shelltype = SHELL_CSH;
  else
    this->shelltype = SHELL_SH;

  //
  // following items are returned on stdout as in:
  // export DISPARM_KEY=1234
  // See also above
  //
  this->key       = -1;     // key: pointer to next item in pool
  this->status    = '0';
  this->committed = 0;      // the number of times the string is given out
  this->rc        = "OK";   // return code
  this->value     = "";     // the string to give out
  this->linenr    = 0;      // the line number in the pool

  this->debug     = 0;

  int rc = posix_memalign((void**)&buffer, 4096, 512);
  if (rc != 0)
    std::cerr << "Disparm: posix_memalign fails" << std::endl;
}

// destructor
disparm::~disparm()
{
  free(this->buffer);
}

void disparm::removelock()
{
  unlink(this->path_link.c_str());
  // wwvv not removing lock directory
  // rmdir(this->path_lock.c_str());
}

void disparm::setpath_pool(std::string pool)
{
  this->path_pool = pool;
}

void disparm::setpath_link(std::string path_link)
{
  this->path_link = path_link;
}

void disparm::setpath_lock(std::string path_lock)
{
  this->path_lock = path_lock;
}

void disparm::setbaselinkfilename(std::string baselinkfilename)
{
  this->baselinkfilename = baselinkfilename;
}

int disparm::setpath_lockdir(std::string lockdir)
{
  this->path_lockdir = lockdir;
  int rc = mkdir(path_lockdir.c_str(),0777);
  if (rc ==0)
    return 0;
  else
  {
    if (errno == EEXIST)
    {
      struct stat buf;
      stat(path_lockdir.c_str(),&buf);
      if (S_ISDIR(buf.st_mode))
        return 0;
    }
    else
      return -1;
  }
  return 0;  // never reached
}

//
// Create a name for the directory to be used as lock: this->path_lock
// This directory will reside in this->path_lockdir
// We try to generate a name for a linkfile that is unique.
// In the name is also the creation time.
// The link file will be placed in the lock directory
//
void disparm::createlinkfilename()
{
  if (this->debug)
    this->debugger("Before createlinkfilename:");
  char hostname[1025];
  gethostname(hostname,1024);
  char poolname_c[this->path_pool.length()+1];
  strcpy(poolname_c,this->path_pool.c_str());
  std::string poolnamebase = basename(poolname_c);
  this->lockname=poolnamebase+".lock.disparm";
  this->setpath_lock(this->path_lockdir + "/" + this->lockname);
  std::string base = this->path_lock + "/" + poolnamebase + ".link.";
  this->setbaselinkfilename(base);
  this->setpath_link(   base
                         + to_string(time(0))
                         + "." 
                         + hostname 
                         + "." 
                         + to_string(getpid()));
  if (this->debug)
    this->debugger("After createlinkfilename:");
}

void disparm::setinputfilename(std::string inputfilename)
{
  this->inputfilename = inputfilename;
}

void disparm::setkey(std::streampos key)
{
  this->key = key;
}

std::streampos disparm::getkey(void)
{
  return this->key;
}

void disparm::setmaxiter(int maxiter)
{
  this->maxiter = maxiter;
}

//
// Create poolfile from input file
//
// On return, s wil contain the output of createrc
//
int disparm::create(std::string &s)
{
  if (this->debug)
    this->debugger("Before create:");
  std::ifstream inp;
  std::istream *pinp;
  s = "";
  //
  // open inputfile, take care that stdin is also allowed
  //
  if (this->inputfilename == "-")
    pinp = &std::cin;
  else
  {
    inp.open(this->inputfilename.c_str());
    if (!inp.is_open())
    {
      std::cerr<<"Cannot open '"<<inputfilename<<"'"<<std::endl;
      this->rc = "FAIL";
      s = this->createrc();
      return 1;
    }
    pinp = &inp;
  }
  //
  // open poolfile
  //
  std::string line;
  std::fstream out;
  out.open(this->path_pool.c_str(),std::fstream::in | std::fstream::out
                               |  std::fstream::trunc);

  //
  // fill poolfile
  //
  token tok;
  tok.setfile(out);
  std::streampos p;
  unsigned int linenr = 0;
  while (getline(*pinp,line))
  {
    tok.setlinenr(++linenr);
    tok.setcommitted(0);
    tok.clearstatus();
    tok.setvalue(line);
    tok.writetoken(p);
  }
  tok.writelocpointer(0);

  //
  // close input file, only if it is not stdin
  //
  if (pinp == &inp)
    inp.close();
  out.close();

  this->key = 0;
  s = createrc();
  if (this->debug)
    this->debugger("After create:");

  return 0;
}

//
// get next string from pool
//
int disparm::next(std::string& s)
{
  if (this->debug)
    this->debugger("Before next:");
  int rc = this->realnext();
  s = this->createrc();
  if (this->debug)
    this->debugger("After next:");
  return rc;
}

//
// create string to be eval'd by the shell
//
std::string disparm::createrc(void)
{
  std::string s = "";
  s = s + shellenv(to_string(this->key),      this->keyname,      this->shelltype);
  s = s + shellenv(to_string(this->committed),this->committedname,this->shelltype);
  s = s + shellenv(to_string(this->value),    this->valuename,    this->shelltype);
  s = s + shellenv(to_string(this->rc),       this->rcname,       this->shelltype);
  s = s + shellenv(to_string(this->linenr),   this->linename,     this->shelltype);
  return s;
}

// 
// get the next string from the pool
//
int disparm::realnext()
{
  std::fstream file;
  bool good = 1;

  this->key       = 0;
  this->status    = '0';
  this->committed = 0;
  this->rc        = "OK";
  this->linenr    = 0;
  this->value     = "";

  //
  // get lock, return if that fails
  //
  if (this->getlock()!=0)
  {
    this->rc = "FAIL";
    return 1;
  }

  //
  // read next string from pool
  // take care of ready records, maximum iterations
  // allowed and some kinds of mishaps
  //
  // open first using O_DIRECT, to force a fresh cache.
  int f = open(this->path_link.c_str(),O_DIRECT|O_RDONLY);
  int bytesread = read(f,buffer,512);
  if (bytesread <0)
  {
    std::cerr << "How strange: cannot read from poolfile." << std::endl;
  }
  close(f);
  file.open (this->path_link.c_str(), std::fstream::in | std::fstream::out);
  if (file.is_open())
  {
    token       tok;
    std::string line;
    tok.setfile(file);
    std::streampos locpointer, posused, posnext;

    //
    // get pointer to next string
    //
    locpointer = tok.readlocpointer();
    if (locpointer < 0)
    {
      std::cerr << "Corrupt parameterfile: cannot find locpointer\n" << std::endl;
      this->rc = "FAIL";
      good = 0;
    }
    //
    // read next string from pool
    //
    if (good)
    {
      int rc = tok.readnext(locpointer,posused,posnext);
      if (rc != 0)
      {
        std::cerr << __FILE__ << ":" << __LINE__ << std::endl;
        std::cerr << "Corrupt or empty parameterfile: cannot read next line\n" << std::endl;
        this->rc = "FAIL";
        good = 0;
      }
    }
    //
    // keep reading until a valid record is found
    // take appropriate action if no such record exists
    //
    if (good)
    {
      std::streampos locpointerold = locpointer;
      bool allready = 1;  // to learn if all records are
                          // ready
      while (1)
      {
        if (!tok.getready())
        {
          allready = 0;
          if (tok.getcommitted() >= this->maxiter)
          {
            good = 0;
            this->rc = "MAX_EXCEEDED";
            break;
          }
          this->key = posused;
          this->status = tok.getstatus();
          this->committed = tok.getcommitted();
          this->value = tok.getvalue();
          this->linenr = tok.getlinenr();
          tok.setcommitted(tok.getcommitted()+1);
          tok.writetoken(posused);
          tok.writelocpointer(posused);
          break;
        }
        locpointer = posnext;
        if (locpointer == locpointerold)
        {
          //
          // so we are cycling, nothing to find, it seems
          //
          good = 0;
          if (allready)
            this->rc = "ALL_READY";
          else
            this->rc = "MAX_EXCEEDED";
          break;
        }
        int rc = tok.readnext(locpointer,posused,posnext);
        if (rc != 0)
        {
          std::cerr << __FILE__ << ":" << __LINE__ << std::endl;
          std::cerr << "Corrupt or empty parameterfile: cannot read next line\n" << std::endl;
          this->rc = "FAIL";
          good = 0;
          break;
        }
      }
    }
  }
  else
  {
    std::cerr << __FILE__ << ":" << __LINE__ << std::endl;
    std::cerr<< this->arg << ": Strange error: cannot open '"<<path_link<<"'"<<std::endl;
    return 1;
  }

  file.close();
  this->removelock();
  return !good;
}

// 
// get a lock on the poolfile
//
// The normal methods don't work on an nfs file system.
// Luckily, there is a method based on the creation
// of a directory, which is an atomic operation.
// If we can create the directory, we are sure that no 
// other instantiation of this program was also successful.
// After successful creation of the lock directory, we create
// a hard link to the poolfile, and use this link to read from and
// write to. We do not use the poolfile itself to read and write,
// we can never be sure in a nfs filesystem of the status of a 
// file. It appears that creating a new hard link suffices to
// get an up to date version of the poolfile.
//
int disparm::getlock(void)
{
  struct stat buf;
  int rc;
  int sleeptime=2;

  int iter = 0;
  int ntryremovelock = 0;
  while(1)
  {
    rc  = stat(this->path_pool.c_str(),&buf);
    std::string er="";
    switch (errno)
    {
      case (EACCES): er="Search permission is denied for one of the  directories  in  the path prefix of path.  (See also path_resolution(7).)";
                     break;

      case(EBADF):  er="fd is bad.";
                    break;

      case(EFAULT): er="Bad address.";
                    break;

      case(ELOOP):  er="Too many symbolic links encountered while traversing the path.";
                    break;

      case(ENAMETOOLONG):er="File name too long.";
                         break;

      case(ENOENT): er="A component of path does not exist, or path is an empty string.";
                    break;

      case(ENOMEM): er="Out of memory (i.e., kernel memory).";
                    break;

      case(ENOTDIR):
              er="A component of the path prefix of path is not a directory.";
              break;

      case(EOVERFLOW):er="(stat())  path refers to a file whose size cannot be represented in the type off_t.  This can occur when an application  compiled on a 32-bit platform without -D_FILE_OFFSET_BITS=64 calls stat() on a file whose size exceeds (2<<31)-1 bits";
                      break;

    }
    if (rc !=0 )
    {
      perror("Disparm: Error stat poolfile:");
      std::cerr << __FILE__ << ":" << __LINE__ << std::endl;
      std::cerr<<er<<std::endl;
      std::cerr<< this->arg << ": Cannot stat the pool file:'"<<this->path_pool.c_str()<<"'"<< std::endl;
      return 1;
    }
    this->createlinkfilename();
    //
    // Try to create the lock directory
    //
    rc = mkdir(this->path_lock.c_str(), 0700);
    // it seems that relying on the atomicity of 
    // mkdir does not work on the Lisa system.
    // therefore, we skip the test and rely on the creation of 
    // links
    rc = 0;
    if (rc == 0)
    {
      // Creation of lock directory was successful, now create link
      // to the poolfile
      // As an extra check: check if the number of hard links to the 
      // poolfile equals 2.
      //
      // Strange: sometimes, when many threads are involved on one node,
      // the poolfile seems to be gone, get error: "no such file or directory"
      // I guess this is a race condition in the NFS filesystem
      // so, I'll try 10 times, and then give up.
      //
      int tries = 0;
      while (1)
      {
        tries++;
        int rc = link(this->path_pool.c_str(),this->path_link.c_str());
        if (rc < 0)
        {
          if (tries < 10)
          {
            sleep(1);
          }
          else
          {
            perror("Disparm: Error making link:");
            std::cerr << __FILE__ << ":" << __LINE__ << std::endl;
            std::cerr<< this->arg << ": Cannot make link '"<<this->path_link<<"' to '"
              <<this->path_pool<<"'"<<std::endl<<" Is poolname '"<<path_pool
              <<"' correct?"<<std::endl;

            this->removelock();
            return 1;
          }
        }
        else
          break;
      }
      //
      // check if number of hard links equals 2
      //
      struct stat linkbuf;
      rc = stat(this->path_link.c_str(),&linkbuf);
      if (rc !=0)
      {
        std::cerr << this->arg << ": Cannot stat just created link'"<<this->path_link<<"'"<<std::endl;
        this->removelock();
        return 1;
      }
      if (linkbuf.st_nlink == 2)
      {
        break;
      }
      else
      {
        rc = unlink(this->path_link.c_str());
      }
    }
    //
    // wait some time before trying again
    //
    iter++;
#ifdef VERBOSE
    std::cerr << iter <<": waiting for lock " << std::endl;
#endif
    sleep(sleeptime);
    if (iter > 9)
    {
#ifdef VERBOSE
      std::cerr << "Trying to remove locks" << std::endl;
#endif
      //
      // remove stale link files
      //
      this->remove_lock_and_links();
      ntryremovelock ++;
      if (ntryremovelock > 100)
      {
        std::cerr << this->arg << ": Cannot get lock after many tries." << std::endl;
        this->removelock();
        return 1;
      }
      iter = 0;
    }
  }
  return 0;
}

//
// mark string as ready
//
int disparm::ready()
{
  if (this->getlock()!=0)
  {
    return 1;
  }

  //
  // if no key given as parameter, read it from environment
  //
  if (this->key < 0)
  {
    char * s = getenv(this->keyname.c_str());
    if (s == NULL || *s == 0)
    {
      this->removelock();
      std::cerr << this->arg << ": No key given" << std::endl;
      return 1;
    }
    else
    {
      this->key = from_string<long long>(std::string(s));
    }
  }

  std::streampos locpointer = this->key;
  std::fstream file;

  // open first using O_DIRECT, to force a fresh cache.
  int f = open(this->path_link.c_str(),O_DIRECT|O_RDONLY);
  int bytesread = read(f,buffer,512);
  if (bytesread <0)
  {
    std::cerr << this->arg << ": cannot read from poolfile." << std::endl;
    return 1;
  }
  close(f);
  file.open (this->path_link.c_str(), std::fstream::in | std::fstream::out);
  if (file.is_open())
  {
    token tok;
    std::string line;
    tok.setfile(file);
    std::streampos posused,posnext;
    //
    // read next record
    //
    int rc = tok.readnext(locpointer,posused,posnext);
    if (rc != 0)
    {
      std::cerr << __FILE__ << ":" << __LINE__ << std::endl;
      std::cerr << this->arg << ": Corrupt or empty parameterfile: cannot read next line\n" << std::endl;
      file.close();
      this->removelock();
      return 1;
    }
    //
    // mark record as ready
    //
    tok.ready();
    //
    // write record back
    //
    tok.writetoken(locpointer);
  }
  else
  {
    std::cerr << this->arg << ": Cannot read " << this->path_link << std::endl;
  }
  file.close();
  this->removelock();
  return 0;
}

//
// remove poolfile
//
int disparm::REMOVE()
{
  if (this->getlock()!=0)
  {
    return 1;
  }
  this->removelock();

  int rc = unlink(this->path_pool.c_str());
  if (rc)
  {
    std::cerr << this->arg << ": Cannot remove " << this->path_pool << std::endl;
    return 1;
  }
  return 0;
}

//
// return statistics about the pool
// no wait for lock, so be careful interpretating the results
//
stats disparm::getstats()
{
  stats statistics;
  token tok;
  std::fstream file;
  statistics.valid     =  0;
  statistics.records   = -1;
  statistics.committed = -1;
  statistics.ready     = -1;
  statistics.commax    = -1;


  file.open (this->path_pool.c_str(), std::fstream::in | std::fstream::out);

  if (file.is_open())
  {
    tok.setfile(file);
    std::streampos pos = 0;
    std::streampos posused;
    std::streampos posnext;
    statistics.valid     = 1;
    statistics.records   = 0;
    statistics.committed = 0;
    statistics.ready     = 0;
    statistics.commax    = 0;
    
    while(1)
    {
      tok.readnext(pos, posused,posnext);
      if (pos != posused)
        break;
      if (tok.getready())
        statistics.ready++;
      if (tok.getcommitted()>0)
        statistics.committed++;
      if (tok.getcommitted() > statistics.commax)
        statistics.commax = tok.getcommitted();
      statistics.records++;
      pos = posnext;
    }
    file.close();
  }
  else
  {
    std::cerr << this->arg << ": Cannot open poolfile '"<<this->path_pool<<"'"<<std::endl;
  }

  return statistics;
}

int disparm::echounhandled(void)
{
  token tok;
  std::fstream file;

  file.open (this->path_pool.c_str(), std::fstream::in | std::fstream::out);

  if (file.is_open())
  {
    tok.setfile(file);
    std::streampos pos = 0;
    std::streampos posused;
    std::streampos posnext;
    std::cerr << "linenr:committed:value"<< std::endl;
    
    while(1)
    {
      tok.readnext(pos, posused,posnext);
      if (pos != posused)
        break;

      if (!tok.getready()&&tok.getcommitted()>0)
      {
        std::cerr << tok.getlinenr() << ":"<<tok.getcommitted()<<":" << tok.getvalue() << std::endl;
      }
      pos = posnext;
    }
  }
  else
  {
    std::cerr << this->arg << ": Cannot open pool file '"<<this->path_pool<<"'"<<std::endl;
    return 1;
  }

  file.close();
  return 0;
}


int disparm::remove_lock_and_links(const std::string path, const std::string dirl, const std::string path_lock, int timer)
{
  // removes lock and hard links in directory "dirl" to file "path" 
  // in directory "dir", if they are older than "time" seconds. 
  // Method: 
  // if lock is older than timer seconds, remove it.
  // and:
  // using readdir, examine all components of dirl and see
  // if they have the same inode as "path". If so, remove, unless
  // the realname is the same as the realname of "path"
  //
  DIR *pdir;
  struct dirent *pent;
  int rc;
  struct stat buf;

  pdir = opendir(dirl.c_str());
  if (!pdir)
  {
    std::cerr << this->arg << ": Cannot read link directory '"<<dirl<<"'"<<std::endl;
    return 1;
  }
  rc = stat(path.c_str(),&buf);
  if (rc !=0)
  {
    std::cerr << this->arg << ": Cannot stat '"<<path<<"'"<<std::endl;
    return 1;
  }
  ino_t inodenr = buf.st_ino;
  char* rpathc = (char*)malloc(PATH_MAX+1);
  rpathc = realpath(path.c_str(),rpathc);
  std::string rpath = rpathc;
  std::string rpath1;
  
  while ((pent=readdir(pdir)))
  {
    struct stat buf;
    std::string name = pent->d_name;
    if (name == ".") continue;
    if (name == "..") continue;
    rpathc = realpath((dirl+"/"+name).c_str(),rpathc);
    rpath1 = rpathc;
#ifdef VERBOSE
    std::cerr<<"TD: looking at '"<<rpath1<<"'"<<std::endl;
#endif
    if (rpath != rpath1)
    {
      int rc = stat(rpath1.c_str(),&buf);
      if (rc != 0)
      {
        std::cerr << this->arg << ": Cannot stat '"<<rpath1<<"'"<<" this cannot happen!" << std::endl;
        free(rpathc);
        return 1;
      }
      if (buf.st_ino == inodenr)
      {
#ifdef VERBOSE
        std::cerr << "time: " << time(0) << " " << buf.st_mtime << " " << timer << std::endl;
#endif
        if (time(0) > buf.st_mtime + timer)
        {
          int rc = unlink(rpath1.c_str());
          if (rc == 0)
          {
#ifdef VERBOSE
            std::cerr<<"Unlinked '"<<rpath1<<"'"<<std::endl;
#endif
          }
          else
          {
#ifdef VERBOSE
            std::cerr<<"Could not unlink '"<<rpath1<<"'"<<std::endl;
#endif
          }
        }
        else
        {
#ifdef VERBOSE
          std::cerr << "TD: not unlinking because time: '"<<rpath1<<"'"<<std::endl;
#endif
        }
      }
    }
    else
    {
#ifdef VERBOSE
      std::cerr<<"TD: not unlinking '"<<rpath1<<"'"<<std::endl;
#endif
    }
  }
  free(rpathc);
  // look if lock can be removed, if yes: and make it so 
  rc = stat(path_lock.c_str(),&buf);
  if (rc == 0)
  {
    if (time(0) > buf.st_mtime + timer)
    {
#ifdef VERBOSE
      std::cerr << "Trying to remove '"<<path_lock<<"'"<<std::endl;
#endif
      // wwvv not removing lock directory
      // rc = rmdir(path_lock.c_str());
      // It could happen that rmdir failed, for example another process
      // already deleted it, or path is invalid etc.
      // We ignore this, relying on the fact that eventually
      // getlock will time out.
    }
  }
  return 0;
}


//
// remove lock and links
// method: do a glob on baselinkfilename and unlink
// the files.
//
// Only remove if links are older than 60 seconds.
//
void disparm::remove_lock_and_links()
{
  if (this->debug)
    this->debugger("Before remove_lock_and_links:");

  remove_lock_and_links(this->path_pool,this->path_lockdir,this->path_lock,60);
  if (this->debug)
    this->debugger("Before remove_lock_and_links:");
}

void disparm::setdebug()
{
  this->debug = 1;
}
void disparm::debugger(const std::string s)
{
  std::cerr << s << std::endl;
  std::cerr << "baselinkfilename: '" << this->baselinkfilename << "'" << std::endl;
  std::cerr << "committed:        '" << this->committed << "'" << std::endl;
  std::cerr << "committedname:    '" << this->committedname << "'" << std::endl;
  std::cerr << "definputfilename: '" << this->definputfilename << "'" << std::endl;
  std::cerr << "defpath_pool:     '" << this->defpath_pool << "'" << std::endl;
  std::cerr << "defpath_lockdir:  '" << this->defpath_lockdir << "'" << std::endl;
  std::cerr << "inputfilename:    '" << this->inputfilename << "'" << std::endl;
  std::cerr << "key:              '" << this->key << "'" << std::endl;
  std::cerr << "keyname:          '" << this->keyname << "'" << std::endl;
  std::cerr << "linename:         '" << this->linename << "'" << std::endl;
  std::cerr << "linenr:           '" << this->linenr << "'" << std::endl;
  std::cerr << "lockname:         '" << this->lockname << "'" << std::endl;
  std::cerr << "maxiter:          '" << this->maxiter << "'" << std::endl;
  std::cerr << "path_link:        '" << this->path_link << "'" << std::endl;
  std::cerr << "path_lock:        '" << this->path_lock << "'" << std::endl;
  std::cerr << "path_lockdir:     '" << this->path_lockdir << "'" << std::endl;
  std::cerr << "path_pool:        '" << this->path_pool << "'" << std::endl;
  std::cerr << "pmarker:          '" << this->pmarker << "'" << std::endl;
  std::cerr << "pmarkerlen:       '" << this->pmarkerlen << "'" << std::endl;
  std::cerr << "pointerlen:       '" << this->pointerlen << "'" << std::endl;
  std::cerr << "rc:               '" << this->rc << "'" << std::endl;
  std::cerr << "rcname:           '" << this->rcname << "'" << std::endl;
  std::cerr << "shelltype:        '" << this->shelltype << "'" << std::endl;
  std::cerr << "status:           '" << this->status << "'" << std::endl;
  std::cerr << "value:            '" << this->value << "'" << std::endl;
  std::cerr << "valuename:        '" << this->valuename << "'" << std::endl;
}

void disparm::setarg(int argc, char*argv[])
{
  this->arg = argv[0];
  for (int i=1; i<argc; i++)
    this->arg += std::string(" ") + std::string(argv[i]);
  this->arg += " pid:" + to_string(getpid());
}

//
// Just to teach me again how to make an object 
// suitable for the << operator
//
std::ostream& operator <<(std::ostream &os,disparm &obj)
{
  stats sta;
  sta = obj.getstats();
  if ( sta.valid )
  {
    os << "Total:     " << sta.records   << "\n" 
       << "Committed: " << sta.committed << "\n"
       << "Ready:     " << sta.ready     << "\n"
       << "Max:       " << sta.commax;
  }
  return os;
}