У меня есть задание, в котором сначала нужно создать следующее дерево процессов:
здесь
Мне удалось это сделать, но после этого мне также нужно создать 5 потоков в процессе P8 (которые в другой задаче должны быть синхронизированы определенным образом) и 6 потоков в процессе P4. Основная задача на этом этапе - синхронизировать определенные потоки из P8 и P4 следующим образом: поток 1 из P4 должен заканчиваться раньше, чем поток 1 из P8, но поток 4 из P4 может начинаться только после того, как закончился поток 1 из P8 (начало / окончание). поток означает, что я должен напечатать сообщение). Я попытался сделать следующее, и это имело смысл для меня:
Это основная программа:
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <pthread.h>
#include <semaphore.h>
#include <fcntl.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <sys/wait.h>
#include "a2_helper.h"
int thread_numbers[35];
int started = 0;
int finished = 0;
int barr_count = 0;
sem_t *msem;
sem_t *sem_aux;
sem_t *sem_thd;
sem_t *sem_one;
sem_t *sem_two;
typedef struct thd_s
{
int id;
pthread_t thd;
pthread_mutex_t *lock;
pthread_cond_t *cond;
int process_nr;
}THD;
void *thread_function_diff(void *args)
{
THD *thd = (THD *)args;
if (thd->process_nr == 8 && thd->id == 1)
{
info(BEGIN, thd->process_nr, thd->id);
sem_wait(sem_one);
info(END, thd->process_nr, thd->id);
sem_post(sem_two);
}
else if (thd->process_nr == 4 && thd->id == 1)
{
info(BEGIN, thd->process_nr, thd->id);
info(END, thd->process_nr, thd->id);
sem_post(sem_one);
}
else if (thd->process_nr == 4 && thd->id == 4)
{
sem_wait(sem_two);
info(BEGIN, thd->process_nr, thd->id);
info(END, thd->process_nr, thd->id);
}
else
{
info(BEGIN, thd->process_nr, thd->id);
info(END, thd->process_nr, thd->id);
}
pthread_exit(0);
}
void *thread_function_barr(void *args)
{
THD *thd = (THD *)args;
sem_wait(msem);
//
// if (thd->id == 31 || thd->id == 32 || thd->id == 33 || thd->id == 34)
// {
// info(BEGIN, thd->process_nr, thd->id);
// sem_post(sem_thd);
// sem_wait(sem_aux);
// info(END, thd->process_nr, thd->id);
// }
// else
// {
// if (thd->id == 13)
// {
// sem_wait(sem_thd);
// sem_wait(sem_thd);
// sem_wait(sem_thd);
// sem_wait(sem_thd);
// info(BEGIN, thd->process_nr, thd->id);
// info(END, thd->process_nr, thd->id);
// sem_post(sem_aux);
// sem_post(sem_aux);
// sem_post(sem_aux);
// sem_post(sem_aux);
// }
// else
// {
info(BEGIN, thd->process_nr, thd->id);
info(END, thd->process_nr, thd->id);
// }
// }
sem_post(msem);
pthread_exit(0);
}
void *thread_function(void *args)
{
THD *thd = (THD *)args;
pthread_mutex_lock(thd->lock);
while (started == 0 && thd->id == 3)
{
pthread_cond_wait(thd->cond, thd->lock);
}
info(BEGIN, thd->process_nr, thd->id);
pthread_mutex_unlock(thd->lock);
info(END, thd->process_nr, thd->id);
pthread_mutex_lock(thd->lock);
if (thd->id == 3)
{
finished = 1;
pthread_cond_signal(thd->cond);
}
pthread_mutex_unlock(thd->lock);
pthread_exit(0);
}
void *thread_function5(void *args)
{
THD *thd = (THD *)args;
started = 1;
info(BEGIN, thd->process_nr, thd->id);
pthread_cond_signal(thd->cond);
pthread_mutex_lock(thd->lock);
while (finished == 0)
{
pthread_cond_wait(thd->cond, thd->lock);
}
info(END, thd->process_nr, thd->id);
pthread_mutex_unlock(thd->lock);
pthread_exit(0);
}
int main()
{
init();
info(BEGIN, 1, 0);
sem_one = sem_open("sem_one", O_CREAT, 0644, 0);
sem_two = sem_open("sem_two", O_CREAT, 0644, 0);
pid_t pid2 = -1, pid3 = -1, pid8 = -1;
pid8 = fork();
if (pid8 == 0)
{
info(BEGIN, 8, 0);
THD thds[5];
pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
for (int i = 4; i >= 0; i--)
{
thds[i].id = i + 1;
thds[i].process_nr = 8;
thds[i].lock = &lock;
thds[i].cond = &cond;
if (thds[i].id == 5)
{
pthread_create(&(thds[i].thd), NULL, thread_function5, &thds[i]);
}
else if (thds[i].id == 1)
{
pthread_create(&(thds[i].thd), NULL, thread_function_diff, &thds[i]);
}
else
{
pthread_create(&(thds[i].thd), NULL, thread_function, &thds[i]);
}
}
for (int i = 4; i >= 0; i--)
{
pthread_join(thds[i].thd, NULL);
}
pthread_cond_destroy(&cond);
pthread_mutex_destroy(&lock);
info(END, 8, 0);
exit(0);
}
else
{
wait(NULL);
}
pid2 = fork();
if (pid2 == 0)
{
info(BEGIN, 2, 0);
pid_t pid4 = -1, pid5 = -1, pid9 = -1;
pid4 = fork();
if (pid4 == 0)
{
info(BEGIN, 4, 0);
THD thds[6];
for (int i = 5; i >= 0; i--)
{
thds[i].id = i + 1;
thds[i].process_nr = 4;
thds[i].lock = NULL;
thds[i].cond = NULL;
pthread_create(&(thds[i].thd), NULL, thread_function_diff, &thds[i]);
}
for (int i = 5; i >= 0; i--)
{
pthread_join(thds[i].thd, NULL);
}
info(END, 4, 0);
exit(0);
}
else
{
wait(NULL);
}
pid5 = fork();
if (pid5 == 0)
{
info(BEGIN, 5, 0);
THD thds[36];
sem_unlink("thdsemaphore");
sem_unlink("auxsemaphore");
sem_unlink("barriersemaphore");
msem = sem_open("barriersemaphore", O_CREAT, 0644, 5);
sem_aux = sem_open("auxsemaphore", O_CREAT, 0644, 0);
sem_thd = sem_open("thdsemaphore", O_CREAT, 0644, 0);
pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER;
for (int i = 35; i >= 0; i--)
{
if (i + 1 == 13)
{
thread_numbers[i] = 36;
}
else
{
thread_numbers[i] = i + 1;
}
thds[i].id = i + 1;
thds[i].process_nr = 5;
thds[i].lock = &lock;
thds[i].cond = NULL;
pthread_create(&(thds[i].thd), NULL, thread_function_barr, &thds[i]);
}
for (int i = 35; i >= 0; i--)
{
pthread_join(thds[i].thd, NULL);
}
pid_t pid6 = -1;
pid6 = fork();
if (pid6 == 0)
{
info(BEGIN, 6, 0);
info(END, 6, 0);
exit(0);
}
else
{
wait(NULL);
}
pthread_mutex_destroy(&lock);
sem_close(sem_thd);
sem_unlink("thdsemaphore");
sem_close(sem_aux);
sem_unlink("auxsemaphore");
sem_close(msem);
sem_unlink("barriersemaphore");
info(END, 5, 0);
exit(0);
}
else
{
wait(NULL);
}
pid9 = fork();
if (pid9 == 0)
{
info(BEGIN, 9, 0);
info(END, 9, 0);
exit(0);
}
else
{
wait(NULL);
}
info(END, 2, 0);
exit(0);
}
else
{
wait(NULL);
}
pid3 = fork();
if (pid3 == 0)
{
info(BEGIN, 3, 0);
pid_t pid7 = -1;
pid7 = fork();
if (pid7 == 0)
{
info(BEGIN, 7, 0);
info(END, 7, 0);
exit(0);
}
else
{
wait(NULL);
}
info(END, 3, 0);
exit(0);
}
else
{
wait(NULL);
}
sem_close(sem_one);
sem_close(sem_two);
sem_unlink("sem_one");
sem_unlink("sem_two");
info(END, 1, 0);
return 0;
}
Проблема в том, что если я попытаюсь сделать это таким образом, Поток 1 из P8 будет бесконечно ждать получения разрешения на семафор, но это разрешение дается Потоком 1 из P4, который никогда не получит шанса на запуск, потому что в основной функции, я должен убедиться, что основной поток любого процесса не должен заканчиваться раньше, чем любой другой поток, и все процессы должны ожидать завершения своих дочерних процессов.
Это a2_helper.h:
#ifndef __A2_HELPER_H__
#define __A2_HELPER_H__
#define BEGIN 1
#define END 2
void init();
int info(int action, int processNr, int threadNr);
#endif
Это a2_helper.c:
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <semaphore.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <pthread.h>
#include <sys/prctl.h>
#include <signal.h>
#include "a2_helper.h"
#define SEM_NAME "A2_HELPER_SEM_17871"
#define SERVER_PORT 1988
#define XSTR(s) STR(s)
#define STR(s) #s
#define CHECK(c) if(!(c)){perror("info function failed at line " XSTR(__LINE__)); break;}
int initialized = 0;
int info(int action, int processNr, int threadNr){
int msg[6];
int sleepTime = 0;
int sockfd = -1;
struct sockaddr_in serv_addr;
sem_t *sem = SEM_FAILED;
int err = -1;
if(initialized == 0){
printf("init() function not called\n");
return -1;
}
do{
CHECK((sem = sem_open(SEM_NAME, 0)) != SEM_FAILED);
//prepare the message
msg[0] = action;
msg[1] = processNr;
msg[2] = threadNr;
msg[3] = getpid();
msg[4] = getppid();
msg[5] = pthread_self();
CHECK((sockfd = socket(AF_INET, SOCK_STREAM, 0)) >= 0);
memset(&serv_addr, 0, sizeof(serv_addr));
serv_addr.sin_family = AF_INET;
serv_addr.sin_port = htons(SERVER_PORT);
CHECK(sem_wait(sem) == 0);
err = -2;
if(connect(sockfd, (struct sockaddr*)&serv_addr, sizeof(serv_addr)) >= 0){
CHECK(write(sockfd, msg, sizeof(msg)) == sizeof(msg));
CHECK(read(sockfd, &sleepTime, sizeof(sleepTime)) == sizeof(sleepTime));
printf("[T] ");
}else{
printf("[ ] ");
}
printf("%s P%d T%d pid=%d ppid=%d tid=%d\n", msg[0]==BEGIN?"BEGIN":" END ", msg[1], msg[2], msg[3], msg[4], msg[5]);
CHECK(sem_post(sem) == 0);
err = -1;
usleep(sleepTime);
err = 0;
}while(0);
if(sockfd >= 0){
close(sockfd);
}
if(err==-2){
sem_post(sem);
}
return err;
}
void atfork_prepare(){
sem_t *sem = SEM_FAILED;
do{
CHECK((sem = sem_open(SEM_NAME, O_CREAT, 0644, 1)) != SEM_FAILED);
CHECK(sem_wait(sem) == 0);
}while(0);
}
void atfork_parent(){
sem_t *sem = SEM_FAILED;
do{
CHECK((sem = sem_open(SEM_NAME, O_CREAT, 0644, 1)) != SEM_FAILED);
CHECK(sem_post(sem) == 0);
}while(0);
}
void atfork_child(){
prctl(PR_SET_PDEATHSIG, SIGHUP);
}
void init(){
sem_t *sem = SEM_FAILED;
if(initialized != 0){
printf("init() function already called\n");
return;
}
do{
pthread_atfork(atfork_prepare, atfork_parent, atfork_child);
sem_unlink(SEM_NAME);
CHECK((sem = sem_open(SEM_NAME, O_CREAT, 0644, 1)) != SEM_FAILED);
initialized = 1;
}while(0);
}
a2_helper.c и a2_helper.h уже были реализованы, и я должен только использовать их.