[C언어] Thread

프로세스(Proccess)

사전적 의미: 컴퓨터에서 연속적으로 실행되고 있는 컴퓨터 프로그램

 

프로그램을 실행 시키면 정적인 프로그램이 동적으로 변하여 프로그램이 돌아가고 있는 상태이다. 모든 프로그램은 운영체제가 실행되기 위한 메모리 공간을 할당해 줘야 실행될 수 있다. 그래서 프로그램을 실행하는 순간 파일은 컴퓨터 메모리에 올라가게 되고, 운영체제로부터 시스템 자원(CPU)을 할당받아 프로그램 코드를 실행시켜 사용할 수 있게 된다.

즉, 메모리에 올라와 실행되고 있는 독립적인 개체이자 시스템 자원을 할당받는 작업의 단위이다.

 

스레드(Thread)

사전적 의미 : 프로세스 내에서 실행되는 여러 흐름의 단위

 

기술이 발전됨에 따라 프로그램이 복잡해지면서 프로세스 작업 하나만을 사용해서 프로그램을 실행하기에는 한계가 있었다. 하나의 프로세스 안에서 여러 가지 작업들이 동시에 진행되는 것을 멀티 스레드라고 부르며, 작업 흐름들을 스레드라고 한다.

 

일반적으로 하나의 프로그램은 하나 이상의 프로세스를 가지고 있고, 하나의 프로세스는 하나 이상의 스레드를 갖는다.

• 프로세스는 독립된 메모리 공간을 사용하며, 하나의 실행 단위로 관리된다. 독립적이기 때문에 (Code, Data, Heap, Stack) 메모리 영역을 다른 프로세스와 공유하지 않는다.
• 스레드는 프로세스 내에서 Stack만 독립적으로 할당 받고, 그 이외의 (Code, Data, Heap) 메모리 영역을 다른 스레드와 공유한다. 스레드 간 자원을 공유하기 때문에 동기화 문제가 발생할 수 있지만, 스레드는 프로세스보다 메모리 사용량이 적고 생성 비용이 낮아 성능이 더 효율적이다.

 

프로세스 vs 스레드

 

1. 메모리

• 스레드는 프로세스 내에서 Stack만 독립적으로 할당받고, 그 이외의 (Code, Data, Heap) 메모리 영역을 다른 스레드와 공유한다.

2. Context-Switching

• 프로세스 간 Context-Switching 시에는 많은 자원손실이 발생한다.
• 스레드 간 Context-Switching에서는 메모리를 공유하고 있는 만큼 부담을 덜 수 있다.

 

스레드의 생명 주기

1. 생성 : pthread_create()를 호출하면 새로운 스레드가 생성되고, 지정된 함수가 실행된다.
2. 실행 : 스레드가 작업을 수행한다.
3. 종료 : 스레드 함수가 반환되거나 pthread_exit()이 호출되면 스레드가 종료된다.
   • 스레드가 종료되었지만 리소스 해제가 되지 않은 상태인 좀비 스레드가 된다.
     → pthread_join()이나 pthread_detach()를 사용하는 것이 필요하다.

---

1. pthread_create()

 

✅ 반환값

• 성공시 스레드를 초기화함과 동시에 0을 리턴한다.
• 실패시 스레드 인자가 설정되지 않고 오류 번호를 반환한다.

ERROR
EAGAIN	현재 리소스가 없거나, 최대 스레드 개수를 초과한 경우 커널 전체에 최대 프로세스 및 쓰레드 개수를 확인하려면 /proc/sys/kernel/thread-max 최대 PID 수를 확인하려면 /proc/sys/kernel/pid-max 를 확인
EINVAL	attr이 유효하지 않음
EPERM	스케줄링 정책을 설정할 수 있는 권한이 없을 경우

 

✅ pthread_create()

• 호출한 프로세스 내에서 새 스레드를 시작한다.
• 새 스레드는 start_routine() 호출로 실행한다.

#include <pthread.h>

int pthread_create(pthread_t *restrict thread,
                   const pthread_attr_t *restrict attr,
                   void *(*start_routine)(void *),
                   void *restrict arg);

• thread : 성공적으로 함수가 호출되면, 스레드 ID가 저장된다.
• attr : pthread_attr_t 구조체를 가리키며, NULL을 지정하면 기본값으로 스레드가 생성된다.
  • 스레드 생성 시점에 사용하여 새 스레드의 속성을 결정할 수 있다.
  • 스레드 속성을 지정하기 전, pthread_attr_init() 함수로 구조체를 초기화 해야 한다.
  • 구조체 주소를 넘겨주면 내부적으로는 pthread_attr_init() 함수가 호출된다.

 

• start_routine : 어떤 로직을 수행할 지 함수 포인터를 매개변수로 받는다.
• arg : start_routine() 에 전달될 인자

실시간 스케줄링 정책을 사용중이지 않다면, 어느 스레드가 다음으로 실행될지 불확실하다.

 

✅ 새 스레드 종료 방법

1. pthread_exist() 호출한다.
2. start_routine()에서 반환하며, pthread_exist()를 호출하는 것과 동등하다.
3. 프로세스 내의 어느 스레드가 exit()를 호출하거나 메인 스레드가 main()에서 반환을 수행하면, 프로세스 내 모든 스레드들이 종료한다.

---

2. pthread_join()

 

✅ 반환값

• 성공시 0을, 실패시 오류 번호를 리턴한다.

 

✅ pthread_join()

• 생성한 스레드가 끝날때까지 기다리며, 대상 스레드가 이미 종료되었다면 즉시 리턴한다.
• 성공적으로 함수가 호출되면 호출자에게는 대상 스레드가 종료했다는 것이 보장된다.

#include <pthread.h>

int pthread_join(pthread_t thread, void **retval);

• thread : 종료하기를 기다릴 스레드를 지정한다.
• retval : pthread_create에서 start_routine이 반환하는 반환값을 저장한다.
  • NULL이 아닌 경우, 대상 스레드의 종료 상태를 retval이 가리키는 위치로 복사한다.
  • 대상 스레드가 취소됐으면, retval이 가리키는 위치에 PTHREAD_CANCELED가 들어간다.

기다리는 스레드가 pthread_detach() 함수를 통해서 detached 상태가 되었거나, pthread_create()로 실행될 때 PTHREAD_CREATE_DETACHED 특성으로 실행되었다면 pthread_join()으로 기다릴 수 없게 된다.

---

3. pthread_detach()

 

✅ 반환값

• 성공시 0을, 실패시 오류 번호를 리턴한다.

ERROR	설명
EINVAL	스레드가 합류 가능한 스레드가 아니다.
ESRCH	ID가 thread인 스레드를 찾을 수 없다.

 

 

✅ pthread_detach()

• 식별자 thread를 가지는 스레드를 메인 스레드에서 분리시킨다.
• pthread_create()시 pthread_attr_t에 detachstate를 지정해 줌으로써 detach상태로 생성할 수 도 있다

#include <pthread.h>

int pthread_detach(pthread_t thread);

• thread : 분리시킬 스레드를 지정한다.
  • 스레드가 종료되는 즉시 스레드의 모든 자원을 free 할 것을 보증한다.

---

4. pthread_exit()

 

✅ 반환값

• 항상 성공하며, 호출자에게 반환하지 않는다.

 

✅ pthread_exit()

#include <pthread.h>

return void pthread_exit(void *retval);

 

• 호출 스레드를 종료하고 retval을 통해 어떤 값을 반환한다.
• 같은 프로세스 내의 다른 스레드가 pthread_join()을 호출해서 그 값을 얻을 수 있다.

- 스레드가 종료할 때는 프로세스 공유 자원들(뮤텍스, 조건 변수, 세마포어, 파일 디스크립터)이 해제되지 않는다.
- main() 함수의 주 스레드가 pthread_exit() 함수를 호출할 경우에는 주 스레드는 종료하지만 다른 스레드는 생존한다.

---

5. pthread_shelf()

 

✅ 반환값

• 항상 성공하며, 호출 스레드의 ID를 반환한다.

 

✅ pthread_shelf()

#include <pthread.h>

pthread_t pthread_self(void);

• 호출 스레드의 ID를 반환한다.

---

6. pthread_cancel()

 

✅ 반환값

• 성공시 0을, 실패시 오류 번호를 반환한다.

ERROR	설명
ESRCH	ID가 thread 인 스레드를 찾을 수 없다.

 

✅ pthread_cancel()

 

스레드에게 취소 요청을 보낸다.

#include <pthread.h>

int pthread_cancel(pthread_t thread);

• 취소 요청은 취소 상태와 취소 유형 두 가지로 나뉜다.

 

취소 상태

• 취소 상태는 pthread_setcancelstate()로 결정하며, 활성(새 스레드의 기본값)이거나 비활성일 수 있다.
• 스레드의 상태를 PTHREAD_CREATE_JOINABLE 또는 PTHREAD_CREATE_DETACHED 상태로 변경시키기 위해 사용된다.

 

취소 유형

• 스레드의 취소 상태가 PTHREAD_CANCEL_ENABLE일 경우 취소의 유형을 결정할 수 있다.
• 취소 유형은 pthread_setcanceltype()으로 결정하며, 비동기나 연기일 수 있다.

 

🔷 grep 명령어 프로그램 (응용)

• 파일에서 특정 문자열을 찾아서 문자열이 존재하는 라인 번호와 해당 라인을 모두 출력하는 프로그램
• Main Thread가 파일의 크기를 고려하여 파일에서 search하는 Thread를 n개 생성
• Search Thread는 임시파일에 찾은 문자열을 저장하고 종료
• Main Thread는 search thread에서 찾은 문자열을 취합하여 화면에 출력

 

🔹 입력

1. 프로그램 실행 시, 인자로 파일명과 찾을 문자열을 입력받는다. 파일명은 1개로 제한한다.
   • 프로그램명 찾을문자열 파일명

🔹 출력

1. 파일에서 찾은 문자열이 존재하는 line번호와 해당 라인 정보를 출력한다. (grep 출력 양식 그대로 사용)
     12: int main()
     15: int nCnt;
     21:     int nTemp;
2. 파일에 찾을 문자열이 존재하지 않는 경우, 문자열이 존재하지 않는다는 문구를 출력
3. 파일이 존재하지 않는 경우, 파일이 없다는 문구를 출력

 

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

#define MAX_RANGE 2000
#define MAX_THREAD 50
#define TEMP_BUF 16

#define CLOSE_FILE(fp) \
	do { \
		if ((fp) != NULL) { \
			if (fclose(fp) == EOF) { \
				printf("[ERROR] fclose: %s\n", strerror(errno)); \
			} \
			(fp) = NULL; \
		} \
	} while (0)

#define REMOVE_FILE(filename) \
	do { \
		if ((filename) != NULL) { \
			if (unlink((filename)) < 0) { \
				printf("[ERROR] unlink: %s\n", strerror(errno)); \
			} \
		} \
	} while (0)

typedef struct grep_arg_t {
	char *keyword;
	char *file_path;
	int num;
	int start_offset;
	int end_offset;
} grep_arg_t;

void *thread_routine(void *data);
int count_line(char *path);
int merge_grep(int thread_count);


int main(int argc, char *argv[]) {

	if(argc != 3) {
		printf("[ERROR] the input argument is invalid\n");
		return -1;
	}

	int line_count;
	if((line_count = count_line(argv[2])) < 0) {
		printf("[ERROR] failed count line\n");
		return -1;
	}

	printf("line_count : %d\n", line_count);

	int thread_count = 1;
	while(1) {
		if(line_count / thread_count > MAX_RANGE && thread_count <= MAX_THREAD) {
			thread_count++;
		} else {
			break;
		}
	}

	printf("thread_count : %d\n", thread_count);

	grep_arg_t grep_arg[thread_count];
	pthread_t p_thread[thread_count];

	int i, unit = line_count / thread_count;
	int remain_line = line_count % thread_count;
	int current_line = 1;
	for(i = 0; i < thread_count; i++) {

		grep_arg[i].keyword = argv[1];
		grep_arg[i].file_path = argv[2];
		grep_arg[i].num = i;
		grep_arg[i].start_offset = current_line;
		grep_arg[i].end_offset = current_line + unit;

		current_line = grep_arg[i].end_offset;

		int ret = pthread_create(&p_thread[i], NULL, thread_routine, &grep_arg[i]);
		if(ret != 0) { // 스레드 생성 실패
			printf("[ERROR] %s\n", strerror(errno));
			break;
		}

		printf("스레드 생성 \n");

	}

	for(i = 0; i < thread_count; i++) {
		if(pthread_join(p_thread[i], NULL) != 0) { // 스레드 종료 기다리기
			printf("[ERROR] %s\n", strerror(errno));
		}
	}

	int found = merge_grep(thread_count);

	if(found < 0) {
		printf("[ERROR] failed merge grep\n");
		return -1;
	} else if(!found) {
		printf("\"%s\" not found in %s\n", argv[1], argv[2]);
	}

	return 0;
}

#define TEMP_BUF 16
void *thread_routine(void *data) {

	if(data == NULL) {
		printf("[ERROR] thread_routine: data is NULL\n");
		return (void *)-1;
	}

	grep_arg_t *arg = (grep_arg_t *)data;

	char *keyword = arg->keyword;
	char *file_path = arg->file_path;
	int num = arg->num;
	int start_offset = arg->start_offset;
	int end_offset = arg->end_offset;

	int pid = pthread_self();

	FILE *rfp = NULL;
	if((rfp = fopen(file_path, "r")) == NULL) {
		printf("[ERROR] %s\n", strerror(errno));
		pthread_exit((void *)(intptr_t) -1);
	}

	char temp_buf[TEMP_BUF];
	snprintf(temp_buf, sizeof(temp_buf), ".temp_%d", num);

	FILE *wfp = NULL;
	if((wfp = fopen(temp_buf, "w")) == NULL) {
		printf("[ERROR] %s\n", strerror(errno));

		REMOVE_FILE(temp_buf);
		CLOSE_FILE(rfp);

		pthread_exit((void *)(intptr_t) -1);
	}

	int current_line = 1;
	while(current_line < start_offset) {
		int ch = 0;
		while(1) {
			ch = fgetc(rfp);
			if(ch == EOF) {
				if(ferror(rfp)) {
					printf("[ERROR] %s\n", strerror(errno));

					CLOSE_FILE(wfp);
					CLOSE_FILE(rfp);
					REMOVE_FILE(temp_buf);
				}
				break;
			} else if(ch == '\n') {
				break;
			}
		}
		current_line ++;
	}

	long line_start_offset = ftell(rfp);

	int index = 0, found = 0, c = 0;
	int keyword_len = strlen(keyword);
	while(1) {
		c = fgetc(rfp);
		if(c == EOF) {
			if(ferror(rfp)) {
				printf("[ERROR] %s\n", strerror(errno));

				CLOSE_FILE(wfp);
				CLOSE_FILE(rfp);
				REMOVE_FILE(temp_buf);

				pthread_exit((void *)(intptr_t) -1);
			}
			break;
		}

		else if(c == '\n') {
			if(found) {
				if(fseek(rfp, line_start_offset, SEEK_SET) != 0) {
					printf("[ERROR] fseek: %s\n", strerror(errno));
					break;
				}

				fprintf(wfp, "%d: ", current_line);

				int ch = 0;
				while(ch != '\n') {
					ch = fgetc(rfp);
					if(ch == EOF) {
						if(ferror(rfp)) {
							printf("[ERROR] %s\n", strerror(errno));

							CLOSE_FILE(wfp);		
							CLOSE_FILE(rfp);
							REMOVE_FILE(temp_buf);

							return (void *) -1;
						}
					}
					fputc(ch, wfp);
				}
				fputc('\n', wfp);

				if(current_line >= end_offset) {
					break;
				}

			}

			current_line++;
			line_start_offset = ftell(rfp);
			found = 0;
			index = 0;

			if(current_line >= end_offset) {
				break;
			}
		}	

		else { // keyword 검색
			if(c == keyword[index++]) {
				if(index == keyword_len) {
					found = 1;
				}
			} else {
				index = 0;
			}
		}

	}

	CLOSE_FILE(wfp);
	CLOSE_FILE(rfp);

	pthread_exit((void *)(intptr_t) -1);
}

int count_line(char *path) {

	if(path == NULL) {
		printf("[ERROR] get_file_size: path is NULL\n");
		return -1;
	}

	FILE *fp;

	if((fp = fopen(path, "r")) < 0) {
		printf("[ERROR] %s\n", strerror(errno));
		return -1;
	}

	int c = 0, line = 0;
	while(1) {
		c = fgetc(fp);
		if(c == EOF) {
			printf("EOF\n");
			if(ferror(fp)) {
				printf("[ERROR] %s\n", strerror(errno));
				errno = 0;
				CLOSE_FILE(fp);
				return -1;
			}
			break;
		} else if(c == '\n') {
			line++;
		}
	}

	CLOSE_FILE(fp);
	return line;
}

int merge_grep(int thread_count) {

	if(thread_count <= 0) {
		printf("[ERROR] merge_grep: thread_count is out of range\n");
		return -1;
	}
	
	int i, found = 0;
	for (i = 0; i < thread_count; i++) {

		char temp_buf[TEMP_BUF];
		snprintf(temp_buf, sizeof(temp_buf), ".temp_%d", i);

		FILE *fp = NULL;
		if((fp = fopen(temp_buf, "r")) < 0) {
			printf("[ERROR] %s\n", strerror(errno));
			REMOVE_FILE(temp_buf);
			continue;
		}

		int c = 0;
		while(1) {

			c = fgetc(fp);
			if(c == EOF) {
				if(ferror(fp)) {
					printf("[ERROR] %s\n", strerror(errno));
					errno = 0;
					CLOSE_FILE(fp);
					return -1;
				}
				break;
			} else {
				if(!found) found = 1; 
				putchar(c);
			}

		}

		CLOSE_FILE(fp);
		REMOVE_FILE(temp_buf);
	}

	return found;
}