Thread Prog in C language (OS independent)

kumars · #1 08-27-2008, 11:48 AM

Dear All,

I am desparately looking for a C prog (OS independent), could any one of you please provide me the program....

Write a program to read from and write into shared memory using threads while satisfying the following conditions:

a) create number of worker threads and a master thread.

b) Master thread should have control on reader and writer threads.

c) Writer thread (worker thread) should return control to the master thread after writing into shared memory.

d) Now the master thread has control, so it should be able to read or write.

e) Reader thread (worker thread) should return control to the master thread after reading from shared memory.

f) Now the master thread has control, so it should be able to read or write.

xpi0t0s · #2 09-18-2008, 05:45 PM

It's easy enough; where are you stuck? Post the code, explain the problem you have and we'll help you solve it.

Or are you just after someone to write the program for you and you're not interested in doing any of it yourself? In which case my consultancy rates are £100/hour, minimum 4 hours, and I'll be happy to write it for you.

kumars · #3 09-19-2008, 04:28 AM

Could you please help me in finding out the reason behind segmentation fault. This program is working fine as long as the command line input is as follows:

./test -r 300 -w 3 -s 100 (gives desired o/p)
./test -r 100 -w 3 -s 200 (gives desired o/p)
./test -r 10 -w 3 -s 10 (gives desired o/p)

But at the same time this program fails the moment I am trying to give following command line input

./test -r 3 -w 4 -s 100 (seg fault)
./test -r 1 -w 4 -s 10 (seg fault)

I have tried my level best to resolve this and to find out the reason behind failing. But could not able to find the solution. While using dbx I came to know the point where it's failing but unfortunately I don't know how to resolve it.

I have mentioned the dbx output below to have a look at it and it may save your time to help me in solving this problem.

midibm14:subh:/home/users/subh/demo #dbx test
Type 'help' for help.
reading symbolic information ...
(dbx) stop in main
[1] stop in main
(dbx) r -r 300 -w 3 -s 100
[1] stopped in main at line 14 ($t1)
14 parse_args(argc, argv);
(dbx) c
test: Shared Memory Test Program

Number of writers thread = 3
Number of readers thread = 300
Shared Memory Size in Bytes = 100

execution completed
(dbx) q
midibm14:subh:/home/users/subh/demo #dbx test
Type 'help' for help.
reading symbolic information ...
(dbx) stop in main
[1] stop in main
(dbx) r -r 1 -w 4 -s 100
[1] stopped in main at line 14 ($t1)
14 parse_args(argc, argv);
(dbx) cont
test: Shared Memory Test Program

Number of writers thread = 4
Number of readers thread = 1
Shared Memory Size in Bytes = 100

Shmat Failed Here *** : Shared Memory Attach Error : Too many open files

execution completed
(dbx)

It's failing when I tried to increase the "-w option" in the 2nd execution.

Let me summarise the problem once again. I can run this program with the writer thread ("-w option") less than or equal to 3. But I can't run this program with the writer thread ("-w option") 4 or more.

May I request you to look into the problem and please provide me the solution.

Code:

/*********************************************/
/****************** head.h *********************/
/*********************************************/

#include <stdio.h>
#include <errno.h>
#include <string.h>
#include <stdlib.h>
#include <unistd.h>
#include <pthread.h>
#include <sys/ipc.h>
#include <sys/shm.h>
#include <sys/sem.h>
#include <sys/wait.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <sys/limits.h>
#include <sys/signal.h>


/********************* User Defined Data ***************/

#define MAX_THREAD_NUMBER       30000
#define MAX_WRITER_NUMBER       100
#define MAX_READER_NUMBER       999
#define DEFAULT_NUM_READERS     2
#define DEFAULT_NUM_WRITERS     2
#define SHMEM_MODE              (SHM_R | SHM_W)
/*#define DEFAULT_SHMEM_SIZE    2000000000*/
#define DEFAULT_SHMEM_SIZE      200
#define MB                      (1024*1024)
#define MAX_SHMEM_NUMBER        100




#define USAGE   "\nUsage: %s [-r num_readers] [-w num_writers] [-s shmem_size]\n
\n" \
                "\t-r num_readers  number of reader thread to create\n" \
                "\t-w num_writers  number of writer thread to create\n" \
                "\t-s buffer_size  size of shmem segments in bytes\n"   \
                "\t                must be less than 256MB\n\n"



/************ Global Functions ****************/

void parse_args(int , char **);
void *reader(void *);
void *writer(void *);
void error(const char *, int);
void release();
/*int sys_error(const char *, int);*/


/************ Global Variables ****************/

pthread_t *writer_th;
pthread_t *reader_th;


int              thread_hold[MAX_WRITER_NUMBER];
pthread_mutex_t  cond_mutex[MAX_WRITER_NUMBER];
pthread_mutex_t  mutex_r[MAX_WRITER_NUMBER];
pthread_cond_t   cond_var[MAX_WRITER_NUMBER];


int           *read_count[MAX_WRITER_NUMBER];
char          *shmptr[MAX_READER_NUMBER];
unsigned long *checksum[MAX_WRITER_NUMBER];
unsigned long cksum[MAX_WRITER_NUMBER];


int      shmem_size = DEFAULT_SHMEM_SIZE;
pid_t    parent_pid;


int      num_readers = DEFAULT_NUM_READERS;
int      buffer_size = DEFAULT_SHMEM_SIZE;
int      num_writers = DEFAULT_NUM_WRITERS;
int      shmid[MAX_THREAD_NUMBER + MAX_WRITER_NUMBER];

Code:

/*********************************************/
/****************** test.c *********************/
/*********************************************/

#include "head.h"


int main(int argc, char **argv)
{
        pthread_attr_t  newattr;

        int    i, j, k;
        size_t Size;

        unsigned long *ulptr;


        parse_args(argc, argv);


        printf("%s: Shared Memory Test Program\n\n", *argv);

        printf("\tNumber of  writers thread   = %d\n", num_writers);
        printf("\tNumber of  readers thread   = %d\n", num_readers);
        printf("\tShared Memory Size in Bytes = %d\n", buffer_size);
        printf("\n");



        for(i = 0; i < num_writers; i++) {
        j = i * 3;

        Size = sizeof(int);

        if((shmid[j] = shmget(IPC_PRIVATE, Size, SHMEM_MODE)) < 0)
                perror("Shmget Error");


        if((long)(read_count[i] = (int *) shmat(shmid[j], 0, 0)) == -1)
                perror("Shmat Error");


        *(read_count[i]) = 0;
        j++;
        Size = sizeof(unsigned long) * num_readers;

        if((shmid[j] = shmget(IPC_PRIVATE, Size, SHMEM_MODE)) < 0)
                perror("Checksum Shmget Failed");


        if((long)(checksum[i] = (unsigned long *) \
                                         shmat(shmid[j], 0, 0)) == -1)
                perror("Shmat Failed : Shared Memory Attach Error ");


        ulptr = checksum[i];

        for(k = 0; k < num_readers; k++)
        {
                *ulptr = 0;
                ulptr++;
        }

        Size = buffer_size;
        j++;

        if((shmid[j] = shmget(IPC_PRIVATE, Size, SHMEM_MODE)) < 0)
                perror("Shmptr Shmget Failed");


        if((long)(shmptr[i] = shmat(shmid[j], 0, 0)) == -1)
                perror("Shmat Failed Here *** : Shared Memory Attach Error ");

        }

}


void parse_args(int argc, char **argv)
{
        int         i;
        int         errflag = 0;
        char        *program_name = *argv;
        extern char *optarg;


        while((i = getopt(argc, argv, "r:s:w:?")) != EOF) {
                switch (i) {
                        case 'r':
                                num_readers = atoi(optarg);
                                break;
                        case 's':
                                buffer_size = atoi(optarg);
                                break;
                        case 'w':
                                num_writers = atoi(optarg);
                                break;
                        case '?':
                                errflag++;
                                break;
                }
        }

        if(num_writers >= MAX_WRITER_NUMBER) {
                errflag++;
                fprintf(stderr, "ERROR: num_writers must be less\
                                       than %d\n", MAX_WRITER_NUMBER);
        }

        if(num_readers >= MAX_READER_NUMBER) {
                errflag++;
                fprintf(stderr, "ERROR: num_readers must be less\
                                        than %d\n", MAX_READER_NUMBER);
        }

        i = num_readers * num_writers;

        if(i >= MAX_THREAD_NUMBER) {
                errflag++;
                fprintf(stderr, "ERROR: maximun threads number\
                         must be less than %d\n", MAX_THREAD_NUMBER);
        }

        if(errflag) {
                fprintf(stderr, USAGE, program_name);
                exit (2);
        }
}

xpi0t0s · #4 09-19-2008, 01:36 PM

Dunno about this shm stuff I'm afraid.

Used dummies to replace the shm stuff with malloc viz:
int shmget(int,size_t sz,int) { return (int)sz; }
void *shmat(int sz,int,int) { return malloc(sz); }

Got a load of warnings about pointer truncations... Are you getting warnings that you're ignoring? If not try compiling with -w4 (I think that switches full pedantic mode on, or is it --pedantic? anyway, the flag'll be in the docs.)

Fixed the warnings and the program ran just fine (used VS2005 debugger which spots most things).

Perhaps someone who knows about shm can help.

kumars · #5 09-19-2008, 05:06 PM

Here is the complete program to avoid any errors and warnings. As per the requirement I have to create 3 shared segments per writer thread. When I am doing it for 3 writer threads, it runs fine. But when I am trying to create it for 4 or more writer threads (again 3 segments for each writer i.e 12 shared segments, it fails).
Could you please have a look at it and let me know where I am doing wrong....

Code:

/****************************************************/
/********************** header.h ********************/
/****************************************************/
/****************************************************/
/********************** header.h ********************/
/****************************************************/

#include <stdio.h>
#include <errno.h>
#include <string.h>
#include <stdlib.h>
#include <unistd.h>
#include <pthread.h>
#include <sys/ipc.h>
#include <sys/shm.h>
#include <sys/sem.h>
#include <sys/wait.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <sys/signal.h>


/********************* User Defined Data ***************/

#define MAX_THREAD_NUMBER       5000
#define MAX_WRITER_NUMBER       100
#define MAX_READER_NUMBER       400
#define DEFAULT_NUM_READERS     2
#define DEFAULT_NUM_WRITERS     2
#define SHMEM_MODE              (SHM_R | SHM_W)
#define DEFAULT_SHMEM_SIZE      20000
#define MB                      (1024*1024)
#define MAX_SHMEM_NUMBER        11



#define USAGE   "\nUsage: %s [-r num_readers] [-w num_writers] [-s shmem_size]\n
\n" \
                "\t-r num_readers  number of reader thread to create\n" \
                "\t-w num_writers  number of writer thread to create\n" \
                "\t-s buffer_size  size of shmem segments in bytes\n"   \
                "\t                must be less than 256MB\n\n"



/************ Global Functions ****************/

int  parse_args(int , char **);
int *reader(void *);
int *writer(void *);
int  error(const char *, int);
int  release();
/*int sys_error(const char *, int);*/


/************ Global Variables ****************/

pthread_t *writer_th;
pthread_t *reader_th;


int              thread_hold[MAX_WRITER_NUMBER];
pthread_mutex_t  cond_mutex[MAX_WRITER_NUMBER];
pthread_mutex_t  mutex_r[MAX_WRITER_NUMBER];
pthread_cond_t   cond_var[MAX_WRITER_NUMBER];


int           *read_count[MAX_WRITER_NUMBER];
char          *shmptr[MAX_WRITER_NUMBER];
unsigned long *checksum[MAX_WRITER_NUMBER];
unsigned long cksum[MAX_WRITER_NUMBER];


int      shmem_size = DEFAULT_SHMEM_SIZE;
pid_t    parent_pid;


int      num_readers = DEFAULT_NUM_READERS;
int      buffer_size = DEFAULT_SHMEM_SIZE;
int      num_writers = DEFAULT_NUM_WRITERS;
int      shmid[MAX_THREAD_NUMBER + MAX_WRITER_NUMBER];



/****************************************************/
/********************** main.c **********************/
/****************************************************/
#include "header.h"

int main(int argc, char **argv)
{
        pthread_attr_t  newattr;

        int    i, j, k;
        size_t Size;
        unsigned long *ulptr;

        parse_args(argc, argv);
 
        printf("%s: Shared Memory Test Program\n\n", *argv);
        printf("\tNumber of  writers thread   = %d\n", num_writers);
        printf("\tNumber of  readers thread   = %d\n", num_readers);
        printf("\tShared Memory Size in Bytes = %d\n", buffer_size);
        printf("\n");

        for(i = 0; i < num_writers; i++) {
        j = i * 3;

        Size = sizeof(int);

        if((shmid[j] = shmget(IPC_PRIVATE, Size, SHMEM_MODE)) < 0)
                perror("Shmget Error");

        if((long)(read_count[i] = (int *) shmat(shmid[j], 0, 0)) == -1)
                perror("Shmat Error");

        *(read_count[i]) = 0;
        j++;
        Size = sizeof(unsigned long) * num_readers;

        if((shmid[j] = shmget(IPC_PRIVATE, Size, SHMEM_MODE)) < 0)
                perror("Checksum Shmget Failed");

        if((long)(checksum[i] = (unsigned long *) \
                                         shmat(shmid[j], 0, 0)) == -1)
                perror("Shmat Failed : Shared Memory Attach Error ");

        ulptr = checksum[i];

        for(k = 0; k < num_readers; k++)
        {
                *ulptr = 0;
                ulptr++;
        }
        Size = buffer_size;
        j++;

        if((shmid[j] = shmget(IPC_PRIVATE, Size, SHMEM_MODE)) < 0)
                perror("Shmptr Shmget Failed");

        if((long)(shmptr[i] = shmat(shmid[j], 0, 0)) == -1)
                perror("Shmat Failed Here *** : Shared Memory Attach Error ");

        }

        writer_th = (pthread_t *) malloc((size_t) (num_writers \
                                                   * sizeof(pthread_t)));
        reader_th = (pthread_t *) malloc((size_t) (num_writers \
                                     * num_readers * sizeof(pthread_t)));

        for(i = 0; i < num_writers; i++) {

        if(pthread_mutex_init(&mutex_r[i] , (pthread_mutexattr_t *)NULL) != 0)
                perror("Can't Initialise Mutex");

        if(pthread_mutex_init (&cond_mutex[i], (pthread_mutexattr_t *)NULL)!=0)
                perror("Can't Initialize Cond_Mutex");

        if(pthread_cond_init (&cond_var[i], (pthread_condattr_t *)NULL)!=0)
                perror("Cond_Var Failed");

        thread_hold[i] = 1;
        }
        if(pthread_attr_init(&newattr))
               perror("Attr_Init Failed");

        if(pthread_attr_setdetachstate(&newattr, PTHREAD_CREATE_UNDETACHED))
                perror("Attr_Setdetachstate Failed");

        for(i = 0; i < num_writers; i++)
        {

                if(pthread_create(&writer_th[i], &newattr, writer, \
                                                        (void *) (long)i))
                        perror("Writer Failed");

                k = i * num_readers;
                for(j = k; j < (k + num_readers) ; j++)
                {
                        if(pthread_create(&reader_th[j], &newattr, reader,\
                                                        (void *) (long)j))

                        perror("Reader Failed");
                }
        }
        for(i = 0; i < num_writers; i++)
        {
               if(pthread_join(writer_th[i], NULL)) {
                        printf("writer_th: pthread_join return: %d\n",i);

                        perror("Pthread_join Bad Status");
                }

                k = i * num_readers;

                for(j = k; j < (k + num_readers) ; j++)
                {
                        if(pthread_join(reader_th[j], NULL)) {
                        perror("Pthread_join Bad Status");
                        }
                }
        }

        for(i = 0; i < num_writers; i++)
        {
                ulptr = checksum[i];
                for(j = 0; j < num_readers; j++) {

                        if(cksum[i] != *ulptr )
                                error("checksums do not match", __LINE__);

                }
        }


        printf("\n\tMain: Readers calculated segment successfully\n");

        release();
        printf("\nsuccessful!\n");

        return(0);
}

The other 3 functions (files) continues in the next post. I am not allowed to write more than 10000 characters. I don't know why...????

kumars · #6 09-19-2008, 05:08 PM

Code:

/****************************************************/
/******************** reader.c **********************/
/****************************************************/

#include "header.h"

void *reader(void *parm)
{
        int num_p = (int) (long)parm;
        unsigned long cksum_r = 0;
        int     i, num_r, num_w;
        char    *ptr;
        unsigned long *ulptr_r;


        num_r = num_p % num_readers;
        num_w = num_p - num_r;
        num_w = num_w / num_readers;
        ptr = shmptr[num_w];
        ulptr_r = checksum[num_w];

        if(pthread_mutex_lock (&cond_mutex[num_w]))
                perror("Can't Take Cond Lock");

        while(thread_hold[num_w])
        {
                if(pthread_cond_wait(&cond_var[num_w], &cond_mutex[num_w]))
                        perror("cond_wait failed");
        }
        if(pthread_mutex_unlock(&cond_mutex[num_w]))
                perror("Release cond Lock Failed");

        if(pthread_mutex_lock(&mutex_r[num_w]))
                perror("Can't take read Lock");

        (*(read_count [num_w]))++;

        if(pthread_mutex_unlock(&mutex_r[num_w]))
                perror("Can't Release Read Lock");

        for(i = 0; i < buffer_size; i++)
                cksum_r += *ptr++;

        if(pthread_mutex_lock(&mutex_r[num_w]))
                perror("Can't Take Read Lock");

        (*(read_count[num_w]))--;

        if(pthread_mutex_unlock(&mutex_r[num_w]))
                perror("Can't Release 1 Read Lock");

        *ulptr_r = cksum_r;

        printf("\tReader (%d) of Writer (%d): checksum %04ld\n",\
                                                num_r,num_w,cksum_r);

        return NULL;
}

/****************************************************/
/******************** writer.c **********************/
/****************************************************/

#include "header.h"

void *writer(void *parm)
{
        int num_w = (int) (long)parm;
        unsigned long cksum_w = 0;
        char  data = 0;
        char  *ptr;
        int count = 0;

        data = num_w;

        for(ptr = shmptr[num_w]; ptr < (shmptr[num_w]+buffer_size); ptr++) {
                *ptr = data++;
                cksum_w += *ptr;

        }

        if(pthread_mutex_lock(&cond_mutex[num_w]))
                perror("Mutex_lock Failed");

        thread_hold[num_w] = 0;

        if(pthread_cond_broadcast(&cond_var[num_w]))
                perror("cond_signal Failed");

        if(pthread_mutex_unlock(&cond_mutex[num_w]))
                perror("Mutex_unlock Failed");

        cksum[num_w] = cksum_w;
        printf("\n\tWriter (%d): shared memory checksum %04ld\n", \
                                                         num_w, cksum_w);
        return NULL;
}

/****************************************************/
/******************** parseargs.c *******************/
/****************************************************/

#include "header.h"

int parse_args(int argc, char **argv)
{
        int         i;
        int         errflag = 0;
        char        *program_name = *argv;
        extern char *optarg;

        while((i = getopt(argc, argv, "r:s:w:?")) != EOF) {
                switch (i) {
                        case 'r':
                                num_readers = atoi(optarg);
                                break;
                        case 's':
                                buffer_size = atoi(optarg);
                                break;
                        case 'w':
                                num_writers = atoi(optarg);
                                break;
                        case '?':
                                errflag++;
                                break;
                }
        }
        if(num_writers >= MAX_WRITER_NUMBER) {
                errflag++;
                fprintf(stderr, "ERROR: num_writers must be less\
                                       than %d\n", MAX_WRITER_NUMBER);
        }
        if(num_readers >= MAX_READER_NUMBER) {
                errflag++;
                fprintf(stderr, "ERROR: num_readers must be less\
                                        than %d\n", MAX_READER_NUMBER);
        }
        i = num_readers * num_writers;

        if(i >= MAX_THREAD_NUMBER) {
                errflag++;
                fprintf(stderr, "ERROR: maximun threads number\
                         must be less than %d\n", MAX_THREAD_NUMBER);
        }

        if(errflag) {
                fprintf(stderr, USAGE, program_name);
                exit (2);
        }
}

int release()
{
        int i, j;

        for(i = 0; i < num_writers; i++) {
        if(pthread_mutex_destroy(&cond_mutex[i]) != 0)
                perror("Can't destroy cond_mutex");

        if(pthread_mutex_destroy(&mutex_r[i]) != 0)
                perror("Can't destroy mutex_r");
        }

        for(i = 0; i < num_writers; i++) {

        j = i * 3;
        if(shmctl(shmid[j], IPC_RMID, 0) < 0)
                perror("Read_count shmctl Failed");
        j++;
        if(shmctl(shmid[j], IPC_RMID, 0) < 0)
                perror("checksum shmctl failed");
        j++;
        if(shmctl(shmid[j], IPC_RMID, 0) < 0)
                perror("shmptr shmctl failed");

        }
}

int sys_error(const char *msg, int line)
{
        char syserr_msg [256];

        sprintf(syserr_msg, "%s: %s\n", msg, strerror (errno));
        error(syserr_msg, line);
}

int error(const char *msg, int line)
{
        fprintf(stderr, "ERROR [line: %d] %s\n", line, msg);
        if(line>= 260)
        release();
        exit(-1);
}

The remaining 3 files are here....

xpi0t0s · #7 09-20-2008, 01:49 PM

It's not OS independent by the way. Doesn't work on Windows in VS2005.
(Maybe it works in some other compiler.)

xpi0t0s · #8 09-20-2008, 01:51 PM

I guess you can't post more than 10K because that's a heck of a lot of code to ask someone to go through. I don't see why you posted the whole lot anyway, if the problem isn't solved in the first part then it's not likely to help to post the rest of it.

kumars · #9 09-21-2008, 05:40 AM

The code which I posted definitely is not going to compile on window. It requires minimal changes that I can send you on Monday and I think you'll be able to compile it on MS-VS 2005 also.

kumars · #10 10-03-2008, 05:24 AM

Dear All,

Any Clue on this, I need to implement the features mentioned below in my program (posted in prev message)

Imagining that there are four threads - 2 writers and 2 readers and we have one shared memory
segment with one block in it.

All threads are contending for the same block of shared memory.
The Writers will write just one message each and the readers will attempt to read, process and respond.
A typical run would look like this :-

1.Writer 1 & Writer 2 read in a unique buffer from two different files and store these in malloced blocks of memory.

2. Writer 1 contends for the shared memory segment with the other 3 threads and if he get it - locks the segment.

3. Writer 1 writes the message to the segment along with a header defining the length of the buffer
he has written,his thread ID, and a flag showing it is an 'inbound' operation and then appends the
buffer he has read from file.
He then unlocks the segment. And immediately starts polling to get back a response.

4. Reader 1 or Reader 2 contend to read from shared memory. Reader 1 wins the race and locks the block

5. Reader 1 reads in the buffer, parses the header to make sure it is a 'inbound' request,
IF it is a 'inbound' requests he reads in the message as defined by length field in the header
then xors the buffer or rot13's the buffer and moves to point 6.
ELSE He unlocks the buffer because it's not an inbound request.

6. Reader 1 then clears that block of shared memory and writes back a header which has in it
The threadid of the sender, the length of the xor'd or rot13'd buffer and a flag indicating it is a 'outbound' operation.
He the writes the buffer onto the back of this and unlocks the shared memory block.

7. Writer 1 is polling for a response. He wins the race with the other 3 threads and locks the block.
He peaks into the shared memory and reads the header.
IF the operation is 'outbound' i.e from a reader thread and the threadid is his then he moves to point 8.
ELSE he unlocks the blocks because it is not a response for him and or it is an 'inbound' request from Writer 2.

8. Writer 1 xors or rot13's the response message and then compares it with the message he sent which he stored in
malloced memory in point 1.
IF it matches he signals success by writing a suitable message to standard out.
IF it doesn't he writes the block to a error file along with the block he sent and exits the process witn a suitable error code and
error message to standard error.

When both writer 1 and writer 2 have written their message, got back a response and check it is correct.

Of course all of this is happening at once. So writer 1 and writer 2 are both trying to lock the blocks and send 'inbound' messages
and readers 1 and readers 2 are simultaneously trying to lock the same block to read from it, process the message and respond.