Commit d3805219 authored by Thomas Jahns's avatar Thomas Jahns 🤸
Browse files

Fix logic/variable scoping problems in pio backend code.

parent 713bd44f
...@@ -54,7 +54,7 @@ fileIDTest(void *a, void *fileID) ...@@ -54,7 +54,7 @@ fileIDTest(void *a, void *fileID)
return ((bFiledataPA *)a)->fileID == (int)(intptr_t)fileID; return ((bFiledataPA *)a)->fileID == (int)(intptr_t)fileID;
} }
int nPrefStreams = 4; static int nPrefStreams = 4;
/***************************************************************/ /***************************************************************/
...@@ -62,7 +62,6 @@ static bFiledataPA * ...@@ -62,7 +62,6 @@ static bFiledataPA *
initBFiledataPA(char *filename, size_t bs, int nc) initBFiledataPA(char *filename, size_t bs, int nc)
{ {
bFiledataPA *bfd; bFiledataPA *bfd;
int i;
xdebug ( "filename=%s, buffersize=%zu, ncollectors=%d, nPrefetchStreams=%d", xdebug ( "filename=%s, buffersize=%zu, ncollectors=%d, nPrefetchStreams=%d",
filename, bs, nc, nPrefStreams ); filename, bs, nc, nPrefStreams );
...@@ -77,7 +76,7 @@ initBFiledataPA(char *filename, size_t bs, int nc) ...@@ -77,7 +76,7 @@ initBFiledataPA(char *filename, size_t bs, int nc)
bfd->ctrlBlks = xcalloc((size_t)nPrefStreams, sizeof (bfd->ctrlBlks[0])); bfd->ctrlBlks = xcalloc((size_t)nPrefStreams, sizeof (bfd->ctrlBlks[0]));
for ( i = 0; i < nPrefStreams; i++ ) for (int i = 0; i < nPrefStreams; i++ )
{ {
bfd->ctrlBlks[i].aio_fildes = bfd->handle; bfd->ctrlBlks[i].aio_fildes = bfd->handle;
bfd->ctrlBlks[i].aio_buf = bfd->fb->buffer + (size_t)i * bs; bfd->ctrlBlks[i].aio_buf = bfd->fb->buffer + (size_t)i * bs;
...@@ -246,11 +245,6 @@ void pioWriterAIO(void) ...@@ -246,11 +245,6 @@ void pioWriterAIO(void)
{ {
bFiledataPA *bfd; bFiledataPA *bfd;
listSet * bibBFiledataPA; listSet * bibBFiledataPA;
long amount, buffersize;
char *messageBuffer, *pMB, *filename, *temp;
int messagesize, source, tag, id;
struct fileOpTag rtag;
MPI_Status status;
MPI_Comm commNode = commInqCommNode (); MPI_Comm commNode = commInqCommNode ();
int nProcsCollNode = commInqSizeNode () - commInqSizeColl (); int nProcsCollNode = commInqSizeNode () - commInqSizeColl ();
bool * sentFinalize, doFinalize; bool * sentFinalize, doFinalize;
...@@ -262,15 +256,17 @@ void pioWriterAIO(void) ...@@ -262,15 +256,17 @@ void pioWriterAIO(void)
sentFinalize = xmalloc((size_t)nProcsCollNode * sizeof (sentFinalize[0])); sentFinalize = xmalloc((size_t)nProcsCollNode * sizeof (sentFinalize[0]));
for ( ;; ) for ( ;; )
{ {
xmpiStat ( MPI_Probe ( MPI_ANY_SOURCE, MPI_ANY_TAG, commNode, MPI_Status status;
&status ), &status ); xmpiStat(MPI_Probe(MPI_ANY_SOURCE, MPI_ANY_TAG, commNode,
&status ), &status);
source = status.MPI_SOURCE; int source = status.MPI_SOURCE;
tag = status.MPI_TAG; int tag = status.MPI_TAG;
rtag = decodeFileOpTag(tag); struct fileOpTag rtag = decodeFileOpTag(tag);
xmpi ( MPI_Get_count ( &status, MPI_CHAR, &messagesize )); int messagesize;
xmpi (MPI_Get_count(&status, MPI_CHAR, &messagesize));
xdebug ( "receive message from source=%d, id=%d, command=%d ( %s ), " xdebug ( "receive message from source=%d, id=%d, command=%d ( %s ), "
"messagesize=%d", source, rtag.id, rtag.command, "messagesize=%d", source, rtag.id, rtag.command,
...@@ -279,48 +275,48 @@ void pioWriterAIO(void) ...@@ -279,48 +275,48 @@ void pioWriterAIO(void)
switch (rtag.command) switch (rtag.command)
{ {
case IO_Open_file: case IO_Open_file:
{
char *messageBuffer = xmalloc((size_t)messagesize
* sizeof (messageBuffer[0]));
char *pMB = messageBuffer;
messageBuffer = ( char *) xmalloc ( messagesize * xmpi(MPI_Recv(messageBuffer, messagesize, MPI_CHAR, source,
sizeof ( messageBuffer[0] )); tag, commNode, &status ));
pMB = messageBuffer;
xmpi ( MPI_Recv ( messageBuffer, messagesize, MPI_CHAR, source,
tag, commNode, &status ));
filename = strtok ( pMB, token ); char *filename = strtok(pMB, token);
pMB += ( strlen ( filename ) + 1 ); pMB += (strlen(filename) + 1);
temp = strtok ( pMB, token ); char *temp = strtok(pMB, token);
buffersize = strtol ( temp, NULL, 16 ); long buffersize = strtol(temp, NULL, 16);
pMB += ( strlen ( temp ) + 1 ); pMB += (strlen(temp) + 1);
amount = ( long ) ( messageBuffer + messagesize - pMB ); size_t amount = (size_t)(messageBuffer + messagesize - pMB);
xdebug("command %s, filename=%s, buffersize=%ld, amount=%ld", xdebug("command %s, filename=%s, buffersize=%ld, amount=%zd",
command2charP[rtag.command], filename, buffersize, amount); command2charP[rtag.command], filename, buffersize, amount);
if (!(bfd = listSetGet(bibBFiledataPA, fileIDTest, if (!(bfd = listSetGet(bibBFiledataPA, fileIDTest,
(void *)(intptr_t)rtag.id))) (void *)(intptr_t)rtag.id)))
{ {
listSetForeach(bibBFiledataPA, elemCheck, filename); listSetForeach(bibBFiledataPA, elemCheck, filename);
bfd = initBFiledataPA(filename, buffersize, nProcsCollNode); bfd = initBFiledataPA(filename, (size_t)buffersize, nProcsCollNode);
if ((id = listSetAdd(bibBFiledataPA, bfd)) < 0) int id;
xabort("fileID=%d not unique", rtag.id); if ((id = listSetAdd(bibBFiledataPA, bfd)) < 0)
bfd->fileID = id; xabort("fileID=%d not unique", rtag.id);
} bfd->fileID = id;
else }
if (strcmp(filename, bfd->name) != 0) else
xabort("filename is not consistent, fileID=%d", rtag.id); if (strcmp(filename, bfd->name) != 0)
xabort("filename is not consistent, fileID=%d", rtag.id);
bfd->currOpIndex = bfd->nextOpIndex;
bfd->nextOpIndex = ( bfd->nextOpIndex + 1 ) % nPrefStreams;
xassert(amount >= 0); bfd->currOpIndex = bfd->nextOpIndex;
memcpy((void *)bfd->ctrlBlks[bfd->currOpIndex].aio_buf, bfd->nextOpIndex = ( bfd->nextOpIndex + 1 ) % nPrefStreams;
pMB, (size_t)amount);
writePA ( bfd, amount ); memcpy((void *)bfd->ctrlBlks[bfd->currOpIndex].aio_buf,
pMB, (size_t)amount);
free ( messageBuffer ); writePA(bfd, amount);
free(messageBuffer);
}
break; break;
case IO_Send_buffer: case IO_Send_buffer:
...@@ -373,11 +369,7 @@ void pioWriterAIO(void) ...@@ -373,11 +369,7 @@ void pioWriterAIO(void)
sentFinalize[source] = true; sentFinalize[source] = true;
doFinalize = true; doFinalize = true;
for ( collID = 0; collID < nProcsCollNode; collID++ ) for ( collID = 0; collID < nProcsCollNode; collID++ )
if ( !sentFinalize[collID] ) doFinalize &= sentFinalize[collID];
{
doFinalize = false;
break;
}
if ( doFinalize ) if ( doFinalize )
{ {
if (!listSetIsEmpty(bibBFiledataPA)) if (!listSetIsEmpty(bibBFiledataPA))
......
...@@ -245,7 +245,10 @@ pioSendOpen(const char *filename) ...@@ -245,7 +245,10 @@ pioSendOpen(const char *filename)
{ {
remoteFileBuf *afd; remoteFileBuf *afd;
static long buffersize = 0; static long buffersize = 0;
int root = 0, id, iret, messageLength = 32; int root = 0, id, iret;
enum {
messageLength = 32,
};
char message[messageLength]; char message[messageLength];
MPI_Comm commCollectors = commInqCommColl(); MPI_Comm commCollectors = commInqCommColl();
......
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment