pio_interface.c 26 KB
Newer Older
1
2
3
4
#ifdef HAVE_CONFIG_H
#  include "config.h"
#endif

5
#include <limits.h>
Deike Kleberg's avatar
Deike Kleberg committed
6
#include <stdlib.h>
7
#include <stdio.h>
8
#include <stdarg.h>
9
10
11

#ifdef USE_MPI
#include <mpi.h>
12
#include <yaxt.h>
13
14
#endif

15
#include "cdi.h"
16
#include "cdipio.h"
17
#include "dmemory.h"
Deike Kleberg's avatar
Deike Kleberg committed
18
#include "pio_util.h"
Deike Kleberg's avatar
Deike Kleberg committed
19
#include "vlist_var.h"
20
21

#ifdef USE_MPI
Deike Kleberg's avatar
Deike Kleberg committed
22
#include "namespace.h"
23
#include "pio.h"
24
#include "pio_serialize.h"
25
#include "pio_interface.h"
Deike Kleberg's avatar
Deike Kleberg committed
26
#include "pio_comm.h"
Deike Kleberg's avatar
Deike Kleberg committed
27
#include "pio_rpc.h"
28
#include "pio_client.h"
Deike Kleberg's avatar
Deike Kleberg committed
29
#include "pio_server.h"
30
#include "resource_handle.h"
31
#include "cdi_int.h"
Deike Kleberg's avatar
Deike Kleberg committed
32
#include "vlist.h"
33

Deike Kleberg's avatar
Deike Kleberg committed
34
extern resOps streamOps;
Deike Kleberg's avatar
Deike Kleberg committed
35

36

37
static struct rdmaWin
38
39
{
  size_t size;
40
  unsigned char *buffer, *head;
41
  MPI_Win win;
42
  int postSet, refuseFuncCall;
43
  MPI_Group ioGroup;
44
  int dictSize, dictDataUsed, dictRPCUsed, dict;
45
46
} *txWin = NULL;

Deike Kleberg's avatar
Deike Kleberg committed
47

48
49
50
const char * const funcMap[numRPCFuncs] = {
  "streamOpen",
  "streamDefVlist",
51
52
  "streamClose",
  "streamDefTimestep",
53
};
Deike Kleberg's avatar
Deike Kleberg committed
54

55
float cdiPIOpartInflate_;
Deike Kleberg's avatar
Deike Kleberg committed
56

Deike Kleberg's avatar
Deike Kleberg committed
57
/****************************************************/
Deike Kleberg's avatar
Deike Kleberg committed
58

59
static int cmp ( const void * va, const void * vb )
Deike Kleberg's avatar
Deike Kleberg committed
60
{
61
62
    const int ** a, ** b;

Deike Kleberg's avatar
Deike Kleberg committed
63
64
  a = ( const int ** ) va;
  b = ( const int ** ) vb;
65

Deike Kleberg's avatar
Deike Kleberg committed
66
67
68
  return (( **a < **b ) - ( **a > **b ));
}

69
70
71
72
73
74
75
76
77
void memcpyPackFunc(void *dataDesc, void *buf, int size, int *pos,
                    void *context)
{
  struct memCpyDataDesc *p = dataDesc;
  xassert(size >= *pos && (size_t)(size - *pos) >= p->obj_size);
  memcpy((unsigned char *)buf + *pos, p->obj, p->obj_size);
  *pos += (int)p->obj_size;
}

Deike Kleberg's avatar
Deike Kleberg committed
78
/****************************************************/
79

80
81
82
static void
mapProblems(int problemSizes[], int * problemMapping, int nProblems,
            int nWriter, double * w)
Deike Kleberg's avatar
Deike Kleberg committed
83
84
85
86
87
88
89
{
  int *ip[nProblems];
  int dummy[nProblems];
  int buckets[nWriter];
  int currCapacity, nextCapacity;
  double meanBucket[nWriter];
  int sum = 0;
Thomas Jahns's avatar
Thomas Jahns committed
90
  int writerIdx = 0;
Deike Kleberg's avatar
Deike Kleberg committed
91
  int i, j;
92

Deike Kleberg's avatar
Deike Kleberg committed
93
94
95
96
97
98
99
100
101
102
103

  for ( i = 0; i < nProblems; i++ )
    {
      ip[i] = &problemSizes[i];
      sum += problemSizes[i];
    }

  qsort ( ip, nProblems, sizeof ( int * ), cmp );

  for ( i = 0; i < nProblems; i++ )
    dummy[i] = ip[i] - problemSizes;
104
105

  for ( j = 0; j < nWriter; j++ )
Deike Kleberg's avatar
Deike Kleberg committed
106
107
    meanBucket[j] = ( double ) sum * ( * ( w + j ));

Thomas Jahns's avatar
Thomas Jahns committed
108
109
  memset(buckets, 0, sizeof (buckets));

Deike Kleberg's avatar
Deike Kleberg committed
110
111
112
  for ( i = 0; i < nProblems; i++ )
    {
      currCapacity = INT_MIN;
113

Deike Kleberg's avatar
Deike Kleberg committed
114
      for ( j = 0; j < nWriter; j++ )
115
116
	{
	  nextCapacity = meanBucket[j] - ( buckets[j] + ( *ip[i] ));
117

118
119
120
121
122
123
	  if ( nextCapacity > currCapacity )
	    {
	      currCapacity = nextCapacity;
	      writerIdx = j;
	    }
	}
Thomas Jahns's avatar
Thomas Jahns committed
124
      problemMapping[ dummy[i] ] = writerIdx;
Deike Kleberg's avatar
Deike Kleberg committed
125
126
      buckets[writerIdx] +=  *ip[i];
    }
Deike Kleberg's avatar
Deike Kleberg committed
127

128
129
130
131
132
133
134
135
136
  if ( ddebug )
    {
      xprintArray3 (  "problemSizes = ", problemSizes, nProblems, DATATYPE_INT );
      xprintArray3 ( "vector of indices, qsort of problemSizes", dummy,
                    nProblems, DATATYPE_INT );
      xprintArray3 ( "problemMapping", problemMapping, nProblems, DATATYPE_INT );
      xprintArray3 ( "meanBucket", meanBucket, nWriter, DATATYPE_FLT );
      xprintArray3 ( "actual buckets", buckets, nWriter, DATATYPE_INT );
    }
Deike Kleberg's avatar
Deike Kleberg committed
137
138
}

139
/****************************************************/
Deike Kleberg's avatar
Deike Kleberg committed
140
141
142
143
144

/**
   @brief is encapsulated in CDI library.

   @param vSizes array with number of levels for all var_t 's in order of streams
145

Deike Kleberg's avatar
Deike Kleberg committed
146
147
   @param sSizes array with number of var_t for all stream_t 's

148
   @param varMapping return value, array with ranks of I/O PEs assigned to var_t 's
Deike Kleberg's avatar
Deike Kleberg committed
149
150
151
in order of vSizes

   @param nStreams number of stream_t 's
152
153

   @param nodeSizes array of number of I/O PEs on each physical nodes, increasing
Deike Kleberg's avatar
Deike Kleberg committed
154
155
156
   order of ranks assumed

   @param nNodes number of physical nodes hosting I/O PEs
157

Deike Kleberg's avatar
Deike Kleberg committed
158
159
   @return
*/
160

161
162
163
static void
varMapGen(int *vSizes, int *sSizes, int *varMapping,
          int nStreams, int *nodeSizes, int nNodes)
Deike Kleberg's avatar
Deike Kleberg committed
164
{
165

Deike Kleberg's avatar
Deike Kleberg committed
166
167
168
169
170
  int weightsStreams[nStreams];
  int streamMapping[nStreams];
  int nPEs = 0, nVars = 0;
  int i, j, k, offset = 0, offsetN = 0;
  double * w;
171

Deike Kleberg's avatar
Deike Kleberg committed
172
173
174
  int * weightsVarsNode;
  int * varMappingNode;
  int nVarsNode, summandRank = 0;
175
  int nProcsColl = commInqNProcsColl ();
Deike Kleberg's avatar
Deike Kleberg committed
176

177
  int buckets[nProcsColl];
Deike Kleberg's avatar
Deike Kleberg committed
178
179
180
181
182
183

  for ( i = 0; i < nStreams; i++ )
    {
      nVars += * ( sSizes + i );
      weightsStreams[i] = 0;
      for ( j = 0; j < * ( sSizes + i ); j++ )
184
	weightsStreams[i] += * ( vSizes + offset++ );
Deike Kleberg's avatar
Deike Kleberg committed
185
186
    }

Deike Kleberg's avatar
Deike Kleberg committed
187
  w = ( double * ) xmalloc ( nNodes * sizeof ( double ));
Deike Kleberg's avatar
Deike Kleberg committed
188
189
190
191
192
193
194
195
196
197
198
199
200
  for ( j = 0; j < nNodes; j++ )
    nPEs += * ( nodeSizes + j );

  for ( j = 0; j < nNodes; j++ )
    w[j] = ( double ) * ( nodeSizes + j ) / ( double ) nPEs;

  mapProblems ( weightsStreams, streamMapping, nStreams, nNodes, w );
  free ( w );

  for ( i = 0; i < nNodes; i++ )
    {
      nVarsNode = 0;
      for ( j = 0; j < nStreams; j++ )
201
	if ( * ( streamMapping + j ) == i )
202
203
	  nVarsNode += * ( sSizes + j );

Deike Kleberg's avatar
Deike Kleberg committed
204
205
206
      weightsVarsNode = ( int * ) xmalloc ( nVarsNode * sizeof ( int ));
      varMappingNode = ( int * ) xmalloc ( nVarsNode * sizeof ( int ));
      w = ( double * ) xmalloc ( * ( nodeSizes + i ) * sizeof ( double ));
Deike Kleberg's avatar
Deike Kleberg committed
207
208
209
210
      offset = 0;
      offsetN = 0;

      for ( j = 0; j < nStreams; j++ )
211
212
213
214
215
	if ( * ( streamMapping + j ) == i )
	  for ( k = 0; k < * ( sSizes + j ); k ++ )
	    * ( weightsVarsNode + offsetN++ ) = * ( vSizes + offset++ );
	else
	  offset += * ( sSizes + j );
Deike Kleberg's avatar
Deike Kleberg committed
216
217

      for ( j = 0; j < * ( nodeSizes + i ); j ++ )
218
	w[j] = 1.0 / ( double ) * ( nodeSizes + i );
Deike Kleberg's avatar
Deike Kleberg committed
219

220
      mapProblems ( weightsVarsNode, varMappingNode, nVarsNode,
221
		    * ( nodeSizes + i ),  w );
Deike Kleberg's avatar
Deike Kleberg committed
222
223
224
225
226

      offset = 0;
      offsetN = 0;

      for ( j = 0; j < nStreams; j++ )
227
228
	if ( * ( streamMapping + j ) == i )
	  for ( k = 0; k < * ( sSizes + j ); k ++ )
229
	    * ( varMapping + offset ++ ) =
230
              commCollID2RankGlob ( * ( varMappingNode + offsetN ++ ) +
231
                                    summandRank );
232
233
	else
	  offset += * ( sSizes + j );
Deike Kleberg's avatar
Deike Kleberg committed
234
235
236
237
238
239
240

      summandRank += * ( nodeSizes + i );

      free ( w );
      free ( varMappingNode );
      free ( weightsVarsNode );
    }
241

242
  if ( ddebug )
Deike Kleberg's avatar
Deike Kleberg committed
243
    {
Deike Kleberg's avatar
Deike Kleberg committed
244
      xprintArray ( "varMapping", varMapping, nVars, DATATYPE_INT  );
245
      for ( i = 0; i < nProcsColl; i++ )
246
	buckets[i] = 0;
Deike Kleberg's avatar
Deike Kleberg committed
247
      for ( i = 0; i < nVars; i ++ )
248
	buckets[commRankGlob2CollID ( *(varMapping + i ))] += * ( vSizes + i );
249
      xprintArray ( "buckets", buckets, nProcsColl, DATATYPE_INT );
Deike Kleberg's avatar
Deike Kleberg committed
250
    }
251
}
Deike Kleberg's avatar
Deike Kleberg committed
252

253
/************************************************************************/
Deike Kleberg's avatar
Deike Kleberg committed
254

255
256
static void
varsMapNDeco(int nNodes, int *nodeSizes)
Deike Kleberg's avatar
Deike Kleberg committed
257
{
258
  int nStreams, nVars, * resHs, * streamSizes, * varSizes, * varMapping,
Thomas Jahns's avatar
Thomas Jahns committed
259
    * collectsData;
Deike Kleberg's avatar
Deike Kleberg committed
260
  int i, j, k = 0;
Deike Kleberg's avatar
Deike Kleberg committed
261
  int nProcsColl = commInqNProcsColl ();
Deike Kleberg's avatar
Deike Kleberg committed
262

263
264
  xdebug ( "START, nProcsColl=%d", nProcsColl );

Deike Kleberg's avatar
Deike Kleberg committed
265
  nStreams = streamSize ();
266

267
268
  resHs       = xmalloc ( nStreams * sizeof ( resHs[0] ));
  streamSizes = xmalloc ( nStreams * sizeof ( streamSizes[0] ));
269
  collectsData = xmalloc ( nProcsColl * sizeof ( collectsData[0] ));
270
  streamGetIndexList ( nStreams, resHs );
271

Deike Kleberg's avatar
Deike Kleberg committed
272
  for ( i = 0; i < nStreams; i++ )
273
    streamSizes[i] = streamInqNvars ( * ( resHs + i ));
274

Thomas Jahns's avatar
Thomas Jahns committed
275
  nVars = sum_int(nStreams, streamSizes);
276
277
  varSizes   = xmalloc ( nVars * sizeof ( varSizes[0] ));
  varMapping = xmalloc ( nVars * sizeof ( varMapping[0] ));
278

Deike Kleberg's avatar
Deike Kleberg committed
279
280
  for ( i = 0; i < nStreams; i++ )
    for ( j = 0; j < * ( streamSizes + i ); j++ )
281
      varSizes[k++] += vlistInqVarSize ( streamInqVlist ( * ( resHs + i )), j );
282

Deike Kleberg's avatar
Deike Kleberg committed
283
  xassert ( k == nVars );
284

285
  varMapGen ( varSizes, streamSizes, varMapping,
Deike Kleberg's avatar
Deike Kleberg committed
286
	      nStreams, nodeSizes, nNodes );
287
288
289

  k = 0;
  for ( i = 0; i < nStreams; i++ )
290
    for ( j = 0; j < * ( streamSizes + i ); j++ )
291
      {
292
        vlistDefVarIOrank ( streamInqVlist ( * ( resHs + i )), j,
293
                            * ( varMapping + k ));
294
        vlistDefVarIOrank ( streamInqVlistIDorig ( * ( resHs + i )), j,
295
296
                            * ( varMapping + k ));
        collectsData[commRankGlob2CollID ( varMapping[k++] )] = 1;
297
      }
298
299
300

  for ( j = 0; j < nProcsColl; j++ )
    if ( collectsData[j] == 0 )
301
302
303
304
      xabort("AT LEAST ONE COLLECTOR PROCESS IDLES, "
             "CURRENTLY NOT COVERED: "
             "PE%d collects no data",
             commCollID2RankGlob(j));
305

306
307
308
309
310
  if ( varMapping )   free ( varMapping );
  if ( varSizes )     free ( varSizes );
  if ( collectsData ) free ( collectsData );
  if ( streamSizes )  free ( streamSizes );
  if ( resHs )        free ( resHs );
311

312
  xdebug("%s", "RETURN");
Deike Kleberg's avatar
Deike Kleberg committed
313
314
315
316
}

/************************************************************************/

317
static
Deike Kleberg's avatar
Deike Kleberg committed
318
319
void modelWinCleanup ( void )
{
Deike Kleberg's avatar
Deike Kleberg committed
320
  int collID;
Deike Kleberg's avatar
Deike Kleberg committed
321

322
  xdebug("%s", "START");
323
  if (txWin != NULL)
324
325
326
327
328
329
330
331
332
333
334
    {
      for ( collID = 0; collID < commInqNProcsColl (); collID ++ )
        {
          if (txWin[collID].postSet)
            xmpi(MPI_Win_wait(txWin[collID].win));
          xmpi(MPI_Win_free(&txWin[collID].win));
          xmpi ( MPI_Free_mem ( txWin[collID].buffer ));
          xmpi(MPI_Group_free(&txWin[collID].ioGroup));
        }
      free(txWin);
    }
Deike Kleberg's avatar
Deike Kleberg committed
335

336
  xdebug("%s", "RETURN. CLEANED UP MPI_WIN'S");
Deike Kleberg's avatar
Deike Kleberg committed
337
338
339
}

/************************************************************************/
340

341
342
struct collDesc
{
343
  int numDataRecords, numRPCRecords;
344
345
};

346
347
static void
modelWinDefBufferSizes(void)
Deike Kleberg's avatar
Deike Kleberg committed
348
{
349
  int collID, nstreams, * streamIndexList, streamNo, nvars, varID;
350
  int sumWinBufferSize = 0;
351
352
353
  int nProcsColl  = commInqNProcsColl ();
  int rankGlob    = commInqRankGlob ();
  int root = commInqRootGlob ();
354
  struct collDesc *collIndex;
Deike Kleberg's avatar
Deike Kleberg committed
355

356
  xdebug("%s", "START");
357
  xassert(txWin != NULL);
358

Deike Kleberg's avatar
Deike Kleberg committed
359
  nstreams = reshCountType ( &streamOps );
360
  streamIndexList = xmalloc ( nstreams * sizeof ( streamIndexList[0] ));
361
  collIndex = xcalloc(nProcsColl, sizeof (collIndex[0]));
362
  reshGetResHListOfType ( nstreams, streamIndexList, &streamOps );
Deike Kleberg's avatar
Deike Kleberg committed
363
364
  for ( streamNo = 0; streamNo < nstreams; streamNo++ )
    {
365
366
367
      // memory required for data
      int streamID = streamIndexList[streamNo];
      int vlistID = streamInqVlist(streamID);
Deike Kleberg's avatar
Deike Kleberg committed
368
369
370
      nvars = vlistNvars ( vlistID );
      for ( varID = 0; varID < nvars; varID++ )
        {
371
372
373
374
375
376
377
378
          int collID = commRankGlob2CollID(vlistInqVarIOrank(vlistID, varID));
          int collIDchunk;
          {
            int varSize = vlistInqVarSize(vlistID, varID);
            int nProcsModel = commInqNProcsModel();
            collIDchunk = (int)ceilf(cdiPIOpartInflate_
                                     * (varSize + nProcsModel - 1)/nProcsModel);
          }
Deike Kleberg's avatar
Deike Kleberg committed
379
          xassert ( collID != CDI_UNDEFID && collIDchunk > 0 );
380
381
382
383
384
385
          collIndex[collID].numDataRecords += 2;
          txWin[collID].size += (size_t)collIDchunk * sizeof (double)
            /* re-align chunks to multiple of double size */
            + sizeof (double) - 1
            /* one header for data record, one for corresponding part
             * descriptor*/
386
            + 2 * sizeof (struct winHeaderEntry)
387
388
            /* FIXME: heuristic for size of packed Xt_idxlist */
            + sizeof (Xt_int) * collIDchunk * 3;
Deike Kleberg's avatar
Deike Kleberg committed
389
        }
390

391
392
      // memory required for the function calls encoded
      // for remote execution
Deike Kleberg's avatar
Deike Kleberg committed
393
      // once per stream and timestep for all collprocs only on the modelproc root
394
395
      if ( rankGlob == root )
        for ( collID = 0; collID < nProcsColl; collID++ )
396
          {
397
            collIndex[collID].numRPCRecords += numRPCFuncs;
398
            txWin[collID].size +=
399
              numRPCFuncs * sizeof (struct winHeaderEntry)
400
401
402
              + MAXDATAFILENAME
              /* data part of streamDefTimestep */
              + (2 * CDI_MAX_NAME + sizeof (taxis_t));
403
          }
Deike Kleberg's avatar
Deike Kleberg committed
404
    }
405
  for (collID = 0; collID < nProcsColl; ++collID)
406
    {
407
408
      int numRecords = 1 + collIndex[collID].numDataRecords
        + collIndex[collID].numRPCRecords;
409
      txWin[collID].dictSize = numRecords;
410
411
      txWin[collID].dictDataUsed = 1;
      txWin[collID].dictRPCUsed = 0;
412
      /* account for size header */
413
      txWin[collID].size += sizeof (struct winHeaderEntry);
414
415
      txWin[collID].size = roundUpToMultiple(txWin[collID].size,
                                             PIO_WIN_ALIGN);
416
      sumWinBufferSize += txWin[collID].size;
417
    }
418
419
420
  free(collIndex);
  free ( streamIndexList );

421
422
  xdebug("sumWinBufferSize=%zu, MAXWINBUFFERSIZE=%zu", (size_t)sumWinBufferSize,
         (size_t)MAXWINBUFFERSIZE);
Deike Kleberg's avatar
Deike Kleberg committed
423
  xassert ( sumWinBufferSize <= MAXWINBUFFERSIZE );
424
  xdebug("%s", "RETURN");
Deike Kleberg's avatar
Deike Kleberg committed
425
}
426

Deike Kleberg's avatar
Deike Kleberg committed
427
428
429
430

/************************************************************************/


431
static
Deike Kleberg's avatar
Deike Kleberg committed
432
  void modelWinFlushBuffer ( int collID )
Deike Kleberg's avatar
Deike Kleberg committed
433
{
Deike Kleberg's avatar
Deike Kleberg committed
434
  int nProcsColl = commInqNProcsColl ();
Deike Kleberg's avatar
Deike Kleberg committed
435

Deike Kleberg's avatar
Deike Kleberg committed
436
437
  xassert ( collID                >= 0         &&
            collID                < nProcsColl &&
438
            txWin != NULL      &&
Thomas Jahns's avatar
Thomas Jahns committed
439
            txWin[collID].buffer     != NULL      &&
440
            txWin[collID].size >= 0         &&
441
            txWin[collID].size <= MAXWINBUFFERSIZE);
442
  memset(txWin[collID].buffer, 0, txWin[collID].size);
443
  txWin[collID].head = txWin[collID].buffer
444
    + txWin[collID].dictSize * sizeof (struct winHeaderEntry);
445
  txWin[collID].refuseFuncCall = 0;
446
447
  txWin[collID].dictDataUsed = 1;
  txWin[collID].dictRPCUsed = 0;
Deike Kleberg's avatar
Deike Kleberg committed
448
449
450
}


Deike Kleberg's avatar
Deike Kleberg committed
451
452
453
/************************************************************************/


Deike Kleberg's avatar
Deike Kleberg committed
454
455
456
static
void modelWinCreate ( void )
{
Deike Kleberg's avatar
Deike Kleberg committed
457
  int collID, ranks[1];
458
  int nProcsColl = commInqNProcsColl ();
Deike Kleberg's avatar
Deike Kleberg committed
459

460
  xdebug("%s", "START");
461
  txWin = xmalloc(nProcsColl * sizeof (txWin[0]));
Deike Kleberg's avatar
Deike Kleberg committed
462

Deike Kleberg's avatar
Deike Kleberg committed
463
  modelWinDefBufferSizes ();
464
  ranks[0] = commInqNProcsModel ();
Deike Kleberg's avatar
Deike Kleberg committed
465

466
467
468
  MPI_Info no_locks_info;
  xmpi(MPI_Info_create(&no_locks_info));
  xmpi(MPI_Info_set(no_locks_info, "no_locks", "true"));
Deike Kleberg's avatar
Deike Kleberg committed
469
  for ( collID = 0; collID < nProcsColl; collID ++ )
Deike Kleberg's avatar
Deike Kleberg committed
470
    {
471
      xassert(txWin[collID].size > 0);
472
      txWin[collID].buffer = NULL;
473
      xmpi(MPI_Alloc_mem((MPI_Aint)txWin[collID].size, MPI_INFO_NULL,
474
475
                         &txWin[collID].buffer));
      xassert ( txWin[collID].buffer != NULL );
476
      txWin[collID].head = txWin[collID].buffer
477
        + txWin[collID].dictSize * sizeof (struct winHeaderEntry);
478
      xmpi(MPI_Win_create(txWin[collID].buffer, (MPI_Aint)txWin[collID].size, 1,
479
                          no_locks_info, commInqCommsIO(collID),
480
                          &txWin[collID].win));
Thomas Jahns's avatar
Thomas Jahns committed
481
482
483
484
      MPI_Group commGroup;
      xmpi(MPI_Comm_group(commInqCommsIO(collID), &commGroup));
      xmpi(MPI_Group_incl(commGroup, 1, ranks, &txWin[collID].ioGroup));
      xmpi(MPI_Group_free(&commGroup));
Deike Kleberg's avatar
Deike Kleberg committed
485
    }
486
487
488

  xmpi(MPI_Info_free(&no_locks_info));

489
  xdebug("%s", "RETURN, CREATED MPI_WIN'S");
Deike Kleberg's avatar
Deike Kleberg committed
490
491
492
493
}

/************************************************************************/

494
static void
495
modelWinEnqueue(int collID,
496
                struct winHeaderEntry header, const void *data,
497
                valPackFunc packFunc)
Deike Kleberg's avatar
Deike Kleberg committed
498
{
499
500
  struct winHeaderEntry *winDict
    = (struct winHeaderEntry *)txWin[collID].buffer;
501
  int targetEntry;
502
503
504
505
  if (header.id > 0 || header.id == PARTDESCMARKER)
    targetEntry = (txWin[collID].dictDataUsed)++;
  else
    targetEntry = txWin[collID].dictSize - ++(txWin[collID].dictRPCUsed);
506
  if (header.id > 0)
507
    {
508
      int offset = header.offset
509
510
        = (int)roundUpToMultiple(txWin[collID].head - txWin[collID].buffer,
                                 sizeof (double));
511
512
513
514
      MPI_Comm comm = commInqCommsIO(collID);
      packFunc((void *)data, txWin[collID].buffer, (int)txWin[collID].size,
               &offset, &comm);
      txWin[collID].head = txWin[collID].buffer + offset;
515
    }
516
  else if (header.id == PARTDESCMARKER)
517
    {
518
      Xt_uid uid = header.specific.partDesc.uid;
519
520
521
522
      int offset = -1;
      /* search if same uid entry has already been enqueued */
      for (int entry = 2; entry < targetEntry; entry += 2)
        {
523
          xassert(winDict[entry].id == PARTDESCMARKER);
524
          if (winDict[entry].specific.partDesc.uid == uid)
525
            {
526
              offset = winDict[entry].offset;
527
528
529
530
531
532
533
              break;
            }
        }
      if (offset == -1)
        {
          /* not yet used partition descriptor, serialize at
           * current position */
534
          int position = header.offset
535
            = (int)(txWin[collID].head - txWin[collID].buffer);
536
537
538
539
          MPI_Comm comm = commInqCommsIO(collID);
          packFunc((void *)data, txWin[collID].buffer, (int)txWin[collID].size,
                   &position, &comm);
          txWin[collID].head = txWin[collID].buffer + position;
540
541
542
        }
      else
        /* duplicate entries are copied only once per timestep */
543
        header.offset = offset;
544
    }
545
  else
546
    {
547
548
549
550
551
552
      int position = header.offset
        = (int)(txWin[collID].head - txWin[collID].buffer);
      MPI_Comm comm = commInqCommsIO(collID);
      packFunc((void *)data, txWin[collID].buffer, (int)txWin[collID].size,
               &position, &comm);
      txWin[collID].head = txWin[collID].buffer + position;
553
    }
554
  winDict[targetEntry] = header;
Deike Kleberg's avatar
Deike Kleberg committed
555
556
}

557
558
559
560
561
562
563
564
565
566
567
static void
cdiPio_xt_idxlist_pack_wrap(void *data, void *buf, int size, int *pos,
                            void *context)
{
  MPI_Comm comm = *(MPI_Comm *)context;
  int pack_size = xt_idxlist_get_pack_size((Xt_idxlist)data, comm);
  xassert(size >= *pos && pack_size <= size - *pos);
  xt_idxlist_pack((Xt_idxlist)data, (unsigned char *)buf,
                  size, pos, comm);
}

568
569
570
571
572
573
574
575
576
577
578
static inline void
collWait(int collID)
{
  if (txWin[collID].postSet)
    {
      xmpi(MPI_Win_wait(txWin[collID].win));
      txWin[collID].postSet = 0;
      modelWinFlushBuffer(collID);
    }
}

579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
static inline void
collProbe(int collID)
{
  if (txWin[collID].postSet)
    {
      int flag;
      xmpi(MPI_Win_test(txWin[collID].win, &flag));
      if (flag)
        {
          txWin[collID].postSet = 0;
          modelWinFlushBuffer(collID);
        }
    }
}

void
cdiPioRDMAProgress()
{
  int nProcsColl = commInqNProcsColl();
  for (int collID = 0; collID < nProcsColl; collID++)
    collProbe(collID);
}


603
604
605
606
static void
pioBufferPartData_(int streamID, int varID,
                   const void *packData, valPackFunc packDataFunc,
                   int nmiss, Xt_idxlist partDesc)
607
608
609
{
  int vlistID, collID = CDI_UNDEFID;

Deike Kleberg's avatar
Deike Kleberg committed
610
611
  vlistID  = streamInqVlist ( streamID );
  collID   = commRankGlob2CollID ( vlistInqVarIOrank ( vlistID, varID ));
612
613
  xassert ( collID         >= 0                    &&
            collID         <  commInqNProcsColl () &&
614
            txWin != NULL);
Deike Kleberg's avatar
Deike Kleberg committed
615

616
  collWait(collID);
617

618

619
  struct winHeaderEntry dataHeader
620
    = { .id = streamID, .specific.dataRecord = { varID, nmiss }, .offset = -1 };
621
  modelWinEnqueue(collID, dataHeader, packData, packDataFunc);
622
  {
623
    struct winHeaderEntry partHeader
624
625
      = { .id = PARTDESCMARKER,
          .specific.partDesc = { .uid = xt_idxlist_get_uid(partDesc) },
626
          .offset = 0 };
627
    modelWinEnqueue(collID, partHeader, partDesc, cdiPio_xt_idxlist_pack_wrap);
628
  }
629

630
  txWin[collID].refuseFuncCall = 1;
Deike Kleberg's avatar
Deike Kleberg committed
631
632
}

633
634
635
636
637
638
639
640
641
642
643
644
void
pioBufferPartData(int streamID, int varID, const double *data,
                  int nmiss, Xt_idxlist partDesc)
{
  int chunk = xt_idxlist_get_num_indices(partDesc);
  xassert(chunk <= INT_MAX);
  pioBufferPartData_(streamID, varID,
                     &(struct memCpyDataDesc){data, chunk * sizeof (data[0])},
                     memcpyPackFunc,
                     nmiss, partDesc);
}

645
646
/************************************************************************/

647
void pioBufferFuncCall(struct winHeaderEntry header,
648
                       const void *data, valPackFunc dataPackFunc)
649
650
651
652
{
  int rankGlob = commInqRankGlob ();
  int root = commInqRootGlob ();
  int collID, nProcsColl = commInqNProcsColl ();
653
  int funcID = header.id;
654

655
656
  xassert(funcID >= MINFUNCID && funcID <= MAXFUNCID);
  xdebug("%s, func: %s", "START", funcMap[(-1 - funcID)]);
657

Deike Kleberg's avatar
Deike Kleberg committed
658
659
  if ( rankGlob != root ) return;

660
  xassert(txWin != NULL);
661

Thomas Jahns's avatar
Thomas Jahns committed
662
663
  for (collID = 0; collID < nProcsColl; ++collID)
    {
664
      collWait(collID);
665
666
      xassert(txWin[collID].dictRPCUsed + txWin[collID].dictDataUsed
              < txWin[collID].dictSize);
Thomas Jahns's avatar
Thomas Jahns committed
667
      xassert(txWin[collID].refuseFuncCall == 0);
668
      modelWinEnqueue(collID, header, data, dataPackFunc);
669
    }
670

671
  xdebug("%s", "RETURN");
672
673
}

674
675
#endif

676
677
678
679
680
681

void
cdiPioNoPostCommSetup(void)
{
}

682
683
/*****************************************************************************/

684
685
686
687
/* pioInit definition must currently compile even in non-MPI configurations */
#ifndef MPI_VERSION
#  define MPI_Comm int
#endif
688
689
690
/**
   @brief initializes the MPI_Communicators needed for the
  communication between the calculator PEs and the I/O PEs and within the
691
692
693
694
695
696
697
698
699
  group of I/O PEs.

  commGlob: all PEs

  commPIO: I/O PEs, PEs with highest ranks in commGlob

  commModel: calculating PEs, no I/O PEs

  commsIO[i]:
700
701

  Collective call
702
703
704

  @param comm MPI_Communicator of all calling PEs
  @param nIOP number of I/O PEs
705
706
707
  @param partInflate allow for array partitions on comute
  PE that are at most sized \f$ partInflate * \lceil arraySize /
  numComputePEs \rceil \f$
708
709
  @param postSetupActions function which is called by all I/O servers
  after communicator split
710
711
712
  @return int indicating wether the calling PE is a calcutator (1) or not (0)
*/

Thomas Jahns's avatar
Thomas Jahns committed
713
714
#ifdef USE_MPI
static int pioNamespace_ = -1;
715
static int xtInitByCDI = 0;
Thomas Jahns's avatar
Thomas Jahns committed
716
717
#endif

718
719
MPI_Comm
pioInit(MPI_Comm commGlob, int nProcsIO, int IOMode,
720
721
        int *pioNamespace, float partInflate,
        void (*postCommSetupActions)(void))
722
{
723
#ifdef USE_MPI
Thomas Jahns's avatar
Thomas Jahns committed
724
  int sizeGlob;
Deike Kleberg's avatar
Deike Kleberg committed
725

726
727
  namespaceSwitchSet(NSSWITCH_WARNING, NSSW_FUNC(cdiPioWarning));

Deike Kleberg's avatar
Deike Kleberg committed
728
  if ( IOMode < PIO_MINIOMODE || IOMode > PIO_MAXIOMODE )
Deike Kleberg's avatar
Deike Kleberg committed
729
    xabort ( "IOMODE IS NOT VALID." );
730

731
#ifdef _SX
Deike Kleberg's avatar
Deike Kleberg committed
732
733
  if ( IOMode ==  PIO_ASYNCH )
    xabort ( "PIO_ASYNCH DOES NOT WORK ON SX." );
734
735
#endif

736
  if ((xtInitByCDI = (!xt_initialized() || xt_finalized())))
737
    xt_initialize(commGlob);
738
739
  commInit ();
  commDefCommGlob ( commGlob );
740
  sizeGlob = commInqSizeGlob ();
741

742
743
  if (((IOMode != PIO_NONE && (nProcsIO <= 0 || nProcsIO > sizeGlob - 1)))
      || (IOMode == PIO_NONE && nProcsIO != 1))
744
745
    xabort("DISTRIBUTION OF TASKS ON PROCS IS NOT VALID.\n"
           "nProcsIO=%d, sizeGlob=%d\n", nProcsIO, sizeGlob);
Deike Kleberg's avatar
Deike Kleberg committed
746
747

  commDefNProcsIO ( nProcsIO );
748
  commDefIOMode   ( IOMode );
Deike Kleberg's avatar
Deike Kleberg committed
749
  commDefCommPio  ();
750

751
752
753
  xassert(partInflate >= 1.0);
  cdiPIOpartInflate_ = partInflate;

Deike Kleberg's avatar
Deike Kleberg committed
754
  // JUST FOR TEST CASES WITH ONLY ONE MPI TASK
755
  if ( commInqSizeGlob () == 1 )
756
    {
757
      pioNamespace_ = *pioNamespace = namespaceNew();
Deike Kleberg's avatar
Deike Kleberg committed
758
759
      return commInqCommGlob ();
    }
760

761
762
  if ( commInqIsProcIO ())
    {
763
      serializeSetMPI();
764
765
766
      namespaceSwitchSet(NSSWITCH_ABORT, NSSW_FUNC(cdiAbortC_MPI));
      namespaceSwitchSet(NSSWITCH_FILE_OPEN, NSSW_FUNC(pioFileOpen));
      namespaceSwitchSet(NSSWITCH_FILE_CLOSE, NSSW_FUNC(pioFileClose));
767
      IOServer(postCommSetupActions);
Thomas Jahns's avatar
Thomas Jahns committed
768
      namespaceDelete(0);
769
      namespaceNew();
770
      commDestroy ();
771
772
773
      if (xtInitByCDI)
        xt_finalize();
      return MPI_COMM_NULL;
774
775
    }
  else
776
    cdiPioClientSetup(&pioNamespace_, pioNamespace);
777

778
  xdebug ( "nProcsGlob=%d, RETURN", sizeGlob );
779
  return commInqCommModel ();
780
#else
781
  abort();
782
#endif
783
}
784

785
786
#ifndef MPI_VERSION
#  undef MPI_Comm
Deike Kleberg's avatar
Deike Kleberg committed
787
788
#endif

789
/*****************************************************************************/
Deike Kleberg's avatar
Deike Kleberg committed
790

791
void  pioEndDef ( void )
792
793
{
#ifdef USE_MPI
Deike Kleberg's avatar
Deike Kleberg committed
794
  char   * buffer;
795
  int bufferSize;
796
  int rankGlob = commInqRankGlob ();
797

798
  xdebug("%s", "START");
799

Deike Kleberg's avatar
Deike Kleberg committed
800
  varsMapNDeco ( commInqNNodes (), commInqNodeSizes ());
Deike Kleberg's avatar
Deike Kleberg committed
801

802
  if ( rankGlob < commInqNProcsColl ())
Deike Kleberg's avatar
Deike Kleberg committed
803
    {
804
805
      MPI_Comm comm = commInqCommsIO ( rankGlob );
      reshPackBufferCreate(&buffer, &bufferSize, &comm);
806
807

      xmpi ( MPI_Send ( buffer, bufferSize, MPI_PACKED, commInqNProcsModel (),
808
                        RESOURCES, commInqCommsIO ( rankGlob )));
809

810
      xdebug("%s", "SENT MESSAGE WITH TAG \"RESOURCES\"");
811

Deike Kleberg's avatar
Deike Kleberg committed
812
      reshPackBufferDestroy ( &buffer );
Deike Kleberg's avatar
Deike Kleberg committed
813
    }
Deike Kleberg's avatar
Deike Kleberg committed
814
815

  modelWinCreate ();
816
  namespaceDefResStatus ( STAGE_TIMELOOP );
817
  xdebug("%s", "RETURN");
818
819
820
821
822
823
824
825
#endif
}

/************************************************************************/

void  pioEndTimestepping ( void )
{
#ifdef USE_MPI
826
  xdebug("%s", "START");
827
  namespaceDefResStatus ( STAGE_CLEANUP );
828
  xdebug("%s", "RETURN");
829
#endif
830
831
}

Deike Kleberg's avatar
Deike Kleberg committed
832

833
/****************************************************/
Deike Kleberg's avatar
Deike Kleberg committed
834

Deike Kleberg's avatar
Deike Kleberg committed
835

Deike Kleberg's avatar
Deike Kleberg committed
836
837
838
839
840
841
842
843
844
/**
  @brief is invoked by the calculator PEs, to inform
  the I/O PEs that no more data will be written.

  @param

  @return
*/

Deike Kleberg's avatar
Deike Kleberg committed
845
void pioFinalize ( void )
Deike Kleberg's avatar
Deike Kleberg committed
846
{
847
#ifdef USE_MPI
Deike Kleberg's avatar
Deike Kleberg committed
848
  int collID, ibuffer = 1111;
849
  xdebug("%s", "START");
850
851
852
853

  /* pioNamespace_ is unchanged on I/O servers */
  if (pioNamespace_ == -1)
    return;
Thomas Jahns's avatar
Thomas Jahns committed
854
  namespaceDelete(pioNamespace_);
Deike Kleberg's avatar
Deike Kleberg committed
855
  for ( collID = 0; collID < commInqNProcsColl (); collID++ )
856
    {