mpi.c

Go to the documentation of this file.

 
/* #[ License : */
/*
 *   Copyright (C) 1984-2026 J.A.M. Vermaseren
 *   When using this file you are requested to refer to the publication
 *   J.A.M.Vermaseren "New features of FORM" math-ph/0010025
 *   This is considered a matter of courtesy as the development was paid
 *   for by FOM the Dutch physics granting agency and we would like to
 *   be able to track its scientific use to convince FOM of its value
 *   for the community.
 *
 *   This file is part of FORM.
 *
 *   FORM is free software: you can redistribute it and/or modify it under the
 *   terms of the GNU General Public License as published by the Free Software
 *   Foundation, either version 3 of the License, or (at your option) any later
 *   version.
 *
 *   FORM is distributed in the hope that it will be useful, but WITHOUT ANY
 *   WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
 *   FOR A PARTICULAR PURPOSE.  See the GNU General Public License for more
 *   details.
 *
 *   You should have received a copy of the GNU General Public License along
 *   with FORM.  If not, see <http://www.gnu.org/licenses/>.
 */
/* #] License : */ 
/*
    #[ Includes and variables :
*/
 
/*
#define MPIDEBUGGING
#define MPIDEBUGGING_DELAY_US 10000
*/
 
#include <limits.h>
#include "form3.h"
 
#ifdef MPICH_PROFILING
# include "mpe.h"
#endif
 
#ifdef MPIDEBUGGING
#include "mpidbg.h"
#endif
 
/*[12oct2005 mt]:*/
/*
    Today there was some cleanup, some stuff is moved into another place
    in this file, and PF.packsize is removed and PF_packsize is used
    instead. It is rather difficult to proper comment it, so not all these
    changing are marked by "[12oct2005 mt]"
*/
 
#define PF_PACKSIZE 1600
 
/*
    Size in bytes, will be initialized soon as
    PF_packsize=PF_PACKSIZE/sizeof(int)*sizeof(int); for possible
    future developing we prefer to do this initialization not here,
    but in PF_LibInit:
*/
 
static int PF_packsize = 0;
static MPI_Status PF_status;
LONG PF_maxDollarChunkSize = 0;      /*:[04oct2005 mt]*/
 
static int PF_ShortPackInit(void);
static int PF_longPackInit(void);      /*:[12oct2005 mt]*/
 
#define MPI_ERRCODE_CHECK(err) \
    do { \
        int _tmp_err = (err); \
        if ( _tmp_err != MPI_SUCCESS ) return _tmp_err != 0 ? _tmp_err : -1; \
    } while (0)
 
/*
    #] Includes and variables : 
    #[ PF_RealTime :
*/
 
LONG PF_RealTime(int i)
{
    static double starttime;
    if ( i == PF_RESET ) {
        starttime = MPI_Wtime();
        return((LONG)0);
    }
    return((LONG)( 100. * (MPI_Wtime() - starttime) ) );
}
 
/*
    #] PF_RealTime : 
    #[ PF_LibInit :
*/
 
int PF_LibInit(int *argcp, char ***argvp)
{
    int ret;
    ret = MPI_Init(argcp,argvp);
    if ( ret != MPI_SUCCESS ) return(ret);
    ret = MPI_Comm_rank(PF_COMM,&PF.me);
    if ( ret != MPI_SUCCESS ) return(ret);
    ret = MPI_Comm_size(PF_COMM,&PF.numtasks);
    if ( ret != MPI_SUCCESS ) return(ret);
 
    /* Initialization of packed communications. */
    PF_packsize = PF_PACKSIZE/sizeof(int)*sizeof(int);
    if ( PF_ShortPackInit() ) return -1;
    if ( PF_longPackInit() ) return -1;
 
    {/*Block*/
        int bytes, totalbytes=0;
/*
            There is one problem with maximal possible packing: there is no API to
            convert bytes to the record number. So, here we calculate the buffer
            size needed for storing dollarvars:
 
            LONG PF_maxDollarChunkSize is the size for the portion of the dollar
            variable buffer suitable for broadcasting. This variable should be
            visible from parallel.c
 
            Evaluate PF_Pack(numterms,1,PF_INT):
*/
        if ( ( ret = MPI_Pack_size(1,PF_INT,PF_COMM,&bytes) )!=MPI_SUCCESS )
            return(ret);
 
        totalbytes+=bytes;
/*
            Evaluate PF_Pack( newsize,1,PF_LONG):
*/
        if ( ( ret = MPI_Pack_size(1,PF_LONG,PF_COMM,&bytes) )!=MPI_SUCCESS )
            return(ret);
 
        totalbytes += bytes;
/*
            Now available room is PF_packsize-totalbytes
*/
        totalbytes = PF_packsize-totalbytes;
/*
            Now totalbytes is the size of chunk in bytes.
            Evaluate this size in number of records:
 
            Rough estimate:
*/
        PF_maxDollarChunkSize=totalbytes/sizeof(WORD);
/*
            Go to the up limit:
*/
        do {
            if ( ( ret = MPI_Pack_size(
                        ++PF_maxDollarChunkSize,PF_WORD,PF_COMM,&bytes) )!=MPI_SUCCESS )
                return(ret);
        } while ( bytes<totalbytes );
/*
            Now the chunk size is too large
            And now evaluate the exact value:
*/
        do {
            if ( ( ret = MPI_Pack_size(
                        --PF_maxDollarChunkSize,PF_WORD,PF_COMM,&bytes) )!=MPI_SUCCESS )
                return(ret);
        } while ( bytes>totalbytes );
/*
            Now PF_maxDollarChunkSize is the size of chunk of PF_WORD fitting the
            buffer <= (PF_packsize-PF_INT-PF_LONG)
*/
    }/*Block*/
    return(0);
}
/*
    #] PF_LibInit : 
    #[ PF_LibTerminate :
*/
 
int PF_LibTerminate(int error)
{
    DUMMYUSE(error);
    return(MPI_Finalize());
}
 
/*
    #] PF_LibTerminate : 
    #[ PF_Probe :
*/
 
int PF_Probe(int *src)
{
    int ret, flag;
    if ( *src == PF_ANY_SOURCE ) { /*Blocking call*/
        ret = MPI_Probe(*src,MPI_ANY_TAG,PF_COMM,&PF_status);
        flag = 1;
    }
    else { /*Non-blocking call*/
        ret = MPI_Iprobe(*src,MPI_ANY_TAG,PF_COMM,&flag,&PF_status);
    }
    *src = PF_status.MPI_SOURCE;
    if ( ret != MPI_SUCCESS ) { if ( ret > 0 ) ret *= -1; return(ret); }
    if ( !flag ) return(0);
    return(PF_status.MPI_TAG);
}
 
/*
    #] PF_Probe : 
    #[ PF_ISendSbuf :
*/
 
int PF_ISendSbuf(int to, int tag)
{
    PF_BUFFER *s = PF.sbuf;
    int a = s->active;
    int size = s->fill[a] - s->buff[a];
    int r = 0;
 
    static int finished;
 
    s->fill[a] = s->buff[a];
    if ( s->numbufs == 1 ) {
        r = MPI_Ssend(s->buff[a],size,PF_WORD,MASTER,tag,PF_COMM);
        if ( r != MPI_SUCCESS ) {
            fprintf(stderr,"[%d|%d] PF_ISendSbuf: MPI_Ssend returns: %d \n",
                    PF.me,(int)AC.CModule,r);
            fflush(stderr);
            return(r);
        }
        return(0);
    }
 
    switch ( tag ) { /* things to do before sending */
        case PF_TERM_MSGTAG:
            if ( PF.sbuf->request[to] != MPI_REQUEST_NULL)
                r = MPI_Wait(&PF.sbuf->request[to],&PF.sbuf->retstat[to]);
            if ( r != MPI_SUCCESS ) return(r);
            break;
        default:
            break;
    }
 
    r = MPI_Isend(s->buff[a],size,PF_WORD,to,tag,PF_COMM,&s->request[a]);
 
    if ( r != MPI_SUCCESS ) return(r);
 
    switch ( tag ) { /* things to do after initialising sending */
        case PF_TERM_MSGTAG:
            finished = 0;
            break;
        case PF_ENDSORT_MSGTAG:
            if ( ++finished == PF.numtasks - 1 )
                r = MPI_Waitall(s->numbufs,s->request,s->status);
            if ( r != MPI_SUCCESS ) return(r);
            break;
        case PF_BUFFER_MSGTAG:
            if ( ++s->active >= s->numbufs ) s->active = 0;
            while ( s->request[s->active] != MPI_REQUEST_NULL ) {
                r = MPI_Waitsome(s->numbufs,s->request,&size,s->index,s->retstat);
                if ( r != MPI_SUCCESS ) return(r);
            }
            break;
        case PF_ENDBUFFER_MSGTAG:
            if ( ++s->active >= s->numbufs ) s->active = 0;
            r = MPI_Waitall(s->numbufs,s->request,s->status);
            if ( r != MPI_SUCCESS ) return(r);
            break;
        default:
            return(-99);
            break;
    }
    return(0);
}
 
/*
    #] PF_ISendSbuf : 
    #[ PF_RecvWbuf :
*/
 
int PF_RecvWbuf(WORD *b, LONG *s, int *src)
{
    int i, r = 0;
 
    for (;;) {
        r = MPI_Probe(*src,PF_ANY_MSGTAG,PF_COMM,&PF_status);
        if ( r != MPI_SUCCESS ) { if ( r > 0 ) r *= -1; return(r); }
        if ( PF_status.MPI_TAG != PF_RUNTIME_ERROR_MSGTAG ) break;
        PF_ReceiveRuntimeError();
    }
 
    r = MPI_Recv(b,(int)*s,PF_WORD,PF_status.MPI_SOURCE,PF_status.MPI_TAG,PF_COMM,&PF_status);
    if ( r != MPI_SUCCESS ) { if ( r > 0 ) r *= -1; return(r); }
 
    r = MPI_Get_count(&PF_status,PF_WORD,&i);
    if ( r != MPI_SUCCESS ) { if ( r > 0 ) r *= -1; return(r); }
 
    *s = (LONG)i;
    *src = PF_status.MPI_SOURCE;
    return(PF_status.MPI_TAG);
}
 
/*
    #] PF_RecvWbuf : 
    #[ PF_IRecvRbuf :
*/
 
int PF_IRecvRbuf(PF_BUFFER *r, int bn, int from)
{
    int ret;
    r->type[bn] = PF_WORD;
 
    if ( r->numbufs == 1 ) {
        r->tag[bn] = MPI_ANY_TAG;
        r->from[bn] = from;
    }
    else {
        ret = MPI_Irecv(r->full[bn],(int)(r->stop[bn] - r->full[bn]),PF_WORD,from,
                        MPI_ANY_TAG,PF_COMM,&r->request[bn]);
        if (ret != MPI_SUCCESS) { if(ret > 0) ret *= -1; return(ret); }
    }
    return(0);
}
 
/*
    #] PF_IRecvRbuf : 
    #[ PF_WaitRbuf :
*/
 
int PF_WaitRbuf(PF_BUFFER *r, int bn, LONG *size)
{
    int ret, rsize;
 
    if ( r->numbufs == 1 ) {
        *size = r->stop[bn] - r->full[bn];
        ret = MPI_Recv(r->full[bn],(int)*size,r->type[bn],r->from[bn],r->tag[bn],
                       PF_COMM,&(r->status[bn]));
        if ( ret != MPI_SUCCESS ) { if ( ret > 0 ) ret *= -1; return(ret); }
        ret = MPI_Get_count(&(r->status[bn]),r->type[bn],&rsize);
        if ( ret != MPI_SUCCESS ) { if ( ret > 0 ) ret *= -1; return(ret); }
        if ( rsize > *size ) return(-99);
        *size = (LONG)rsize;
    }
    else {
        while ( r->request[bn] != MPI_REQUEST_NULL ) {
            ret = MPI_Waitsome(r->numbufs,r->request,&rsize,r->index,r->retstat);
            if ( ret != MPI_SUCCESS ) { if ( ret > 0 ) ret *= -1; return(ret); }
            while ( --rsize >= 0 ) r->status[r->index[rsize]] = r->retstat[rsize];
        }
        ret = MPI_Get_count(&(r->status[bn]),r->type[bn],&rsize);
        if ( ret != MPI_SUCCESS ) { if ( ret > 0 ) ret *= -1; return(ret); }
        *size = (LONG)rsize;
    }
    return(r->status[bn].MPI_TAG);
}
 
/*
    #] PF_WaitRbuf : 
    #[ PF_Bcast :
*/
 
int PF_Bcast(void *buffer, int count)
{
    if ( MPI_Bcast(buffer,count,MPI_BYTE,MASTER,PF_COMM) != MPI_SUCCESS )
        return(-1);
    return(0);
}
 
/*
    #] PF_Bcast : 
    #[ PF_Reduce :
*/
 
int PF_Reduce(const void *sendbuf, void *recvbuf, int count, MPI_Datatype type, MPI_Op op, int root)
{
    if ( MPI_Reduce(sendbuf,recvbuf,count,type,op,root,PF_COMM) != MPI_SUCCESS )
        return(-1);
    return(0);
}
 
/*
    #] PF_Reduce : 
    #[ PF_RawSend :
*/
 
int PF_RawSend(int dest, void *buf, LONG l, int tag)
{
    int ret=MPI_Ssend(buf,(int)l,MPI_BYTE,dest,tag,PF_COMM);
    if ( ret != MPI_SUCCESS ) return(-1);
    return(0);
}
/*
    #] PF_RawSend : 
    #[ PF_RawRecv :
*/
 
LONG PF_RawRecv(int *src,void *buf,LONG thesize,int *tag)
{
    MPI_Status stat;
    int ret=MPI_Recv(buf,(int)thesize,MPI_BYTE,*src,MPI_ANY_TAG,PF_COMM,&stat);
    if ( ret != MPI_SUCCESS ) return(-1);
    if ( MPI_Get_count(&stat,MPI_BYTE,&ret) != MPI_SUCCESS ) return(-1);
    *tag = stat.MPI_TAG;
    *src = stat.MPI_SOURCE;
    return(ret);
}
 
/*
    #] PF_RawRecv : 
    #[ PF_RawProbe :
*/
 
int PF_RawProbe(int *src, int *tag, int *bytesize)
{
    MPI_Status stat;
    int srcval = src != NULL ? *src : PF_ANY_SOURCE;
    int tagval = tag != NULL ? *tag : PF_ANY_MSGTAG;
    int ret = MPI_Probe(srcval, tagval, PF_COMM, &stat);
    if ( ret != MPI_SUCCESS ) return -1;
    if ( src != NULL ) *src = stat.MPI_SOURCE;
    if ( tag != NULL ) *tag = stat.MPI_TAG;
    if ( bytesize != NULL ) {
        ret = MPI_Get_count(&stat, MPI_BYTE, bytesize);
        if ( ret != MPI_SUCCESS ) return -1;
    }
    return 0;
}
 
/*
    #] PF_RawProbe : 
    #[ PF_RawIsend :
*/
 
int PF_RawIsend(int dest, const void *buf, int count, MPI_Datatype type, int tag, MPI_Request *request)
{
    int ret = MPI_Isend(buf, count, type, dest, tag, PF_COMM, request);
    if ( ret != MPI_SUCCESS ) return(-1);
    return(0);
}
 
/*
    #] PF_RawIsend : 
    #[ PF_RawWaitAll :
*/
 
int PF_RawWaitAll(int count, MPI_Request *request, MPI_Status *status)
{
    int ret = MPI_Waitall(count, request, status);
    if ( ret != MPI_SUCCESS ) return(-1);
    return(0);
}
 
/*
    #] PF_RawWaitAll : 
    #[ PF_Discard :
*/
 
int PF_Discard(int *src, int *tag)
{
    enum { DEFAULT_BUF_SIZE = 1024 };
    MPI_Status stat;
    int count;
    void *buf;
    char default_buf[DEFAULT_BUF_SIZE];
    int srcval = src != NULL ? *src : PF_ANY_SOURCE;
    int tagval = tag != NULL ? *tag : PF_ANY_MSGTAG;
    int ret = MPI_Probe(srcval, tagval, PF_COMM, &stat);
    if ( ret != MPI_SUCCESS ) return -1;
    if ( src != NULL ) *src = stat.MPI_SOURCE;
    if ( tag != NULL ) *tag = stat.MPI_TAG;
    ret = MPI_Get_count(&stat, MPI_BYTE, &count);
    if ( ret != MPI_SUCCESS ) return -1;
    buf = count <= DEFAULT_BUF_SIZE ? default_buf : Malloc1(count, "PF_Discard");
    ret = MPI_Recv(buf, count, MPI_BYTE, stat.MPI_SOURCE, stat.MPI_TAG, PF_COMM, MPI_STATUS_IGNORE);
    if ( buf != default_buf ) M_free(buf, "PF_Discard");
    if ( ret != MPI_SUCCESS ) return -1;
    return 0;
}
 
/*
    #] PF_Discard : 
    #[ The pack buffer :
        #[ Variables :
*/
 
/*
 * The pack buffer with the fixed size (= PF_packsize).
 */
static UBYTE *PF_packbuf  = NULL;
static UBYTE *PF_packstop = NULL;
static int    PF_packpos  = 0;
 
/*
        #] Variables : 
        #[ PF_ShortPackInit :
*/
 
static int PF_ShortPackInit(void)
{
    PF_packbuf = (UBYTE *)Malloc1(sizeof(UBYTE) * PF_packsize, "PF_ShortPackInit");
    if ( PF_packbuf == NULL ) return -1;
    PF_packstop = PF_packbuf + PF_packsize;
    return 0;
}
 
/*
        #] PF_ShortPackInit : 
        #[ PF_InitPackBuf :
*/
 
static inline int PF_InitPackBuf(void)
{
/*
        This is definitely not the best place for allocating the
        buffer! Moved to PF_LibInit():
 
    if ( PF_packbuf == 0 ) {
        PF_packbuf = (UBYTE *)Malloc1(sizeof(UBYTE)*PF.packsize,"PF_InitPackBuf");
        if ( PF_packbuf == 0 ) return(-1);
        PF_packstop = PF_packbuf + PF.packsize;
    }
*/
    PF_packpos = 0;
    return(0);
}
 
/*
        #] PF_InitPackBuf : 
        #[ PF_PrintPackBuf :
*/
 
int PF_PrintPackBuf(char *s, int size)
{
#ifdef NOMESPRINTYET
/*
        The use of printf should be discouraged. The results are flushed to
        the output at unpredictable moments. We should use printf only
        during startup when MesPrint doesn't have its buffers and output
        channels initialized.
*/
    int i;
    printf("[%d] %s: ",PF.me,s);
    for(i=0;i<size;i++) printf("%d ",PF_packbuf[i]);
    printf("\n");
#else
    MesPrint("[%d] %s: %a",PF.me,s,size,(WORD *)(PF_packbuf));
#endif
    return(0);
}
 
/*
        #] PF_PrintPackBuf : 
        #[ PF_PreparePack :
*/
 
int PF_PreparePack(void)
{
    return PF_InitPackBuf();
}
 
/*
        #] PF_PreparePack : 
        #[ PF_Pack :
*/
 
int PF_Pack(const void *buffer, size_t count, MPI_Datatype type)
{
    int err, bytes;
 
    if ( count > INT_MAX ) return -99;
 
    err = MPI_Pack_size((int)count, type, PF_COMM, &bytes);
    MPI_ERRCODE_CHECK(err);
    if ( PF_packpos + bytes > PF_packstop - PF_packbuf ) return -99;
 
    err = MPI_Pack((void *)buffer, (int)count, type, PF_packbuf, PF_packsize, &PF_packpos, PF_COMM);
    MPI_ERRCODE_CHECK(err);
 
    return 0;
}
 
/*
        #] PF_Pack : 
        #[ PF_Unpack :
*/
 
int PF_Unpack(void *buffer, size_t count, MPI_Datatype type)
{
    int err;
 
    if ( count > INT_MAX ) return -99;
 
    err = MPI_Unpack(PF_packbuf, PF_packsize, &PF_packpos, buffer, (int)count, type, PF_COMM);
    MPI_ERRCODE_CHECK(err);
 
    return 0;
}
 
/*
        #] PF_Unpack : 
        #[ PF_PackString :
*/
 
int PF_PackString(const UBYTE *str)
{
    int ret,buflength,bytes,length;
/*
        length will be packed in the beginning.
        Decrement buffer size by the length of the field "length":
*/
    if ( ( ret = MPI_Pack_size(1,PF_INT,PF_COMM,&bytes) ) != MPI_SUCCESS )
        return(ret);
    buflength = PF_packsize - bytes;
/*
        Calculate the string length (INCLUDING the trailing zero!):
*/
    for ( length = 0; length < buflength; length++ ) {
        if ( str[length] == '\0' ) {
            length++; /* since the trailing zero must be accounted */
            break;
        }
    }
/*
        The string "\0!\0" is used as an image of the NULL.
*/
    if ( ( str[0] == '\0' ) /* empty string */
      && ( str[1] == '!' )  /* Special case? */
      && ( str[2] == '\0' ) /* Yes, pass 3 initial symbols */
            ) length += 2;  /* all 3 characters will be packed */
    length++;               /* Will be decremented in the following loop */
/*
        The problem: packed size of byte may be not equal 1! So first, suppose
        it is 1, and if this is not the case decrease the length of the string
        until it fits the buffer:
*/
    do {
        if ( ( ret = MPI_Pack_size(--length,PF_BYTE,PF_COMM,&bytes) )
            != MPI_SUCCESS ) return(ret);
    } while ( bytes > buflength );
/*
        Note, now if str[length-1] == '\0' then the string fits to the buffer
        (INCLUDING the trailing zero!);if not, the rest must be packed further!
 
        Pack the length to PF_packbuf:
*/
    if ( ( ret = MPI_Pack(&length,1,PF_INT,PF_packbuf,PF_packsize,
            &PF_packpos,PF_COMM) ) != MPI_SUCCESS ) return(ret);
/*
        Pack the string to PF_packbuf:
*/
    if ( ( ret = MPI_Pack((UBYTE *)str,length,PF_BYTE,PF_packbuf,PF_packsize,
            &PF_packpos,PF_COMM) ) != MPI_SUCCESS ) return(ret);
    return(length);
}
 
/*
        #] PF_PackString : 
        #[ PF_UnpackString :
*/
 
int PF_UnpackString(UBYTE *str)
{
    int ret,length;
/*
        Unpack the length:
*/
    if(  (ret = MPI_Unpack(PF_packbuf,PF_packsize,&PF_packpos,
            &length,1,PF_INT,PF_COMM))!= MPI_SUCCESS )
                return(ret);
/*
        Unpack the string:
*/
    if ( ( ret = MPI_Unpack(PF_packbuf,PF_packsize,&PF_packpos,
            str,length,PF_BYTE,PF_COMM) ) != MPI_SUCCESS ) return(ret);
/*
        Now if str[length-1]=='\0' then the whole string
        (INCLUDING the trailing zero!) was unpacked ;if not, the rest
        must be unpacked to str+length.
*/
    return(length);
}
 
/*
        #] PF_UnpackString : 
        #[ PF_Send :
*/
 
int PF_Send(int to, int tag)
{
    int err;
    err = MPI_Ssend(PF_packbuf, PF_packpos, MPI_PACKED, to, tag, PF_COMM);
    MPI_ERRCODE_CHECK(err);
    return 0;
}
 
/*
        #] PF_Send : 
        #[ PF_Receive :
*/
 
int PF_Receive(int src, int tag, int *psrc, int *ptag)
{
    int err;
    MPI_Status status;
    PF_InitPackBuf();
    err = MPI_Recv(PF_packbuf, PF_packsize, MPI_PACKED, src, tag, PF_COMM, &status);
    MPI_ERRCODE_CHECK(err);
    if ( psrc ) *psrc = status.MPI_SOURCE;
    if ( ptag ) *ptag = status.MPI_TAG;
    return 0;
}
 
/*
        #] PF_Receive : 
        #[ PF_Broadcast :
*/
 
int PF_Broadcast(void)
{
    int err;
/*
 * If PF_SHORTBROADCAST is defined, then the broadcasting will be performed in
 * 2 steps. First, the size of the buffer will be broadcast, then the buffer of
 * exactly used size. This should be faster with slow connections, but slower on
 * SMP shmem MPI because of the latency.
 */
#ifdef PF_SHORTBROADCAST
    int pos = PF_packpos;
#endif
    if ( PF.me != MASTER ) {
        err = PF_InitPackBuf();
        if ( err ) return err;
    }
#ifdef PF_SHORTBROADCAST
    err = MPI_Bcast(&pos, 1, MPI_INT, MASTER, PF_COMM);
    MPI_ERRCODE_CHECK(err);
    err = MPI_Bcast(PF_packbuf, pos, MPI_PACKED, MASTER, PF_COMM);
#else
    err = MPI_Bcast(PF_packbuf, PF_packsize, MPI_PACKED, MASTER, PF_COMM);
#endif
    MPI_ERRCODE_CHECK(err);
    return 0;
}
 
/*
        #] PF_Broadcast : 
    #] The pack buffer : 
    #[ Long pack stuff :
        #[ Explanations :
 
    The problems here are:
        1. We need to send/receive long dollar variables. For
    preprocessor-defined dollarvars we used multiply
    packing/broadcasting  (see parallel.c:PF_BroadcastPreDollar())
    since each variable must be broadcast immediately. For run-time
    the changed dollar variables, collecting and broadcasting are
    performed at the end of the module and all modified dollarvars
    are transferred "at once", that is why the size of packed and
    transferred buffers may be really very large.
        2. There is some strange feature of MPI_Bcast() on Altix MPI
    implementation, namely, sometimes it silently fails with big
    buffers. For better performance, it would be useful to send one
    big buffer instead of several small ones (since the latency is more
    important than the bandwidth). That is why we need two different
    sets of routines: for long point-to-point communication we collect
    big re-allocatable buffer, the corresponding routines have the
    prefix PF_longSingle, and for broadcasting we pack data into
    several smaller buffers, the corresponding routines have the
    prefix PF_longMulti.
        Note, from portability reasons we cannot split large packed
    buffer into small chunks, send them and collect back on the other
    side, see "Advice to users" on page 180 MPI--The Complete Reference
    Volume1, second edition.
        OPTIMIZING:
        We assume, for most communications, the single buffer of size
    PF_packsize is enough.
 
        How does it work:
        For point-to-point, there is one big re-allocatable
    buffer PF_longPackBuf with two integer positions: PF_longPackPos
    and PF_longPackTop (due to re-allocatable character of the buffer,
    it is better to use integers rather than pointers).
        Each time of re-allocation, the size of the buffer
    PF_longPackBuf is incremented by the same size of a "standard" chunk
    PF_packsize.
        For broadcasting there is one linked list (PF_longMultiRoot),
    which contains either positions of a chunk of PF_longPackBuf, or
    it's own buffer. This is done for better memory utilisation:
    longSingle and longMulti are never used simultaneously.
        When a new cell is needed for LongMulti packing, we increment
    the counter PF_longPackN and just follow the list. If it is not
    possible, we allocate the cell's own buffer and link it to the end
    of the list PF_longMultiRoot.
        When PF_longPackPos is reallocated, we link new chunks into
    existing PF_longMultiRoot list before the first longMulti allocated
    cell's own buffer. The pointer PF_longMultiLastChunk points to the last
    cell of PF_longMultiRoot containing the pointer to the chunk of
    PF_longPackBuf.
        Initialization PF_longPackBuf is made by the function
    PF_longSingleReset(). In the begin of the PF_longPackBuf it packs
    the size of the last sent buffer. Upon sending, the program checks,
    whether there was at list one re-allocation (PF_longPackN>1) .
    If so, the sender first packs and sends small buffer
    (PF_longPackSmallBuf) containing one integer number -- the
    _negative_ new size of the send buffer. Getting the buffer, a
    receiver unpacks one integer and checks whether it is <0 . If so,
    the receiver will repeat receiving, but first it checks whether
    it has enough buffer and increase it, if necessary.
        Initialization PF_longMultiRoot is made by the function
    PF_longMultiReset(). In the begin of the first chunk it packs
    one integer -- the number 1. Upon sending, the program checks,
    how many cells were packed (PF_longPackN). If more than 1, the
    sender packs to the next cell the integer PF_longPackN, than
    packs PF_longPackN pairs of integers -- the information about how many
    times chunk on each cell was accessed by the packing procedure,
    this information is contained by the nPacks field of the cell
    structure, and how many non-complete items was at the end of this
    chunk the structure field lastLen. Then the sender sends first
    this auxiliary chunk.
    The receiver unpacks the integer from obtained chunk and, if this
    integer is more than 1, it gets more chunks, unpacking information
    from the first auxiliary chunk into the corresponding nPacks
    fields. Unpacking information from multiple chunks, the receiver
    knows, when the chunk is expired and it must switch to the next cell,
    successively decrementing corresponding nPacks field.
 
    XXX: There are still some flaws:
    PF_LongSingleSend/PF_LongSingleReceive may fail, for example, for data
    transfers from the master to many slaves. Suppose that the master sends big
    data to slaves, which needs an increase of the buffer of the receivers. For
    the first data transfer, the master sends the new buffer size as the first
    message, and then sends the data as the second message, because
    PF_LongSinglePack records the increase of the buffer size on the master. For
    the next time, however, the master sends the data without sending the new
    buffer size, and then MPI_Recv fails due to the data overflow.
    In parallel.c, they are used for the communication from slaves to the
    master. In this case, this problem does not occur because the master always
    has enough buffer.
    The maximum size that PF_LongMultiBroadcast can broadcast is limited to
    around 320kB because the current implementation tries to pack all
    information of chained buffers into one buffer, whose size is PF_packsize
    = 1600B.
 
        #] Explanations : 
        #[ Variables :
*/
 
typedef struct longMultiStruct {
    UBYTE *buffer; /* NULL if */
    int bufpos;    /* if >=0, PF_longPackBuf+bufpos is the chunk start */
    int packpos;   /* the current position */
    int nPacks;    /* How many times PF_longPack operates on this cell */
    int lastLen;   /* if > 0, the last packing didn't fit completely to this
                        chunk, only lastLen items was packed, the rest is in
                        the next cell. */
    struct longMultiStruct *next;  /* next linked cell, or NULL */
} PF_LONGMULTI;
 
static UBYTE *PF_longPackBuf = NULL;
static void *PF_longPackSmallBuf = NULL;
static int PF_longPackPos = 0;
static int PF_longPackTop = 0;
static PF_LONGMULTI *PF_longMultiRoot = NULL;
static PF_LONGMULTI *PF_longMultiTop = NULL;
static PF_LONGMULTI *PF_longMultiLastChunk = NULL;
static int PF_longPackN = 0;
 
/*
        #] Variables : 
        #[ Long pack private functions :
        #[ PF_longMultiNewCell :
*/
 
static inline int PF_longMultiNewCell(void)
{
/*
        Allocate a new cell:
*/
    PF_longMultiTop->next = (PF_LONGMULTI *)
            Malloc1(sizeof(PF_LONGMULTI),"PF_longMultiCell");
    if ( PF_longMultiTop->next == NULL ) return(-1);
/*
        Allocate a private buffer:
*/
    PF_longMultiTop->next->buffer=(UBYTE*)
            Malloc1(sizeof(UBYTE)*PF_packsize,"PF_longMultiChunk");
    if ( PF_longMultiTop->next->buffer == NULL ) return(-1);
/*
        For the private buffer position is -1:
*/
    PF_longMultiTop->next->bufpos = -1;
/*
        This is the last cell in the chain:
*/
    PF_longMultiTop->next->next = NULL;
/*
        packpos and nPacks are not initialized!
*/
    return(0);
}
 
/*
        #] PF_longMultiNewCell : 
        #[ PF_longMultiPack2NextCell :
*/
static inline int PF_longMultiPack2NextCell(void)
{
/*
        Is there a free cell in the chain?
*/
    if ( PF_longMultiTop->next == NULL ) {
/*
            No, allocate the new cell with a private buffer:
*/
        if ( PF_longMultiNewCell() ) return(-1);
    }
/*
        Move to the next cell in the chain:
*/
    PF_longMultiTop = PF_longMultiTop->next;
/*
        if >=0, the cell buffer is the chunk of PF_longPackBuf, initialize it:
*/
    if ( PF_longMultiTop->bufpos > -1 )
        PF_longMultiTop->buffer = PF_longPackBuf+PF_longMultiTop->bufpos;
/*
    else -- the cell has it's own private buffer.
    Initialize the cell fields:
*/
    PF_longMultiTop->nPacks  = 0;
    PF_longMultiTop->lastLen = 0;
    PF_longMultiTop->packpos = 0;
    return(0);
}
 
/*
        #] PF_longMultiPack2NextCell : 
        #[ PF_longMultiNewChunkAdded :
*/
 
static inline int PF_longMultiNewChunkAdded(int n)
{
/*
        Store the list tail:
*/
    PF_LONGMULTI *MemCell = PF_longMultiLastChunk->next;
    int pos = PF_longPackTop;
 
    while ( n-- > 0 ) {
/*
            Allocate a new cell:
*/
        PF_longMultiLastChunk->next = (PF_LONGMULTI *)
                Malloc1(sizeof(PF_LONGMULTI),"PF_longMultiCell");
        if ( PF_longMultiLastChunk->next == NULL ) return(-1);
/*
            Update the Last Chunk Pointer:
*/
        PF_longMultiLastChunk = PF_longMultiLastChunk->next;
/*
            Initialize the new cell:
*/
        PF_longMultiLastChunk->bufpos = pos;
        pos += PF_packsize;
        PF_longMultiLastChunk->buffer = NULL;
        PF_longMultiLastChunk->packpos = 0;
        PF_longMultiLastChunk->nPacks  = 0;
        PF_longMultiLastChunk->lastLen = 0;
    }
/*
        Hitch the tail:
*/
    PF_longMultiLastChunk->next = MemCell;
    return(0);
}
 
/*
        #] PF_longMultiNewChunkAdded : 
        #[ PF_longCopyChunk :
*/
 
static inline void PF_longCopyChunk(int *to, int *from, int n)
{
    NCOPYI(to,from,n)
/*  for ( ; n > 0; n-- ) *to++ = *from++; */
}
 
/*
        #] PF_longCopyChunk : 
        #[ PF_longAddChunk :
 
    The chunk must be increased by n*PF_packsize.
*/
 
static int PF_longAddChunk(int n, int mustRealloc)
{
    UBYTE *newbuf;
    if ( ( newbuf = (UBYTE*)Malloc1(sizeof(UBYTE)*(PF_longPackTop+n*PF_packsize),
                "PF_longPackBuf") ) == NULL ) return(-1);
/*
        Allocate and chain a new cell for longMulti:
*/
    if ( PF_longMultiNewChunkAdded(n) ) return(-1);
/*
        Copy the content to the new buffer:
*/
    if ( mustRealloc ) {
        PF_longCopyChunk((int*)newbuf,(int*)PF_longPackBuf,PF_longPackTop/sizeof(int));
    }
/*
        Note, PF_packsize is multiple by sizeof(int) by construction!
*/
    PF_longPackTop += (n*PF_packsize);
/*
        Free the old buffer and store the new one:
*/
    M_free(PF_longPackBuf,"PF_longPackBuf");
    PF_longPackBuf = newbuf;
/*
        Count number of re-allocs:
*/
    PF_longPackN += n;
    return(0);
}
 
/*
        #] PF_longAddChunk : 
        #[ PF_longMultiHowSplit :
 
    "count" of "type" elements in an input buffer occupy "bytes" bytes.
    We know from the algorithm, that it is too many. How to split
    the buffer so that the head fits to rest of a storage buffer?*/
static inline int PF_longMultiHowSplit(int count, MPI_Datatype type, int bytes)
{
    int ret, items, totalbytes;
 
    if ( count < 2 ) return(0); /* Nothing to split */
/*
        A rest of a storage buffer:
*/
    totalbytes = PF_packsize - PF_longMultiTop->packpos;
/*
    Rough estimate:
*/
    items = (int)((double)totalbytes*count/bytes);
/*
        Go to the up limit:
*/
    do {
        if ( ( ret = MPI_Pack_size(++items,type,PF_COMM,&bytes) )
            !=MPI_SUCCESS ) return(ret);
    } while ( bytes < totalbytes );
/*
        Now the value of "items" is too large
        And now evaluate the exact value:
*/
    do {
        if ( ( ret = MPI_Pack_size(--items,type,PF_COMM,&bytes) )
            !=MPI_SUCCESS ) return(ret);
        if ( items == 0 ) /* Nothing about MPI_Pack_size(0) == 0 in standards! */
            return(0);
    } while ( bytes > totalbytes );
    return(items);
}
/*
        #] PF_longMultiHowSplit : 
        #[ PF_longPackInit :
*/
 
static int PF_longPackInit(void)
{
    int ret;
    PF_longPackBuf = (UBYTE*)Malloc1(sizeof(UBYTE)*PF_packsize,"PF_longPackBuf");
    if ( PF_longPackBuf == NULL ) return(-1);
/*
        PF_longPackTop is not initialized yet, use in as a return value:
*/
    ret = MPI_Pack_size(1,MPI_INT,PF_COMM,&PF_longPackTop);
    if ( ret != MPI_SUCCESS ) return(ret);
 
    PF_longPackSmallBuf =
            (void*)Malloc1(sizeof(UBYTE)*PF_longPackTop,"PF_longPackSmallBuf");
 
    PF_longPackTop = PF_packsize;
    PF_longMultiRoot =
            (PF_LONGMULTI *)Malloc1(sizeof(PF_LONGMULTI),"PF_longMultiRoot");
    if ( PF_longMultiRoot == NULL ) return(-1);
    PF_longMultiRoot->bufpos = 0;
    PF_longMultiRoot->buffer = NULL;
    PF_longMultiRoot->next   = NULL;
    PF_longMultiLastChunk    = PF_longMultiRoot;
 
    PF_longPackPos = 0;
    PF_longMultiRoot->packpos = 0;
    PF_longMultiTop = PF_longMultiRoot;
    PF_longPackN = 1;
    return(0);
}
 
/*
        #] PF_longPackInit : 
        #[ PF_longMultiPreparePrefix :
*/
 
static inline int PF_longMultiPreparePrefix(void)
{
    int ret;
    PF_LONGMULTI *thePrefix;
    int i = PF_longPackN;
/*
        Here we have PF_longPackN>1!
        New cell (at the list end) to create the auxiliary chunk:
*/
    if ( PF_longMultiPack2NextCell() ) return(-1);
/*
        Store the pointer to the chunk we will proceed:
*/
    thePrefix = PF_longMultiTop;
/*
        Pack PF_longPackN:
*/
    ret = MPI_Pack(&(PF_longPackN),
                   1,
                   MPI_INT,
                   thePrefix->buffer,
                   PF_packsize,
                   &(thePrefix->packpos),
                   PF_COMM);
    if ( ret != MPI_SUCCESS ) return(ret);
/*
        And start from the beginning:
*/
    for ( PF_longMultiTop = PF_longMultiRoot; i > 0; i-- ) {
/*
            Pack number of Pack hits:
*/
        ret = MPI_Pack(&(PF_longMultiTop->nPacks),
                       1,
                       MPI_INT,
                       thePrefix->buffer,
                       PF_packsize,
                       &(thePrefix->packpos),
                       PF_COMM);
/*
                Pack the length of the last fit portion:
*/
        ret |= MPI_Pack(&(PF_longMultiTop->lastLen),
                        1,
                        MPI_INT,
                        thePrefix->buffer,
                        PF_packsize,
                        &(thePrefix->packpos),
                        PF_COMM);
/*
                Check the size -- not necessary, MPI_Pack did it.
*/
        if (  ret != MPI_SUCCESS ) return(ret);
/*
            Go to the next cell:
*/
        PF_longMultiTop = PF_longMultiTop->next;
    }
 
    PF_longMultiTop = thePrefix;
/*
        PF_longMultiTop is ready!
*/
    return(0);
}
 
/*
        #] PF_longMultiPreparePrefix : 
        #[ PF_longMultiProcessPrefix :
*/
 
static inline int PF_longMultiProcessPrefix(void)
{
    int ret,i;
/*
        We have PF_longPackN records packed in PF_longMultiRoot->buffer,
        pairs nPacks and lastLen. Loop through PF_longPackN cells,
        unpacking these integers into proper fields:
*/
    for ( PF_longMultiTop = PF_longMultiRoot, i = 0; i < PF_longPackN; i++ ) {
/*
            Go to th next cell, allocating, when necessary:
*/
        if ( PF_longMultiPack2NextCell() ) return(-1);
/*
            Unpack the number of Pack hits:
*/
        ret = MPI_Unpack(PF_longMultiRoot->buffer,
                            PF_packsize,
                            &(  PF_longMultiRoot->packpos),
                            &(PF_longMultiTop->nPacks),
                            1,
                            MPI_INT,
                            PF_COMM);
        if ( ret != MPI_SUCCESS ) return(ret);
/*
            Unpack the length of the last fit portion:
*/
        ret = MPI_Unpack(PF_longMultiRoot->buffer,
                            PF_packsize,
                            &(  PF_longMultiRoot->packpos),
                            &(PF_longMultiTop->lastLen),
                            1,
                            MPI_INT,
                            PF_COMM);
        if ( ret != MPI_SUCCESS ) return(ret);
    }
    return(0);
}
 
/*
        #] PF_longMultiProcessPrefix : 
        #[ PF_longSingleReset :
*/
 
static inline int PF_longSingleReset(int is_sender)
{
    int ret;
    PF_longPackPos=0;
    if ( is_sender ) {
        ret = MPI_Pack(&PF_longPackTop,1,MPI_INT,
            PF_longPackBuf,PF_longPackTop,&PF_longPackPos,PF_COMM);
        if ( ret != MPI_SUCCESS ) return(ret);
        PF_longPackN = 1;
    }
    else {
        PF_longPackN=0;
    }
    return(0);
}
 
/*
        #] PF_longSingleReset : 
        #[ PF_longMultiReset :
*/
 
static inline int PF_longMultiReset(int is_sender)
{
    int ret = 0, theone = 1;
    PF_longMultiRoot->packpos = 0;
    if ( is_sender ) {
        ret = MPI_Pack(&theone,1,MPI_INT,
            PF_longPackBuf,PF_longPackTop,&(PF_longMultiRoot->packpos),PF_COMM);
        PF_longPackN = 1;
    }
    else {
        PF_longPackN = 0;
    }
    PF_longMultiRoot->nPacks = 0;   /* The auxiliary field is not counted */
    PF_longMultiRoot->lastLen = 0;
    PF_longMultiTop = PF_longMultiRoot;
    PF_longMultiRoot->buffer = PF_longPackBuf;
    return ret;
}
 
/*
        #] PF_longMultiReset : 
        #] Long pack private functions : 
        #[ PF_PrepareLongSinglePack :
*/
 
int PF_PrepareLongSinglePack(void)
{
    return PF_longSingleReset(1);
}
 
/*
        #] PF_PrepareLongSinglePack : 
        #[ PF_LongSinglePack :
*/
 
int PF_LongSinglePack(const void *buffer, size_t count, MPI_Datatype type)
{
    int ret, bytes;
    /* XXX: Limited by int size. */
    if ( count > INT_MAX ) return -99;
    ret = MPI_Pack_size((int)count,type,PF_COMM,&bytes);
    if ( ret != MPI_SUCCESS ) return(ret);
 
    while ( PF_longPackPos+bytes > PF_longPackTop ) {
        if ( PF_longAddChunk(1, 1) ) return(-1);
    }
/*
        PF_longAddChunk(1, 1) means, the chunk must
        be increased by 1 and re-allocated
*/
    ret = MPI_Pack((void *)buffer,(int)count,type,
                   PF_longPackBuf,PF_longPackTop,&PF_longPackPos,PF_COMM);
    if ( ret != MPI_SUCCESS ) return(ret);
    return(0);
}
 
/*
        #] PF_LongSinglePack : 
        #[ PF_LongSingleUnpack :
*/
 
int PF_LongSingleUnpack(void *buffer, size_t count, MPI_Datatype type)
{
    int ret;
    /* XXX: Limited by int size. */
    if ( count > INT_MAX ) return -99;
    ret = MPI_Unpack(PF_longPackBuf,PF_longPackTop,&PF_longPackPos,
                     buffer,(int)count,type,PF_COMM);
    if ( ret != MPI_SUCCESS ) return(ret);
    return(0);
}
 
/*
        #] PF_LongSingleUnpack : 
        #[ PF_LongSingleSend :
*/
 
int PF_LongSingleSend(int to, int tag)
{
    int ret, pos = 0;
/*
        Note, here we assume that this function couldn't be used
        with to == PF_ANY_SOURCE!
*/
    if ( PF_longPackN > 1 ) {
        /* The buffer was incremented, pack send the new size first: */
        int tmp = -PF_longPackTop;
/*
            Negative value means there will be the second buffer
*/
        ret = MPI_Pack(&tmp, 1,PF_INT,
                       PF_longPackSmallBuf,PF_longPackTop,&pos,PF_COMM);
        if ( ret != MPI_SUCCESS ) return(ret);
        ret = MPI_Ssend(PF_longPackSmallBuf,pos,MPI_PACKED,to,tag,PF_COMM);
        if ( ret != MPI_SUCCESS ) return(ret);
    }
    ret = MPI_Ssend(PF_longPackBuf,PF_longPackPos,MPI_PACKED,to,tag,PF_COMM);
    if ( ret != MPI_SUCCESS ) return(ret);
    return(0);
}
 
/*
        #] PF_LongSingleSend : 
        #[ PF_LongSingleReceive :
*/
 
int PF_LongSingleReceive(int src, int tag, int *psrc, int *ptag)
{
    int ret, missed, oncemore;
    MPI_Status status;
    PF_longSingleReset(0);
    do {
        ret = MPI_Recv(PF_longPackBuf,PF_longPackTop,MPI_PACKED,src,tag,
                       PF_COMM,&status);
        if ( ret != MPI_SUCCESS ) return(ret);
/*
            The source and tag must be specified here for the case if
            MPI_Recv is performed more than once:
*/
        src = status.MPI_SOURCE;
        tag = status.MPI_TAG;
        if ( psrc ) *psrc = status.MPI_SOURCE;
        if ( ptag ) *ptag = status.MPI_TAG;
/*
            Now we got either small buffer with the new PF_longPackTop,
            or just a regular chunk.
*/
        ret = MPI_Unpack(PF_longPackBuf,PF_longPackTop,&PF_longPackPos,
                         &missed,1,MPI_INT,PF_COMM);
        if ( ret != MPI_SUCCESS ) return(ret);
 
        if ( missed < 0 ) { /* The small buffer was received. */
            oncemore = 1; /* repeat receiving afterwards */
                          /* Reallocate the buffer and get the data */
            missed = -missed;
/*
                restore after unpacking small from buffer:
*/
            PF_longPackPos = 0;
        }
        else {
            oncemore = 0;  /* That's all, no repetition */
        }
        if ( missed > PF_longPackTop ) {
            /*
             * The room must be increased. We need a re-allocation for the
             * case that there is no repetition.
             */
            if ( PF_longAddChunk( (missed-PF_longPackTop)/PF_packsize, !oncemore ) )
                return(-1);
        }
    } while ( oncemore );
    return(0);
}
 
/*
        #] PF_LongSingleReceive : 
        #[ PF_PrepareLongMultiPack :
*/
 
int PF_PrepareLongMultiPack(void)
{
    return PF_longMultiReset(1);
}
 
/*
        #] PF_PrepareLongMultiPack : 
        #[ PF_LongMultiPackImpl :
*/
 
int PF_LongMultiPackImpl(const void*buffer, size_t count, size_t eSize, MPI_Datatype type)
{
    int ret, items;
 
    /* XXX: Limited by int size. */
    if ( count > INT_MAX ) return -99;
 
    ret = MPI_Pack_size((int)count,type,PF_COMM,&items);
    if ( ret != MPI_SUCCESS ) return(ret);
 
    if ( PF_longMultiTop->packpos + items <= PF_packsize ) {
        ret = MPI_Pack((void *)buffer,(int)count,type,PF_longMultiTop->buffer,
                       PF_packsize,&(PF_longMultiTop->packpos),PF_COMM);
        if ( ret != MPI_SUCCESS ) return(ret);
        PF_longMultiTop->nPacks++;
        return(0);
    }
/*
        The data do not fit to the rest of the buffer.
        There are two possibilities here: go to the next cell
        immediately, or first try to pack some portion. The function
        PF_longMultiHowSplit() returns the number of items could be
        packed in the end of the current cell:
*/
    if ( ( items = PF_longMultiHowSplit((int)count,type,items) ) < 0 ) return(items);
 
    if ( items > 0 ) {   /* store the head */
        ret = MPI_Pack((void *)buffer,items,type,PF_longMultiTop->buffer,
                       PF_packsize,&(PF_longMultiTop->packpos),PF_COMM);
        if ( ret != MPI_SUCCESS ) return(ret);
        PF_longMultiTop->nPacks++;
        PF_longMultiTop->lastLen = items;
    }
/*
        Now the rest should be packed to the new cell.
        Slide to the new cell:
*/
    if ( PF_longMultiPack2NextCell() ) return(-1);
    PF_longPackN++;
/*
        Pack the rest to the next cell:
*/
    return(PF_LongMultiPackImpl((char *)buffer+items*eSize,count-items,eSize,type));
}
 
/*
        #] PF_LongMultiPackImpl : 
        #[ PF_LongMultiUnpackImpl :
*/
 
int PF_LongMultiUnpackImpl(void *buffer, size_t count, size_t eSize, MPI_Datatype type)
{
    int ret;
 
    /* XXX: Limited by int size. */
    if ( count > INT_MAX ) return -99;
 
    if ( PF_longPackN < 2 ) { /* Just unpack the buffer from the single cell */
        ret = MPI_Unpack(
                    PF_longMultiTop->buffer,
                    PF_packsize,
                    &(PF_longMultiTop->packpos),
                    buffer,
                    count,type,PF_COMM);
        if ( ret != MPI_SUCCESS ) return(ret);
        return(0);
    }
/*
        More than one cell is in use.
*/
    if ( ( PF_longMultiTop->nPacks > 1 )     /* the cell is not expired */
        ||          /* The last cell contains exactly required portion: */
        ( ( PF_longMultiTop->nPacks == 1 ) && ( PF_longMultiTop->lastLen == 0 ) )
    ) {    /* Just unpack the buffer from the current cell */
        ret = MPI_Unpack(
                    PF_longMultiTop->buffer,
                    PF_packsize,
                    &(PF_longMultiTop->packpos),
                    buffer,
                    count,type,PF_COMM);
        if ( ret != MPI_SUCCESS ) return(ret);
        (PF_longMultiTop->nPacks)--;
        return(0);
    }
    if ( ( PF_longMultiTop->nPacks == 1 ) && ( PF_longMultiTop->lastLen != 0 ) ) {
/*
            Unpack the head:
*/
        ret = MPI_Unpack(
                    PF_longMultiTop->buffer,
                    PF_packsize,
                    &(PF_longMultiTop->packpos),
                    buffer,
                    PF_longMultiTop->lastLen,type,PF_COMM);
        if ( ret != MPI_SUCCESS ) return(ret);
/*
            Decrement the counter by read items:
*/
        count -= PF_longMultiTop->lastLen;
        if ( count <= 0 ) return(-1);  /*Something is wrong! */
/*
            Shift the output buffer position:
*/
        buffer = (char *)buffer + PF_longMultiTop->lastLen * eSize;
        (PF_longMultiTop->nPacks)--;
    }
/*
        Here PF_longMultiTop->nPacks == 0
*/
    if ( ( PF_longMultiTop = PF_longMultiTop->next ) == NULL ) return(-1);
    return(PF_LongMultiUnpackImpl(buffer,count,eSize,type));
}
 
/*
        #] PF_LongMultiUnpackImpl : 
        #[ PF_LongMultiBroadcast :
*/
 
int PF_LongMultiBroadcast(void)
{
    int ret, i;
 
    if ( PF.me == MASTER ) {
/*
            PF_longPackN is the number of packed chunks. If it is more
            than 1, we have to pack a new one and send it first
*/
        if ( PF_longPackN > 1 ) {
            if ( PF_longMultiPreparePrefix() ) return(-1);
            ret = MPI_Bcast((void*)PF_longMultiTop->buffer,
                            PF_packsize,MPI_PACKED,MASTER,PF_COMM);
            if ( ret != MPI_SUCCESS ) return(ret);
/*
                PF_longPackN was not incremented by PF_longMultiPreparePrefix()!
*/
        }
/*
            Now we start from the beginning:
*/
        PF_longMultiTop = PF_longMultiRoot;
/*
            Just broadcast all the chunks:
*/
        for ( i = 0; i < PF_longPackN; i++ ) {
            ret = MPI_Bcast((void*)PF_longMultiTop->buffer,
                            PF_packsize,MPI_PACKED,MASTER,PF_COMM);
            if ( ret != MPI_SUCCESS ) return(ret);
            PF_longMultiTop = PF_longMultiTop->next;
        }
        return(0);
    }
/*
        else - the slave
*/
    PF_longMultiReset(0);
/*
        Get the first chunk; it can be either the only data chunk, or
        an auxiliary chunk, if the data do not fit the single chunk:
*/
    ret = MPI_Bcast((void*)PF_longMultiRoot->buffer,
                    PF_packsize,MPI_PACKED,MASTER,PF_COMM);
    if ( ret != MPI_SUCCESS ) return(ret);
 
    ret = MPI_Unpack((void*)PF_longMultiRoot->buffer,
                     PF_packsize,
                     &(PF_longMultiRoot->packpos),
                     &PF_longPackN,1,MPI_INT,PF_COMM);
    if ( ret != MPI_SUCCESS ) return(ret);
/*
        Now in PF_longPackN we have the number of cells used
        for broadcasting. If it is >1, then we have to allocate
        enough cells, initialize them and receive all the chunks.
*/
    if ( PF_longPackN < 2 ) /* That's all, the single chunk is received. */
        return(0);
/*
        Here we have to get PF_longPackN chunks. But, first,
        initialize cells by info from the received auxiliary chunk.
*/
    if ( PF_longMultiProcessPrefix() ) return(-1);
/*
        Now we have free PF_longPackN cells, starting
        from PF_longMultiRoot->next,  with properly initialized
        nPacks and lastLen fields. Get chunks:
*/
    for ( PF_longMultiTop = PF_longMultiRoot->next, i = 0; i < PF_longPackN; i++ ) {
        ret = MPI_Bcast((void*)PF_longMultiTop->buffer,
                        PF_packsize,MPI_PACKED,MASTER,PF_COMM);
        if ( ret != MPI_SUCCESS ) return(ret);
        if ( i == 0 ) {   /* The first chunk, it contains extra "1". */
            int tmp;
/*
                Extract this 1 into tmp and forget about it.
*/
            ret = MPI_Unpack((void*)PF_longMultiTop->buffer,
                             PF_packsize,
                             &(PF_longMultiTop->packpos),
                             &tmp,1,MPI_INT,PF_COMM);
            if ( ret != MPI_SUCCESS ) return(ret);
        }
        PF_longMultiTop = PF_longMultiTop->next;
    }
/*
        multiUnPack starts with PF_longMultiTop, skip auxiliary chunk in
        PF_longMultiRoot:
*/
    PF_longMultiTop = PF_longMultiRoot->next;
    return(0);
}
 
/*
        #] PF_LongMultiBroadcast : 
    #] Long pack stuff : 
*/