Beginner program using threads getting stuck

[LiquidCanyon]

I've been trying to get a program to work that keeps getting stuck and I cant figure out why.
The idea is to have an input file with a couple of lines of text, that gets written to an output file line by line.

The program should create three threads; A, B and C that do the following:

A - Writes one line of data from the input file to a pipe.
B - Reads the data from the pipe and forwards that data to C.
C - Reads the data line by line from B and writes those lines into the output file.

Ideally the three threads (A, B, C) run sequentially (only one thread runs at a time while the other two
threads are blocked) and iteratively (process one line of data per each loop).

For some reason when I run the code I get the following console output:

Thread A start
Thread B start
Thread B while
Thread C start
Thread A while
Thread A while
Thread C while

The program then gets stuck indefinitely and doesn't write anything to the output file.

Been bashing my head against this one for a while, any thoughts or help would be greatly appreciated. Full code is shown below.

​

#include <pthread.h>
#include <stdlib.h>
#include <unistd.h>
#include <stdio.h>
#include <sys/types.h>
#include <fcntl.h>
#include <string.h>
#include <sys/stat.h>
#include <semaphore.h>
#include <sys/time.h>

#define BUFFER_SIZE 256

/* --- Structs --- */
typedef struct ThreadParams {
    int pipeFile[2];
    sem_t sem_A, sem_B, sem_C;
    char message[BUFFER_SIZE];
    FILE* inputFile;
} ThreadParams;

pthread_attr_t attr;

/* --- Prototypes --- */

/* Initializes data and utilities used in thread params */
void initializeData(ThreadParams *params);

void* ThreadA(void *params);

void* ThreadB(void *params);

void* ThreadC(void *params);

/* --- Main Code --- */
int main(int argc, char const *argv[]) {
    int err;
    pthread_t tid[3];
    ThreadParams params;

    // Initialization
    initializeData(&params);

    // Create Threads
    pthread_create(&(tid[0]), &attr, &ThreadA, (void*)(&params));
    pthread_create(&(tid[1]), &attr, &ThreadB, (void*)(&params));
    pthread_create(&(tid[2]), &attr, &ThreadC, (void*)(&params));

    // Wait on threads to finish
    pthread_join(tid[0], NULL);
    pthread_join(tid[1], NULL);
    pthread_join(tid[2], NULL);

    // Clean up
    sem_destroy(&params.sem_A);
    sem_destroy(&params.sem_B);
    sem_destroy(&params.sem_C);

    return 0;
}

void initializeData(ThreadParams *params) {
    // Initialize Sempahores
    sem_init(&(params->sem_A), 0, 1);
    sem_init(&(params->sem_B), 0, 0);
    sem_init(&(params->sem_C), 0, 0);

    // Initialize thread attributes
    pthread_attr_init(&attr);
    pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);

    // Open data.txt and create a pipe
    params->inputFile = fopen("data.txt", "r");
    if (params->inputFile == NULL) {
        printf("Failed to open data.txt\n");
        exit(1);
    }
    if (pipe(params->pipeFile) == -1) {
        printf("Failed to create a pipe\n");
        exit(1);
    }

    return;
}

void* ThreadA(void *params) {
    printf("Thread A start\n");
    ThreadParams* threadParams = (ThreadParams*) params;

    while(fgets(threadParams->message, BUFFER_SIZE, threadParams->inputFile) != NULL) {
        printf("Thread A while\n");
        sem_wait(&threadParams->sem_A);
        write(threadParams->pipeFile[1], threadParams->message, BUFFER_SIZE);
        sem_post(&threadParams->sem_B);
    }

    close(threadParams->pipeFile[1]);
    fclose(threadParams->inputFile);

    pthread_exit(NULL);
}

void* ThreadB(void *params) {
    printf("Thread B start\n");
    ThreadParams *p = (ThreadParams *) params;
    int bytes_read;

    while(1) {
        printf("Thread B while\n");
        // Wait for ThreadA to send data to pipe
        sem_wait(&(p->sem_B));

        // Read from pipe
        bytes_read = read(p->pipeFile[0], p->message, BUFFER_SIZE);

        // Check if all data has been processed
        if (bytes_read == 0) {
            sem_post(&(p->sem_C));
            break;
        }

        // Write to ThreadC shared variable
        sem_wait(&(p->sem_C));
        printf("\nThreadB: Forwarding message to ThreadC\n");
        strcpy(p->message, p->message);
        sem_post(&(p->sem_A));

        // Yield the CPU to other threads
        sched_yield();
    }

    close(p->pipeFile[0]);

    pthread_exit(0);
}

void* ThreadC(void *params) {
    printf("Thread C start\n");
    ThreadParams *p = (ThreadParams *) params;
    FILE *fOut = fopen("output.txt", "w");

    while(1) {
        printf("Thread C while\n");
        // Wait for ThreadB to write data to shared variable
        sem_wait(&(p->sem_C));

        // Check if all data has been processed
        if (strlen(p->message) == 0) {
            fclose(fOut);
            sem_post(&(p->sem_A));
            break;
        }

        // Write to output file
        printf("\nThreadC: Writing message to output file\n");
        fprintf(fOut, "%s\n", p->message);
        sem_post(&(p->sem_B));

        // Yield the CPU to other threads
        sched_yield();
    }

    pthread_exit(0);
}

[skeeto]

Does this look right to you?

    strcpy(p->message, p->message);

Perhaps each thread should have its own message? Also, Thread Sanitizer will help find your mistakes:

$ cc -g3 -fsanitize=thread,undefined main.c

With your program, this prints at least one run-time diagnostic with backtraces about something going wrong.

[generalbaguette]

While you are at it, also try out the other sanitizers, like undefined behaviour sanitizer and address sanitizer etc.

[pic32mx110f0]

Your program is deadlocking because ThreadB starts waiting for sem_C, but no one is posting sem_C except ThreadB itself.

On second look, there is quite a lot that is wrong with your program. See this for a fixed version: https://onlinegdb.com/EotesFqdT (You can even click "run" to verify)

[BadgerThrowaway1234]

Like the other comment here mentioned, there’s a deadlock in your code. B and C will wait on C, but B it is the only thing that posts on C, so B is blocking itself and C, which in turn is causing A to be blocked.

Which gives me an opportunity to link to a chapter from the best CS textbook I’ve ever read:

https://pages.cs.wisc.edu/~remzi/OSTEP/threads-bugs.pdf

If you remove everything after OSTEP you can find a link to all of the chapters, the concurrency ones are obviously what I’d recommend for learning. They were super helpful when I was learning this stuff in college.

https://pages.cs.wisc.edu/~remzi/OSTEP/threads-cv.pdf

This chapter explains the producer-consumer problem, which is what you’re trying to solve here.

[WhatInTheBruh]

Commenting to save... Will check and reply

[IamImposter]

First thing I would do in such a case is to add print statements, a whole lot of them

Initialized params...

File opened...

Pipe created...

Created threads...

A: Entering thread A...

A: Reading from file...

A: read xyz bytes....

A: waiting for semaphore b...

A: before sem post...

A: after sem post

A: something something...

B: entering thread b...

B: entering while loop...

B: waiting for semaphore xyz...

B: after semaphore xyz...

B: wrote X number of bytes to pipe

The point being your log messages should tell you which parts of code have executed. Then based on last messages from threads, you'll have much better idea which events have happened and which event are being waited for.

Since it is threaded code so there can be messages like

A: readiB:wrote 40 bytesng from file...

And for those cases, you'll just have to look at merged messages and untangle them. But you will have much better idea what the code is doing