per-file thread safety in ROMIO

[wkliao]

I notice the thread lock and unlock calls are commented out for file open in the ROMIO driver, as shown below.

ompi/ompi/mca/io/romio321/src/io_romio321_file_open.c

Lines 51 to 54 in c4d0752

	// OPAL_THREAD_LOCK (&mca_io_romio321_mutex);
	ret = ROMIO_PREFIX(MPI_File_open)(comm, filename, amode, ompi_info,
	&data->romio_fh);
	// OPAL_THREAD_UNLOCK (&mca_io_romio321_mutex);

In order to support per-file thread safety, these lock and unlock calls must be uncommented, as ROMIO internally uses a few global variables need to be protected by a mutex.

I tested a short program below that lets each process to spawns 6 threads and each thread uses MPI_COMM_SELF to open a file. Without the lock protection, the test program failed.

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

#define NTHREADS 6

#define ERR \
    if (err != MPI_SUCCESS) { \
        int errorStringLen; \
        char errorString[MPI_MAX_ERROR_STRING]; \
        MPI_Error_string(err, errorString, &errorStringLen); \
        printf("Error at line %d: %s\n",__LINE__, errorString); \
        nerrs++; \
    }

/* pthread barrier object */
static pthread_barrier_t barr;

typedef struct {
    int  id;         /* globally unique thread ID */
    char fname[256]; /* output file name base */
} thread_arg;

/*----< thread_func() >------------------------------------------------------*/
static
void* thread_func(void *arg)
{
    char filename[256];
    int id, err, nerrs=0, *ret;
    MPI_File fh;

    /* make a unique file name for each thread */
    id = ((thread_arg*)arg)->id;
    sprintf(filename, "%s.%d", ((thread_arg*)arg)->fname, id);

    err = MPI_File_open(MPI_COMM_SELF, filename, MPI_MODE_CREATE | MPI_MODE_RDWR,
                        MPI_INFO_NULL, &fh); ERR

    err = MPI_File_close(&fh); ERR

    /* return number of errors encountered */
    ret = (int*)malloc(sizeof(int));
    *ret = nerrs;

    return ret; /* same as pthread_exit(ret); */
}

/*----< main() >-------------------------------------------------------------*/
int main(int argc, char **argv) {
    extern int optind;
    char filename[256];
    int  i, nerrs=0, rank, providedT;
    pthread_t threads[NTHREADS];

    MPI_Init_thread(&argc, &argv, MPI_THREAD_MULTIPLE, &providedT);

    MPI_Comm_rank(MPI_COMM_WORLD, &rank);

    if (providedT != MPI_THREAD_MULTIPLE) {
        if (!rank)
            printf("Error: MPI does not support MPI_THREAD_MULTIPLE\n");
        MPI_Finalize();
        return 0;
    }

    if (argc == 1) strcpy(filename, "testfile");
    else           strcpy(filename, argv[1]);

    /* initialize thread barrier */
    pthread_barrier_init(&barr, NULL, NTHREADS);

    /* create threads, each calls thread_func() */
    for (i=0; i<NTHREADS; i++) {
        thread_arg t_arg[NTHREADS]; /* must be unique to each thread */
        t_arg[i].id = i + rank * NTHREADS;
        sprintf(t_arg[i].fname, "%s",filename);
        if (pthread_create(&threads[i], NULL, thread_func, &t_arg[i])) {
            fprintf(stderr, "Error in %s line %d creating thread %d\n",
                    __FILE__, __LINE__, i);
            nerrs++;
        }
    }

    /* wait for all threads to finish */
    for (i=0; i<NTHREADS; i++) {
        void *ret;
        if (pthread_join(threads[i], (void**)&ret)) {
            fprintf(stderr, "Error in %s line %d joining thread %d\n",
                    __FILE__, __LINE__, i);
        }
        nerrs += *(int*)ret;
        free(ret);
    }

    pthread_barrier_destroy(&barr);

    MPI_Finalize();
    return (nerrs > 0);
}

[jsquyres]

@ggouaillardet It looks like 4272b57 commented these lines out (i.e., the commit that originally updated ROMIO to 3.2.1).

Do you know / remember why you commented these lines out?

[ggouaillardet]

The lock was too coarse and could cause some deadlocks (see the example below, to be ran with 2 MPI tasks).

This reproducer evidences there is a need for some locking.
So instead of locking the global mca_io_romio321_mutex should we instead lock comm->c_lock ?

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <pthread.h>
#include <poll.h>
#include <mpi.h>

#define NTHREADS 2

#define ERR \
    if (err != MPI_SUCCESS) { \
        int errorStringLen; \
        char errorString[MPI_MAX_ERROR_STRING]; \
        MPI_Error_string(err, errorString, &errorStringLen); \
        printf("Error at line %d: %s\n",__LINE__, errorString); \
        nerrs++; \
    }

MPI_Comm comm0;
MPI_Comm comm1;
int rank;

/*----< thread_func() >------------------------------------------------------*/
static
void* thread_func(void *arg)
{
    char filename[256];
    int err, nerrs=0, *ret;
    long id;
    MPI_File fh;

    /* make a unique file name for each thread */
    id = (long)arg;
    sprintf(filename, "io.%d", id);

    poll(NULL, 0, 2000*(id^rank));

    err = MPI_File_open((0==id)?comm0:comm1, filename, MPI_MODE_CREATE | MPI_MODE_RDWR,
                        MPI_INFO_NULL, &fh); ERR

    err = MPI_File_close(&fh); ERR

    /* return number of errors encountered */
    ret = (int*)malloc(sizeof(int));
    *ret = nerrs;

    return ret; /* same as pthread_exit(ret); */
}

/*----< main() >-------------------------------------------------------------*/
int main(int argc, char **argv) {
    int  i, nerrs=0, providedT;
    pthread_t threads[NTHREADS];

    MPI_Init_thread(&argc, &argv, MPI_THREAD_MULTIPLE, &providedT);

    MPI_Comm_rank(MPI_COMM_WORLD, &rank);

    if (providedT != MPI_THREAD_MULTIPLE) {
        if (!rank)
            printf("Error: MPI does not support MPI_THREAD_MULTIPLE\n");
        MPI_Finalize();
        return 0;
    }

    MPI_Comm_dup(MPI_COMM_WORLD, &comm0);
    MPI_Comm_dup(MPI_COMM_WORLD, &comm1);

    /* create threads, each calls thread_func() */
    for (i=0; i<NTHREADS; i++) {
        if (pthread_create(&threads[i], NULL, thread_func, (void *)(long)i) < 0) {
            fprintf(stderr, "Error in %s line %d creating thread %d\n",
                    __FILE__, __LINE__, i);
            nerrs++;
        }
    }

    /* wait for all threads to finish */
    for (i=0; i<NTHREADS; i++) {
        void *ret;
        if (pthread_join(threads[i], (void**)&ret)) {
            fprintf(stderr, "Error in %s line %d joining thread %d\n",
                    __FILE__, __LINE__, i);
        }
        nerrs += *(int*)ret;
        free(ret);
    }

    MPI_Finalize();
    return (nerrs > 0);
}

[wkliao]

Hi, @ggouaillardet
I am trying to understand the logic of your test program.
What is the purpose of the call to poll() before MPI_File_open?
FYI. I tested the program and it ran fine, with the opal thread lock codes uncomments.
It looks to me the two threads on each MPI process will not block each other.

[wkliao]

Locking comm->c_lock sounds a better solution.

[edgargabriel]

I am trying to understand why locking on the communicator would be necessary. Collective MPI File operations have the same rules according to my understanding as 'regular' collective communication operations, i.e the order on how they are posted must be identical on all processes. With that being said, any intermixing of message on the same communicator that I can only see occurring is if the order is not deterministic/identical.

[wkliao]

Hi, @ggouaillardet
MPI standard 3.1 has the following statement in Chapter 12.4.1.

Advice to users. It is the user’s responsibility to prevent races when threads within the same application post conflicting communication calls. The user can make sure that two threads in the same process will not issue conflicting communication calls by using distinct communicators at each thread. (End of advice to users.)

As @edgargabriel mentioned, MPI_File_open should be treated like a regular MPI
collective communication operation. So your test program is doing the practice
against the above advice. I guess the hanging is an expected outcome
by MPI standard.

FYI, MPICH is using a global mutex, not a mutex inside a communicator, to
wrap around MPI_File_open. Interestingly, your test program does not hang
if compiled with MPICH 3.3. I don't know how MPICH does the trick.

[ggouaillardet]

@wkliao

@edgargabriel is 100% right with respect to the MPI standard, but your conclusion is wrong.
I mean your program is incorrect (because of concurrent calls to MPI_COMM_SELF) but mine is a valid one (thread 0 on comm0 and thread 1 on comm1).

With the Open MPI global lock, my program hangs.
The poll() call is here to force this condition

• thread 0 on rank 0 issues a collective on comm0
• thread 1 on rank 1 issues a collective on comm1
• then other thread hang trying to acquire the Open MPI global lock
• thread 0 on rank 0 waits for thread 0 on rank 1
• thread 1 on rank 1 waits for thread 1 on rank 0
  => deadlock

Locking a communicator is indeed not a correct solution.

What we do wrong is we do not implement the three macros below from ompi/mca/io/romio321/romio/mpi-io/mpioimpl.h but we use a global lock instead (more to come ...)

#else /* not ROMIO_INSIDE_MPICH */
/* Any MPI implementation that wishes to follow the thread-safety and
   error reporting features provided by MPICH must implement these 
   four functions.  Defining these as empty should not change the behavior 
   of correct programs */
#define ROMIO_THREAD_CS_ENTER()
#define ROMIO_THREAD_CS_EXIT()
#define ROMIO_THREAD_CS_YIELD()

my program does not hang in MPICH because MPI_File_open() indirectly calls some collective operations such as MPI_Gather() that will then call MPI_Wait() that will ultimately calls something equivalent to ROMIO_THREAD_CS_YIELD() that allows thread 1 on rank 0 and thread 0 on rank 1 to acquire the global lock, and hence solve the deadlock.

[wkliao]

Thanks for the explanation. Now I have a clear picture of your test program.
I assume you will implement those 3 macros soon.

[ggouaillardet]

@wkliao I just issued #6958

• the global lock is now used via the romio macros
• MPI_Allreduce() is now a wrapper that calls PMPI_Allreduce() without holding the global lock

this is really a proof of concept, but so far, it is enough to get my test completes.

[edgargabriel]

@ggouaillardet just to make sure: based on your understand, is there a reason why a per-file handle lock should not work? This is the model currently used by ompio. I ran your test and it passes with ompio as far as I can see.

[wkliao]

Hi, @ggouaillardet

Thanks for the patch. I assume the wrapper approach will be applied to all collective
calls made in ROMIO. (There are other collective, such as MPI_Gather, MPI_Bcast,
MPI_Comm_dup, etc. may be called during MPI_File_open, depending on MPI-IO
hints and file systems.)

I wonder why MPICH does not need that wrapper to pass your test program.

[ggouaillardet]

@edgargabriel as long as the code is thread-safe (I could spot some bits in ROMIO that uses global variables and hence is not thread safe) and a given file internally uses a unique communicator, I do not see any objections with a per-file handle lock.

@wkliao as I previously explained, collective in MPICH (basically) calls ROMIO_THREAD_CS_YIELD() and that allow other threads to progress.