Interprocess Communication in a Farmer-Worker setup

Question

The following code is my first C program I have made for an university assignment about Interprocess Communication. This involved creating a parent process, the farmer, and then creating a certain number of forked processes, the workers, and we had to let the workers calculate something and use message queues to communicate it from the parent to the worker and vica versa. The computations were chosen to be Mandelbrot computations by the teacher. The goal of the assignment was to get familiar with the POSIX libraries.

These were the functional requirements, in my own words:

• The program must use the settings from settings.h
• The farmer starts NROF_WORKER worker processes
• The message queues have a size of MQ_MAX_MESSAGES messages
• The names of the message queues had to include the student name and the student id
• The worker queue needs to be as full as possible
• The farmer queue needs to be as empty as possible
• Busy waiting is not allowed
• We have to use a rsleep(10000) call to simulate some kind of random waiting, implementation given.
• Lastly, another implicit requirement is that we cannot use extra processes or threads, we are only allowed to use the 1 farmer process and the NROF_WORKER worker processes.

The following implementations were given:

• The code that outputs the image on the screen
• The Mandelbrot computation

The Makefile that was provided:

#
#

BINARIES = worker farmer

CC = gcc
CFLAGS = -Wall -g -c
LDLIBS = -lrt -lX11

all:    $(BINARIES)

clean:
    rm -f *.o $(BINARIES)

worker: worker.o 

farmer: farmer.o output.o

worker.o: worker.c settings.h common.h

output.o: output.c output.h settings.h

farmer.o: farmer.c output.h settings.h common.h

This is the relevant code:

settings.h

#ifndef _SETTINGS_H_
#define _SETTINGS_H_

// remove the comments for the output you like: either graphical (X11) output
// or storage in a BMP file (or both)
#define WITH_X11
//#define WITH_BMP

// settings for interprocess communications
// (note: be sure that /proc/sys/fs/mqueue/msg_max >= MQ_MAX_MESSAGES)
#define NROF_WORKERS    64
#define MQ_MAX_MESSAGES 64

// settings for the fractal computations
#define INFINITY        10.0
#define MAX_ITER        512

// settings for graphics
#define X_PIXEL         880
#define Y_PIXEL         660
#define X_LOWERLEFT     -2.0
#define Y_LOWERLEFT     -1.0
#define STEP            0.003
//#define X_LOWERLEFT     -0.65
//#define Y_LOWERLEFT     -0.5
//#define STEP            0.0001

// lower left pixel (0,0) has coordinate
//                  (X_LOWERLEFT, Y_LOWERLEFT)
// upperright pixel (X_PIXEL-1,Y_PIXEL-1) has coordinate
//                  (X_LOWERLEFT+((X_PIXEL-1)*STEP),Y_LOWERLEFT+((Y_PIXEL-1)*STEP))

#endif

output.h

#ifndef _OUTPUT_H_
#define _OUTPUT_H_

extern void output_init (void);
extern void output_draw_pixel (int x, int y, int color);
extern void output_end (void);

#endif

settings.h

/* 
 *
 * Contains definitions which are commonly used by the farmer and the workers
 *
 */

#ifndef _COMMON_H_
#define _COMMON_H_

#include "settings.h"

#define STUDENT_NAME "FrankVanHeeswijk"

typedef struct
{
    int y;
} MQ_FARMER_REQUEST_MESSAGE;

typedef struct
{
    int y;
    int x_colors[X_PIXEL];
} MQ_WORKER_RESULT_MESSAGE;

#endif

farmer.c

/* 
 *
 */

#include <stdio.h>
#include <stdlib.h>
#include <stdbool.h>
#include <string.h>
#include <sys/wait.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <errno.h>    
#include <unistd.h>         // for execlp
#include <mqueue.h>         // for mq

#include "settings.h"
#include "output.h"
#include "common.h"

static char mq_farmer_request_name[80];
static char mq_worker_result_name[80];

static void fork_children(pid_t children_IDs[])
{
    int i;
    for (i = 0; i < NROF_WORKERS; i++)
    {
        pid_t processID = fork();
        if (processID < 0)
        {   
            perror("fork() failed: " + processID);
        }
        else
        {
            if (processID == 0)
            {
                execlp("./worker", "worker", mq_farmer_request_name, mq_worker_result_name, NULL);
                perror("execlp() failed");
            }
            children_IDs[i] = processID;
        }
    }
}

static void kill_children(pid_t children_IDs[])
{
    int i;
    for (i = 0; i < NROF_WORKERS; i++)
    {
        waitpid(children_IDs[i], NULL, 0);
    }
}

static void process_worker_result_message(MQ_WORKER_RESULT_MESSAGE worker_result_message)
{                                       
    int x;
    for (x = 0; x < X_PIXEL; x++)
    {
        output_draw_pixel(x, worker_result_message.y, worker_result_message.x_colors[x]);
    }                                                           
}

int main (int argc, char* argv[])
{
    if (argc != 1)
    {
        fprintf (stderr, "%s: invalid arguments\n", argv[0]);
    }

    output_init ();

    //create message queues
    sprintf(mq_farmer_request_name, "/mq_farmer_request_%s_%d", STUDENT_NAME, getpid());
    sprintf(mq_worker_result_name, "/mq_worker_result_%s_%d", STUDENT_NAME, getpid());

    struct mq_attr attr;

    attr.mq_maxmsg = MQ_MAX_MESSAGES;
    attr.mq_msgsize = sizeof(MQ_FARMER_REQUEST_MESSAGE);
    mqd_t mq_farmer_request = mq_open(mq_farmer_request_name, O_WRONLY | O_CREAT | O_EXCL, 0600, &attr);
    if (mq_farmer_request < 0)
    {
        perror("error opening farmer request message queue in farmer");
    }

    attr.mq_maxmsg = MQ_MAX_MESSAGES;
    attr.mq_msgsize = sizeof(MQ_WORKER_RESULT_MESSAGE);
    mqd_t mq_worker_result = mq_open(mq_worker_result_name, O_RDONLY | O_CREAT | O_EXCL, 0600, &attr);
    if (mq_worker_result < 0)
    {
        perror("error opening worker result message queue in farmer");
    }

    //create children
    pid_t children_IDs[NROF_WORKERS];
    fork_children(children_IDs);

    //send & receive data
    MQ_FARMER_REQUEST_MESSAGE farmer_request_message;
    MQ_WORKER_RESULT_MESSAGE worker_result_message;

    //keep farmer request queue as filled as possible, keep worker result queue as empty as possible
    int msg_max = Y_PIXEL;
    int msg_num_received = 0;
    int msg_num_sent = 0;
    while (msg_num_sent < msg_max && msg_num_received < msg_max)
    {
        //fill up farmer request queue to the max
        int get_farmer_request_attr_result = mq_getattr(mq_farmer_request, &attr);
        if (get_farmer_request_attr_result < 0)
        {
            perror("error getting farmer request attr in farmer");
            exit(EXIT_FAILURE);
        }
        while (attr.mq_curmsgs < attr.mq_maxmsg && msg_num_sent < msg_max)
        {
            //send message
            farmer_request_message.y = msg_num_sent;
            int sent = mq_send(mq_farmer_request, (char*)&farmer_request_message, sizeof(farmer_request_message), 0);
            if (sent < 0)
            {
                perror("error sending message in farmer");
                exit(EXIT_FAILURE);
            }
            msg_num_sent++;

            get_farmer_request_attr_result = mq_getattr(mq_farmer_request, &attr);
            if (get_farmer_request_attr_result < 0)
            {
                perror("error getting farmer request attr in farmer");
                exit(EXIT_FAILURE);
            }
        }

        //empty worker result queue
        int get_worker_result_attr_result = mq_getattr(mq_worker_result, &attr);
        if (get_worker_result_attr_result < 0)
        {
            perror("error getting worker result attr in farmer");
            exit(EXIT_FAILURE);
        }
        while (attr.mq_curmsgs > 0)
        {
            //take one message
            int received_bytes = mq_receive(mq_worker_result, (char*)&worker_result_message, sizeof(worker_result_message), NULL);
            if (received_bytes < 0)
            {
                perror("error receiving message in farmer");
                exit(EXIT_FAILURE);
            }
            msg_num_received++;

            process_worker_result_message(worker_result_message);

            //because we took one message, we can now send another one
            if (msg_num_sent < msg_max)
            {
                farmer_request_message.y = msg_num_sent;
                int sent = mq_send(mq_farmer_request, (char*)&farmer_request_message, sizeof(farmer_request_message), 0);
                if (sent < 0)
                {
                    perror("error sending message in farmer");
                    exit(EXIT_FAILURE);
                }
                msg_num_sent++;
            }

            get_worker_result_attr_result = mq_getattr(mq_worker_result, &attr);
            if (get_worker_result_attr_result < 0)
            {
                perror("error getting worker result attr in farmer");
                exit(EXIT_FAILURE);
            }
        }
    }

    //stop children
    int i;
    for (i = 0; i < NROF_WORKERS; i++)
    {
        farmer_request_message.y = -1;
        int sent = mq_send(mq_farmer_request, (char*)&farmer_request_message, sizeof(farmer_request_message), 0);
        if (sent < 0)
        {
            perror("error sending message in farmer");
        }
    }

    kill_children(children_IDs);

    //close message queues
    int closed_farmer = mq_close(mq_farmer_request);
    if (closed_farmer < 0)
    {
        perror("failed to close farmer request queue in farmer");
    }
    int closed_worker = mq_close(mq_worker_result);
    if (closed_worker < 0)
    {
        perror("failed to close worker result queue in farmer");
    }

    int unlink_farmer = mq_unlink(mq_farmer_request_name);
    if (unlink_farmer < 0)
    {
        perror("failed to unlink farmer request queue in farmer");
    }
    int unlink_worker = mq_unlink(mq_worker_result_name);
    if (unlink_worker < 0)
    {
        perror("failed to unlink worker result queue in farmer");
    }

    output_end();

    return EXIT_SUCCESS;
}

worker.c

/* 
 *
 */

#include <stdio.h>
#include <stdlib.h>
#include <stdbool.h>
#include <string.h>
#include <errno.h>          // for perror()
#include <unistd.h>         // for getpid()
#include <mqueue.h>         // for mq-stuff
#include <time.h>           // for time()
#include <complex.h>

#include "settings.h"
#include "common.h"


static double 
complex_dist (complex a)
{
    // distance of vector 'a'
    // (in fact the square of the distance is computed...)
    double re, im;

    re = __real__ a;
    im = __imag__ a;
    return ((re * re) + (im * im));
}

static int 
mandelbrot_point (double x, double y)
{
    int     k;
    complex z;
    complex c;

    z = x + y * I;     // create a complex number 'z' from 'x' and 'y'
    c = z;

    for (k = 0; (k < MAX_ITER) && (complex_dist (z) < INFINITY); k++)
    {
        z = z * z + c;
    }

    //                                    2
    // k >= MAX_ITER or | z | >= INFINITY

    return (k);
}

/*
 * rsleep(int t)
 *
 * The calling thread will be suspended for a random amount of time
 * between 0 and t microseconds
 * At the first call, the random generator is seeded with the current time
 */
static void rsleep (int t)
{
    static bool first_call = true;

    if (first_call == true)
    {
        srandom (time(NULL) % getpid());
        first_call = false;
    }
    usleep (random () % t);
}

int main (int argc, char* argv[])
{
    //open message queues
    char* mq_farmer_request_name = argv[1];
    char* mq_worker_result_name = argv[2];

    mqd_t mq_farmer_request = mq_open(mq_farmer_request_name, O_RDONLY);
    if (mq_farmer_request < 0)
    {
        perror("error opening farmer request message queue in worker");
    }
    mqd_t mq_worker_result = mq_open(mq_worker_result_name, O_WRONLY);
    if (mq_worker_result < 0)
    {
        perror("error opening worker result message queue in worker");
    }

    //read messages
    MQ_FARMER_REQUEST_MESSAGE farmer_request_message;
    MQ_WORKER_RESULT_MESSAGE worker_result_message;

    while (true)
    {
        int received_bytes = mq_receive(mq_farmer_request, (char*)&farmer_request_message, sizeof(farmer_request_message), NULL);
        if (received_bytes < 0)
        {
            perror("error receiving message in worker");
            break;
        }

        if (farmer_request_message.y < 0)
        {
            break;
        }

        rsleep(10000);

        worker_result_message.y = farmer_request_message.y;
        int x;
        for (x = 0; x < X_PIXEL; x++)
        {   
            double mx = (X_LOWERLEFT + x * STEP);
            double my = (Y_LOWERLEFT + worker_result_message.y * STEP);
            worker_result_message.x_colors[x] = mandelbrot_point(mx, my);
        }

        int sent = mq_send(mq_worker_result, (char*)&worker_result_message, sizeof(worker_result_message), 0);
        if (sent < 0)
        {
            perror("error sending message in worker");
            break;
        }
    }

    int closed_farmer = mq_close(mq_farmer_request);
    if (closed_farmer < 0)
    {
        perror("failed to close farmer request queue in worker");
    }
    int closed_worker = mq_close(mq_worker_result);
    if (closed_worker < 0)
    {
        perror("failed to close worker result queue in worker");
    }

    return EXIT_SUCCESS;
}

In order to help you understand the intended implementation in the farmer, I have attached this pseudo-code to understand how I want to achieve a maximally filled worker queue and a minimally filled farmer queue:

while (not_all_messages_sent) {
    while (mq_farmer_request not full) {
         send request to mq_farmer_request;
    }

    while (mq_worker_result not empty) {
        receive result from mq_worker_result;
        process result;
        if (not_all_messages_sent) {
            send request to mq_farmer_request;
        }
    }
}

I'd like to have a review on all aspects of my code. I have a Java background so I am very used to putting everything that is reusable into methods and I hate hard-coding, but these seems less applicable in C. I wouldn't mind to have more methods in my code.

You can view the full code and extract a working copy over at my Github repository.

Answer 1

Overall this is some very nicely written C, well done. Some stuff I noted aside from @Morwenn's good points:

• Portability: <mqueue.h> is a POSIX C library. Unfortunately this restrains the platforms that you compile this for (I couldn't compile this on my Mac without some fiddling around). There are two ways you could fix this:

  1. Include the header with your package and use that to compile with. I'm not sure how portable that header is though. If it is portable then I would probably go with this option.
  2. Rewrite your code using <fcntl.h>, <signal.h>, <sys/types.h>, and <time.h>.

• Push your code: Looking at your Makefile, you don't have some things that I would consider a necessity. Going off of my own C Makefile boilerplate, Here is some stuff I would change:

  1. Add more CFLAGS:

     CFLAGS = -Werror -Wall -Wextra -pedantic-errors -Wformat=2 -Wno-import -Wimplicit -Wmain -Wchar-subscripts -Wsequence-point -Wmissing-braces -Wparentheses -Winit-self -Wswitch-enum -Wstrict-aliasing=2 -Wundef -Wshadow -Wpointer-arith -Wbad-function-cast -Wcast-qual -Wcast-align -Wwrite-strings -Wstrict-prototypes -Wold-style-definition -Wmissing-prototypes -Wmissing-declarations -Wredundant-decls -Wnested-externs -Winline -Wdisabled-optimization -Wunused-macros -Wno-unused
     

     These are all the flags that I always use for all of my projects. Sometimes I use even more. I would recommend that you take a look at all of the warning options sometime and fine-tune them for your own needs.

  2. Specify your compiler version.

     CC = gcc-4.9
     

     Right now you are using whatever the OS has set for the default in its PATH variable. This could lead to some compatibility problems, since earlier versions of a compiler obviously won't have support for later standards. Which leads me to my next point...

• Standards: You are using what appears to be the -std=gnu89 as @Morwenn said. Please no. Those are 25 year old standards.

  

  Your code should be updated for more modern standards. At a minimum, I think you should be using -std=gnu99, but why not go the extra mile and use the -std=gnu11 standards?

• Stress-test, and then some: write some code to test that your code works in a variety of cases. Push your code to it's breaking point. Find out what makes it break and fix it. This is how you develop "bullet-proof" software. You may have to incorporate some fuzzing to fully test what your code can handle.

---

Some other more minor stuff:

• Is there a reason that you need a whole struct for one value?

  typedef struct
  {
      int y;
  } MQ_FARMER_REQUEST_MESSAGE;
  

  Could this not work in its place?

  typedef int MQ_FARMER_REQUEST_MESSAGE;  // just needs shorthand typename
  

• Your naming conventions aren't what I would consider normal. Structure names are usually written in PascalCase.

• Always declare your variables within your for loops.

  for (int i = 0; i < NROF_WORKERS; ++i)
  

  This is universally considered a good practice, and is the main reason (I think) that you should be using the C99 standards (since the C89 syntax doesn't support it).

• You return ((re * re) + (im * im)) from your complex_dist() function. I would use the pow() function from <math.h> (there are some other places where I would use this in your code as well). Why? Because the function handles errors in a way that simply multiplying two values together couldn't. Please note however, that this will result in slightly less efficient code.

• The variable names in your mandelbrot_point() could maybe be better. If you can't find more suitable names, then more documentation is definitely needed.

• You may want to consider using Doxygen to generate documentation your code. It's what I use and I helps me with commenting stuff a lot.

• From Code Complete, 2nd Edition, p. 761:

  Use only one data declaration per line.

  It’s easier to modify declarations because each declaration is self-contained.

  [...]

  It’s easier to find specific variables because you can scan a single column rather than reading each line. It’s easier to find and fix syntax errors because the line number the compiler gives you has only one declaration on it.

• You don't have to return 0/EXIT_SUCCESS at the end of main(), just like you wouldn't bother putting return; at the end of a void-returning function. The C standard knows how frequently this is used, and lets you not bother.

  C99 & C11 §5.1.2.2(3)

  ...reaching the } that terminates the main() function returns a value of 0.

• Prefer snprintf() to sprintf().

---

_{There might be more stuff I missed, but this is all I think I could review without actually compiling the code and running some tests.}

Answer 2

Generally speaking, the code looks good. It seems to be clean and to achieve what it is meant to achieve. Since your Makefile was given, as well as the compiler, I will review your code in function of what is legal with the -std=gnu89 compiler option, which means POSIX C89 + parts of C99 + some other compiler extensions. Here is a list of the allowed extensions.

• First of all, your header guard names (_COMMON_H_, _OUTPUT_H_) start with an underscore followed by a capital letter, and is therefore a name reserved to the implementation (compiler + standard library). The simpler fix is to remove the leading underscore: COMMON_H_, INPUT_H_.

• You use perror to display an error message when some computation fails. Do you want to keep it as is, or do you also want to make you program abort on errors? Currently, I don't think that resuming the execution after an error will lead to expected behaviour. While it is not that important if closing a resource fails, there ought to be some problems if opening one fails.

• complex_dist is a bad name since it actually returns the square of the distance. You should change it to complex_square_dist or something equivalent. Note that the "distance" of a complex number is its absolute value that can be computed with cabs. Since you only need the square distance, I would simply rename your function.

• if (first_call == true) is not the best way to check for a boolean value. You should write if (first_call) instead.

• If there is no other possible outcome, returning EXIT_SUCCESS is useless, you could simply return 0 instead. Generally speaking, it is good to return EXIT_SUCCESS when you know that your program can also return error codes. Removing it may serve as kind of documentation to say "this program never returns error codes".

• You are retrieving information from argv without checking first if they exist. You should use conditions such as if (argc < 2) printf("you did something wring"); to tell when the command line is not long enough and how it should be to work correctly.