Кольцевой буфер C Семафор с общей памятью ReadPointer WritePointer Not Right

Question

Я хочу создать копировщик файлов, используя парадигму синхронизации Reader-Writer.Писатель инициализирует оба мьютекса.FullMutex обозначает, сколько буферов доступно для записи, а FreeMutex обозначает, сколько буферов доступно для чтения.Писатель ждет, когда Блок заполнен.WritePointer и ReadPointer используют кольцевой буфер.Поэтому я использовал Mod Operation.Размер блока = M.Размер буфера = B.Есть N количество буферов.Итак, М = Н * Б.Размер файла = 2M.И поэтому BufferCount фактически продвигает File Pointer.Когда все байты записаны, я выдаю FileEnding = 1.

Команды компиляции -

g++ Writer.c -o Writer -lpthread -lrt
g++ Reader.c -o Reader -lpthread -lrt

И в двух разных командных приглашениях открыты и команды выдают -

./Writer
./Reader

Теперь я не знаю, почему ReadPointer и WritePointer идут таким образом?

Вот Writer.c

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <semaphore.h>
#include <fcntl.h>
#include <unistd.h>
#include <sys/stat.h>
#include <sys/mman.h>

int main(void){
    char FileName[128]="aaa.txt";
    struct MemData{
        sem_t FullMutex;
        sem_t FreeMutex;
        int ReadPointer;
        int WritePointer;
        int FileEnding;
        char Data[512000];//MEMORY BLOCK SIZE: 500 KB
    };
    int SD;
    struct MemData *M;
    int NumberOfBuffers=10;
    //int BufferSize=51200;//FILE BUFFER SIZE 50 KB
    int BufferSize=2;//EXPERIMENATION
    unsigned char Buf[BufferSize];
    int BufferCount=0;

    SD= shm_open("/program.shared", O_RDWR|O_CREAT, S_IREAD|S_IWRITE);
    if(SD< 0){
        printf("\nshm_open() error \n");
        return EXIT_FAILURE;
    }
    fchmod(SD, S_IRWXU|S_IRWXG|S_IRWXO);
    if(ftruncate(SD, sizeof(MemData))< 0){
        printf ("ftruncate() error \n");
        return EXIT_FAILURE;
    }
    M=(struct MemData*)mmap(NULL, sizeof(MemData), PROT_READ|PROT_WRITE, MAP_SHARED, SD, 0);
    if(M== MAP_FAILED){
        printf("mmap() error");
        return EXIT_FAILURE;
    }else{
        sem_init(&M->FullMutex, 1, 0);
        sem_init(&M->FreeMutex, 1, NumberOfBuffers);
        M->FileEnding=0;
        M->ReadPointer=0;
        M->WritePointer=0;
        memset(M->Data, '\0', strlen(M->Data));
    }

    FILE *FP= fopen(FileName, "rb");
    if(FP!= NULL){
        struct stat StatBuf;
        if(stat(FileName, &StatBuf)==-1){
            printf("failed to fstat %s\n", FileName);
            exit(EXIT_FAILURE);
        }
        long long FileSize=StatBuf.st_size;
        printf("\nFile Size: %lld", FileSize);
        long long FilePosition=ftell(FP);
        FilePosition=ftell(FP);
        long long CopyableMemorySize=FileSize-FilePosition;
        printf("\nCopyable File Size: %lld", CopyableMemorySize);
        int NumberOfFileBuffers=CopyableMemorySize/BufferSize;
        printf("\nNumber Of File Buffers: %d\n", NumberOfFileBuffers);

        //WRITE
        while(1){
            sem_wait(&M->FreeMutex);
            fseek(FP, BufferCount*BufferSize, SEEK_SET);
            fread(Buf, sizeof(unsigned char), BufferSize, FP);
            int FreeMutexValue;
            sem_getvalue(&M->FreeMutex, &FreeMutexValue);
            int FullMutexValue;
            sem_getvalue(&M->FullMutex, &FullMutexValue);
            printf("\nMutexes-Free: %d and Full: %d", FreeMutexValue, FullMutexValue);
            printf("\nBuffer Writing: %d", BufferCount);
            memcpy(&M->Data[M->WritePointer*BufferSize], &Buf, sizeof(Buf)*sizeof(unsigned char));
            BufferCount++;
            M->WritePointer=(M->WritePointer+1)%NumberOfBuffers;
            if(BufferCount>=NumberOfFileBuffers && M->WritePointer==M->ReadPointer){
                M->FileEnding=1;
                break;
            }
            sem_post(&M->FullMutex);
        }
        fclose(FP);
    }
    //close(SD);
    return 0;
}

Вот Reader.c

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <semaphore.h>
#include <fcntl.h>
#include <unistd.h>
#include <sys/stat.h>
#include <sys/mman.h>

int main(void){
    char FileName[128]="bbb.txt";
    struct MemData{
        sem_t FullMutex;
        sem_t FreeMutex;
        int ReadPointer;
        int WritePointer;
        int FileEnding;
        char Data[512000];//MEMORY BLOCK SIZE: 500 KB
    };
    int SD;
    struct MemData *M;
    int NumberOfBuffers=10;
    //int BufferSize=51200;//FILE BUFFER SIZE 50 KB
    int BufferSize=2;//EXPERIMENATION
    unsigned char Buf[BufferSize];
    int BufferCount=0;
    SD= shm_open("/program.shared", O_RDWR|O_CREAT, S_IREAD|S_IWRITE);
    if(SD< 0){
        printf("\nshm_open() error \n"); 
        return EXIT_FAILURE;
    }
    M=(struct MemData*)mmap(NULL, sizeof(MemData), PROT_READ|PROT_WRITE, MAP_SHARED, SD, 0);
    if(M== MAP_FAILED){
        printf("mmap() error");
        return EXIT_FAILURE;
    }

    FILE *FP= fopen(FileName, "wb");
    if(FP!= NULL){

        //READ
        while(1){
            sem_wait(&M->FullMutex);
            int FreeMutexValue;
            sem_getvalue(&M->FreeMutex, &FreeMutexValue);
            int FullMutexValue;
            sem_getvalue(&M->FullMutex, &FullMutexValue);
            printf("\nMutexes-Free: %d and Full: %d", FreeMutexValue, FullMutexValue);
            printf("\nBuffer Writing: %d", BufferCount);
            printf("\nReadPointer: %d", M->ReadPointer);
            printf("\nWritePointer: %d", M->WritePointer);
            fseek(FP, BufferCount*BufferSize, SEEK_SET);
            fwrite(&M->Data[M->ReadPointer*BufferSize], sizeof(unsigned char), BufferSize, FP);
            BufferCount++;
            M->ReadPointer=(M->ReadPointer+1)%NumberOfBuffers;
            if(M->FileEnding){
                fclose(FP);
                break;
            }
            sem_post(&M->FreeMutex);
        }
    }
    munmap(M,sizeof(MemData));
    close(SD);
    return 0;
}

Выход для Writer.c:

File Size: 50
Copyable File Size: 50
Number Of File Buffers: 25

Mutexes-Free: 9 and Full: 0
Buffer Writing: 0
Mutexes-Free: 8 and Full: 1
Buffer Writing: 1
Mutexes-Free: 7 and Full: 2
Buffer Writing: 2
Mutexes-Free: 6 and Full: 3
Buffer Writing: 3
Mutexes-Free: 5 and Full: 4
Buffer Writing: 4
Mutexes-Free: 4 and Full: 5
Buffer Writing: 5
Mutexes-Free: 3 and Full: 6
Buffer Writing: 6
Mutexes-Free: 2 and Full: 7
Buffer Writing: 7
Mutexes-Free: 1 and Full: 8
Buffer Writing: 8
Mutexes-Free: 0 and Full: 9
Buffer Writing: 9
Mutexes-Free: 0 and Full: 8
Buffer Writing: 10
Mutexes-Free: 2 and Full: 6
Buffer Writing: 11
Mutexes-Free: 1 and Full: 7
Buffer Writing: 12
Mutexes-Free: 0 and Full: 8
Buffer Writing: 13
Mutexes-Free: 0 and Full: 8
Buffer Writing: 14
Mutexes-Free: 0 and Full: 8
Buffer Writing: 15
Mutexes-Free: 0 and Full: 8
Buffer Writing: 16
Mutexes-Free: 1 and Full: 7
Buffer Writing: 17
Mutexes-Free: 0 and Full: 8
Buffer Writing: 18
Mutexes-Free: 0 and Full: 8
Buffer Writing: 19
Mutexes-Free: 0 and Full: 8
Buffer Writing: 20
Mutexes-Free: 0 and Full: 8
Buffer Writing: 21
Mutexes-Free: 0 and Full: 8
Buffer Writing: 22
Mutexes-Free: 0 and Full: 8
Buffer Writing: 23
Mutexes-Free: 0 and Full: 8
Buffer Writing: 24

Выход для Reader.c:

Mutexes-Free: 0 and Full: 9
Buffer Writing: 0
ReadPointer: 0
WritePointer: 0
Mutexes-Free: 1 and Full: 8
Buffer Writing: 1
ReadPointer: 1
WritePointer: 0
Mutexes-Free: 1 and Full: 8
Buffer Writing: 2
ReadPointer: 2
WritePointer: 1
Mutexes-Free: 2 and Full: 7
Buffer Writing: 3
ReadPointer: 3
WritePointer: 1
Mutexes-Free: 2 and Full: 6
Buffer Writing: 4
ReadPointer: 4
WritePointer: 2
Mutexes-Free: 1 and Full: 7
Buffer Writing: 5
ReadPointer: 5
WritePointer: 4
Mutexes-Free: 1 and Full: 8
Buffer Writing: 6
ReadPointer: 6
WritePointer: 5
Mutexes-Free: 1 and Full: 8
Buffer Writing: 7
ReadPointer: 7
WritePointer: 7
Mutexes-Free: 1 and Full: 8
Buffer Writing: 8
ReadPointer: 8
WritePointer: 7
Mutexes-Free: 1 and Full: 7
Buffer Writing: 9
ReadPointer: 9
WritePointer: 8
Mutexes-Free: 1 and Full: 8
Buffer Writing: 10
ReadPointer: 0
WritePointer: 9
Mutexes-Free: 1 and Full: 7
Buffer Writing: 11
ReadPointer: 1
WritePointer: 0
Mutexes-Free: 1 and Full: 8
Buffer Writing: 12
ReadPointer: 2
WritePointer: 1
Mutexes-Free: 1 and Full: 7
Buffer Writing: 13
ReadPointer: 3
WritePointer: 2
Mutexes-Free: 1 and Full: 8
Buffer Writing: 14
ReadPointer: 4
WritePointer: 3
Mutexes-Free: 1 and Full: 8
Buffer Writing: 15
ReadPointer: 5
WritePointer: 4

Входной файл (aaa.txt) содержитэти строки -

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

Выходной файл (bbb.txt) содержит эти строки -

1
2
3
4
5
6
7
8
9
10
11
12
13
14

Answer 1

У вас есть ряд небольших проблем и неправильное использование указателя, но даже исправляя их, ваш писатель / читатель по-прежнему не будет согласован при копировании файла aaa.txt в bbb.txt.Зачем?Логика мьютекса может принимать разные пути, когда вы читаете файл и записываете его в общую память, а затем извлекаете его из памяти.

Давайте сначала посмотрим на количество необходимых небольших исправлений.Если вы не предоставите typedef, вам нужно будет использовать struct MemData в качестве типа с sizeof, или лучше использовать сам указатель.Далее, strlen(M->Data) недопустимо и должно быть sizeof (M->Data)); при использовании с:

    memset (M->Data, '\0', sizeof (M->Data));

Далее у вас проблемы с указателем при использовании &Buf in:

memcpy(&M->Data[M->WritePointer*BufferSize], &Buf, sizeof(Buf)*sizeof(unsigned char));

Bufуже указатель и sizeof(unsigned char) == 1.Кроме того, если вы позже выделите Buf вместо использования VLA unsigned char Buf[BufferSize];, sizeof (Buf) будет sizeof (a_pointer).Лучше использовать:

        memcpy (&M->Data[M->WritePointer*BufferSize], Buf, BufferSize);

В Writer.c после того, как вы получили FileSize с StatBuf.st_size, нет необходимости вызывать ftell (FP), так как вы не переместили индикатор положения файла сfseek где-нибудь в этом коде.Аналогично в Writer.c нет необходимости вызывать fseek (FP, BufferCount*BufferSize, SEEK_SET);, так как ваше чтение будет продолжать перемещать индикатор вперед с каждым чтением.(у вас та же проблема с fseek() в Reader.c)

Дополнительно Buf - это неиспользуемая переменная в Reader.c.

Кроме того, вы не проверяете возврат fread (Buf, 1, BufferSize, FP); или fwrite (&M->Data[M->ReadPointer * BufferSize], 1, BufferSize, FP);.В то время как в случае файловых операций вы должны иметь конкурирующее чтение и запись Buffersize байтов, это не гарантируется, поэтому необходимо проверить возврат (особенно при чтении 2-байтовых блоков с odd (51 байт) входной файл).Там будет 1-байтовое чтение или запись где-нибудь, и если вы будете слепо писать 2-байтовые каждый раз, что-то не получится.

Что приводит нас к сути вашей проблемы при написании полногоbbb.txt (или запись большего количества байтов, чем в aaa.txt) с использованием M->FileEnding в качестве общего флага, указывающего конец ввода.Здесь есть проблема со счетом, которую я позволю вам разобраться.Он начинается, когда вы буферизуете 10 (чтение из файла и) пишет в Data перед началом чтения.Ваш Writer.c счетчик на 10 опережает счетчик на Reader.c до начала Reader.c.Вот почему на большинстве запусков у вас есть 16 итераций Reader, когда Writer устанавливает M->FileEnding, заставляя Reader записать первые 32 байта bbb.txt и выйти.(тогда при достаточном количестве запусков вы на самом деле будете писать Buffer Writing: 22 раз или Buffer Writing: 27 раз, что приведет к 58-байтному bbb.txt (со всеми записанными числами 1-20 и затем несколькими паразитными 0 с в конце).

Я предоставил ссылку. Рассмотрим 5 причин большого жира, почему мьютексы сосут большое время (2-я статья), чтобы помочь объяснить подводные камни, которые неуклонно охватывают все случаи в различных путях, которые может принять многопоточный код.у вас есть как можно больше основ логики, но все же могут быть проблемы, при которых буферы заполняются, когда и выгружаются, когда и когда установлен и распознан флаг M->FileEnding. Найдите время, чтобы вернуться к логике и убедиться, что вашFullMutex и FreeMutex логика делает то, что предназначено. У меня нет серебряной пули для вас там, на макушке, и потребуется отладка за итерацией, какая возможная комбинация может бытьнабор - который я оставляю вам.

Answer 2

MemData была изменена, чтобы отслеживать, сколько буферов было записано или прочитано.

Вот Writer.c

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <math.h>
#include <semaphore.h>
#include <fcntl.h>
#include <unistd.h>
#include <sys/stat.h>
#include <sys/mman.h>

int main(void){
    char FileName[128]="aaa.txt";
    struct MemData{
        sem_t FullMutex;
        sem_t FreeMutex;
        int ReadPointer;
        int WritePointer;
        int NumberOfFileBuffers;
        char Data[512000];//MEMORY BLOCK SIZE: 500 KB
    };
    int SD;
    struct MemData *M;
    int NumberOfBuffers=10;
    //int BufferSize=51200;//FILE BUFFER SIZE 50 KB
    int BufferSize=2;//EXPERIMENATION
    unsigned char Buf[BufferSize];
    int BufferCount=0;

    SD= shm_open("/program.shared", O_RDWR|O_CREAT, S_IREAD|S_IWRITE);
    if(SD< 0){
        printf("\nshm_open() error \n");
        return EXIT_FAILURE;
    }
    fchmod(SD, S_IRWXU|S_IRWXG|S_IRWXO);
    if(ftruncate(SD, sizeof(struct MemData))< 0){
        printf ("ftruncate() error \n");
        return EXIT_FAILURE;
    }
    M=(struct MemData*)mmap(NULL, sizeof(struct MemData), PROT_READ|PROT_WRITE, MAP_SHARED, SD, 0);
    if(M== MAP_FAILED){
        printf("mmap() error");
        return EXIT_FAILURE;
    }else{
        FILE *FP= fopen(FileName, "rb");
        if(FP!= NULL){
            struct stat StatBuf;
            if(stat(FileName, &StatBuf)==-1){
                printf("failed to fstat %s\n", FileName);
                exit(EXIT_FAILURE);
            }
            long long FileSize=StatBuf.st_size;
            printf("\nFile Size: %lld", FileSize);
            long long FilePosition=ftell(FP);
            FilePosition=ftell(FP);
            long long CopyableMemorySize=FileSize-FilePosition;
            printf("\nCopyable File Size: %lld", CopyableMemorySize);
            int NumberOfFileBuffers=ceil(CopyableMemorySize/BufferSize);
            printf("\nNumber Of File Buffers: %d\n", NumberOfFileBuffers);

            //INITIALIZATION
            sem_init(&M->FullMutex, 1, 0);
            sem_init(&M->FreeMutex, 1, NumberOfBuffers);
            M->ReadPointer=0;
            M->WritePointer=0;
            M->NumberOfFileBuffers=NumberOfFileBuffers;
            memset(M->Data, '\0', sizeof(M->Data));

            //WRITE
            while(1){
                sem_wait(&M->FreeMutex);
                fseek(FP, BufferCount*BufferSize, SEEK_SET);
                fread(Buf, sizeof(unsigned char), BufferSize, FP);
                int FreeMutexValue;
                sem_getvalue(&M->FreeMutex, &FreeMutexValue);
                int FullMutexValue;
                sem_getvalue(&M->FullMutex, &FullMutexValue);
                printf("\nMutexes-Free: %d and Full: %d\n", FreeMutexValue, FullMutexValue);
                printf("\nBuffer Writing: %d\n", BufferCount);
                memcpy(&M->Data[M->WritePointer*BufferSize], Buf, BufferSize);
                BufferCount++;
                M->WritePointer=(M->WritePointer+1)%NumberOfBuffers;
                sem_post(&M->FullMutex);                
                if(BufferCount==M->NumberOfFileBuffers){
                    fclose(FP);
                    break;
                }
                //sem_post(&M->FullMutex);                
            }
        }
        close(SD);
    }    
    return 0;
}

Вот Reader.c

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <semaphore.h>
#include <semaphore.h>
#include <fcntl.h>
#include <unistd.h>
#include <sys/stat.h>
#include <sys/mman.h>

int main(void){
    char FileName[128]="bbb.txt";
    struct MemData{
        sem_t FullMutex;
        sem_t FreeMutex;
        int ReadPointer;
        int WritePointer;
        int NumberOfFileBuffers;
        char Data[512000];//MEMORY BLOCK SIZE: 500 KB
    };
    int SD;
    struct MemData *M;
    int NumberOfBuffers=10;
    //int BufferSize=51200;//FILE BUFFER SIZE 50 KB
    int BufferSize=2;//EXPERIMENATION
    //unsigned char Buf[BufferSize];
    int BufferCount=0;
    SD= shm_open("/program.shared", O_RDWR|O_CREAT, S_IREAD|S_IWRITE);
    if(SD< 0){
        printf("\nshm_open() error \n"); 
        return EXIT_FAILURE;
    }
    M=(struct MemData*)mmap(NULL, sizeof(struct MemData), PROT_READ|PROT_WRITE, MAP_SHARED, SD, 0);
    if(M== MAP_FAILED){
        printf("mmap() error");
        return EXIT_FAILURE;
    }else{
        FILE *FP= fopen(FileName, "wb");
        if(FP!= NULL){

            //READ
            while(1){
                sem_wait(&M->FullMutex);
                int FreeMutexValue;
                sem_getvalue(&M->FreeMutex, &FreeMutexValue);
                int FullMutexValue;
                sem_getvalue(&M->FullMutex, &FullMutexValue);
                printf("\nMutexes-Free: %d and Full: %d", FreeMutexValue, FullMutexValue);
                printf("\nBuffer Writing: %d", BufferCount);
                printf("\nReadPointer: %d and WritePointer: %d", M->ReadPointer, M->WritePointer);
                printf("\nBuffer Writing: %d\n", BufferCount);
                fseek(FP, BufferCount*BufferSize, SEEK_SET);
                fwrite(&M->Data[M->ReadPointer*BufferSize], sizeof(unsigned char), BufferSize, FP);
                BufferCount++;
                M->ReadPointer=(M->ReadPointer+1)%NumberOfBuffers;
                sem_post(&M->FreeMutex);
                if(BufferCount==M->NumberOfFileBuffers){
                    fclose(FP);
                    break;
                }
                //sem_post(&M->FreeMutex);
            }
        }
        munmap(M,sizeof(struct MemData));
        close(SD);
    }
    return 0;
}