ring_buffer.c

来自「this code is about the doubly link list 」· C语言 代码 · 共 215 行

#include <stdio.h>
#include <stdlib.h>
#include <errno.h> 
#include <pthread.h>
#include <string.h>

/*
 * The program reads from a text file, each of the threads takes that message
 * and writes it to the Ring Buffer after attaching its thread-id to it. Finally,
 * the Ring Buffer is dumped to the file out.txt. The command line argument is the name of
 * the input message file.
 */
 
struct ring_buffer
{
	char * write_pointer;
	char * buff_start;
	int wrapped;
	unsigned long size;
};

/** Function Prototypes **/
void init_buffer(unsigned long);
int load_data(void*, unsigned int*);
void print_buffer_state();
char* mem_allocate(unsigned long);
int dump_buffer(const char*);

/** Global Variables **/
struct ring_buffer global_ring;
pthread_mutex_t mutex;
char file_contents[300000]; 
  
/** Fucntions **/
char* mem_allocate(unsigned long mem_size)
{
    char* data;
    data = (char*)calloc(mem_size, sizeof(unsigned char));
    if(data == NULL)
    {
        perror( "calloc" );
        exit(1);
    }
    return data;
}

void init_buffer(unsigned long mem_size)
{
    if(pthread_mutex_init(&mutex, 0) != 0)
    {
        perror("pthread_mutex_init");
        exit(1);
    }
    global_ring.size = mem_size;
    global_ring.buff_start = mem_allocate(mem_size);
    global_ring.write_pointer = global_ring.buff_start;
    global_ring.wrapped = 0;
}

int load_data(void* data, unsigned int* len)
{
    int temp = 0;
    
    if(len == NULL || *len == 0)
    {
        return -1;
    }
    
    if(*len > global_ring.size)
    {
        *len = global_ring.size;
    }
    
    if(pthread_mutex_lock(&mutex) != 0)
    {
        perror("pthread_mutex_lock");
        exit(1);        
    }    
    temp = (global_ring.buff_start + global_ring.size) - global_ring.write_pointer;
    if(*len > temp)
    {
        memcpy(global_ring.write_pointer, data, temp);
        memcpy(global_ring.buff_start, (char*)data+temp, *len-temp);
        global_ring.write_pointer = global_ring.buff_start + *len-temp;
        global_ring.wrapped = 1;
    }
    else
    {
        memcpy(global_ring.write_pointer, data, *len);
        global_ring.write_pointer += *len;
    }
        
    if(pthread_mutex_unlock(&mutex) != 0)
    {
        perror("pthread_mutex_lock");
        exit(1);        
    }
    return 0; 
}

int dump_buffer(const char* file_to_dump)
{
    FILE* fp ;
    int temp = 0;
    int len;
  
    if ((fp = fopen(file_to_dump, "a")) == NULL)
        return -1;
       
    pthread_mutex_lock(&mutex);
          
    if(global_ring.wrapped == 1)
    {
        fwrite( global_ring.buff_start, sizeof(unsigned char), global_ring.size, fp);
    }
    else
    {
        len = global_ring.write_pointer - global_ring.buff_start + 1;
        fwrite(global_ring.buff_start, sizeof(unsigned char), len, fp);   
    }
    fclose(fp);
    pthread_mutex_unlock(&mutex);
    return 0;
}

void print_buffer_state()
{
    printf("\n/***** Printing the buffer before dump****/\n");
    printf("Buffer data from Start pointer:\n%s\n", global_ring.buff_start);
    printf("/******************************/\n\n");
}

int read_input(char* filename)
{
    FILE *inputFilePtr;
    char *iReturn;
    inputFilePtr = fopen(filename, "r");
    if(inputFilePtr == NULL)
    {
        printf("ERROR: File Not Found\n");
        return -1;
    }
    iReturn = fgets(file_contents, 209600, inputFilePtr);
    
    if (iReturn == NULL) /* End of file reached */
        return -1;
    return 0;
}

void* loader_function(void* a)
{
    unsigned int len, current_thread, old_state = 0;
    char* local_msg;
    
    current_thread = pthread_self();
    local_msg = (char*)calloc(sizeof(file_contents)+32 , sizeof(unsigned char));    
      
    if(sprintf(local_msg, "%s:%d", file_contents, current_thread) <0)
    {
        printf(":(\n");
    }
    printf("Message loaded by thread %d to the buffer = \"%s\"\n", pthread_self(), local_msg);
    len = strlen(local_msg);
    load_data(local_msg, &len);
    pthread_exit(0);
}

void threaded_test(char* filename)
{
    pthread_t threads[2];
    int i,j, ret,toDump = 0;
    
    int data_read = read_input(filename);
    
    if(data_read < 0)
        exit(1);
    printf("FILE_CONTENTS = %s\n", file_contents);
    
    if(pthread_mutex_init(&mutex, 0) != 0)
    {
        perror("pthread_mutex_init");
        exit(1);
    }
    init_buffer(20);
    
    /***** Threads Created for to load the data *****/
    
    for(j = 0; j<2; j++)
    {
        pthread_create (&threads[j], NULL, &loader_function, NULL);
    }
    
    for(j = 0; j<2; j++)
    {
        pthread_join(threads[j], NULL);
    }
   
    print_buffer_state();
    dump_buffer("out.txt");
    printf("Buffer dumped to file \"out.txt\"\n\n");
}

int main(int argc, char** argv)
{
    if(argc!=2)
    {
        printf("Usage: ./test <input-file>\n");
        exit(1);
    }
    else
    {
        threaded_test(argv[1]);
    }
}