Producer/Consumer - problems creating threads

[chitvan1988]

#include<stdio.h>

#include<stdlib.h>
#include <sys/un.h>
#include <netinet/in.h>
#include<string.h>

#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/sem.h>
#include<sys/shm.h>

#include<pthread.h>

#include <sys/types.h>          /* See NOTES */
#include <sys/socket.h>
#include<unistd.h>



int newsockfd;


//variables for shared and semaphore

int sem_id,shmid;

int mysemid[100];

void *sharedptr;

struct mystruct

{

	sem_t mysem;
	pthread_t tid;

	long int count;

	int flag;

};


union semun

{

	int val; /* Value for SETVAL */

	struct semid_ds *buf; /* Buffer for IPC_STAT, IPC_SET */

	unsigned short *array; /* Array for GETALL, SETALL */

	struct seminfo *__buf; /* Buffer for IPC_INFO

	(Linux specific) */

}sem_union;


//end of special purpose variables

void *mythreadfunc(void *arg)

{

	int n;

	int segment;

	char buffer[4];

	memset(buffer,0,sizeof(buffer));

	printf("going to read");

	fflush(stdout);

	while( n= read(newsockfd, buffer, 4))

	{

	

		segment=atoi(buffer);

		memset(buffer,0,sizeof(buffer));

		struct sembuf sem_b;

		/*sem_b.sem_num=0;

		sem_b.sem_op=-1;

		sem_b.sem_flg=SEM_UNDO;*/
		

//		printf("inside server--received segment-->%d--data is-->",segment);
				printf("inside server--received segment-->%d\n>",segment);
				fflush(stdout);
		/*if(semop(mysemid[segment],&sem_b,1)==-1)

		{

			fprintf(stderr,"semaphore lock failed");
			sleep(10);
			continue;

		}*/
		printf("threadid-->%d---count-->%ld\n",(int)((((struct mystruct *)sharedptr)+segment)->tid),(((struct mystruct *)sharedptr)+segment)->count);
		

		//(((struct mystruct *)sharedptr)+segment)->flag=0;


		/*sem_b.sem_num=0;

		sem_b.sem_op=1;

		sem_b.sem_flg=SEM_UNDO;



	if(semop(mysemid[segment],&sem_b,1)==-1)

	{	

	

			fprintf(stderr,"semaphore unlock failed");
			sleep(100);

	}*/

	}

}


int main(int argc,char **argv)

{



	int sockfd, portno, clilen;

	struct sockaddr_in serv_addr, cli_addr;

	int n;

	if (argc < 2)

	{

		fprintf(stderr,"ERROR, no port provided\n");

		exit(1);

	}

	sockfd = socket(PF_INET, SOCK_STREAM, 0);

	if (sockfd < 0)
	{

		perror("ERROR creating socket");
		exit(0);
	}

	
	memset( &serv_addr,0, sizeof(serv_addr));

	portno = atoi(argv[1]);

	
	serv_addr.sin_family =AF_INET;

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

		perror("ERROR on binding");
		exit(0);
	}


	listen(sockfd,50);

	clilen = sizeof(cli_addr);


	int i=0;

	for ( i=0;i<100 ;i++ )

	{

		if(i==0)

			mysemid[i]=semget((key_t)102,1,0666|IPC_CREAT);

		else

			mysemid[i]=semget((key_t)i,1,0666|IPC_CREAT);

		if(mysemid[i]<0)

		{

			perror("semid error occured");

			exit(1);

		}

	
		sem_union.val=1;

		if((semctl(mysemid[i],0,SETVAL,sem_union))<0)

		{

			perror("semid error occured initioalization");

			exit(1);

		}

	}//end for

	
	key_t key = 1234;

	shmid=shmget(key,100*sizeof(struct mystruct),0644|IPC_CREAT);

	
	if(shmid==-1)

	printf("shmid error");

	
	sharedptr=shmat(shmid,(void*)0,0);

	if(sharedptr==(void *)-1)

	{

		fprintf(stderr,"shared memory error");

		fflush(stderr);
		exit(0);

	}

	memset(sharedptr,0,100*sizeof(struct mystruct));
	sem_init(&((((struct mystruct *)sharedptr)+5)->mysem), 0, 1);

	while(1)

	{

		newsockfd = accept(sockfd,(struct sockaddr *) &cli_addr,&clilen);

		printf("connection accepted");

		fflush(stdout);


		if (newsockfd < 0)
		{

			perror("ERROR on accept");
			exit(0);
		}


		printf("connection accepted");

		fflush(stdout);

		
		pthread_t thread_ID;

		int value;	

		pthread_create(&thread_ID , NULL, mythreadfunc , &value ) ;

	}


	for(i=0;i<100;i++)

	{

		semctl(mysemid[i],0,IPC_RMID);

	}

	return 0;

} //end main

[chitvan1988]

server-->

#include<stdio.h>

#include<stdlib.h>
#include <sys/un.h>
#include <netinet/in.h>
#include<string.h>

#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/sem.h>
#include<sys/shm.h>

#include<pthread.h>

#include <sys/types.h>          /* See NOTES */
#include <sys/socket.h>
#include<unistd.h>
#include<semaphore.h>



int newsockfd;


//variables for shared and semaphore

int sem_id,shmid;

int mysemid[100];

void *sharedptr;

struct mystruct

{

	sem_t mysem;
	pthread_t tid;

	long int count;

	int flag;

};


union semun

{

	int val; /* Value for SETVAL */

	struct semid_ds *buf; /* Buffer for IPC_STAT, IPC_SET */

	unsigned short *array; /* Array for GETALL, SETALL */

	struct seminfo *__buf; /* Buffer for IPC_INFO

	(Linux specific) */

}sem_union;


//end of special purpose variables

void *mythreadfunc(void *arg)

{

	int n;
	int val;

	int segment;

	char buffer[4];

	memset(buffer,0,sizeof(buffer));

	printf("going to read");

	fflush(stdout);

	while( n= read(newsockfd, buffer, 4))

	{
		
	

		segment=atoi(buffer);

		memset(buffer,0,sizeof(buffer));

		struct sembuf sem_b;

		
		printf("inside server--received segment-->%d\n>",segment);
		fflush(stdout);
		
		printf("threadid-->%ld---count-->%ld\n",(long int)((((struct mystruct *)sharedptr)+segment)->tid),		(((((struct mystruct *) sharedptr)+segment)->count)));
		
		sem_getvalue(&((((struct mystruct *)sharedptr)+segment)->mysem), &val);
		printf("sem value is-->%d\n",val);
		fflush(stdout);
		
		if( sem_post(&((((struct mystruct *)sharedptr)+segment)->mysem))<0)
		{
				
					perror("sem_post");
		}
		
		sem_getvalue(&((((struct mystruct *)sharedptr)+segment)->mysem), &val);
		printf("sem value after posting is-->%d\n",val);
		fflush(stdout);
		

		
	}

}


int main(int argc,char **argv)

{



	int sockfd, portno, clilen;

	struct sockaddr_in serv_addr, cli_addr;

	int n;

	if (argc < 2)

	{

		fprintf(stderr,"ERROR, no port provided\n");

		exit(1);

	}

	sockfd = socket(PF_INET, SOCK_STREAM, 0);

	if (sockfd < 0)
	{

		perror("ERROR creating socket");
		exit(0);
	}

	
	memset( &serv_addr,0, sizeof(serv_addr));

	portno = atoi(argv[1]);

	
	serv_addr.sin_family =AF_INET;

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

		perror("ERROR on binding");
		exit(0);
	}


	listen(sockfd,50);

	clilen = sizeof(cli_addr);


	int i=0;

	

	
	key_t key = 1234;

	shmid=shmget(key,100*sizeof(struct mystruct),0644|IPC_CREAT);

	
	if(shmid==-1)

	printf("shmid error");

	
	sharedptr=shmat(shmid,(void*)0,0);

	if(sharedptr==(void *)-1)

	{

		fprintf(stderr,"shared memory error");

		fflush(stderr);
		exit(0);

	}


	memset(sharedptr,0,100*sizeof(struct mystruct));
	for(i=0;i<100;i++)
	{
		sem_init(&((((struct mystruct *)sharedptr)+i)->mysem), 1, 1);
	}

	while(1)

	{

		newsockfd = accept(sockfd,(struct sockaddr *) &cli_addr,&clilen);

		printf("connection accepted");

		fflush(stdout);


		if (newsockfd < 0)
		{

			perror("ERROR on accept");
			exit(0);
		}


		printf("connection accepted");

		fflush(stdout);

		
		pthread_t thread_ID;

		int value;	

		pthread_create(&thread_ID , NULL, mythreadfunc , &value ) ;

	}


	for(i=0;i<100;i++)

	{

		semctl(mysemid[i],0,IPC_RMID);

	}

	return 0;

} //end main

[chitvan1988]

client-->

#include<stdio.h>

#include<stdlib.h>

#include<string.h>

#include<sys/sem.h>

#include<sys/shm.h>

#include<pthread.h>

#include<unistd.h>

#include<stdio.h>

#include<stdlib.h>
#include <sys/un.h>
#include <netinet/in.h>
#include<string.h>

#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/sem.h>
#include<sys/shm.h>

#include<pthread.h>

#include <sys/types.h>          /* See NOTES */
#include <sys/socket.h>
#include<unistd.h>
#include<semaphore.h>



int sockfd, portno, clilen;

int sem_id,shmid;

void *sharedptr;

int mysemid[100];

long int count=0;





struct mystruct

{

	sem_t mysem;
	pthread_t tid;

	long int count;

	int flag;

};



void *mythreadfunc(void *arg)

{


	while(1)

	{



		char buffer[4];
		int val;

		memset(buffer,0,4);

		int segment=rand()%100;
		//int segment=5;
		//int segment=1;

		sprintf(buffer,"%d", segment);

		printf("\n outer threadid is-->%ld segment-->%d, buffer =%s\n",(long int)(pthread_self()),segment, buffer);
		sem_getvalue(&((((struct mystruct *)sharedptr)+segment)->mysem), &val);
		printf("sem value is-->%d\n",val);
		fflush(stdout);


		if( sem_wait(&((((struct mystruct *)sharedptr)+segment)->mysem))<0)
		{
			perror("semwait failure");
		
		
		}
		
		 
				


			

			{

				((((struct mystruct *)sharedptr)+segment)->tid)=(pthread_self());

				((((struct mystruct *)sharedptr)+segment)->count)=count;

				((((struct mystruct *)sharedptr)+segment)->flag)=1;

				fflush(stdout);
				printf("count-->%ld\n",count);
				fflush(stdout);

				count++;
				

				write(sockfd,buffer,4);

				
				printf("\n---> done main work\n");

				fflush(stdout);


				
				
				
				//break;

			}
			//sleep(2);
			continue;
	

	}//outer while
	printf("reached end of while");

}//end function


int main(int argc,char **argv)

{

	struct sockaddr_in serv_addr, cli_addr;

	int n;

	if (argc < 2)

	{

		fprintf(stderr,"ERROR, no port provided\n");

		exit(1);

	}

	sockfd = socket(PF_INET, SOCK_STREAM, 0);


	if(sockfd<0)
	{

		perror("\n ERROR creating socket  : ");
		exit(0);
	}



	struct linger linger;

	linger.l_onoff=1;

	linger.l_linger=1;

	setsockopt(sockfd,SOL_SOCKET,SO_LINGER, &linger,sizeof(linger));


	memset(&serv_addr, 0, sizeof(serv_addr));

	portno = atoi(argv[1]);
	

	serv_addr.sin_family =AF_INET;

	serv_addr.sin_addr.s_addr =inet_addr("127.0.0.1");

	serv_addr.sin_port = htons(portno);

	
	if(connect(sockfd, (struct sockaddr *)&serv_addr, sizeof(serv_addr))< 0)
	{
		perror("\n ERROR in connect socket  : ");
		exit(0);
	}



	int i=0;

	
	

	key_t key = 1234;

	shmid=shmget(key,100*sizeof(struct mystruct),0644|IPC_CREAT);

	if(shmid==-1)
	{

		printf("shmid error");
		exit(0);
	}

	sharedptr=shmat(shmid,(void*)0,0);



	if(sharedptr==(void *)-1)

	{

		fprintf(stderr,"shared memory error");

		fflush(stderr);

	}

	pthread_t thread_ID[100];

	int value;



	i=0;
	//mythreadfunc(&value);

	for(i=0;i<5;i++)

	{

		if((pthread_create(&thread_ID[i] , NULL, mythreadfunc , &value ))<0)

		{
			perror("\n\n\n\nWARNING\n\n\n");

			fflush(stdout);

		}

		sleep(1);

	}
	//while(1);
	pthread_exit(NULL);

}//end main