source: trunk/disparm.cpp

/*
   Copyright 2012 Willem Vermin, SARA

   Licensed under the Apache License, Version 2.0 (the "License");
   you may not use this file except in compliance with the License.
   You may obtain a copy of the License at

       http://www.apache.org/licenses/LICENSE-2.0

   Unless required by applicable law or agreed to in writing, software
   distributed under the License is distributed on an "AS IS" BASIS,
   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   See the License for the specific language governing permissions and
   limitations under the License.
 */

#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;
}