pstream.c

/*
  This file is part of CDO. CDO is a collection of Operators to
  manipulate and analyse Climate model Data.

  Copyright (C) 2003-2012 Uwe Schulzweida, Uwe.Schulzweida@zmaw.de
  See COPYING file for copying and redistribution conditions.

  This program is free software; you can redistribute it and/or modify
  it under the terms of the GNU General Public License as published by
  the Free Software Foundation; version 2 of the License.

  This program is distributed in the hope that it will be useful,
  but WITHOUT ANY WARRANTY; without even the implied warranty of
  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  GNU General Public License for more details.
*/

#if  defined  (HAVE_CONFIG_H)
#  include "config.h"
#endif

#if defined (_OPENMP)
#  include <omp.h>
#endif

#include <stdio.h>
#include <string.h>
#include <stdarg.h>
#include <errno.h>
#include <sys/stat.h> /* stat */

FILE *popen(const char *command, const char *type);
int pclose(FILE *stream);

#include <cdi.h>
#include "cdo.h"
#include "cdo_int.h"
#include "dtypes.h"
#include "modules.h"
#include "pstream_int.h"
#include "cdo_int.h"
#include "util.h"
#include "pipe.h"
#include "error.h"
#include "dmemory.h"


extern int timer_read, timer_write;

static int PSTREAM_Debug = 0;

#define  MAX_PSTREAMS  4096

static int _pstream_max = MAX_PSTREAMS;

static void pstream_initialize(void);

static int _pstream_init = FALSE;

#if  defined  (HAVE_LIBPTHREAD)
#include <pthread.h>
#include "pthread_debug.h"

static int pthreadScope = 0;

static pthread_mutex_t streamOpenReadMutex  = PTHREAD_MUTEX_INITIALIZER;
static pthread_mutex_t streamOpenWriteMutex = PTHREAD_MUTEX_INITIALIZER;

static pthread_once_t _pstream_init_thread = PTHREAD_ONCE_INIT;
static pthread_mutex_t _pstream_mutex;

#  define PSTREAM_LOCK           pthread_mutex_lock(&_pstream_mutex);
#  define PSTREAM_UNLOCK         pthread_mutex_unlock(&_pstream_mutex);
#  define PSTREAM_INIT                               \
   if ( _pstream_init == FALSE ) pthread_once(&_pstream_init_thread, pstream_initialize);

#else

#  define PSTREAM_LOCK
#  define PSTREAM_UNLOCK
#  define PSTREAM_INIT                               \
   if ( _pstream_init == FALSE ) pstream_initialize();

#endif


typedef struct _pstreamPtrToIdx {
  int idx;
  pstream_t *ptr;
  struct _pstreamPtrToIdx *next;
} pstreamPtrToIdx;


static pstreamPtrToIdx *_pstreamList  = NULL;
static pstreamPtrToIdx *_pstreamAvail = NULL;


static
void pstream_list_new(void)
{
  assert(_pstreamList == NULL);

  _pstreamList = (pstreamPtrToIdx *) malloc(_pstream_max*sizeof(pstreamPtrToIdx));
}

static
void pstream_list_delete(void)
{
  if ( _pstreamList ) free(_pstreamList);
}

static
void pstream_init_pointer(void)
{
  int  i;
  
  for ( i = 0; i < _pstream_max; i++ )
    {
      _pstreamList[i].next = _pstreamList + i + 1;
      _pstreamList[i].idx  = i;
      _pstreamList[i].ptr  = 0;
    }

  _pstreamList[_pstream_max-1].next = 0;

  _pstreamAvail = _pstreamList;
}

static
pstream_t *pstream_to_pointer(int idx)
{
  pstream_t *pstreamptr = NULL;

  PSTREAM_INIT

  if ( idx >= 0 && idx < _pstream_max )
    {
      PSTREAM_LOCK

      pstreamptr = _pstreamList[idx].ptr;

      PSTREAM_UNLOCK
    }
  else
    Error("pstream index %d undefined!", idx);

  return (pstreamptr);
}

/* Create an index from a pointer */
static
int pstream_from_pointer(pstream_t *ptr)
{
  int      idx = -1;
  pstreamPtrToIdx *newptr;

  if ( ptr )
    {
      PSTREAM_LOCK

      if ( _pstreamAvail )
	{
	  newptr        = _pstreamAvail;
	  _pstreamAvail = _pstreamAvail->next;
	  newptr->next  = 0;
	  idx	        = newptr->idx;
	  newptr->ptr   = ptr;
      
	  if ( PSTREAM_Debug )
	    Message("Pointer %p has idx %d from pstream list", ptr, idx);
	}
      else
	Warning("Too many open pstreams (limit is %d)!", _pstream_max);

      PSTREAM_UNLOCK
    }
  else
    Error("Internal problem (pointer %p undefined)", ptr);

  return (idx);
}

static
void pstream_init_entry(pstream_t *pstreamptr)
{
  pstreamptr->self       = pstream_from_pointer(pstreamptr);

  pstreamptr->isopen     = TRUE;
  pstreamptr->ispipe     = FALSE;
  pstreamptr->fileID     = -1;
  pstreamptr->vlistID    = -1;
  pstreamptr->tsID       = -1;
  pstreamptr->filetype   = -1;
  pstreamptr->name       = NULL;
  pstreamptr->tsID0      = 0;
  pstreamptr->mfiles     = 0;
  pstreamptr->nfiles     = 0;
  pstreamptr->varID      = -1;
  pstreamptr->name       = NULL;
  pstreamptr->mfnames    = NULL;
  pstreamptr->varlist    = NULL;
#if  defined  (HAVE_LIBPTHREAD)
  pstreamptr->pipe       = NULL;
  //  pstreamptr->rthreadID  = 0;
  //  pstreamptr->wthreadID  = 0;
#endif
}

static
pstream_t *pstream_new_entry(void)
{
  pstream_t *pstreamptr;

  pstreamptr = (pstream_t *) malloc(sizeof(pstream_t));

  if ( pstreamptr ) pstream_init_entry(pstreamptr);

  return (pstreamptr);
}

static
void pstream_delete_entry(pstream_t *pstreamptr)
{
  int idx;

  idx = pstreamptr->self;

  PSTREAM_LOCK

  free(pstreamptr);

  _pstreamList[idx].next = _pstreamAvail;
  _pstreamList[idx].ptr  = 0;
  _pstreamAvail   	 = &_pstreamList[idx];

  PSTREAM_UNLOCK

  if ( PSTREAM_Debug )
    Message("Removed idx %d from pstream list", idx);
}

static
void pstream_initialize(void)
{
  char *env;

#if  defined  (HAVE_LIBPTHREAD)
  /* initialize global API mutex lock */
  pthread_mutex_init(&_pstream_mutex, NULL);
#endif

  env = getenv("PSTREAM_DEBUG");
  if ( env ) PSTREAM_Debug = atoi(env);

  env = getenv("PSTREAM_MAX");
  if ( env ) _pstream_max = atoi(env);

  if ( PSTREAM_Debug )
    Message("PSTREAM_MAX = %d", _pstream_max);

  pstream_list_new();
  atexit(pstream_list_delete);

  pstream_init_pointer();

  _pstream_init = TRUE;
}

static
int pstreamFindID(const char *name)
{
  int pstreamID;
  pstream_t *pstreamptr;

  for ( pstreamID = 0; pstreamID < _pstream_max; pstreamID++ )
    {
      pstreamptr = pstream_to_pointer(pstreamID);

      if ( pstreamptr )
	if ( pstreamptr->name )
	  if ( strcmp(pstreamptr->name, name) == 0 ) break;
    }

  if ( pstreamID == _pstream_max ) pstreamID = -1;

  return (pstreamID);
}


int pstreamIsPipe(int pstreamID)
{
  pstream_t *pstreamptr;

  pstreamptr = pstream_to_pointer(pstreamID);

  return (pstreamptr->ispipe);
}


int pstreamOpenRead(const char *argument)
{
  char *operatorArg = NULL;
  char *operatorName = NULL;
  int ispipe = FALSE;
  int fileID;
  int pstreamID;
  pstream_t *pstreamptr;

  PSTREAM_INIT

  pstreamptr = pstream_new_entry();
  if ( ! pstreamptr ) Error("No memory");

  pstreamID = pstreamptr->self;

  ispipe = argument[0] == '-';

  if ( ispipe )
    {
#if  defined  (HAVE_LIBPTHREAD)
      char *newarg;
      char *pipename = (char *) malloc(16);
      int rval;
      pthread_t thrID;
      pthread_attr_t attr;
      struct sched_param param;
      size_t len;
      size_t stacksize;
      int status;

      operatorArg = getOperator(argument);
      operatorName = getOperatorName(operatorArg);
      free(operatorArg);

      len = strlen(argument);
      newarg = (char *) malloc(len+16);
      strcpy(newarg, argument);
      sprintf(pipename, "(pipe%d.%d)", processSelf() + 1, processInqChildNum() + 1);
      newarg[len] = ' ';
      strcpy(&newarg[len+1], pipename);

      pstreamptr->ispipe = TRUE;
      pstreamptr->name   = pipename;
      pstreamptr->rthreadID = pthread_self();
      pstreamptr->pipe   = pipeNew();
 
      if ( ! cdoSilentMode )
	fprintf(stderr, "%s: Started child process \"%s\".\n", processInqPrompt(), newarg+1);

      status = pthread_attr_init(&attr);
      if ( status ) SysError("pthread_attr_init failed for '%s'\n", newarg+1);
      status = pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);
      if ( status ) SysError("pthread_attr_setdetachstate failed for '%s'\n", newarg+1);
      /*
      param.sched_priority = 0;
      status = pthread_attr_setschedparam(&attr, &param);
      if ( status ) SysError("pthread_attr_setschedparam failed for '%s'\n", newarg+1);
      */
      /* status = pthread_attr_setinheritsched(&attr, PTHREAD_EXPLICIT_SCHED); */
      /* if ( status ) SysError("pthread_attr_setinheritsched failed for '%s'\n", newarg+1); */

      pthread_attr_getscope(&attr, &pthreadScope);

      /* status = pthread_attr_setscope(&attr, PTHREAD_SCOPE_PROCESS); */
      /* if ( status ) SysError("pthread_attr_setscope failed for '%s'\n", newarg+1); */
      /* If system scheduling scope is specified, then the thread is scheduled against all threads in the system */
      /* pthread_attr_setscope(&attr, PTHREAD_SCOPE_SYSTEM); */

      status = pthread_attr_getstacksize(&attr, &stacksize);
      if ( stacksize < 2097152 )
	{
	  stacksize = 2097152;
	  pthread_attr_setstacksize(&attr, stacksize);
	}
      rval = pthread_create(&thrID, &attr, operatorModule(operatorName), newarg);
      if ( rval != 0 )
	{
	  errno = rval;
	  SysError("pthread_create failed for '%s'\n", newarg+1);
	}

      /* free(operatorName); */
      processAddStream(pstreamID);
      /*      pipeInqInfo(pstreamID); */
      if ( PSTREAM_Debug ) Message("pipe %s", pipename);
#else
      cdoAbort("Cannot use pipes, pthread support not compiled in!");
#endif
    }
  else
    {
      extern int cdoDefaultFileType/*, cdoDefaultInstID*/;
      size_t len, i;
      int nfiles = 1, j;
      char *filename = NULL;
      const char *pch;

      len = strlen(argument);

      for ( i = 0; i < len; i++ )
	if ( argument[i] == ':' ) break;

      if ( i < len )
	{
	  pch = &argument[i+1];
	  len -= (i+1);
	  if ( len && ( memcmp(argument, "filelist:", 9) == 0 || 
			memcmp(argument, "flist:", 6) == 0 ) )
	    {
	      for ( i = 0; i < len; i++ ) if ( pch[i] == ',' ) nfiles++;

	      if ( nfiles == 1 )
		{
		  char line[4096];
		  FILE *fp, *fp2;
		  fp = fopen(pch, "r");
		  if ( fp == NULL ) cdoAbort("Open failed on %s", pch);

		  if ( cdoVerbose )
		    cdoPrint("Reading file names from %s", pch);

		  /* find number of files */
		  nfiles = 0;
		  while ( readline(fp, line, 4096) )
		    {
		      if ( line[0] == '#' || line[0] == '\0' ||
			   line[0] == ' ' ) continue;

		      fp2 = fopen(line, "r" );
		      if ( fp2 == NULL ) cdoAbort("Open failed on %s", line);
		      fclose(fp2);
		      nfiles++;
		      if ( cdoVerbose )
			cdoPrint("File number %d is %s", nfiles, line);
		    }

		  if ( nfiles == 0 ) cdoAbort("No imput file found in %s", pch);

		  pstreamptr->mfiles = nfiles;
		  pstreamptr->mfnames = (char **) malloc(nfiles*sizeof(char *));
		  
		  rewind(fp);

		  nfiles = 0;
		  while ( readline(fp, line, 4096) )
		    {
		      if ( line[0] == '#' || line[0] == '\0' ||
			   line[0] == ' ' ) continue;

		      pstreamptr->mfnames[nfiles] = strdupx(line);
		      nfiles++;
		    }

		  fclose(fp);
		}
	      else
		{
		  char line[65536];

		  pstreamptr->mfiles = nfiles;
		  pstreamptr->mfnames = (char **) malloc(nfiles*sizeof(char *));
		  
		  strcpy(line, pch);
		  for ( i = 0; i < len; i++ ) if ( line[i] == ',' ) line[i] = 0;

		  i = 0;
		  for ( j = 0; j < nfiles; j++ )
		    {
		      pstreamptr->mfnames[j] = strdupx(&line[i]);
		      i += strlen(&line[i]) + 1;
		    }
		}
	    }
	  else if ( len && memcmp(argument, "ls:", 3) == 0 )
	    {
	      char line[4096];
	      char command[4096];
	      char *fnames[16384];
	      FILE *pfp;

	      strcpy(command, "ls ");
	      strcat(command, pch);

	      pfp = popen(command, "r");
	      if ( pfp == 0 )
		SysError("popen %s failed", command);

	      nfiles = 0;
	      while ( readline(pfp, line, 4096) )
		{
		  if ( nfiles >= 16384 ) cdoAbort("Too many input files (limit: 16384)");
		  fnames[nfiles++] = strdupx(line);
		}

	      pclose(pfp);

	      pstreamptr->mfiles = nfiles;
	      pstreamptr->mfnames = (char **) malloc(nfiles*sizeof(char *));

	      for ( j = 0; j < nfiles; j++ )
		pstreamptr->mfnames[j] = fnames[j];
	    }
	}

      if ( pstreamptr->mfiles )
	{
	  len = strlen(pstreamptr->mfnames[0]);
	  filename = (char *) malloc(len+1);
	  strcpy(filename, pstreamptr->mfnames[0]);
	  pstreamptr->nfiles = 1;
	}
      else
	{
	  len = strlen(argument);
	  /*
	  if ( cdoExpMode == CDO_EXP_REMOTE )
	    {
	      char datapath[] = "/scratch/localA/m214003/data/";
	      len += strlen(datapath);

	      filename = (char *) malloc(len+1);

	      strcpy(filename, datapath);
	      strcat(filename, argument);
	    }
	  else
	  */
	    {
	      filename = (char *) malloc(len+1);

	      strcpy(filename, argument);
	    }
	}

      if ( PSTREAM_Debug ) Message("file %s", filename);

#if  defined  (HAVE_LIBPTHREAD)
      pthread_mutex_lock(&streamOpenReadMutex);
#endif
      fileID = streamOpenRead(filename);
      if ( fileID < 0 ) cdiError(fileID, "Open failed on >%s<", filename);

      if ( cdoDefaultFileType == CDI_UNDEFID )
	cdoDefaultFileType = streamInqFiletype(fileID);
      /*
      if ( cdoDefaultInstID == CDI_UNDEFID )
	cdoDefaultInstID = streamInqInstID(fileID);
      */
      cdoInqHistory(fileID);
#if  defined  (HAVE_LIBPTHREAD)
      pthread_mutex_unlock(&streamOpenReadMutex);
#endif

      pstreamptr->mode   = 'r';
      pstreamptr->name   = filename;
      pstreamptr->fileID = fileID;
    }

  if ( pstreamID < 0 ) cdiError(pstreamID, "Open failed on %s", argument);
  
  return (pstreamID);
}

static
void query_user_exit(const char *argument)
{
  /* modified code from NCO */
#define USR_RPL_MAX_LNG 10 /* Maximum length for user reply */
#define USR_RPL_MAX_NBR 10 /* Maximum number of chances for user to reply */
  char usr_rpl[USR_RPL_MAX_LNG];
  int usr_rpl_int;
  short nbr_itr=0;
  size_t usr_rpl_lng = 0;

  /* Initialize user reply string */
  usr_rpl[0]='z';
  usr_rpl[1]='\0';

  while ( !(usr_rpl_lng == 1 && 
	    (*usr_rpl == 'o' || *usr_rpl == 'O' || *usr_rpl == 'e' || *usr_rpl == 'E')) )
    {
      if ( nbr_itr++ > USR_RPL_MAX_NBR )
	{
	  (void)fprintf(stdout,"\n%s: ERROR %d failed attempts to obtain valid interactive input.\n",
			processInqPrompt(), nbr_itr-1);
	  exit(EXIT_FAILURE);
	}

      if ( nbr_itr > 1 ) (void)fprintf(stdout,"%s: ERROR Invalid response.\n", processInqPrompt());
      (void)fprintf(stdout,"%s: %s exists ---`e'xit, or `o'verwrite (delete existing file) (e/o)? ",
		    processInqPrompt(), argument);
      (void)fflush(stdout);
      if ( fgets(usr_rpl, USR_RPL_MAX_LNG, stdin) == NULL ) continue;

      /* Ensure last character in input string is \n and replace that with \0 */
      usr_rpl_lng = strlen(usr_rpl);
      if ( usr_rpl_lng >= 1 )
	if ( usr_rpl[usr_rpl_lng-1] == '\n' )
	  {
	    usr_rpl[usr_rpl_lng-1] = '\0';
	    usr_rpl_lng--;
	  }
    }

  /* Ensure one case statement for each exit condition in preceding while loop */
  usr_rpl_int=(int)usr_rpl[0];
  switch(usr_rpl_int)
    {
    case 'E':
    case 'e':
      exit(EXIT_SUCCESS);
      break;
    case 'O':
    case 'o':
      break;
    default:
      exit(EXIT_FAILURE);
      break;
    } /* end switch */
}


int pstreamOpenWrite(const char *argument, int filetype)
{
  int fileID;
  int pstreamID = -1;
  int ispipe;
  pstream_t *pstreamptr;

  PSTREAM_INIT

  ispipe = memcmp(argument, "(pipe", 5) == 0;

  if ( ispipe )
    {
#if  defined  (HAVE_LIBPTHREAD)
      if ( PSTREAM_Debug ) Message("pipe %s", argument);
      pstreamID = pstreamFindID(argument);
      if ( pstreamID == -1 )
	Error("%s not open", argument);

      pstreamptr = pstream_to_pointer(pstreamID);

      pstreamptr->wthreadID = pthread_self();
      pstreamptr->filetype = filetype;
      processAddStream(pstreamID);
#endif
    }
  else
    {
      /* extern int cdoDefaultInstID; */
      char *filename = (char *) malloc(strlen(argument)+1);

      pstreamptr = pstream_new_entry();
      if ( ! pstreamptr ) Error("No memory");

      pstreamID = pstreamptr->self;
  
      if ( PSTREAM_Debug ) Message("file %s", argument);

      if ( filetype == CDI_UNDEFID ) filetype = FILETYPE_GRB;

      if ( cdoInteractive )
	{
	  int rstatus;
	  struct stat stbuf;

	  rstatus = stat(argument, &stbuf);
	  /* If permanent file already exists, query user whether to overwrite or exit */
	  if ( rstatus != -1 ) query_user_exit(argument);
	}

#if  defined  (HAVE_LIBPTHREAD)
      pthread_mutex_lock(&streamOpenWriteMutex);
#endif
      timer_start(timer_write);
      fileID = streamOpenWrite(argument, filetype);
      timer_stop(timer_write);
#if  defined  (HAVE_LIBPTHREAD)
      pthread_mutex_unlock(&streamOpenWriteMutex);
#endif
      if ( fileID < 0 ) cdiError(fileID, "Open failed on %s", argument);

      cdoDefHistory(fileID, commandLine());

      if ( cdoDefaultByteorder != CDI_UNDEFID )
	streamDefByteorder(fileID, cdoDefaultByteorder);

      if ( cdoCompress )
	{
	  if      ( filetype == FILETYPE_GRB )
	    {
	      cdoCompType  = COMPRESS_SZIP;
	      cdoCompLevel = 0;
	    }
	  else if ( filetype == FILETYPE_NC4 || filetype == FILETYPE_NC4C )
	    {
	      cdoCompType  = COMPRESS_ZIP;
	      cdoCompLevel = 1;
	    }
	}

      if ( cdoCompType != COMPRESS_NONE )
	{
	  streamDefCompType(fileID, cdoCompType);
	  streamDefCompLevel(fileID, cdoCompLevel);

	  if ( cdoCompType == COMPRESS_SZIP && (filetype != FILETYPE_GRB && filetype != FILETYPE_NC4 && filetype != FILETYPE_NC4C) )
	    cdoWarning("SZIP compression not available for non GRIB1/netCDF4 data!");

	  if ( cdoCompType == COMPRESS_JPEG && filetype != FILETYPE_GRB2 )
	    cdoWarning("SZIP compression not available for non GRIB2 data!");

	  if ( cdoCompType == COMPRESS_ZIP && (filetype != FILETYPE_NC4 && filetype != FILETYPE_NC4C) )
	    cdoWarning("Deflate compression not available for non netCDF4 data!");
	}
      /*
      if ( cdoDefaultInstID != CDI_UNDEFID )
	streamDefInstID(fileID, cdoDefaultInstID);
      */
      strcpy(filename, argument);

      pstreamptr->mode     = 'w';
      pstreamptr->name     = filename;
      pstreamptr->fileID   = fileID;
      pstreamptr->filetype = filetype;
   }

  return (pstreamID);
}


int pstreamOpenAppend(const char *argument)
{
  int fileID;
  int pstreamID = -1;
  int ispipe;
  pstream_t *pstreamptr;

  ispipe = memcmp(argument, "(pipe", 5) == 0;

  if ( ispipe )
    {
      if ( PSTREAM_Debug ) Message("pipe %s", argument);
      cdoAbort("this operator doesn't work with pipes!");
    }
  else
    {
      char *filename = (char *) malloc(strlen(argument)+1);

      pstreamptr = pstream_new_entry();
      if ( ! pstreamptr ) Error("No memory");

      pstreamID = pstreamptr->self;
  
      if ( PSTREAM_Debug ) Message("file %s", argument);

      timer_start(timer_write);
      fileID = streamOpenAppend(argument);
      timer_stop(timer_write);
      if ( fileID < 0 ) cdiError(fileID, "Open failed on %s", argument);
      /*
      cdoInqHistory(fileID);
      cdoDefHistory(fileID, commandLine());
      */
      strcpy(filename, argument);

      pstreamptr->mode   = 'a';
      pstreamptr->name   = filename;
      pstreamptr->fileID = fileID;
    }

  return (pstreamID);
}


void pstreamClose(int pstreamID)
{
  pstream_t *pstreamptr;

  pstreamptr = pstream_to_pointer(pstreamID);

  if ( pstreamptr == NULL )
    Error("Internal problem, stream %d not open!", pstreamID);

  if ( pstreamptr->ispipe )
    {
#if  defined  (HAVE_LIBPTHREAD)
      pipe_t *pipe;
      int lread = FALSE, lwrite = FALSE;
      pthread_t threadID = pthread_self();

      if      ( pthread_equal(threadID, pstreamptr->rthreadID) ) lread  = TRUE;
      else if ( pthread_equal(threadID, pstreamptr->wthreadID) ) lwrite = TRUE;
      else Error("Internal problem! Close pipe %s", pstreamptr->name);

      if ( lread )
	{
	  pipe = pstreamptr->pipe;
	  pthread_mutex_lock(pipe->mutex);
	  pipe->EOP = TRUE;
	  if ( PSTREAM_Debug )
	    Message("%s read closed", pstreamptr->name);
	  pthread_mutex_unlock(pipe->mutex);     
	  pthread_cond_signal(pipe->tsDef);
	  pthread_cond_signal(pipe->tsInq);
	 
	  pthread_cond_signal(pipe->recInq);
	 
	  pthread_mutex_lock(pipe->mutex);
	  pstreamptr->isopen = FALSE;
	  pthread_mutex_unlock(pipe->mutex);     
	  pthread_cond_signal(pipe->isclosed);

	  pthread_join(pstreamptr->wthreadID, NULL);

	  processAddNvals(pipe->nvals);
	  pipeDelete(pipe);
	  pstream_delete_entry(pstreamptr);
	}
      else
	{
	  pipe = pstreamptr->pipe;
	  pthread_mutex_lock(pipe->mutex);
	  pipe->EOP = TRUE;
	  if ( PSTREAM_Debug )
	    Message("%s write closed", pstreamptr->name);
	  pthread_mutex_unlock(pipe->mutex);     
	  pthread_cond_signal(pipe->tsDef);
	  pthread_cond_signal(pipe->tsInq);

	  pthread_mutex_lock(pipe->mutex);
	  while ( pstreamptr->isopen )
	    {
	      if ( PSTREAM_Debug )
		Message("wait of read close");
	      pthread_cond_wait(pipe->isclosed, pipe->mutex);
	    }
	  pthread_mutex_unlock(pipe->mutex);
	}
#else
      cdoAbort("Cannot use pipes, pthread support not compiled in!");
#endif
    }
  else
    {
      if ( PSTREAM_Debug )
	Message("%s fileID %d\n", pstreamptr->name, pstreamptr->fileID);

      if ( pstreamptr->mode == 'r' )
	{
	  processAddNvals(streamNvals(pstreamptr->fileID));
	}

      streamClose(pstreamptr->fileID);

      if ( cdoExpMode == CDO_EXP_REMOTE )
	{
	  if ( pstreamptr->mode == 'w' )
	    {
	      extern const char *cdojobfiles;
	      FILE *fp = fopen(cdojobfiles, "a");
	      fprintf(fp, "%s\n", pstreamptr->name);
	      fclose(fp);
	    }
	}

      if ( pstreamptr->name )
	{
	  free(pstreamptr->name);
	  pstreamptr->name = NULL;
	}

      if ( pstreamptr->varlist )
	{
	  free(pstreamptr->varlist);
	  pstreamptr->varlist = NULL;
	}

      pstream_delete_entry(pstreamptr);
    }
}


int pstreamInqVlist(int pstreamID)
{
  int vlistID;
  pstream_t *pstreamptr;

  pstreamptr = pstream_to_pointer(pstreamID);

#if  defined  (HAVE_LIBPTHREAD)
  if ( pstreamptr->ispipe )
    {
      vlistID = pipeInqVlist(pstreamptr);
      if ( vlistID == -1 )
	cdoAbort("Couldn't read data from input stream %s!", pstreamptr->name);
    }
  else
#endif
    {
      extern int cdoDefaultTimeType;

      timer_start(timer_read);
      vlistID = streamInqVlist(pstreamptr->fileID);
      timer_stop(timer_read);

      if ( cdoDefaultTimeType != CDI_UNDEFID )
	taxisDefType(vlistInqTaxis(vlistID), cdoDefaultTimeType);

      pstreamptr->vlistID = vlistID;
    }

  if ( vlistNumber(vlistID) == CDI_COMP && cdoStreamNumber() == CDI_REAL )
    cdoAbort("Complex fields are not supported by this operator!");

  if ( vlistNumber(vlistID) == CDI_REAL && cdoStreamNumber() == CDI_COMP )
    cdoAbort("This operator needs complex fields!");

  processDefVarNum(vlistNvars(vlistID), pstreamID);

  return (vlistID);
}


const char *cdoComment(void)
{
  static char comment[256];
  static int init = 0;
  int size = 0;
  extern char CDO_Version[];

  if ( ! init )
    {
      init = 1;

      size = strlen(CDO_Version);

      strncat(comment, CDO_Version, size);
    }

  return (comment);
}

static
void pstreamDefVarlist(pstream_t *pstreamptr, int vlistID)
{
  int varID, nvars;
  int laddoffset, lscalefactor;
  int datatype, filetype;
  varlist_t *varlist;

  filetype = pstreamptr->filetype;

  if ( pstreamptr->vlistID != -1 )
    cdoAbort("Internal problem, vlist already defined!");

  if ( pstreamptr->varlist != NULL )
    cdoAbort("Internal problem, varlist already allocated!");

  nvars = vlistNvars(vlistID);
  varlist = (varlist_t *) malloc(nvars*sizeof(varlist_t));

  for ( varID = 0; varID < nvars; ++varID )
    {
      varlist[varID].gridsize    = gridInqSize(vlistInqVarGrid(vlistID, varID));
      varlist[varID].datatype    = vlistInqVarDatatype(vlistID, varID);
      varlist[varID].missval     = vlistInqVarMissval(vlistID, varID);
      varlist[varID].addoffset   = vlistInqVarAddoffset(vlistID, varID);
      varlist[varID].scalefactor = vlistInqVarScalefactor(vlistID, varID);

      varlist[varID].check_datarange = FALSE;

      laddoffset   = IS_NOT_EQUAL(varlist[varID].addoffset, 0);
      lscalefactor = IS_NOT_EQUAL(varlist[varID].scalefactor, 1);

      datatype = varlist[varID].datatype;

      if ( filetype == FILETYPE_NC || filetype == FILETYPE_NC2 || filetype == FILETYPE_NC4 || filetype == FILETYPE_NC4C )
	{
	  if ( datatype == DATATYPE_UINT8 && (filetype == FILETYPE_NC || filetype == FILETYPE_NC2) )
	    {
	      datatype = DATATYPE_INT16;
	      varlist[varID].datatype = datatype;
	    }

	  if ( datatype == DATATYPE_UINT16 && (filetype == FILETYPE_NC || filetype == FILETYPE_NC2) )
	    {
	      datatype = DATATYPE_INT32;
	      varlist[varID].datatype = datatype;
	    }

	  if ( laddoffset || lscalefactor )
	    {
	      if ( datatype == DATATYPE_INT8   ||
		   datatype == DATATYPE_UINT8  ||
		   datatype == DATATYPE_INT16  ||
		   datatype == DATATYPE_UINT16 )
		varlist[varID].check_datarange = TRUE;
	    }