Run Constant Number of Threads at a Time in C

Question

I am trying to copy files from one directory to another using pthreads. Each thread is responsible for copying exactly one file. The maximum number of threads is specified via a command line argument.

What I want to do is IF the current threads is less than the max threads, create a thread to do work. ELSE, wait for the current threads and when one of them finishes, decrease the number of current threads.

I can't figure out how to wait for a thread via pthread_join without blocking the main thread.

Here is what I have so far:

#include <stdlib.h>
#include <stdio.h>
#include <dirent.h>
#include <stdlib.h>
#include <stdio.h>
#include <dirent.h>
#include <sys/stat.h>
#include <string.h>
#include <errno.h>
#include <fcntl.h>
#include <pthread.h>

#define MAXNAME 80
#define R_FLAGS O_RDONLY
#define W_FLAGS (O_WRONLY | O_CREAT)
#define W_PERMS (S_IRUSR | S_IWUSR)
#define KILOBYTE 0.001

void *copyfilepass(void *arg);

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

    int num_threads;        //number of threads to execute in parallel
    int cur_threads = 0;    //number of threads currently executing
    char filename[MAXNAME]; //will temporarily hold the file name
    DIR* source;            //pointer to the source directory
    DIR* dest;              //pointer to the destination directory
    struct dirent* dentry;  //pointer to the internal structure of the source directory
    struct stat fst;        //stats for each file in a directory
    int error;
    void *status;

    //***BEGIN ERROR CHECKING***

    if (argc != 4) {
        fprintf(stderr, "Usage: %s sourceDir destDir numThreads\n", argv[0]);
        return 1; 
    } 

    //check if source directory name is too long
    if ( snprintf(filename, MAXNAME, "%s", argv[1]) == MAXNAME ) {
        fprintf(stderr, "Source directory name %s too long\n", argv[1]); 
        return 1;
    }
    //check if destination directory name is too long
    if ( snprintf(filename, MAXNAME, "%s", argv[2]) == MAXNAME ) {
        fprintf(stderr, "Source directory name %s too long\n", argv[2]); 
        return 1;
    }

    //check if we can successfully open the source directory
    if( (source = opendir(argv[1])) == NULL ) {
        fprintf(stderr, "Error opening source directory %s\n", argv[1]); 
        return 1;
    }

    //check if we can successfully open the destination directory
    if( (dest = opendir(argv[2])) == NULL ) {
        fprintf(stderr, "Error opening destination directory %s\n", argv[2]); 
        return 1;
    }

    //***END ERROR CHECKING***

    num_threads = atoi(argv[3]);

    while( (dentry = readdir(source)) != NULL ){
        //source path
        char* path = (char*)malloc(sizeof(char) * (strlen(dentry->d_name) + strlen(argv[1]) + 2)); //need '.' + '/' + '\0'
        sprintf(path, "%s%c%s", argv[1], '/', dentry->d_name);

        //destination path
        char* dest_path = (char*)malloc(sizeof(char) * (strlen(dentry->d_name) + strlen(argv[2]) + 2)); //need '.' + '/' + '\0'
        sprintf(dest_path, "%s%c%s", argv[2], '/', dentry->d_name);

        if(!stat(path, &fst)){ //stat() return 0 if successful
            if(S_ISREG(fst.st_mode)){

                int args[3];
                pthread_t tid;

                if ( (args[0] = open(path, R_FLAGS)) == -1 ) {
                    fprintf(stderr, "Failed to open source file %s: %s\n", path, strerror(errno));
                    continue; 
                }
                if ( (args[1] = open(dest_path, W_FLAGS, W_PERMS)) == -1 ) {
                    fprintf(stderr, "Failed to open destination file %s: %s\n", dest_path, strerror(errno));
                    continue;
                }

                if(cur_threads < num_threads) {

                    ++cur_threads;

                    if ( (error = pthread_create((&tid), NULL, copyfilepass, args)) ) {
                        --cur_threads;
                        fprintf(stderr, "Failed to create thread: %s\n", strerror(error));
                        tid = pthread_self();    /* cannot be value for new thread */
                    }

                    printf("file: %.03fKB %s\n", (fst.st_size * KILOBYTE), path);

                }

            }
        }
    }
    //close directory
    closedir(source);

    return 0;

}

Answer 1

Better than spawning and reaping threads is to just create a fixed-size pool at the beginning and have them all consume from a work queue. This will reduce overhead and simplify your code.

By the way, using threads to solve this problem may not improve performance, depending on the filesystem you're operating on. Food for thought.