Subject: one red fish?

From: [email-masked] (Tom Henderson)
Date: Tue, 20 Jul 93 14:40:50 MDT

Hi all,

Tony writes:  

> Desired action
> --------------
> We find a way to make ibraries have control over
> their own communication safety, while permitting multi-threadedness.

OK, I'll give it a shot with a possible pthreads() implementation of 
MPI_COMM_MAKE().  Caveat:  I'm not an expert on pthreads()!!  Also, I have 
used routines from the current draft.  This is a naive implementation-- I'm 
just trying to show something that may work, not something that works 
well...  :-)

What do we want?  

An implementation of MPI_COMM_MAKE(sync_group, comm_group, comm_new) that is 
"safe" in a multi-threaded environment.  As I understand this, it means that 
there must be some way of ensuring that multiple threads cannot "interfere" 
with each other by making "simultaneous" calls to this routine.  It also means 
that multiple processes cannot "interfere" with each other when two or more 
groups make this call and the groups have overlapping membership.  Finally, it 
is permissible for an implementation to use a "hidden" context (not accessible 
to the user) to accomplish all this.  

(If I haven't got this right, read no further...)  

Here's the scheme ("pseudo-code" follows):  

Each process maintains a "shared" variable among all threads.  This shared 
variable is used by the threads to sequentialize access to MPI_COMM_MAKE().  
Let's name the variable mpiCommMakeMutex.  On entry to 
MPI_COMM_MAKE(sync_group, comm_group, comm_new), the calling thread (process) 
attempts to obtain exclusive access to mpiCommMakeMutex using 
pthread_mutex_lock().  If another thread is already executing MPI_COMM_MAKE(), 
this call will block.  If not, the thread will lock mpiCommMakeMutex and 
proceed.  

The calling process then sends a message to process 0 in sync_group that 
contains information describing comm_group.  (If the thread is in process 0, 
it receives these messages.)  These messages are sent using special context 
MPI_COMM_MAKE_CONTEXT.  This special context is not accessible to the user.  
To disambiguate overlapping process groups that have the same "process 0", 
sync_group is used as message tag.  (Note that I am assuming "INTEGER" handles 
here.  If "group" is a struct then we have to use an implementation-dependent 
"group ID" field in this struct.  I use this scheme in the example below.)  
Use of this tag ensures that process 0 will not attempt to receive a message 
on context MPI_COMM_MAKE_CONTEXT from a process that is not in sync_group 
until the current call to MPI_COMM_MAKE() completes.  

After receiving a message from every other process in sync_group, the thread 
in process 0 checks to see if comm_group is the same in each received 
message.  If not, an "error" message is sent to each process in sync_group 
and MPI_COMM_MAKE() returns an error status.  If everything is "OK", a 
"acknowledge" message is sent to each process in sync_group.   After the 
"acknowledge" message, process 0 calls MPI_CONTEXTS_RESERVE(1, &newContext) 
and sends newContext to each process in sync_group.  Each process then calls 
MPI_COMM_BIND(comm_group, newContext, comm_new) to create the new 
communicator.  Each process then releases mpiCommMakeMutex using 
pthread_mutex_unlock() and returns from MPI_COMM_MAKE().  

One note:  the use of MPI_CONTEXTS_RESERVE() is a bit of a kludge.  Probably, 
there should be some internal implementation-dependent way for processes 
{1, ..., (N-1)} to verify that newContext is not currently locally reserved 
and, if so, to reserve a context with VALUE newContext.  If newContext IS 
currently locally reserved, then some negotiation must go on within sync_group 
to generate a unique context.  I have NOT included this level of detail...  

Some implications of the example below:  

1)  The user is responsible for making sure that multiple threads do not 
    accidentally call MPI_COMM_MAKE() in the "wrong" order.  This is 
    consistent with what we have said about threads elsewhere in MPI...  

2)  The user is responsible for making sure that calls to MPI_COMM_MAKE() by 
    groups with common members are made in the correct order to avoid 
    deadlock.  Again, this is consistent with what's going on in collcomm.  

OK, will this work?  If not, what's wrong with it?  Can we fix it?  

Tom Henderson

----------------- "Implementation" of MPI_COMM_MAKE() -----------------------

/* "Implementation-dependent" part of "group" struct typedef. */
typedef struct group
    {
    ...
    int implementationGroupID;
    ...
    } group;

/* "Global" declaration (visible to all threads in a process). */
pthread_mutex_t mpiCommMakeMutex;
...

/*****************************************************************************
** function MPI_COMM_MAKE()
**
** Note that I am "assuming" an error status passed as an OUT argument...  
** Error handling is oversimplified.  I use only point-to-point communication 
** routines.  
**
*****************************************************************************/
void MPI_COMM_MAKE(group sync_group, group comm_group, comm comm_new, 
                   int *status)
    {
    pthread_mutex_t mpiCommMakeMutex;                  /* "shared" variable */
    int i, myRank, syncGroupSize, syncGroupID, commGroupID;
    int newContext, reqMsg, ackMsg, errFlag;
    receiveHandle rcvStatus;

    /* Lock "shared" variable (blocks until lock succeeds). */
    pthread_mutex_lock(&mpiCommMakeMutex);

    /* Get implementation-dependent group IDs. */
    syncGroupID = sync_group.implementationGroupID;
    commGroupID = comm_group.implementationGroupID;

    /* Code for process 0 in sync_group and other processes in sync_group. */
    MPI_GROUP_RANK(sync_group, &myRank);
    if (myRank == 0)
        {
        /* Receive "request" message from each process in sync_group. */
        /* This message should contain commGroupID. */
        MPI_GROUP_SIZE(sync_group, &syncGroupSize);
        errFlag = 0;
        for (i=1; i<syncGroupSize; i++)
            {
            MPI_RECVC(&reqMsg, 1, MPI_INT, MPI_ANY_SOURCE, syncGroupID, 
                      MPI_COMM_MAKE_CONTEXT, rcvStatus);
            /* Set error flag if wrong comm_group sent by a process. */
            if (reqMsg != commGroupID)
                errFlag = 1; 
            }
        /* Send "acknowledge" message or "error" message. */
        if (errFlag == 0)
            {
            /* "Acknowledge" message (continue, everything's OK...). */
            ackMsg = MPI_OK;
            for (i=1; i<syncGroupSize; i++)
                {
                MPI_SENDC(&ackMsg, 1, MPI_INT, i, syncGroupID, 
                          MPI_COMM_MAKE_CONTEXT);
                }
            }
        else
            {
            /* Handle error (wrong comm_group sent by a process). */
            ackMsg = MPI_ERROR_BAD_COMM_GROUP;
            for (i=1; i<syncGroupSize; i++)
                {
                MPI_SENDC(&ackMsg, 1, MPI_INT, i, syncGroupID, 
                          MPI_COMM_MAKE_CONTEXT);
                }
            *status = MPI_ERROR_BAD_COMM_GROUP;
            return;
            }
        /* Get new context and send to other processes. */
        MPI_CONTEXTS_RESERVE(1, &newContext);
        for (i=1; i<syncGroupSize; i++)
            {
            MPI_SENDC(&newContext, 1, MPI_INT, i, syncGroupID, 
                      MPI_COMM_MAKE_CONTEXT);
            }
        }  /* end of myRank==0 if */
    else
        {
        /* Send "request" message containing commGroupID to process 0 in */
        /* sync_group. */
        reqMsg = commGroupID;
        MPI_SENDC(&reqMsg, 1, MPI_INT, 0, syncGroupID, MPI_COMM_MAKE_CONTEXT);
        /* Receive "acknowledge" message from process 0. */
        MPI_RECVC(&ackMsg, 1, MPI_INT, 0, syncGroupID, MPI_COMM_MAKE_CONTEXT, 
                  rcvStatus);
        /* Check for error... */
        if (ackMsg != MPI_OK)
            {
            *status = ackMsg;
            return;
            }
        /* Receive new context from process 0. */
        MPI_RECVC(&newContext, 1, MPI_INT, 0, syncGroupID, 
                  MPI_COMM_MAKE_CONTEXT, rcvStatus);
        }  /* end of myRank==0 else */

    /* All processes in sync_group bind newContext to comm_group to create */
    /* new communicator. */
    MPI_COMM_BIND(comm_group, newContext, comm_new);

    /* Release lock. */
    pthread_mutex_unlock(&mpiCommMakeMutex);

    /* Set error status to "OK" and return. */
    *status = MPI_OK;
    return;
    }  /* end of function MPI_COMM_MAKE() */