---

dataservermpi.cc

Go to the documentation of this file.

//**************************************************************************
#include "dataservermpi.h"
//**************************************************************************
#ifdef MPI
#include "mpi.h"
#endif
//**************************************************************************
DataServerMPI::DataServerMPI() : DataServer()
{
}
//**************************************************************************
DataServerMPI::DataServerMPI(int port) : DataServer()
{
Start(port);
}
//**************************************************************************
DataServerMPI::~DataServerMPI()
{
#ifdef MPI
MPI_Comm_free(&comm_dataserver);
#endif
}
//**************************************************************************
int DataServerMPI::Start(int port)
{
#ifdef MPI
MPI_Comm_dup(MPI_COMM_WORLD, &comm_dataserver);
MPI_Comm_rank(comm_dataserver, &procID);
MPI_Comm_size(comm_dataserver, &numCPUs);
#else
procID = 0;
numCPUs = 1;
#endif
portNum = port+procID;
char hostname[100];
int hostnamelength;
char (*rbuf)[100] = NULL;
#ifdef MPI
if (isMaster())
        ALLOC1D(&rbuf, numCPUs);
MPI_Get_processor_name(hostname,&hostnamelength);
MPI_Gather(hostname, 100, MPI_BYTE, rbuf, 100, MPI_BYTE, 0, comm_dataserver);
if (isMaster())
        {
         ofstream file("DataServer.out");
         for (int i = 0; i < numCPUs; i++)
                file << "Proc " << i << ": " << rbuf[i] << " => " << port+i << endl << flush;
         file.close();
         FREE1D(&rbuf, numCPUs);
        }
#endif
return DataServer::Start(portNum);
}
//**************************************************************************
int DataServerMPI::Wait(char *key)
{
if (isMaster())
        DataServer::Wait(key);
Barrier();
return 0;
}
//**************************************************************************
int DataServerMPI::Post(char *key)
{
if (isMaster())
        DataServer::Post(key);
return 0;
}
//**************************************************************************
int DataServerMPI::Synchronize(char *filename)
{
int update_flag;
if (isMaster())
        update_flag = IsDirty(0);
BroadcastVariable(&update_flag);

if (!update_flag)
        return POSSE_SUCCESS;

for (int i = 0; i < NumKeys(); i++)
        {
         if (keytable->isViewableVar(i) && keytable->isWritable(i))
                BroadcastData(keytable->key[i].ptr, keytable->key[i].elem_size);
         if (keytable->isWritable(i) && (callback || keytable->hasCallback(i)))
                {
                 update_flag = IsDirtyID(i, 0);
                 BroadcastVariable(&update_flag);
                 if (update_flag)
                        {
                         if (callback)
                                callback(Key(i));
                         if (keytable->hasCallback(i))
                                keytable->Callback(i);
                        }
                }
        }

if (isMaster())
        DataServer::Synchronize(filename);

Barrier();

return POSSE_SUCCESS;
}
//**************************************************************************
void DataServerMPI::BroadcastData(void *var, int size)
{
#ifdef MPI
MPI_Bcast(var, size, MPI_BYTE, 0, comm_dataserver);
#endif
}
//**************************************************************************
void DataServerMPI::Barrier()
{
#ifdef MPI
MPI_Barrier(comm_dataserver);
#endif
}
//**************************************************************************
int DataServerMPI::RegisterInternal(char *keyword, RemoteSocket *C, int &error_flag, bool reg_flag)
{
// Register_Variable(C, error_flag, reg_flag, keyword, "MPI_nprocs", "ro", numCPUs);
// Register_Variable(C, error_flag, reg_flag, keyword, "MPI_procID", "ro", procID);
REGISTER_VARIABLE("MPI_nprocs", "ro", numCPUs);
REGISTER_VARIABLE("MPI_procID", "ro", procID);

return 0;
}
//**************************************************************************

---