How to create processes and use them with fork() and pipe()

[dhayden]

Here's an implementation of what I wrote in my earlier post. Note that I had to include my own version of strdup because I'm using cygwin and g++ -std=c++14, and that combo doesn't have strdup.

Also, because it was getting late, I didn't deal with commas in the input file (the one that gives the initial values of input vars).

As with Duthomhas, the biggest hassle was simply parsing the input file.

#include<iostream>
#include <map>
#include <string>
#include <vector>
#include <unistd.h>
#include <cstring>
#include <fstream>

using std::string;
using std::map;
using std::cout;
using std::cin;
using std::vector;
using std::cerr;
using std::ifstream;

// Data for the "input_var" processes
struct InProc {
    int value = 0;
    vector<int> outputs;
    int run();
};


// Data for the "internal_val" procs. I called them CalcProcs because
// they calculate.  Since they have a value, and output file
// descriptors, and when the time comes, they need to write to all
// output file descriptors, I derived this from InProc.
struct CalcProc: public InProc {
    vector<int> inputs;    // FDs to read from
    
    // The commands to perform: legal values are =,+,-,*,/.
    // After reading each input, it processes one command
    string commands;
    int run();
};

// The main proc. This handles the "write" line.  It has a vector of
// FDs to read an the corresponding process names.
struct MainProc {
    vector<int> inputs;		// input FDs
    vector<string> names;	// the names if the procs to write;
    int run();
};


// I need this on cygwin
char *strdup(const char *src)
{
    char *result = (char *)malloc(strlen(src)+1);
    if (result == nullptr) return nullptr;
    strcpy(result, src);
    return result;
}


// An input proc simply writes its value to all output file descriptors.
// I close the fd after writing to ensure it gets flused to the reader.
int InProc::run()
{
    int result = 0;
    for (int &fd : outputs) {
	result = write(fd, &value, sizeof(value));
	if (result != sizeof(value)) break;
	close(fd);
    }
    return result;
}

// For each input file descriptor, read the value and do one operation
// from the command string. When you're done, write your value to the
// output file descriptors
int CalcProc::run()
{
    int result=0;
    
    size_t idx = 0;
    for (int &fd : inputs) {
	int arg;
	result = read(fd, &arg, sizeof(arg));
	if (result != sizeof(arg)) {
	    cerr << "CalcProc::run(): read returned " << result << '\n';
	    return 1;
	}
	close(fd);

	switch(commands[idx++]) {
	case '=':
	    value = arg;
	    break;
	case '+':
	    value += arg;
	    break;
	case '-':
	    value -= arg;
	    break;
	case '*':
	    value *= arg;
	    break;
	case '/':
	    value /= arg;
	    break;
	default:
	    cerr << "Illegal command " << commands[idx-1] << '\n';
	    return 1;
	}
    }

    // Now write the value to the output FDs. Conveniently,
    // the Inproc::run() function does just that;
    return InProc::run();
}


// The Main process writes the names and values to cout
int MainProc::run()
{
    int value;
    int result = 0;
    for (size_t i = 0; i<inputs.size(); ++i) {
	result = read(inputs[i], &value, sizeof(value));
	if (result != sizeof(value)) break;
	cout << names[i] << " = " << value << '\n';
	close(inputs[i]);
    }
    return result;
}


int
main()
{
    // Data for the inProcs, calcProcs, and main proc.
    map<string, InProc> inProcs; // map name to proc data
    map<string, CalcProc> calcProcs; // map name to proc data
    MainProc mainProc;
    
    string line;
    char *token;
    
    // The config file contains all the config details
    // cin points to the initial values for the input procs.
    ifstream infile("config.txt");
    
    //
    // Read the input_var line and create the input proc data
    // 
    getline(infile, line);
    if (line.size() == 0 || line[line.size()-1] != ';' ||
	line.find("input_var") != 0) {
	cerr << "input_var line is bad\n";
    }

    char *cp = strdup(line.c_str());
    token = strtok(cp, " ,;"); 	// skip "input_var"
    while ((token = strtok(nullptr, " ,;"))) {
	// Create the process in the map and read into its value
	cin >> inProcs[token].value;
    }
    free(cp);
    
    // 
    // Read the internal_var line and create the calc proc data.
    //
    getline(infile, line);
    if (line.size() == 0 || line[line.size()-1] != ';' ||
	line.find("internal_var") != 0) {
	cerr << "internal_var line is bad\n";
    }

    cp = strdup(line.c_str());
    token = strtok(cp, " ,;"); 	// skip "internal_var"
    while ((token = strtok(nullptr, " ,;"))) {
	calcProcs[token];	// create the process data
    }
    free(cp);
    
    //
    // Read the processing lines that tell who writes to whom.
    // 
    while (true) {
	getline(infile, line);
	if (!infile) {
	    cerr << "No write line\n";
	    return 1;
	}
	if (line.find("write") == 0) {
	    break;
	}
	
	cp = strdup(line.c_str());

	// Get the first token
	token = strtok(cp, " ;");
	char cmd;
	if (strchr("+-*/", token[0])) {
	    // it's the optional command. Remember the command
	    // and get the next input proc name
	    cmd = token[0];
	    token = strtok(nullptr, " ;");
	} else {
	    // It's the input proc name. The command is "=" to set the value
	    cmd = '=';
	}

	// The writing process can be an input process or a calc
	// process. Get a pointer to the appropriate one.
	// Now I'm really glad that CalcProc derives from InProc
	InProc *writer;
	auto iter = inProcs.find(token);
	if (iter != inProcs.end()) {
	    writer = &iter->second;
	} else {
	    writer = &calcProcs[token];
	}

	// Skip "->"
	token = strtok(nullptr, " ;");
	if (strcmp(token, "->") != 0) {
	    cerr << "Bad line: " << line << '\n';
	    return 1;
	}
	
	// Get the reading process
	token = strtok(nullptr, " ;");
	CalcProc &reader (calcProcs[token]);

	// Create the pipe
	int fildes[2];
	pipe(fildes);
	
	// Now we have all the info needed. Tell the writing process
	// to write to the pipe. Tell the reading process to read from
	// the pipe and perform the command;
	writer->outputs.push_back(fildes[1]);
	reader.inputs.push_back(fildes[0]);
	reader.commands.push_back(cmd);

	free(cp);
    }

    // Now you have the "write" line
    cp = strdup(line.c_str());
    token = strtok(cp, "(,;");	// skip "write"

    // For each proc name on the "write" line, tell the named proc
    // to write it's value. Tell the main proc to read the value
    // and print it out
    while ((token = strtok(nullptr, ",;)"))) {
	int fildes[2];
	pipe(fildes);
	mainProc.inputs.push_back(fildes[0]);
	mainProc.names.push_back(token);

	// The writing process could be a calcProc or an input proc.
	auto iter = inProcs.find(token);
	InProc *writer;
	if (iter != inProcs.end()) {
	    writer = &iter->second;
	} else {
	    writer = &calcProcs[token];
	}
	writer->outputs.push_back(fildes[1]);
    }
    free(cp);


    // Now let's kick this stuff off.
    for (auto &iter : inProcs) {
	if (fork() == 0) {
	    return iter.second.run();
	}
    }
    for (auto &iter : calcProcs) {
	if (fork() == 0) {
	    return iter.second.run();
	}
    }
    mainProc.run();
    return 0;

}

[Duthomhas]

Non-language-stsndard functions disappear whenever you declare a language standard on the command line.

Strdup() is, however, POSIX standard, so it really is there. Hey it back by defining the following in your code before including any headers:

#define _POSIX_C_SOURCE 200809L

My solution also uses it. I'll post the code later when I get home.

[Duthomhas]

Sorry to be so late. Here, as promised, is my solution. I added a little bonus so that you can watch the processes order themselves (assuming stdout prints in more-or-less the same order as the process mutex gets it).

It is longer that it strictly needs to be, but like I said, I like really obvious utility functions.

Part 1/3

// From forum question about a homework assignment
// http://www.cplusplus.com/forum/general/242972
//
// The assignment is given in the PDF at
// http://www2.cs.uh.edu/~acheng/hw1.f18.pdf
//
// This is -a- possible solution, knowing nothing more than what is given in that PDF and
// additional commentary by the student.
//
// The assignment's goal appears to be to create a static dependency tree (directed acyclic graph)
// by parsing the unfriendly argv[0] input and then instantiating it as subprocesses (nodes) and
// pipes (edges) and reporting the resulting calculations on stdout.
//
// The stated learning objective(s) is to understand how concurrent processes synchronize
// using pipes.
//
// IM(NS)HO, however, it failed that. The effective learning result is how to parse a dependency
// graph. There is no requirement to observe how I/O wait states affect the flow of data through
// the tree. If you wish to it here, change the following line to have a value of '1', compile, 
// and execute the binary a few times.

#define SHOW_DATA_FLOW 0


//-------------------------------------------------------------------------------------------------
// REQUIRED INCLUDES
//-------------------------------------------------------------------------------------------------

#define _POSIX_C_SOURCE 200809L
#include <ctype.h>
#include <iso646.h>
#include <math.h>
#include <stdarg.h>
#include <stdbool.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>

#include <sys/types.h>
#include <unistd.h>


//-------------------------------------------------------------------------------------------------
// TOTALLY UNNECESSARY STUFF TO SHOW THE DATA FLOW
//-------------------------------------------------------------------------------------------------

#if SHOW_DATA_FLOW
  #define DECL               char ivars[ 10 ][ 10 ]; char ovars[ 10 ][ 10 ];
  #define SEND( me, x, to )  printf( "%-6s sends %12g     to      %s\n", me, x, to );
  #define GOT( me, x, from ) printf( "%-6s got   %12g     from    %s\n", me, x, from );
  #define DONE( me )         printf( "%-6s done\n", me );
  #define LINK( output, input, iname ) \
    strcpy( input->name, iname ); \
    strcpy( output->ovars[ output->noutputs ], input ->name ); \
    strcpy( input ->ivars[ input ->ninputs  ], output->name );
#else
  #define DECL
  #define SEND( me, x, to )
  #define GOT( me, x, from )
  #define DONE( me )
  #define LINK( output, input, iname )
#endif


//-------------------------------------------------------------------------------------------------
// USAGE AND ERROR REPORTING
//-------------------------------------------------------------------------------------------------

const char* basename( const char* path )
{
  const char* result = strchr( path, '\0' );
  while (result != path)
    if (strchr( "/\\", *--result))
      return result + 1;
  return result;
}


void usage( const char* argv0 )
{
  printf( "%s%s%s",
    "usage:\n"
    "  ", basename( argv0 ), " DATAGRAPHFILE < INPUTFILE\n"
    "\n"
    "See hw1.f18.pdf for details.\n\n"
  );
  exit( 0 );
}


void error( int line_number, const char* message, ... )
{
  va_list args;
  fprintf( stderr, "[%d] ", line_number );
  va_start( args, message );
  vfprintf( stderr, message, args );
  fprintf( stderr, "\n" );
  va_end( args );
  exit( 1 );
}
#define error( ... ) error( __LINE__, __VA_ARGS__ )


//-------------------------------------------------------------------------------------------------
// I/O UTILITIES
//-------------------------------------------------------------------------------------------------

bool readline( FILE* f, char* s, int n, char delim )
{
  int c, nread = 0;
  while (true)
  {
    c = fgetc( f );
    if ((c == EOF) or (c == delim)) break;
    *s++ = c;
    if (++nread == n) error( "input line too long" );
  }
  *s = '\0';
  return !!nread;
}


double read_double( FILE* f )
{
  double result;
  if (!fscanf( f, "%lf", &result )) result = NAN;
  return result;
}


int skip( FILE* f, const char* chars_to_skip )
{
  int c;
  while (true)
  {
    c = fgetc( f );
    if (c == EOF) error( "unexpected EOF" );
    if (strchr( chars_to_skip, c ) == NULL) break;
  }
  ungetc( c, f );
  return c;
}


//-------------------------------------------------------------------------------------------------
// PIPE WRAPPER
//-------------------------------------------------------------------------------------------------

typedef struct
{
  FILE* reader;
  FILE* writer;
}
Pipe;


Pipe create_pipe()
{
  int  fd[ 2 ];
  Pipe result = { NULL, NULL };

  if (pipe( fd ) == 0)
  {
    result.reader = fdopen( fd[ 0 ], "r" );
    result.writer = fdopen( fd[ 1 ], "w" );
  }
  else error( "pipe() failed!" );

  if (!result.reader or !result.writer)
    error( "could not create pipe" );

  return result;
}


//-------------------------------------------------------------------------------------------------
// VAR TYPE
//-------------------------------------------------------------------------------------------------

typedef enum                  {  input_var,   internal_var,   write_var } vartype;
const char* vartype_names[] = { "input_var", "internal_var", "write" };


typedef struct var
{
  char    name[ 10 ];                                                                              DECL
  vartype type;

  FILE*   inputs [ 10 ];  // read end of a pipe
  int     ninputs;

  FILE*   outputs[ 10 ];  // write end of a pipe
  int     noutputs;

  // input_var
  double  value;

  // internal_var
  char    mathops[ 10 ];
}
var;


var create_var( const char* name, vartype type )
{
  var result;
  strncpy( result.name, name, 9 );
  result.name[ 9 ] = '\0';
  result.type      = type;
  result.ninputs   = 0;
  result.noutputs  = 0;
  return result;
}


var* find_var( const char* name, var* vars, int nvars )
{
  while (nvars--)
    if (strcmp( vars[ nvars ].name, name ) == 0)
      return vars + nvars;
  return NULL;
}


//-------------------------------------------------------------------------------------------------
// CHILD PROCESSES
//-------------------------------------------------------------------------------------------------

int input_var_proc( var* v )
//
// An 'input_var' proces simply prints its stored value to all of its output pipes and terminates
//
{
  while (v->noutputs--)
  {                                                                                                SEND( v->name, v->value, v->ovars[ v->noutputs ] )
    fprintf( v->outputs[ v->noutputs ], "%g\n", v->value );  // '\n' is important!
  }                                                                                                DONE( v->name )
  return 0;
}


int internal_var_proc( var* v )
//
// An 'internal_var' process reads each input pipe, in order, updating value as indicated.
// Once done it outputs that value to all its output pipes, so we just reuse the 'input_var' proc.
//
{
  v->value = 0.0;

  for (int n = 0; n < v->ninputs; n++)
  {
    double value = read_double( v->inputs[ n ] );                                                  GOT( v->name, value, v->ivars[ n ] )

    switch (v->mathops[ n ])
    {
      case ' ': v->value  = value; break;
      case '+': v->value += value; break;
      case '-': v->value -= value; break;
      case '*': v->value *= value; break;
      case '/': v->value /= value; break;
      default:  v->value  = NAN;
    }
  }

  return input_var_proc( v );
}

[Duthomhas]

Part 2/3

//-------------------------------------------------------------------------------------------------
// FORK WRAPPER
//-------------------------------------------------------------------------------------------------
// This utility helps us when forking.
// The child process invokes the correct child process function and terminates with the resulting
// exit code. The parent process gets nothing.

void fork_var( var* v )
{
  int (*proc)( var* )
    = (v->type == input_var)    ? input_var_proc
    : (v->type == internal_var) ? internal_var_proc
    : NULL;
  if (!proc) error( "invalid vartype == %d", v->type );

  switch (fork())
  {
    case -1: error( "could not fork subprocess for variable '%s'", v->name );
    case  0: exit( proc( v ) );
  }
}


//-------------------------------------------------------------------------------------------------
// KNUTH-FISHER-YATES RANDOM SHUFFLE
//-------------------------------------------------------------------------------------------------
// https://en.wikipedia.org/wiki/Fisher%E2%80%93Yates_shuffle


void swap( var* a, var* b ) { var c = *a;  *a = *b;  *b = c; }


int random_less_than( int n )
//
// Return a random value in  0 .. n-1  without modulo bias.
//
{
  int x, N = RAND_MAX / n * n;
  do x = rand(); while (x >= N);
  return x % n;
}


void shuffle_vars( var* vars, int n )
{
  if (n) while (--n) swap( vars + n, vars + random_less_than( n ) );
}


//-------------------------------------------------------------------------------------------------
// DATAFLOW GRAPH PARSING
//-------------------------------------------------------------------------------------------------

const char* SEPARATORS = "\t\n\r ,().";


bool is_mathop( const char* s )
{
  return strchr( "+-*/", *s ) != NULL;
}


bool is_varname( const char* s )
{
  if (!*s) return false;
  while (*s)
    if (!isalnum( *s++ ))
      return false;
  return true;
}


void parse_varnames( char* s, var* vars, int* nvars, vartype type )
//
// This function implements handling 'input_var', 'internal_var', and 'write' statements,
// which are all essentually a list of variable names.
//
{
  int ntokens = 0;
  for (char* token = strtok( s, SEPARATORS ); token; token = strtok( NULL, SEPARATORS ))
  {
    if (!*token) continue;
    if (!ntokens++)
    {
      if (strcmp( token, vartype_names[ type ] ) != 0)
        error( "invalid DATAGRAPH: expected '%s'", vartype_names[ type ] );
    }
    else if (!is_varname( token ))
      error( "invalid DATAGRAPH: syntax error in '%s' statement", vartype_names[ type ] );
    else
    {
      var* v = find_var( token, vars, *nvars );
      var* w = vars + *nvars;
      *nvars += 1;
      *w = create_var( token, type );

      switch (type)
      {
        case input_var:
          if (!v)
          {
            skip( stdin, "\t\n\r ," );
            w->value = read_double( stdin );
            break;
          }

        case internal_var:
          if (v) error( "invalid DATAGRAPH: variable '%s' already exists", token );
          break;

        case write_var:
          if (!v) error( "invalid write statement: variable '%s' does not exist", token );
          {
            Pipe pipe = create_pipe();                                                             LINK( v, w, "main()" )
            v->outputs[ v->noutputs ] = pipe.writer;
            w->inputs [ w->ninputs  ] = pipe.reader;
            v->noutputs += 1;
            w->ninputs  += 1;
          }
          break;

        default:
          error( "internal error: parse_varnames() given invalid type" );
      }
    }
  }
}


bool parse_dependency( char* s, var* vars, int nvars )
//
// This function extracts one dependency from the dependency graph and updates the variables list.
//
// A dependency has the expected form:
//   op? varname0 '->' varname1
//
// This procedure immediately terminates with 'false' if the first token on the line is "write".
// Otherwise the result is 'true' to let the caller know we have not yet parsed "write".
//
{
  char  mathop   = ' ';
  char* varname0 = NULL;
  char* varname1 = NULL;
  int   ntokens  = 0;

  s = strdup( s );

  for (char* token = strtok( s, SEPARATORS ); token; token = strtok( NULL, SEPARATORS ))
  {
    if (!*token) continue;
    if (!ntokens++)
    {
      if (strcmp( token, "write" ) == 0)
        return false;
    }

    if (strcmp( token, "->" ) == 0) continue;

    if (is_mathop( token ))
      mathop = *token;

    else if (is_varname( token ))
    {
      if      (!varname0) varname0 = token;
      else if (!varname1) varname1 = token;
      else error( "invalid DATAGRAPH: syntax error" );
    }

    else error( "invalid DATAGRAPH: syntax error" );
  }

  if (!varname0 or !varname1)
    error( "invalid DATAGRAPH: syntax error" );

  var* v0 = find_var( varname0, vars, nvars );
  var* v1 = find_var( varname1, vars, nvars );

  if (!v0)
    error( "invalid DATAGRAPH: variable name not listed by input_var nor by internal_var" );

  if (v1->type != internal_var)
    error( "invalid DATAGRAPH: cannot use input_var as internal_var" );

  // Update the 'variable' data; connect v0's output to v1's input
  Pipe pipe = create_pipe();                                                                       LINK( v0, v1, v1->name )

  v0->outputs[ v0->noutputs ] = pipe.writer;
  v0->noutputs += 1;

  v1->inputs [ v1->ninputs  ] = pipe.reader;
  v1->mathops[ v1->ninputs  ] = mathop;
  v1->ninputs += 1;

  free( s );
  return true;
}

[Duthomhas]

Part 3/3

//-------------------------------------------------------------------------------------------------
// MAIN
//-------------------------------------------------------------------------------------------------

int main( int argc, char** argv )
{
  srand( (unsigned)time( NULL ) );

  // Modification: argv[1] may also be the NAME OF A FILE containing the "dataflow graph"

  if (argc != 2)                                                                usage( argv[ 0 ] );
  if ((strcmp( argv[ 1 ], "/?" ) == 0) or (strcmp( argv[ 1 ], "--help" ) == 0)) usage( argv[ 0 ] );
  FILE* f;
  if (access( argv[ 1 ], F_OK ) == 0)
  {
    f = fopen( argv[ 1 ], "r" );
    if (!f) error( "could not open DATAGRAPHFILE '%s'", argv[ 1 ] );
  }
  else
  {
    f = tmpfile();
    if (!f) error( "cannot parse DATAGRAPH" );
    fprintf( f, "%s\n", argv[ 1 ] );
    rewind( f );
  }

  // When parsing the graph itself, we separate on ';' instead of newline since the specification
  // uses Pascal-statement notation, and no guarantee was made in the homework assignment that the
  // graph input would be nicely formatted into multiple lines.

  char s[ 1000 ];
  var vars[ 30 ];
  int nvars = 0;

  // input_var
  if (!readline( f, s, sizeof s, ';' ))
    error( "invalid DATAGRAPH: expected 'input_var'" );
  parse_varnames( s, vars, &nvars, input_var );

  // internal_var
  if (!readline( f, s, sizeof s, ';' ))
    error( "invalid DATAGRAPH: expected 'internal_var'" );
  parse_varnames( s, vars, &nvars, internal_var );

  // dependencies
  while (readline( f, s, sizeof s, ';' ))
    if (!parse_dependency( s, vars, nvars ))
      break;

  // write()
  int nwritevars = nvars;
  parse_varnames( s, vars, &nwritevars, write_var );
  
  // Fork all subprocesses
  //   We don't do this in write() because write() is not guaranteed to name all subprocesses.
  //   Also, we will do it in random order to demonstrate that process traversal resolves
  //   itself based on the pipe dependencies, not the order of forking.
  shuffle_vars( vars, nvars );
  for (int n = 0; n < nvars; n++)
    fork_var( vars + n );

  // Print the results for processes listed in write()
  //   We don't need to first call wait() for each subprocesses because
  //   it happens automatically in order to read our pipe ends here:
  for (int n = nvars; n < nwritevars; n++)
  {
    double value = read_double( vars[ n ].inputs[ --vars[ n ].ninputs ] );                         GOT( "main()", value, vars[ n ].ivars[ vars[ n ].ninputs ] )
    printf( "%g\n", value );
  }

  fclose( f );
  return 0;
}

Enjoy.

[Metal488]

Thank you guys so much for all your help and these amazing solutions. I did end up getting something done and turned in, but it was no where near as sophisticated as you guys' solutions. I wonder where you guys learned all this, my past classes certainly didn't teach me things that complex.
Anyway, thank you for taking the time to help me and sorry it took me so long to replay.

[dhayden]

I'd be happy to address by solution if you like, but you'd have to ask more specific questions.