now0930 일지

[태그:] pthread

  • memory based semaphore, unix network programming p241

    이름있는 세마포어 말고 메모리에 올려 사용하는 방법도 있다. 책은 sem_init(…, 두번째 arg,…)에서 두번째 입력하는 숫자를 thread에서 공유할 거면 0, 프로세스간 할거면 1로 하라고 한다. 1로 하여 thread도 잘 된다.

    named semaphore는 포인터를 선언했고 파일에 영역을 만들었다. 이와 다르게 직접 데이터를 선언하고 주소로 넘겨줘야 한다.

    ipcs로 보일 법 한데, 안 보인다. 무엇이 잘못되었는지 모르겠다.

    /* include main */
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <fcntl.h>
#include <semaphore.h>
#include <sys/stat.h>
#include <unistd.h>
#define	NBUFF	 10
#define	SEM_MUTEX	"mutex"	 	/* these are args to px_ipc_name() */
#define	SEM_NEMPTY	"empty"
#define	SEM_NSTORED	"nstored"

int		nitems;					/* read-only by producer and consumer */
struct {	/* data shared by producer and consumer */
  int	buff[NBUFF];
  //포인터 대신 세마포어로 변경.
  //sem_t	*mutex, *nempty, *nstored;
  sem_t	mutex, nempty, nstored;		/* semaphores, not pointers */

} shared;

void	*produce(void *), *consume(void *);

int
main(int argc, char **argv)
{
	pthread_t	tid_produce, tid_consume;

	if (argc != 2)
		perror("usage: prodcons1 <#items>");
	nitems = atoi(argv[1]);
		/* 4create three semaphores */
	//memory based로 변경.
	/*
	shared.mutex = sem_open(SEM_MUTEX, O_CREAT | O_EXCL,
							0660, 1);
	shared.nempty = sem_open(SEM_NEMPTY, O_CREAT | O_EXCL,
							 0660, NBUFF);
	shared.nstored = sem_open(SEM_NSTORED, O_CREAT | O_EXCL,
							  0660, 0);
							  */


	//두번째 파라미터를 0으로 설정해야 thread간 전달.
	//sem_init(&shared.mutex, 1, 1) 설정해도 잘 됨..

	sem_init(&shared.mutex, 0, 1);
	sem_init(&shared.nempty, 0, NBUFF);
	sem_init(&shared.nstored, 0, 0);


		/* 4create one producer thread and one consumer thread */
	//pthread_setconcurrency(2);
	//pthread_setconcurrency();
	pthread_create(&tid_produce, NULL, produce, NULL);
	pthread_create(&tid_consume, NULL, consume, NULL);

		/* 4wait for the two threads */
	pthread_join(tid_produce, NULL);
	pthread_join(tid_consume, NULL);

		/* 4remove the semaphores */
	//메모리 베이스로 변경.
	/*
	sem_unlink(SEM_MUTEX);
	sem_unlink(SEM_NEMPTY);
	sem_unlink(SEM_NSTORED);
	*/


	//ipcs로 확인하기위해 대기
	sleep(10);

	sem_destroy(&shared.mutex);
	sem_destroy(&shared.nempty);
	sem_destroy(&shared.nstored);

	exit(0);
}
/* end main */

/* include prodcons */
void *
produce(void *arg)
{
	int		i;

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

		sem_wait(&shared.nempty);	/* wait for at least 1 empty slot */
		sem_wait(&shared.mutex);

		shared.buff[i % NBUFF] = i;	/* store i into circular buffer */
		sem_post(&shared.mutex);
		sem_post(&shared.nstored);	/* 1 more stored item */


	}
	return(NULL);
}

void *
consume(void *arg)
{
	int		i;
	int tmpnempty, tmpmutex, tmpnstored;

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


		sem_wait(&shared.nstored);		/* wait for at least 1 stored item */
		sem_wait(&shared.mutex);


		if (shared.buff[i % NBUFF] = i)
			printf("buff[%d] = %d\n", i, shared.buff[i % NBUFF]);
		sem_post(&shared.mutex);
		sem_post(&shared.nempty);		/* 1 more empty slot */

	}
	return(NULL);
}
/* end prodcons */

    pi@raspberrypi:~/Project/cCode/IPC $ ipcs

------ Message Queues --------
key        msqid      owner      perms      used-bytes   messages    

------ Shared Memory Segments --------
key        shmid      owner      perms      bytes      nattch     status      

------ Semaphore Arrays --------
key        semid      owner      perms      nsems

    ipcs는 system V 기능이라 posix semaphore는 해당하지 않는다. 보이지 않음이 당연하다.

  • semaphore p223 ~ p238

    name semaphore는 /dev/shm에 저장된다. 이름을 엄한 /tmp/xxx 이런 식으로 만들면 세마포어를 만들 수 없다. segment error로 죽는다. 책에서는 /tmp/로 넣어도 잘 동작해서 확인하는데 오래 걸렸다.

    디버그 하다 중간에 멈추면 /dev/shm에 sem*로 파일을 지울 수 없어 다음 실행에 세마포어를 받을 수 없다. 처음 시작 전 자기 이름으로 된 세마포어 있으면 지워야 한다.

    thread를 돌리면 미친 race condition이 피곤하게 한다. 참 헷갈리는데, 자주보면 익숙해 지겠지. thread를 돌리면 순서는 이제 별 의미 없는 듯 하다. sem_post하는 즉시 자기 thread를 실행하는게 아니라 다른 thread로 가서 프로그램을 실행한다. wait로 thread 이후 행을 잘 막는 게 핵심인 듯 하다.

    /* include main */
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <fcntl.h>
#include <semaphore.h>
#include <sys/stat.h>
#define	NBUFF	 10
#define	SEM_MUTEX	"mutex"	 	/* these are args to px_ipc_name() */
#define	SEM_NEMPTY	"empty"
#define	SEM_NSTORED	"nstored"

int		nitems;					/* read-only by producer and consumer */
struct {	/* data shared by producer and consumer */
  int	buff[NBUFF];
  sem_t	*mutex, *nempty, *nstored;
} shared;

void	*produce(void *), *consume(void *);

int
main(int argc, char **argv)
{
	pthread_t	tid_produce, tid_consume;
	int errno;
	

	if (argc != 2)
		perror("usage: prodcons1 <#items>");
	nitems = atoi(argv[1]);
		/* 4create three semaphores */
	shared.mutex = sem_open(SEM_MUTEX, O_CREAT | O_EXCL,
							0660, 1);
	shared.nempty = sem_open(SEM_NEMPTY, O_CREAT | O_EXCL,
							 0660, NBUFF);
	shared.nstored = sem_open(SEM_NSTORED, O_CREAT | O_EXCL,
							  0660, 0);

		/* 4create one producer thread and one consumer thread */
	//pthread_setconcurrency(2);
	//pthread_setconcurrency();
	pthread_create(&tid_produce, NULL, produce, NULL);
	pthread_create(&tid_consume, NULL, consume, NULL);

		/* 4wait for the two threads */
	pthread_join(tid_produce, NULL);
	pthread_join(tid_consume, NULL);

		/* 4remove the semaphores */
	sem_unlink(SEM_MUTEX);
	sem_unlink(SEM_NEMPTY);
	sem_unlink(SEM_NSTORED);
	exit(0);
}
/* end main */

/* include prodcons */
void *
produce(void *arg)
{
	int		i;
	int tmpnempty, tmpmutex, tmpnstored;

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


		//value check
		//sem_getvalue(shared.nempty, &tmpnempty);
		//sem_getvalue(shared.nstored, &tmpnstored);
		//sem_getvalue(shared.mutex, &tmpmutex);

		sem_wait(shared.nempty);	/* wait for at least 1 empty slot */
		sem_wait(shared.mutex);

		//value check
		//sem_getvalue(shared.nempty, &tmpnempty);
		//sem_getvalue(shared.nstored, &tmpnstored);
		//sem_getvalue(shared.mutex, &tmpmutex);



		shared.buff[i % NBUFF] = i;	/* store i into circular buffer */
		sem_post(shared.mutex);
		sem_post(shared.nstored);	/* 1 more stored item */
		//value check
		//sem_getvalue(shared.nempty, &tmpnempty);
		//sem_getvalue(shared.nstored, &tmpnstored);
		//sem_getvalue(shared.mutex, &tmpmutex);


	}
	return(NULL);
}

void *
consume(void *arg)
{
	int		i;
	int tmpnempty, tmpmutex, tmpnstored;

	for (i = 0; i < nitems; i++) {
		//value check
		//sem_getvalue(shared.nempty, &tmpnempty);
		//sem_getvalue(shared.mutex, &tmpmutex);
		//sem_getvalue(shared.nstored, &tmpnstored);


		sem_wait(shared.nstored);		/* wait for at least 1 stored item */
		sem_wait(shared.mutex);
		//value check
		//sem_getvalue(shared.nempty, &tmpnempty);
		//sem_getvalue(shared.mutex, &tmpmutex);
		//sem_getvalue(shared.nstored, &tmpnstored);


		if (shared.buff[i % NBUFF] = i)
			printf("buff[%d] = %d\n", i, shared.buff[i % NBUFF]);
		sem_post(shared.mutex);
		sem_post(shared.nempty);		/* 1 more empty slot */
		//value check
		//sem_getvalue(shared.nempty, &tmpnempty);
		//sem_getvalue(shared.mutex, &tmpmutex);
		//sem_getvalue(shared.nstored, &tmpnstored);


	}
	return(NULL);
}
/* end prodcons */

    pi@raspberrypi:~/Project/cCode/IPC $ gdb --args ./a.out  3
GNU gdb (Raspbian 8.2.1-2) 8.2.1
Copyright (C) 2018 Free Software Foundation, Inc.
License GPLv3+: GNU GPL version 3 or later <http://gnu.org/licenses/gpl.html>
This is free software: you are free to change and redistribute it.
There is NO WARRANTY, to the extent permitted by law.
Type "show copying" and "show warranty" for details.
This GDB was configured as "arm-linux-gnueabihf".
Type "show configuration" for configuration details.
For bug reporting instructions, please see:
<http://www.gnu.org/software/gdb/bugs/>.
Find the GDB manual and other documentation resources online at:
    <http://www.gnu.org/software/gdb/documentation/>.

For help, type "help".
Type "apropos word" to search for commands related to "word"...
Reading symbols from ./a.out...done.
...

(gdb) info b
Num     Type           Disp Enb Address    What
1       breakpoint     keep y   0x000107b4 in produce at semaphore.c:72
	breakpoint already hit 3 times
2       breakpoint     keep y   0x00010884 in consume at semaphore.c:108
	breakpoint already hit 3 times
(gdb) r
Starting program: /home/pi/Project/cCode/IPC/a.out 3
[Thread debugging using libthread_db enabled]
Using host libthread_db library "/lib/arm-linux-gnueabihf/libthread_db.so.1".
[New Thread 0xb6e31460 (LWP 2842)]
[New Thread 0xb6630460 (LWP 2843)]
[Switching to Thread 0xb6e31460 (LWP 2842)]

Thread 2 "a.out" hit Breakpoint 1, produce (arg=0x0) at semaphore.c:72
72			sem_wait(shared.nempty);	/* wait for at least 1 empty slot */
1: shared = {buff = {0, 0, 0, 0, 0, 0, 0, 0, 0, 0}, mutex = 0xb6ff8000, 
  nempty = 0xb6ff7000, nstored = 0xb6ff6000}
2: *shared.nempty = {
  __size = "\024\000\000\000\200\000\000\000\000\000\000\000\000\000\000", 
  __align = 20}
3: *shared.nstored = {
  __size = "\000\000\000\000\200\000\000\000\000\000\000\000\000\000\000", 
  __align = 0}
4: *shared.mutex = {
  __size = "\002\000\000\000\200\000\000\000\000\000\000\000\000\000\000", 
  __align = 2}
(gdb) c
Continuing.
[Switching to Thread 0xb6630460 (LWP 2843)]

Thread 3 "a.out" hit Breakpoint 2, consume (arg=0x0) at semaphore.c:108
108			sem_wait(shared.nstored);		/* wait for at least 1 stored item */
1: shared = {buff = {0, 0, 0, 0, 0, 0, 0, 0, 0, 0}, mutex = 0xb6ff8000, 
  nempty = 0xb6ff7000, nstored = 0xb6ff6000}
2: *shared.nempty = {
  __size = "\024\000\000\000\200\000\000\000\000\000\000\000\000\000\000", 
  __align = 20}
3: *shared.nstored = {
  __size = "\000\000\000\000\200\000\000\000\000\000\000\000\000\000\000", 
  __align = 0}
4: *shared.mutex = {
  __size = "\002\000\000\000\200\000\000\000\000\000\000\000\000\000\000", 
  __align = 2}
(gdb) c
Continuing.
[Switching to Thread 0xb6e31460 (LWP 2842)]

Thread 2 "a.out" hit Breakpoint 1, produce (arg=0x0) at semaphore.c:72
72			sem_wait(shared.nempty);	/* wait for at least 1 empty slot */
1: shared = {buff = {0, 0, 0, 0, 0, 0, 0, 0, 0, 0}, mutex = 0xb6ff8000, 
  nempty = 0xb6ff7000, nstored = 0xb6ff6000}
2: *shared.nempty = {
  __size = "\022\000\000\000\200\000\000\000\000\000\000\000\000\000\000", 
  __align = 18}
3: *shared.nstored = {
  __size = "\002\000\000\000\200\000\000\000\000\000\000\000\000\000\000", 
  __align = 2}
4: *shared.mutex = {
  __size = "\002\000\000\000\200\000\000\000\000\000\000\000\000\000\000", 
  __align = 2}
(gdb) c
Continuing.
[Switching to Thread 0xb6630460 (LWP 2843)]

Thread 3 "a.out" hit Breakpoint 2, consume (arg=0x0) at semaphore.c:108
108			sem_wait(shared.nstored);		/* wait for at least 1 stored item */
1: shared = {buff = {0, 1, 0, 0, 0, 0, 0, 0, 0, 0}, mutex = 0xb6ff8000, 
  nempty = 0xb6ff7000, nstored = 0xb6ff6000}
2: *shared.nempty = {
  __size = "\022\000\000\000\200\000\000\000\000\000\000\000\000\000\000", 
  __align = 18}
3: *shared.nstored = {
  __size = "\002\000\000\000\200\000\000\000\000\000\000\000\000\000\000", 
  __align = 2}
4: *shared.mutex = {
  __size = "\002\000\000\000\200\000\000\000\000\000\000\000\000\000\000", 
  __align = 2}
(gdb) c
Continuing.
[Switching to Thread 0xb6e31460 (LWP 2842)]

Thread 2 "a.out" hit Breakpoint 1, produce (arg=0x0) at semaphore.c:72
72			sem_wait(shared.nempty);	/* wait for at least 1 empty slot */
1: shared = {buff = {0, 1, 0, 0, 0, 0, 0, 0, 0, 0}, mutex = 0xb6ff8000, 
  nempty = 0xb6ff7000, nstored = 0xb6ff6000}
2: *shared.nempty = {
  __size = "\022\000\000\000\200\000\000\000\000\000\000\000\000\000\000", 
  __align = 18}
3: *shared.nstored = {
  __size = "\002\000\000\000\200\000\000\000\000\000\000\000\000\000\000", 
  __align = 2}
4: *shared.mutex = {
  __size = "\002\000\000\000\200\000\000\000\000\000\000\000\000\000\000", 
  __align = 2}
(gdb) c
Continuing.
buff[1] = 1
[Thread 0xb6e31460 (LWP 2842) exited]
[Switching to Thread 0xb6630460 (LWP 2843)]

Thread 3 "a.out" hit Breakpoint 2, consume (arg=0x0) at semaphore.c:108
108			sem_wait(shared.nstored);		/* wait for at least 1 stored item */
1: shared = {buff = {0, 1, 2, 0, 0, 0, 0, 0, 0, 0}, mutex = 0xb6ff8000, 
  nempty = 0xb6ff7000, nstored = 0xb6ff6000}
2: *shared.nempty = {
  __size = "\022\000\000\000\200\000\000\000\000\000\000\000\000\000\000", 
  __align = 18}
3: *shared.nstored = {
  __size = "\002\000\000\000\200\000\000\000\000\000\000\000\000\000\000", 
  __align = 2}
4: *shared.mutex = {
  __size = "\002\000\000\000\200\000\000\000\000\000\000\000\000\000\000", 
  __align = 2}
(gdb) c
Continuing.
buff[2] = 2
[Thread 0xb6ffa070 (LWP 2841) exited]
[Inferior 1 (process 2841) exited normally]
(gdb)
    https://unix.stackexchange.com/questions/275650/where-is-a-named-semaphore-stored

  • sync, mutex. unix network programming p161

    다시 thread로 돌아왔다.

    #include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <pthread.h>
#define	MAXNITEMS 		1000000
#define	MAXNTHREADS			100
#define MIN(X,Y) ((X)<(Y) ? (X) : (Y))



//gcc -lpthread 옵션으로 컴파일.
int		nitems;			/* read-only by producer and consumer */


//shared를 초기화 하는데,
//맨 처음에 mutext가 있음.
//나머지는 0으로 초기화됨..
struct {
  pthread_mutex_t	mutex;
  int	buff[MAXNITEMS];
  int	nput;
  int	nval;
} shared = { PTHREAD_MUTEX_INITIALIZER };

void	*produce(void *), *consume(void *);


int
main(int argc, char **argv)
{
	int			i, nthreads, count[MAXNTHREADS];
	pthread_t	tid_produce[MAXNTHREADS], tid_consume;
	//앞에서부터 초기화.
	struct test {
		int no1;
		int no2;
	} teststruct = {10};

	if (argc != 3)
		perror("usage: prodcons2 <#items> <#threads>");
	nitems = MIN(atoi(argv[1]), MAXNITEMS);
	nthreads = MIN(atoi(argv[2]), MAXNTHREADS);

	//pthread_setconcurrency();
		/* 4start all the producer threads */
	for (i = 0; i < nthreads; i++) {
		count[i] = 0;
		pthread_create(&tid_produce[i], NULL, produce, &count[i]);
	}

		/* 4wait for all the producer threads */
	for (i = 0; i < nthreads; i++) {
		pthread_join(tid_produce[i], NULL);
		printf("count[%d] = %d\n", i, count[i]);	
	}

		/* 4start, then wait for the consumer thread */
	pthread_create(&tid_consume, NULL, consume, NULL);
	pthread_join(tid_consume, NULL);

	exit(0);
}
/* end main */

/* include producer */
void *
produce(void *arg)
{
	for ( ; ; ) {
		pthread_mutex_lock(&shared.mutex);
		if (shared.nput >= nitems) {
			pthread_mutex_unlock(&shared.mutex);
			return(NULL);		/* array is full, we're done */
		}
		shared.buff[shared.nput] = shared.nval;
		shared.nput++;
		shared.nval++;
		pthread_mutex_unlock(&shared.mutex);
		*((int *) arg) += 1;
	}
}

void *
consume(void *arg)
{
	int		i;

	for (i = 0; i < nitems; i++) {
		if (shared.buff[i] != i)
			printf("buff[%d] = %d\n", i, shared.buff[i]);
	}
	return(NULL);
}
/* end producer */