Почему мой канал не работает в моей многопоточной клиент-серверной программе? - PullRequest
Новая
       18

Почему мой канал не работает в моей многопоточной клиент-серверной программе?

0 голосов
/ 21 декабря 2018

Я реализовал клиент-серверную программу с потоками.Серверная программа должна прочитать от двух клиентов 100 сообщений с сокетом и после этого записать эти данные в канал, чтобы четвертая программа прочитала их.Я успешно прочитал данные из сокета, но не смог записать на канал;мой системный вызов «write» не работает: что я могу сделать? 

#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; 
}

Спасибо за внимание:)

Ответы [ 2 ]

0 голосов
/ 21 декабря 2018

У вас есть как минимум три условия гонки в вашем коде, где данные используются одним потоком, в то время как они могут быть изменены другим.

Этот код создает условия гонки: 

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);
}

Каждый поток использует одни и те же структуры данных m1 и m2, перезаписывая данные друг друга при чтении в m2.Они также одновременно обновляют j, поэтому его значение нельзя доверять ни одному из потоков.

Кроме того, вы не представляете, сколько байт вы действительно прочитали.

Этокод создает другую гонку данных: 

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++;

}

Объедините это с 

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

, и дочерний поток обращается к newsockfd из основного потока, но newsockfd может иметь другое значениек тому времени, когда дочерний поток получает к нему доступ.

Лучший способ: 

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);
}

Обратите внимание, что newsockfd теперь передается значением , а не адресом, поэтомуpthread_create() вызов должен быть: 

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

Это что-то вроде хака, который основан на способности вашей платформы передавать значение int, такое как newsockfd, как void *, но как угодносистема, которую вы сейчас используете, почти наверняка сможет это сделать.

0 голосов
/ 21 декабря 2018

Я думаю, что вы говорите об этом разделе своего кода (изменено пробел): 

    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);

У меня проблемы с определением ожидаемого поведения.Я заметил, что конец чтения канала закрывается сразу после успешного создания канала, что делает канал бесполезным.Это не должно приводить к тому, что последующие write() s до конца записи блокируются бесконечно, а только потому, что вместо этого они должны завершаться с ошибкой EPIPE.

. Если вы хотите связаться с другим процессом по каналу, то выдолжен fork() этот другой процесс после создания канала, но перед закрытием любого конца.Затем родитель и потомок каждый закрывают свою копию конца, который они не собираются использовать.

...