Why my pipe doesn't work in my multithreads client-server program?

Question

I implemented a client-server program with threads. The server programm must read from two clients 100 messages with socket and after it has to write this data on a pipe to make a fourth program read it. I successfully read the data from the socket but i'm not able to write on the pipe; my "write" system call doesn't work: what can I do?

#include <stdio.h>
#include <sys/types.h> 
#include <sys/socket.h>
#include <netinet/in.h>
#include <pthread.h>
#include <arpa/inet.h>
#include <netdb.h> 
#include <sys/time.h>
#include <sys/wait.h>
#include <unistd.h>
#include <string.h>
#include <stdlib.h> 
#include <errno.h> 
#include <signal.h>
#include <time.h> 

#define DIM 100

void logFile(char *msgStatus) {
    FILE *f;
    f = fopen("logFileEs1.log", "a+");
    time_t currentTime;
    char* timeString;
    currentTime = time(NULL);
    timeString = ctime(&currentTime);
    fprintf(f, "%sPID %d. %s: %s\n",timeString, getpid(), msgStatus, strerror(errno));
    fclose(f);
} //function for the creation of the log file
int j=0;

void error(char *msg)
{
    perror(msg);
    exit(1);
}

struct message {  //dichiarazione struct
 time_t timestamp;  
 char g;  //process identifier
 int x;
};

struct message m1[DIM];
struct message m2;

void *func_thread(void *p)
{
    int nfd;
    nfd= *(int*) p;
    int n;  //for reading

    while(read(nfd,&m2,sizeof(m2))!=0) { //reading

        printf("Here is the message: %d from process %d at time %ld     %d\n",m2.x, m2.g, m2.timestamp, j);
        fflush(stdout);
        m1[j]=m2;
        j++;
    }
 pthread_exit(NULL);
}

int main(int argc, char *argv[])
{
 FILE *f;
 f = fopen("logFileEs2.log", "w");
 fclose(f);

    pthread_t id[2];

    void *dummy;
    int iret1, i=0, d, t;
    int pipeState, execState1, data;
    pid_t child;

    int sockfd,newsockfd, portno, clilen;
    portno=6076;

    struct sockaddr_in serv_addr, cli_addr;  //adress of the server and the client

    /*if (argc < 2) {
        fprintf(stderr,"ERROR, no port provided\n");
        exit(1);
    }*/

    sockfd = socket(AF_UNIX, SOCK_STREAM, 0);  //new socket creation
    if (sockfd < 0) 
        error("ERROR opening socket");

    serv_addr.sin_family = AF_UNIX;
    serv_addr.sin_addr.s_addr =inet_addr("127.0.0.1");
    serv_addr.sin_port = htons(portno);
    if (bind(sockfd, (struct sockaddr *) &serv_addr, sizeof(serv_addr)) < 0) 
        error("ERROR on binding");

    listen(sockfd,5);
    printf("Listening\n");

    while(i<2) {
        clilen = sizeof(cli_addr);
        newsockfd = accept(sockfd, (struct sockaddr *) &cli_addr, &clilen);
        if (newsockfd < 0) 
            error("ERROR on accept");


        iret1 = pthread_create(&id[i], NULL, func_thread, &newsockfd);
        if (iret1) {
            perror("pthread_create");
            return -1;
        }

        i++;

    }
    pthread_join(id[0], &dummy);
    pthread_join(id[1], &dummy);

    char readPipe[5]; //string for reading pipe
    char writePipe[5]; //string for writing pipe
    int fd[2]; //file descriptor pipe1


    pipeState = pipe (fd);  //creation first pipe
    if (pipeState < 0) {
        perror("Pipe error");
        return -1;
    }


    close(fd[0]);
    for (t=0; t<DIM; t++) {
        printf("%d\n", m1[t].x); 
    }



    data = write(fd[1], &m1, sizeof(m1));


        if (data < 0) { //if written value is not valid
            perror("Write error\n");
            return -1;
        }
    printf("Data written on pipe\n");
    close(fd[1]);
    printf("Data written on pipe\n");
    fflush(stdout);


    //fd conversion from integer to string
    sprintf(readPipe, "%d", fd[0]);
    sprintf(writePipe, "%d", fd[1]);

    char *function[] = {"M", readPipe, writePipe, NULL};

    child=fork();

    /*if (child1 != 0) {
        logFile("Creation of the child1: ");
    }*/

    if (child < 0) {
        perror  ("Fork error in child1");
        return -1;
    }

    else if (child == 0) {

        execState1=execve("M", function,NULL);
        exit (EXIT_SUCCESS);
    }

    else { wait(NULL);
        exit (EXIT_SUCCESS);
    }


    return 0; 
}

Thanks for attention :)

Answer 1

You have at least three race conditions in your code, where data is used by one thread while it may be modified by another.

This code creates race conditions:

struct message m1[DIM];
struct message m2;

void *func_thread(void *p)
{
    int nfd;
    nfd= *(int*) p;
    int n;  //for reading

    while(read(nfd,&m2,sizeof(m2))!=0) { //reading

        printf("Here is the message: %d from process %d at time %ld     %d\n",m2.x, m2.g, m2.timestamp, j);
        fflush(stdout);
        m1[j]=m2;
        j++;
    }
 pthread_exit(NULL);
}

Every thread shares the same m1 and m2 data structures, overwriting each other's data when they read into m2. They also make simultaneous updates to j, so its value can't be trusted in either thread.

Also, you have no idea how many bytes you've actually read.

This code creates another data race:

while(i<2) {
    clilen = sizeof(cli_addr);
    newsockfd = accept(sockfd, (struct sockaddr *) &cli_addr, &clilen);
    if (newsockfd < 0) 
        error("ERROR on accept");


    iret1 = pthread_create(&id[i], NULL, func_thread, &newsockfd);
    if (iret1) {
        perror("pthread_create");
        return -1;
    }

    i++;

}

Combine that with

void *func_thread(void *p)
{
    int nfd;
    nfd= *(int*) p;

and the child thread is accessing newsockfd from the main thread, but newsockfd may have a different value by the time the child thread accesses it.

A better way:

struct message m1[DIM];
int j = 0
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;

void *func_thread(void *p)
{
    // each thread needs its own struct
    struct message m2;

    // pass the socket by **value**
    int nfd = ( intptr_t ) p;

    for ( ;; )
    {
        // don't put code that needs error checks inside conditions
        // because you can't handle errors, nor in this case partial
        // read results
        ssize_t bytes_read = read( nfd, &m2, sizeof( m2 ) );
        if ( bytes_read == 0 )
        {
            break;
        }
        // really should put code here to handle a partial read()

        printf("Here is the message: %d from process %d at time %ld     %d\n",
            m2.x, m2.g, m2.timestamp, j);
        fflush(stdout);

        // another race condition if this isn't mutex'd
        pthread_mutex_lock( &mutex );

        // get a local copy of the current value of j so
        // the structure assignment can be moved outside
        // the mutex-locked critical section
        int my_j = j;
        j++;
        pthread_mutex_unlock( &mutex );

        // stay within the bounds of the array
        if ( my_j >= DIM )
        {
            break;
        }
        m1[my_j]=m2;
    }

    pthread_exit(NULL);
}

Note that newsockfd is now passed by value, not address, so the pthread_create() call needs to be:

    iret1 = pthread_create(&id[i], NULL, func_thread, ( void * )( intptr_t ) newsockfd);

That's a bit of a hack that relies upon your platform being able to pass an int value such as newsockfd as a void *, but whatever system you're currently using almost certainly can do that.

Answer 2

I think you're talking about this section of your code (whitespace modified):

    int fd[2]; //file descriptor pipe1

    pipeState = pipe(fd);  //creation first pipe
    if (pipeState < 0) {
        perror("Pipe error");
        return -1;
    }

    close(fd[0]);
    for (t=0; t<DIM; t++) {
        printf("%d\n", m1[t].x); 
    }

    data = write(fd[1], &m1, sizeof(m1));

    if (data < 0) { //if written value is not valid
        perror("Write error\n");
        return -1;
    }

    printf("Data written on pipe\n");
    close(fd[1]);
    printf("Data written on pipe\n");
    fflush(stdout);

I'm having trouble determining what behavior is expected. I observe that the read end of the pipe is closed immediately after the pipe is successfully created, which makes the pipe useless. This ought not to cause subsequent write()s to the write end to block indefinitely, but only because they should instead fail with EPIPE.

If you want to communicate with another process over the pipe, then you should fork() that other process after creating the pipe but before closing either end. The parent and child then each close their copy of the end they do not intend to use.