Communicate between threads with pipes

Viewed 46

I am not very good at C++. So what I am trying to do is that I am trying to divide a vector into chunks and send it through the pipe to child threads and they sort and then send it back through pipes. However, I am having trouble sending the data to the child thread, somehow the child thread would read a very large number or zero instead of the array's size.

The thread_arg is the struct where I pass in the arguments for the child thread, this is where I pass in pipes which I used in the childThreadFunction.

I would really appreciate any help/advice!!!

using namespace std;

struct thread_arg {
    int parent_child[2];
    int child_parent[2];
    int n;
};

void *childThreadFunction(void *arguments) {
    cout << "in child thread function" << endl;
    struct thread_arg *args = (struct thread_arg *)arguments;

    // child process takes its own chunk of data and sort it
    close(args->parent_child[1]);
    int n;
    read(args->parent_child[0], &n, sizeof(int));
***this is where it's acting weird***
    cout << "array size: " << n << endl;
    pthread_exit(0);
    return NULL;
}

int main(int argc, char *argv[])
{
    vector<long long> numbers;

    for (int i=0; i<fileNames.size(); i++) {
        // reading each file and adding that to the vector numbers
        fstream file;
        file.open(fileNames[i], ios::in);
        if (!file) {
            fprintf(stderr, "Error: File with the name: %s doesn't exist.", fileNames[i]);
            exit(1);
        }
        string tempNum;
        while(getline(file, tempNum)) {
            if (isNumber(tempNum)) {

                numbers.push_back(stoll(tempNum));
            }
        }
    }
    if (nprocesses == 1) {
        numbers = bubbleSort(numbers);
        for (int i = 0; i < numbers.size(); i++) {
            cout << numbers[i] << endl;
        }
        return 0;
    }

    // downpipe will be from 0 to (nprocesses-1)
    // uppipe will be from nprocesses to 2*nprocesses
    int fd[2*nprocesses][2];
    for (int i=0; i<(2*nprocesses); i++) {
        if (pipe(fd[i]) < 0) {
            fprintf(stderr, "Error with creating the pipes");
            exit(1);
        }
    }

    size_t const chunk_size = numbers.size() / nprocesses;
    if (threads) {  // threads

        vector<thread_arg> thread_args;

        pthread_t thr[nprocesses];

        int s;
        for (int i = 0; i < nprocesses; i++) {
            struct thread_arg child_arg;
            pipe(child_arg.parent_child);
            pipe(child_arg.child_parent);
            thread_args.push_back(child_arg);
            s = pthread_create(&thr[i], NULL, childThreadFunction, (void *)&child_arg);
            if (s != 0) {
                fprintf(stderr, "Error creating the child thread");
            }
        }

        vector<int> sizes;

        for (int i = 0; i < nprocesses; i++) {
            std::vector<long long> splitVec;
            if (i+1 < nprocesses) {
                std::vector<long long> splitVec1(numbers.begin() + (chunk_size * i), numbers.begin() + (chunk_size * (i + 1)));
                splitVec = splitVec1;
            } else {
                std::vector<long long> splitVec2(numbers.begin() + (chunk_size * i), numbers.end());
                splitVec = splitVec2;
            }
            // sending the chuck of numbers to the down pipe for children to read
            //close(thread_args[i].parent_child[0]);
            int n = splitVec.size();
            sizes.push_back(n);
            long long arr[n];
            copy(splitVec.begin(), splitVec.end(), arr);
            write(thread_args[i].parent_child[1], &n, sizeof(int));
            write(thread_args[i].parent_child[1], arr, sizeof(long long)*n);
            close(thread_args[i].parent_child[1]);
        }

        for (int i = 0; i < nprocesses; i++) {
            pthread_join(thr[i], NULL);
        }
    }
}
;

This is the output of the above code, I am supposed to get 25 four times, but instead I get some random large number.

1 Answers

The problem was with how I initialized my arg struct for the thread, I need to dynamically allocate it instead of declaring it on the stack because threads don't have access to the main's stack.

P.S. I understand that we don't need pipes to pass information, it was just my assignment requirement.

Related