read ()를 보장하는 방법 실제로 이름이 지정된 파이프를 통해 쓰기 ()가 보낸 데이터의 100%를 보냅니다.

Question

나는 다음 두 프로그램을 가지고 있습니다. 하나는 독자로서 작가로서 작가로 활동합니다. 작가는 독자가 읽기 위해 데이터의 약 3/4 만 올바르게 보내는 것 같습니다. 모든 데이터가 전송되고 있음을 보장 할 수있는 방법이 있습니까? 나는 그것이 읽고 쓰여지도록 설정했다고 생각하지만 여전히 데이터의 1/4을 놓치는 것 같습니다.

그녀는 작가의 출처입니다

#define pipe "/tmp/testPipe"

using namespace std;

queue<string> sproutFeed;


ssize_t r_write(int fd, char *buf, size_t size) {
   char *bufp;
   size_t bytestowrite;
   ssize_t byteswritten;
   size_t totalbytes;

   for (bufp = buf, bytestowrite = size, totalbytes = 0;
        bytestowrite > 0;
        bufp += byteswritten, bytestowrite -= byteswritten) {
      byteswritten = write(fd, bufp, bytestowrite);
            if(errno == EPIPE)
            {
            signal(SIGPIPE,SIG_IGN);
            }
      if ((byteswritten) == -1 && (errno != EINTR))
         return -1;
      if (byteswritten == -1)
         byteswritten = 0;
      totalbytes += byteswritten;
   }
   return totalbytes;
}


void* sendData(void *thread_arg)
{

int fd, ret_val, count, numread;
string word;
char bufpipe[5];


ret_val = mkfifo(pipe, 0777); //make the sprout pipe

if (( ret_val == -1) && (errno != EEXIST)) 
{
    perror("Error creating named pipe");
    exit(1);
}   
while(1)
{
    if(!sproutFeed.empty())
    {
        string s;
        s.clear();
        s = sproutFeed.front();
        int sizeOfData = s.length();
        snprintf(bufpipe, 5, "%04d\0", sizeOfData); 
        char stringToSend[strlen(bufpipe) + sizeOfData +1];
        bzero(stringToSend, sizeof(stringToSend));                  
        strncpy(stringToSend,bufpipe, strlen(bufpipe));         
        strncat(stringToSend,s.c_str(),strlen(s.c_str()));
        strncat(stringToSend, "\0", strlen("\0"));                  
        int fullSize = strlen(stringToSend);            
        signal(SIGPIPE,SIG_IGN);

        fd = open(pipe,O_WRONLY);
        int numWrite = r_write(fd, stringToSend, strlen(stringToSend) );
        cout << errno << endl;
        if(errno == EPIPE)
        {
        signal(SIGPIPE,SIG_IGN);
        }

        if(numWrite != fullSize )
        {               
            signal(SIGPIPE,SIG_IGN);
            bzero(bufpipe, strlen(bufpipe));
            bzero(stringToSend, strlen(stringToSend));
            close(fd);
        }
        else
        {
            signal(SIGPIPE,SIG_IGN);
            sproutFeed.pop();
            close(fd);
            bzero(bufpipe, strlen(bufpipe));
            bzero(stringToSend, strlen(stringToSend));
        }                   
    }
    else
    {
        if(usleep(.0002) == -1)
        {
            perror("sleeping error\n");
        }
    }
}

}

int main(int argc, char *argv[])
{
    signal(SIGPIPE,SIG_IGN);
    int x;
    for(x = 0; x < 100; x++)
    {
        sproutFeed.push("All ships in the sea sink except for that blue one over there, that one never sinks. Most likley because it\'s blue and thats the mightiest colour of ship. Interesting huh?");
    }
    int rc, i , status;
    pthread_t threads[1];       
    printf("Starting Threads...\n");
    pthread_create(&threads[0], NULL, sendData, NULL);
    rc = pthread_join(threads[0], (void **) &status);

}

그녀는 독자의 출처입니다

#define pipe "/tmp/testPipe"

char dataString[50000];
using namespace std;
char *getSproutItem();

void* readItem(void *thread_arg)
{
    while(1)
    {
        x++;
        char *s = getSproutItem();
        if(s != NULL)
        {
            cout << "READ IN: " << s << endl;
        }
    }
}


ssize_t r_read(int fd, char *buf, size_t size) {
   ssize_t retval;
   while (retval = read(fd, buf, size), retval == -1 && errno == EINTR) ;
   return retval;
}


char * getSproutItem()
{
    cout << "Getting item" << endl;
    char stringSize[4];
    bzero(stringSize, sizeof(stringSize));
    int fd = open(pipe,O_RDONLY);
    cout << "Reading" << endl;

    int numread = r_read(fd,stringSize, sizeof(stringSize));


    if(errno == EPIPE)
    {
        signal(SIGPIPE,SIG_IGN);

    }
    cout << "Read Complete" << endl;

    if(numread > 1)
    {

        stringSize[numread] = '\0'; 
        int length = atoi(stringSize);
        char recievedString[length];
        bzero(recievedString, sizeof(recievedString));
        int numread1 = r_read(fd, recievedString, sizeof(recievedString));
        if(errno == EPIPE)
        {


signal(SIGPIPE,SIG_IGN);
    }       
    if(numread1 > 1)
    {
        recievedString[numread1] = '\0';
        cout << "DATA RECIEVED: " << recievedString << endl;
        bzero(dataString, sizeof(dataString));
        strncpy(dataString, recievedString, strlen(recievedString));
        strncat(dataString, "\0", strlen("\0"));
        close(fd);  
        return dataString;
    }
    else
    {
        return NULL;
    }

}
else
{
    return NULL;
}

close(fd);

}

int main(int argc, char *argv[])
{
        int rc, i , status;
        pthread_t threads[1];       
        printf("Starting Threads...\n");
        pthread_create(&threads[0], NULL, readItem, NULL);
        rc = pthread_join(threads[0], (void **) &status); 

}

Answer 1

당신은 확실히 신호를 잘못 사용하고 있습니다. 스레드는 적어도 제공된 코드에서 완전히 불필요합니다. 문자열 계산은 이상합니다. 얻다 이 책 읽기가 끝날 때까지 키보드를 만지지 마십시오. :)

Answer 2

이름이 지정된 파이프를 통해 데이터를 보내는 데 사용되는 일반적인 방법은 페이로드 길이의 헤더를 압축하는 것입니다. 그런 다음 읽습니다 (fd, header_len); read (rd, data_len); 참고 후자의 read ()는 Data_Len이 읽거나 EOF 될 때까지 루프에서 수행해야합니다. 또한 이름이 지정된 파이프에 여러 작가가 있다면, 쓰기는 원자가 (합리적인 크기 인 한), 즉 여러 작가가 커널 버퍼의 부분 메시지를 사례하지 않습니다.

Answer 3

여기서 무슨 일이 일어나고 있는지 말하기는 어렵습니다. 시스템 통화 중 하나에서 오류를 반환하고 있습니까? 모든 데이터를 성공적으로 전송하고 있다고 확신하십니까?

여기에는 유효하지 않은 코드가있는 것 같습니다.

    int length = atoi(stringSize);
    char recievedString[length];

크기에 대한 비 스탠트 표현식을 사용하여 스택에 배열을 만들 수 없으므로 구문 오류입니다. 실제 버전에서 다른 코드를 사용하고 있습니까?

루프에서 데이터를 읽어야합니까? 때로는 기능이 사용 가능한 데이터의 일부를 반환하고 모든 데이터가 사라질 때까지 반복적으로 호출해야합니다.

시스템 호출이 중단되면 UNIX의 일부 시스템 호출은 Eagain을 반환 할 수 있습니다.

Answer 4

독자 메인 스레드에서 Posix 스레드 신호 처리 시맨틱에 의해 물린 것일 수 있습니다. POSIX 표준을 통해 POSIX 스레드는 반드시 기대되는 스레드가 아니라 신호를 수신 할 수 있습니다. 원하지 않는 경우 신호를 차단합니다.signal(SIG_PIPE,SIG_IGN) 당신의 친구입니다. 독자 메인에 하나를 추가하십시오.

Posix 스레드 처리 시맨틱, POS를 POSIX에 넣습니다. (그러나 POSIX 스레드를 쉽게 구현할 수 있습니다.)

LS로 /TMP의 파이프를 검사 하시겠습니까? 비어 있지 않습니까?