Я делаю домашнюю работу по ОС, и мне нужно реализовать параллельную сортировку слиянием, используя Pthread и семафор для их блокировки и разблокировки.
Вы можете просматривать только имена функций Multi____ и игнорировать Single_____,потому что я уже закончил однопоточную часть.
Я столкнулся с проблемой в многопоточной части. Я сигнализирую главный поток (sem [1]) в строке 227, и он должен перейти в функцию 'void * MultiPartition'.
В этой функции она дает значение arg [id * 2] и arg [id * 2 + 1].
Например, arg [1] даст значение arg [2] и arg [3], а затем сигнализирует, что thread [2] и thread [3] должны начинаться с sem_post.
И, похоже, не работает.
Поэтому я использую cout << "partition id = " << id << ", head = " << head << ", mid = " << mid << ", tail = " << tail << "\n"; в строке 111, чтобы проверить, что происходит.
Это выглядит действительно странно. Иногда он выдает
partition id = 1, head = 0, mid = 7, tail = 15
partition id = 2, head = 0, mid = 3, tail = 7
и зависает, но программа не завершается. Это означает, что мне нужно нажать Ctrl ^ C для выхода из программы.
Иногда он выводит
partition id = 1, head = 0, mid = 7, tail = 15
partition id = 2, head = 0, mid = 3, tail = 7
partition id = 3, head = 8, mid = 11, tail = 15
partition id = 4, head = 0, mid = 1, tail = 3
и тоже зависает.
Мне интересно, куда идут другие потоки?
И если отображается id = 4, обычно он запускает bubble id = 8.
#include <iostream>
#include <pthread.h>
#include <semaphore.h>
#include <fstream>
#include <sys/time.h>
#include <unistd.h>
using namespace std;
//Pthread_create, pthread_exit *don't use pthread_join
//sem_init, sem_wait, sem_post, sem_getvalue, sem_destroy
//Enter input file name: test.txt
//MT sorting used x secs
//ST sorting used x secs
// g++ -o os_hw3.out os_hw3.cpp -pthread
typedef struct{
int head;
int mid;
int tail;
int id;
}arguments;
//declare global variables
sem_t sem[16]; // use id = 1 ~ 15
sem_t final; // the final semaphore signal that indicate all threads finished.
int* s1, * s2; // two array for single and multiple
arguments arg[16];
void swap(int* x, int* y) {
int temp;
temp = *x;
*x = *y;
*y = temp;
}
void *MultiMerge(void* argid) {
int id = *(int*)argid;
sem_wait(&sem[id]);
sem_wait(&sem[id]);
int head = arg[id].head, mid = arg[id].mid, tail = arg[id].tail;
//cout << "merge id = " << id << ", head = " << head << ", mid = " << mid << ", tail = " << tail << "\n";
int lenA = mid - head + 1;
int lenB = tail - (mid + 1) + 1;
int A[lenA];
int B[lenB];
for (int i = 0; i < lenA; i++) {
A[i] = *(s1 + head + i);
}
for (int j = 0; j < lenB; j++) {
B[j] = *(s1 + mid + 1 + j);
}
int i = 0, j = 0, k = 0;
while (i < lenA && j < lenB) {
if (A[i] < B[j]) {
*(s1 + head + k) = A[i];
i++;
}
else {
*(s1 + head + k) = B[j];
j++;
}
k++;
}
while (i < lenA) {
*(s1 + head + k) = A[i];
i++;
k++;
}
while (j < lenA) {
*(s1 + head + k) = B[j];
j++;
k++;
}
sem_post(&sem[id / 2]); // signal the upper level
if (id == 1) {
fstream fout;
fout.open("output1.txt", ios::out);
for (i = 0; i < arg[1].tail + 1; i++)
fout << *(s2 + i);
fout.close();
sem_post(&final);
}
}
void *MultiBubble(void *argid) {
int id = *(int*)argid;
sem_wait(&sem[id]);
//cout << "bubble id = " << id << ", head = " << arg[id].head << ", tail = " << arg[id].tail << "\n";
for (int i = arg[id].tail; i > 0; --i) {
for (int j = arg[id].head; j < i; ++j) {
if (*(s2 + j) > * (s2 + j + 1)) {
swap((s2 + j), (s2 + j + 1));
}
}
}
for (int i = arg[id].head; i <= arg[id].tail; i++) {
cout << *(s2 + i) << " ";
}
cout << "\n";
sem_post(&sem[id / 2]);
}
void *MultiPartition(void* argid) {
int id = *(int*)argid;
sem_wait(&sem[id]);
int head = arg[id].head, mid = arg[id].mid, tail = arg[id].tail;
cout << "partition id = " << id << ", head = " << head << ", mid = " << mid << ", tail = " << tail << "\n";
arg[id * 2].head = arg[id].head;
arg[id * 2].tail = arg[id].mid;
arg[id * 2].mid = (arg[id * 2].head + arg[id * 2].tail) / 2;
arg[id * 2 + 1].head = arg[id].mid + 1;
arg[id * 2 + 1].tail = arg[id].tail;
arg[id * 2 + 1].mid = (arg[id * 2 + 1].head + arg[id * 2 + 1].tail) / 2;
sem_post(&sem[id * 2]);
sem_post(&sem[id * 2 + 1]);
}
void SingleMerge(int* s1, int head, int mid, int tail) {
int lenA = mid - head + 1;
int lenB = tail - (mid + 1) + 1;
int A[lenA];
int B[lenB];
for (int i = 0; i < lenA; i++) {
A[i] = *(s1 + head + i);
}
for (int j = 0; j < lenB; j++) {
B[j] = *(s1 + mid + 1 + j);
}
int i = 0, j = 0, k = 0;
while (i < lenA && j < lenB) {
if (A[i] < B[j]) {
*(s1 + head + k) = A[i];
i++;
}
else {
*(s1 + head + k) = B[j];
j++;
}
k++;
}
while (i < lenA) {
*(s1 + head + k) = A[i];
i++;
k++;
}
while (j < lenA) {
*(s1 + head + k) = B[j];
j++;
k++;
}
}
int SingleBubble(int* s1, int head, int tail) {
for (int i = tail; i > 0; --i) {
for (int j = head; j < i; ++j) {
if (*(s1 + j) > *(s1 + j + 1)) {
swap((s1 + j), (s1 + j + 1));
}
}
}
}
void SinglePartition(int* s1, int head, int tail, int times) {
if (head <= tail) {
int mid = (head + tail) / 2;
if (times < 3) {
SinglePartition(s1, head, mid, ++times);
SinglePartition(s1, mid + 1, tail, ++times);
}
else {
SingleBubble(s1, head, tail);
}
SingleMerge(s1, head, mid, tail);
}
}
int main() {
char filename[100];
int num;
struct timeval Tstart, Tend;
cout << "Enter the input file name: ";
cin >> filename;
fstream file, fout;
file.open(filename, ios::in);
if (!file) {
cout << "Read File Error.\n";
return -1;
}
else {
file >> num;
s1 = new int[num];
s2 = new int[num];
for (int i = 0; i < num; i++) {
file >> *(s1 + i);
*(s2 + i) = *(s1 + i);
}
file.close();
}
//SINGLE THREAD
gettimeofday(&Tstart, 0);
SinglePartition(s1, 0, num - 1, 0);
gettimeofday(&Tend, 0);
fout.open("output2.txt", ios::out);
for (int i = 0; i < num; i++)
fout << *(s1 + i) << " ";
fout.close();
double Tdifference = (Tend.tv_sec - Tstart.tv_sec) + (Tend.tv_usec - Tstart.tv_usec) / 1000000.0;
cout << "Single thread cost " << Tdifference << " s\n";
//MULTI THREAD
gettimeofday(&Tstart, 0);
arg[1].head = 0;
arg[1].tail = num - 1;
arg[1].mid = (arg[1].head + arg[1].tail) / 2;
pthread_t thread[16];
sem_init(&final, 0, 0);
sem_post(&sem[1]);
for (int i = 1; i < 16; i++){
arg[i].id = i;
sem_init(&sem[i], 0, 0);
if (i < 8) {
if(i == 1)
sem_post(&sem[1]); // call the master thread
pthread_create(&thread[i], NULL, MultiPartition, &arg[i].id);
}
else
pthread_create(&thread[i], NULL, MultiBubble, &arg[i].id);
}
for (int i = 7; i > 0; i--) {
pthread_create(&thread[i], NULL, MultiMerge, &arg[i].id);
}
sem_wait(&final);
gettimeofday(&Tend, 0);
Tdifference = (Tend.tv_sec - Tstart.tv_sec) + (Tend.tv_usec - Tstart.tv_usec) / 1000000.0;
cout << "Multi thread cost " << Tdifference << " s\n";
delete s1, s2;
for (int i = 0; i < 16; i++)
sem_destroy(&sem[i]);
sem_destroy(&final);
return 0;
}