19 #define _XOPEN_SOURCE 700 28 #include "sion_error_handler.h" 41 #include "sion_hints.h" 44 #include "sion_generic_buddy.h" 76 sion_int64 *chunksize,
77 sion_int32 *fsblksize,
94 sion_int64 lchunksize, lstartpointer, lglobalrank, new_fsblocksize, helpint64, apiflag;
95 sion_int64 *sion_tmpintfield = NULL;
96 sion_int32 *sion_tmpintfield_map = NULL, helpint32;
97 sion_int32 *sion_tmpintfield_buddy32 = NULL;
98 sion_int64 *sion_tmpintfield_buddy64 = NULL;
99 void *comm_group=NULL;
107 DPRINTFP((2,
"_sion_paropen_generic_one_file", rank,
"enter parallel open of file %s in mode %d #tasks=%d\n", fname, (
int) flags_store->mask, ntasks));
108 DPRINTFP((32,
"_sion_paropen_generic_one_file", rank,
"sizeof: int=%d long=%d longlong=%d sion_int32=%d sion_int64=%d\n",
sizeof(
int),
sizeof(
long),
109 sizeof(
long long),
sizeof(sion_int32),
sizeof(sion_int64)));
113 filenum = *filenumber;
116 if(flag& _SION_INTERNAL_FLAG_NORMAL ) comm_group=sion_gendata->comm_data_local;
117 if(flag& _SION_INTERNAL_FLAG_BUDDY_NORMAL ) comm_group=sion_gendata->comm_data_local;
118 if(flag& _SION_INTERNAL_FLAG_BUDDY_SEND ) comm_group=buddy_data->buddy_send.commgroup;
119 if(flag& _SION_INTERNAL_FLAG_BUDDY_COLL ) comm_group=buddy_data->buddy_coll.commgroup;
120 if(flag& _SION_INTERNAL_FLAG_BUDDY_READ ) comm_group=buddy_data->groups[buddy_data->currentgroup]->commgroup;
124 if (flag&_SION_INTERNAL_FLAG_BUDDY_SEND) do_open_file=0;
125 if ( (flag&_SION_INTERNAL_FLAG_BUDDY_COLL) && (rank>0) ) do_open_file=0;
126 if ( (flag&_SION_INTERNAL_FLAG_BUDDY_READ) && (rank>0) ) do_open_file=0;
128 DPRINTFP((2,
"_sion_paropen_generic_one_file", rank,
"do_open_file=%d\n", do_open_file));
131 if (sion_filedesc == NULL) {
132 _sion_errorprint(SION_ID_NOT_VALID,_SION_ERROR_ABORT,
"_sion_paropen_omp: cannot allocate filedescriptor structure of size %lu (sion_filedesc), aborting ...\n",
133 (
unsigned long)
sizeof(sion_filedesc));
136 sion_filedesc->
fname = strdup(fname);
139 sion_filedesc->
sid=sid;
146 if (flags_store->mask&_SION_FMODE_WRITE) {
149 DPRINTFP((32,
"_sion_paropen_generic_one_file", rank,
" starting open for write #tasks=%d\n", ntasks));
153 return(_sion_errorprint(SION_ID_NOT_VALID,_SION_ERROR_RETURN,
"sion_paropen: wrong number of tasks specific: ntasks=%d (<0), returning ...\n", (
int) ntasks));
157 if ((chunksize != NULL) && (*chunksize<0)) {
158 return(_sion_errorprint(SION_ID_NOT_VALID,_SION_ERROR_RETURN,
"sion_paropen: ((chunksize != NULL) && (*chunksize<0)), returning ...\n"));
162 if ((flag & _SION_INTERNAL_FLAG_NORMAL ) && (globalrank != NULL) && (*globalrank<0)) {
163 return(_sion_errorprint(SION_ID_NOT_VALID,_SION_ERROR_RETURN,
"sion_paropen: ((globalrank != NULL) && (*globalrank<0)), returning ...\n"));
171 sion_filedesc->
rank = rank;
173 sion_filedesc->
ntasks = ntasks;
174 sion_filedesc->
nfiles = nfiles;
176 sion_filedesc->
prefix = strdup(prefix);
177 sion_filedesc->
compress = flags_store->mask&_SION_FMODE_COMPRESS;
178 sion_filedesc->
usecoll = (flags_store->mask&_SION_FMODE_COLLECTIVE)>0;
179 sion_filedesc->
collmergemode = (flags_store->mask&_SION_FMODE_COLLECTIVE_MERGE)>0;
180 sion_filedesc->
collmsa = !!_sion_flags_get(flags_store,
"collmsa");
181 sion_filedesc->
usebuddy = (flags_store->mask&_SION_FMODE_BUDDY)>0;
183 sion_filedesc->
buddylevel = atoi(_sion_flags_get(flags_store,
"buddy")->val);
191 return(_sion_errorprint(SION_ID_NOT_VALID,_SION_ERROR_RETURN,
"sion_paropen_generic: cannot open %s for writing, aborting ...\n", fname));
196 if((new_fsblocksize<0) || (new_fsblocksize>SION_MAX_FSBLOCKSIZE)) new_fsblocksize=SION_DEFAULT_FSBLOCKSIZE;
200 sion_gendata->apidesc->barrier_cb(comm_group);
205 sion_gendata->apidesc->bcastr_cb(&new_fsblocksize, comm_group, _SION_INT64, 1, 0);
206 *fsblksize=new_fsblocksize;
209 DPRINTFP((32,
"_sion_paropen_generic_one_file", rank,
"setting fsblksize to %lld\n", new_fsblocksize));
218 _sion_keyval_check_env(sion_filedesc, flags_store->mask);
220 sion_gendata->apidesc->bcastr_cb(&sion_filedesc->
keyvalmode, comm_group, _SION_INT32, 1, 0);
224 if ((flags_entry = _sion_flags_get(flags_store,
"collsize"))) {
225 sion_filedesc->
collsize = atoi(flags_entry->val);
227 _sion_coll_check_env(sion_filedesc);
231 ( flag&_SION_INTERNAL_FLAG_BUDDY_NORMAL )
232 || ( flag&_SION_INTERNAL_FLAG_BUDDY_SEND )
233 || ( flag&_SION_INTERNAL_FLAG_BUDDY_COLL )
243 sion_gendata->apidesc->bcastr_cb(&sion_filedesc->
usecoll, comm_group, _SION_INT32, 1, 0);
244 sion_gendata->apidesc->bcastr_cb(&sion_filedesc->
collsize, comm_group, _SION_INT32, 1, 0);
245 sion_gendata->apidesc->bcastr_cb(&sion_filedesc->
collmergemode, comm_group, _SION_INT32, 1, 0);
249 if(sion_gendata->apidesc->level!=SION_GENERIC_API_LEVEL_FULL) {
250 _sion_errorprint(SION_ID_NOT_VALID,_SION_ERROR_WARN,
"sion_paropen_generic: requested coalescing I/O but API does not support this mode, falling back to individual mode ...\n");
257 _sion_hints_check_env(sion_filedesc);
259 sion_gendata->apidesc->bcastr_cb(&sion_filedesc->
usehints, comm_group, _SION_INT32, 1, 0);
260 sion_gendata->apidesc->bcastr_cb(&sion_filedesc->
hinttype, comm_group, _SION_INT32, 1, 0);
262 DPRINTFTS(rank,
"before alloc");
268 DPRINTFTS(rank,
"after alloc");
271 lchunksize = (sion_int64) *chunksize;
272 lglobalrank = (sion_int64) *globalrank;
274 DPRINTFP((32,
"_sion_paropen_generic_one_file", rank,
"lchunksize=%lld lglobalrank=%lld\n", lchunksize,lglobalrank));
276 DPRINTFTS2(rank,
"before gather");
277 sion_gendata->apidesc->gatherr_cb(&lchunksize, sion_filedesc->
all_chunksizes, comm_group, _SION_INT64, 1, 0);
278 sion_gendata->apidesc->gatherr_cb(&lglobalrank, sion_filedesc->
all_globalranks, comm_group, _SION_INT64, 1, 0);
282 sion_filedesc->
coll_capability=sion_gendata->apidesc->get_capability_cb(comm_group);
286 DPRINTFTS2(rank,
"after gather");
288 DPRINTFTS(rank,
"before calculate");
289 DPRINTFP((32,
"_sion_paropen_generic_one_file", rank,
"chunksizes[%d - 1]=%ld\n", ntasks,(
long) sion_filedesc->
all_chunksizes[ntasks - 1]));
290 if (!sion_filedesc->
usecoll) _sion_calculate_startpointers(sion_filedesc);
292 if (sion_filedesc->
collmergemode) _sion_calculate_startpointers_collective_merge(sion_filedesc);
293 else if (sion_filedesc->
collmsa) _sion_calculate_startpointers_collective_msa(sion_filedesc);
294 else _sion_calculate_startpointers_collective(sion_filedesc);
296 DPRINTFTS(rank,
"after calculate");
299 DPRINTFTS(rank,
"before open");
306 return(_sion_errorprint(SION_ID_NOT_VALID,_SION_ERROR_RETURN,
"sion_paropen_generic: cannot open %s for writing, aborting ...\n", fname));
309 sion_filedesc->
fileptr = sion_fileptr;
311 sion_gendata->apidesc->barrier_cb(comm_group);
312 DPRINTFTS(rank,
"after open");
319 _sion_apply_hints(sion_filedesc,SION_HINTS_ACCESS_TYPE_METADATABLOCK1);
321 DPRINTFTS(rank,
"before writeh");
323 DPRINTFTS(rank,
"after writeh");
331 DPRINTFTS(rank,
"before setp(0)");
334 DPRINTFTS(rank,
"after setp(0)");
339 DPRINTFTS(rank,
"before scatter");
340 sion_gendata->apidesc->scatterr_cb(sion_filedesc->
all_startpointers, &sion_filedesc->
startpos, comm_group, _SION_INT64, 1, 0);
341 DPRINTFTS(rank,
"after scatter");
344 DPRINTFTS(rank,
"before scatter");
345 sion_gendata->apidesc->scatterr_cb(sion_filedesc->
all_chunksizes, &sion_filedesc->
chunksize, comm_group, _SION_INT64, 1, 0);
346 DPRINTFTS(rank,
"after scatter");
350 sion_gendata->apidesc->scatterr_cb(sion_filedesc->
all_coll_collsize, &sion_filedesc->
collsize, comm_group, _SION_INT32, 1, 0);
357 sion_gendata->apidesc->bcastr_cb(&sion_filedesc->
globalskip, comm_group, _SION_INT64, 1, 0);
359 DPRINTFP((32,
"_sion_paropen_generic_one_file", rank,
" start position is %10lld %10.4f MB chunksize=%10lld %10.4f MB\n",
365 DPRINTFTS(rank,
"before setp");
366 sion_gendata->apidesc->barrier_cb(comm_group);
375 sion_gendata->apidesc->barrier_cb(comm_group);
378 _sion_apply_hints(sion_filedesc,SION_HINTS_ACCESS_TYPE_CHUNK);
380 DPRINTFTS(rank,
"after setp");
381 DPRINTFP((32,
"_sion_paropen_generic_one_file", rank,
" ending open for write #tasks=%d filepos=%lld\n", ntasks,
_sion_file_get_position(sion_filedesc->
fileptr)));
384 else if (flags_store->mask&_SION_FMODE_READ) {
387 DPRINTFP((32,
"_sion_paropen_generic_one_file", rank,
" starting open for read #tasks=%d\n", ntasks));
391 DPRINTFTS(rank,
"before openR");
393 DPRINTFTS(rank,
"after openR");
395 DPRINTFP((32,
"_sion_paropen_generic_one_file", rank,
" cannot open %s for reading, aborting ...\n", fname));
396 return(_sion_errorprint(SION_ID_NOT_VALID,_SION_ERROR_RETURN,
"_sion_paropen_generic_one_file: cannot open %s for reading, aborting ...\n", fname));
401 sion_gendata->apidesc->barrier_cb(comm_group);
404 sion_filedesc->
fileptr = sion_fileptr;
405 sion_filedesc->
rank = rank;
406 sion_filedesc->
ntasks = ntasks;
409 sion_filedesc->
nfiles = nfiles;
410 sion_filedesc->
collmsa = !!_sion_flags_get(flags_store,
"collmsa");
411 sion_filedesc->
usebuddy = (flags_store->mask&_SION_FMODE_BUDDY)>0;
413 sion_filedesc->
buddylevel = atoi(_sion_flags_get(flags_store,
"buddy")->val);
418 if ( flag&_SION_INTERNAL_FLAG_BUDDY_READ ) {
426 DPRINTFP((32,
"_sion_paropen_generic_one_file", rank,
" create buddy mapping ntasks=%d filentasks=%d\n",ntasks,sion_filedesc->
ntasks));
429 sion_tmpintfield_buddy32 = (sion_int32 *) malloc(ntasks *
sizeof(sion_int32));
430 if (sion_tmpintfield_buddy32 == NULL) {
431 return(_sion_errorprint(SION_ID_NOT_VALID,_SION_ERROR_RETURN,
"_sion_paropen_generic_one_file: cannot allocate temporary memory of size %lu (sion_tmpintfield_buddy), aborting ...\n",
432 (
unsigned long) ntasks *
sizeof(sion_int32)));
434 for (j = 0; j < ntasks; j++) sion_tmpintfield_buddy32[j]=-1;
435 sion_tmpintfield_buddy64 = (sion_int64 *) malloc(ntasks *
sizeof(sion_int64));
436 if (sion_tmpintfield_buddy64 == NULL) {
437 free(sion_tmpintfield_buddy32);
438 return(_sion_errorprint(SION_ID_NOT_VALID,_SION_ERROR_RETURN,
"_sion_paropen_generic_one_file: cannot allocate temporary memory of size %lu (sion_tmpintfield_buddy), aborting ...\n",
439 (
unsigned long) ntasks *
sizeof(sion_int64)));
441 for (j = 0; j < ntasks; j++) sion_tmpintfield_buddy64[j]=-1;
442 sion_tmpintfield_map = (sion_int32 *) malloc(ntasks *
sizeof(sion_int32));
443 if (sion_tmpintfield_map == NULL) {
444 free(sion_tmpintfield_buddy32);
445 free(sion_tmpintfield_buddy64);
446 return(_sion_errorprint(SION_ID_NOT_VALID,_SION_ERROR_RETURN,
"_sion_paropen_generic_one_file: cannot allocate temporary memory of size %lu (sion_tmpintfield_map), aborting ...\n",
447 (
unsigned long) ntasks *
sizeof(sion_int32)));
449 for (j = 0; j < ntasks; j++) sion_tmpintfield_map[j]=-1;
452 helpint32=buddy_data->groups[buddy_data->currentgroup]->filelrank;
453 sion_gendata->apidesc->gatherr_cb(&helpint32, sion_tmpintfield_map, comm_group, _SION_INT32, 1, 0);
457 for (j = 0; j < ntasks; j++)
458 DPRINTFP((64,
"_sion_paropen_generic_one_file", rank,
" buddy map[%d]=%d\n", j, (
int) sion_tmpintfield_map[j]));
465 if (rc!=SION_SUCCESS) {
466 free(sion_tmpintfield_buddy32);
467 free(sion_tmpintfield_buddy64);
468 return(_sion_errorprint(SION_ID_NOT_VALID,_SION_ERROR_RETURN,
"_sion_paropen_generic_one_file: cannot read header from file %s, aborting ...\n", fname));
470 DPRINTFP((32,
"_sion_paropen_generic_one_file", rank,
471 " read, after read of fix header part endianness=0x%x blksize=%d ntasks=%d\n", sion_filedesc->
endianness, sion_filedesc->
fsblksize, sion_filedesc->
ntasks));
473 DPRINTFTS(rank,
"before alloc");
476 DPRINTFTS(rank,
"after alloc");
479 if (rc!=SION_SUCCESS) {
480 free(sion_tmpintfield_buddy32);
481 free(sion_tmpintfield_buddy64);
482 return(_sion_errorprint(SION_ID_NOT_VALID,_SION_ERROR_RETURN,
"_sion_paropen_generic_one_file: cannot read header from file %s, aborting ...\n", fname));
485 if ((flags_entry = _sion_flags_get(flags_store,
"collsize"))) {
486 sion_filedesc->
collsize = atoi(flags_entry->val);
488 _sion_coll_check_env(sion_filedesc);
492 if (!sion_filedesc->
usecoll) _sion_calculate_startpointers(sion_filedesc);
494 if (sion_filedesc->
collmergemode) _sion_calculate_startpointers_collective_merge(sion_filedesc);
495 else if (sion_filedesc->
collmsa) _sion_calculate_startpointers_collective_msa(sion_filedesc);
496 else _sion_calculate_startpointers_collective(sion_filedesc);
498 DPRINTFTS(rank,
"after calculate");
501 _sion_keyval_check_env(sion_filedesc, flags_store->mask);
506 sion_gendata->apidesc->bcastr_cb(&sion_filedesc->
keyvalmode, comm_group, _SION_INT32, 1, 0);
509 sion_gendata->apidesc->bcastr_cb(&sion_filedesc->
usecoll, comm_group, _SION_INT32, 1, 0);
510 sion_gendata->apidesc->bcastr_cb(&sion_filedesc->
collsize, comm_group, _SION_INT32, 1, 0);
513 DPRINTFP((32,
"_sion_paropen_generic_one_file", rank,
" usecoll=%d\n", sion_filedesc->
usecoll));
517 if (! (flag&_SION_INTERNAL_FLAG_BUDDY_READ) ) {
518 sion_gendata->apidesc->scatterr_cb(sion_filedesc->
all_coll_collsize, &sion_filedesc->
collsize, comm_group, _SION_INT32, 1, 0);
523 if(rank==0)
for (j = 0; j < ntasks; j++)
if(sion_tmpintfield_map[j]>=0) sion_tmpintfield_buddy32[j]=sion_filedesc->
all_coll_collsize[sion_tmpintfield_map[j]];
524 sion_gendata->apidesc->scatterr_cb(sion_tmpintfield_buddy32, &sion_filedesc->
collsize, comm_group, _SION_INT32, 1, 0);
526 if(rank==0)
for (j = 0; j < ntasks; j++)
if(sion_tmpintfield_map[j]>=0) sion_tmpintfield_buddy32[j]=sion_filedesc->
all_coll_collector[sion_tmpintfield_map[j]];
527 sion_gendata->apidesc->scatterr_cb(sion_tmpintfield_buddy32, &sion_filedesc->
collector, comm_group, _SION_INT32, 1, 0);
535 sion_gendata->apidesc->bcastr_cb(&sion_filedesc->
globalskip, comm_group, _SION_INT64, 1, 0);
538 sion_gendata->apidesc->bcastr_cb(&sion_filedesc->
endianness, comm_group, _SION_INT32, 1, 0);
539 sion_gendata->apidesc->bcastr_cb(&sion_filedesc->
swapbytes, comm_group, _SION_INT32, 1, 0);
540 sion_gendata->apidesc->bcastr_cb(&sion_filedesc->
fsblksize, comm_group, _SION_INT32, 1, 0);
541 sion_gendata->apidesc->bcastr_cb(&sion_filedesc->
ntasks, comm_group, _SION_INT32, 1, 0);
542 sion_gendata->apidesc->bcastr_cb(&sion_filedesc->
fileversion, comm_group, _SION_INT32, 1, 0);
543 sion_gendata->apidesc->bcastr_cb(&sion_filedesc->
nfiles, comm_group, _SION_INT32, 1, 0);
544 sion_gendata->apidesc->bcastr_cb(&sion_filedesc->
filenumber, comm_group, _SION_INT32, 1, 0);
545 sion_gendata->apidesc->bcastr_cb(&sion_filedesc->
flag1, comm_group, _SION_INT32, 1, 0);
546 sion_gendata->apidesc->bcastr_cb(&sion_filedesc->
flag2, comm_group, _SION_INT32, 1, 0);
547 sion_gendata->apidesc->bcastr_cb(&sion_filedesc->
maxusedchunks, comm_group, _SION_INT32, 1, 0);
549 DPRINTFP((32,
"_sion_paropen_generic_one_file", rank,
552 DPRINTFTS(rank,
"after bcast");
555 DPRINTFTS(rank,
"before scatter");
556 if (! (flag&_SION_INTERNAL_FLAG_BUDDY_READ) ) {
557 sion_gendata->apidesc->scatterr_cb(sion_filedesc->
all_chunksizes, &sion_filedesc->
chunksize, comm_group, _SION_INT64, 1, 0);
558 sion_gendata->apidesc->scatterr_cb(sion_filedesc->
all_startpointers, &sion_filedesc->
startpos, comm_group, _SION_INT64, 1, 0);
559 sion_gendata->apidesc->scatterr_cb(sion_filedesc->
all_globalranks, &helpint64, comm_group, _SION_INT64, 1, 0);sion_filedesc->
globalrank=(sion_int32) helpint64;
562 if(rank==0)
for (j = 0; j < ntasks; j++)
if(sion_tmpintfield_map[j]>=0) sion_tmpintfield_buddy64[j]=sion_filedesc->
all_chunksizes[sion_tmpintfield_map[j]];
563 sion_gendata->apidesc->scatterr_cb(sion_tmpintfield_buddy64, &sion_filedesc->
chunksize, comm_group, _SION_INT64, 1, 0);
565 if(rank==0)
for (j = 0; j < ntasks; j++)
if(sion_tmpintfield_map[j]>=0) sion_tmpintfield_buddy64[j]=sion_filedesc->
all_startpointers[sion_tmpintfield_map[j]];
566 sion_gendata->apidesc->scatterr_cb(sion_tmpintfield_buddy64, &sion_filedesc->
startpos, comm_group, _SION_INT64, 1, 0);
568 if(rank==0)
for (j = 0; j < ntasks; j++)
if(sion_tmpintfield_map[j]>=0) sion_tmpintfield_buddy64[j]=sion_filedesc->
all_globalranks[sion_tmpintfield_map[j]];
569 sion_gendata->apidesc->scatterr_cb(sion_tmpintfield_buddy64, &helpint64, comm_group, _SION_INT64, 1, 0);sion_filedesc->
globalrank=(sion_int32) helpint64;
572 DPRINTFTS(rank,
"after scatter");
576 sion_tmpintfield = (sion_int64 *) malloc(sion_filedesc->
ntasks *
sizeof(sion_int64));
577 if (sion_tmpintfield == NULL) {
578 free(sion_tmpintfield_buddy32);
579 return(_sion_errorprint(SION_ID_NOT_VALID,_SION_ERROR_RETURN,
"_sion_paropen_generic_one_file: cannot allocate temporary memory of size %lu (sion_tmpintfield), aborting ...\n",
580 (
unsigned long) ntasks *
sizeof(sion_int64)));
584 for (j = 0; j < sion_filedesc->
ntasks; j++)
585 DPRINTFP((2048,
"_sion_paropen_generic_one_file", rank,
" read, blockcount on task %02d is %10ld\n", j, (
long) sion_tmpintfield[j]));
589 DPRINTFTS(rank,
"before scatter");
590 if (! (flag&_SION_INTERNAL_FLAG_BUDDY_READ) ) {
591 sion_gendata->apidesc->scatterr_cb(sion_tmpintfield, &helpint64, comm_group, _SION_INT64, 1, 0);
594 if(rank==0)
for (j = 0; j < ntasks; j++)
if(sion_tmpintfield_map[j]>=0) sion_tmpintfield_buddy64[j]=sion_tmpintfield[sion_tmpintfield_map[j]];
595 sion_gendata->apidesc->scatterr_cb(sion_tmpintfield_buddy64, &helpint64, comm_group, _SION_INT64, 1, 0);
597 DPRINTFTS(rank,
"after scatter");
599 DPRINTFP((32,
"_sion_paropen_generic_one_file", rank,
" lastchunknr on task %02d is %10ld\n", rank, (
long) sion_filedesc->
lastchunknr));
603 DPRINTFTS(rank,
"before scatter");
604 if (! (flag&_SION_INTERNAL_FLAG_BUDDY_READ) ) {
605 sion_gendata->apidesc->scatterr_cb(sion_tmpintfield, &helpint64, comm_group, _SION_INT64, 1, 0);
608 if(rank==0)
for (j = 0; j < ntasks; j++)
if(sion_tmpintfield_map[j]>=0) sion_tmpintfield_buddy64[j]=sion_tmpintfield[sion_tmpintfield_map[j]];
609 sion_gendata->apidesc->scatterr_cb(sion_tmpintfield_buddy64, &helpint64, comm_group, _SION_INT64, 1, 0);
611 DPRINTFTS(rank,
"after scatter");
622 DPRINTFTS(rank,
"before setp");
623 sion_gendata->apidesc->barrier_cb(comm_group);
637 if(sion_tmpintfield) free(sion_tmpintfield);
638 if(sion_tmpintfield_map) free(sion_tmpintfield_map);
639 if(sion_tmpintfield_buddy32) free(sion_tmpintfield_buddy32);
640 if(sion_tmpintfield_buddy64) free(sion_tmpintfield_buddy64);
642 sion_gendata->apidesc->barrier_cb(comm_group);
643 DPRINTFTS(rank,
"after setp");
646 return(_sion_errorprint(SION_ID_NOT_VALID,_SION_ERROR_RETURN,
"sion_paropen_multi_mpi: unknown file mode"));
652 *fileptr=sion_filedesc->
fileptr->fileptr;
660 if(fileptr!=NULL) *fileptr=NULL;
671 DPRINTFP((32,
"_sion_paropen_generic_one_file", rank,
" start position on task %02d is at end of sion_paropen_generic %10lld\n", rank,
674 DPRINTFP((2,
"_sion_paropen_generic_one_file", rank,
"leave parallel open of file %s in mode 0x%lx #tasks=%d\n", fname, (
long) flags_store->mask, ntasks));
707 int rc = SION_SUCCESS;
709 sion_int64 helpint64;
710 sion_int64 *sion_tmpintfield = NULL;
712 void *comm_group=NULL;
716 return(_sion_errorprint(SION_NOT_SUCCESS,_SION_ERROR_RETURN,
"_sion_parclose_generic: invalid sion_filedesc, aborting %d ...\n", sid));
720 return(_sion_errorprint(SION_NOT_SUCCESS,_SION_ERROR_RETURN,
"_sion_parclose_generic: sion file with sid=%d was not opened by a sion_paropen\n", sid));
723 DPRINTFP((2,
"_sion_parclose_generic", rank,
"enter parallel close sid=%d\n", sid));
726 if(flag& _SION_INTERNAL_FLAG_NORMAL ) comm_group=sion_gendata->comm_data_local;
727 if(flag& _SION_INTERNAL_FLAG_BUDDY_SEND ) comm_group=buddy_data->buddy_send.commgroup;
728 if(flag& _SION_INTERNAL_FLAG_BUDDY_COLL ) comm_group=buddy_data->buddy_coll.commgroup;
729 if(flag& _SION_INTERNAL_FLAG_BUDDY_READ ) comm_group=buddy_data->groups[buddy_data->currentgroup]->commgroup;
733 if (flag&_SION_INTERNAL_FLAG_BUDDY_SEND) do_close_file=0;
734 if ( (flag&_SION_INTERNAL_FLAG_BUDDY_COLL) && (rank>0) ) do_close_file=0;
735 if ( (flag&_SION_INTERNAL_FLAG_BUDDY_READ) && (rank>0) ) do_close_file=0;
742 DPRINTFP((32,
"_sion_parclose_generic", rank,
" parallel close (read mode) sid=%d, call fclose on file\n", sid));
766 DPRINTFP((32,
"_sion_parclose_generic", rank,
" parallel close (write mode) sid=%d, call fclose on file\n", sid));
775 sion_gendata->apidesc->barrier_cb(comm_group);
777 DPRINTFP((32,
"_sion_parclose_generic", rank,
" parallel close sid=%d: lastchunknr=%d globalskip=%lld\n", sid, sion_filedesc->
lastchunknr,
779 for (blknum = 0; blknum <= sion_filedesc->
lastchunknr; blknum++) {
780 DPRINTFP((1024,
"_sion_parclose_generic", rank,
" parallel close sid=%d: local block %02d -> %10lld bytes\n", sid, blknum,
785 sion_tmpintfield = (sion_int64 *) malloc(sion_filedesc->
ntasks *
sizeof(sion_int64));
786 if (sion_tmpintfield == NULL) {
787 return(_sion_errorprint(SION_NOT_SUCCESS,_SION_ERROR_RETURN,
"_sion_parclose_generic: cannot allocate temporary memory of size %lu (sion_tmpintfield), aborting ...\n",
788 (
unsigned long) sion_filedesc->
ntasks *
sizeof(sion_int64)));
793 DPRINTFTS2(rank,
"before gather");
795 sion_gendata->apidesc->gatherr_cb(&helpint64, sion_tmpintfield, comm_group, _SION_INT64, 1, 0);
799 for (blknum = 0; blknum < sion_filedesc->
ntasks; blknum++)
801 sion_filedesc->
maxusedchunks = (int) sion_tmpintfield[blknum];
803 sion_gendata->apidesc->bcastr_cb(&sion_filedesc->
maxusedchunks, comm_group, _SION_INT32, 1, 0);
804 DPRINTFTS2(rank,
"after gather");
818 for (blknum = 0; blknum < sion_filedesc->
maxusedchunks; blknum++) {
819 if (blknum <= sion_filedesc->lastchunknr) {
820 helpint64 = sion_filedesc->
blocksizes[blknum];
826 DPRINTFTS2(rank,
"before gather");
827 sion_gendata->apidesc->gatherr_cb(&helpint64, sion_tmpintfield, comm_group, _SION_INT64, 1, 0);
828 DPRINTFTS2(rank,
"after gather");
831 for (lrank = 0; lrank < ntasks; lrank++)
832 DPRINTFP((2048,
"_sion_parclose_generic", rank,
" parallel close sid=%d: write total chunksize for block %d: %2lld rank=%d\n", sid, blknum,
833 sion_tmpintfield[lrank], lrank));
841 if (mapping != NULL) {
847 DPRINTFP((32,
"_sion_parclose_generic", rank,
" parallel close (write mode) sid=%d, call fclose on file\n", sid));
855 if(sion_tmpintfield) free(sion_tmpintfield);
860 _sion_free_filedesc(sion_filedesc);
861 sion_filedesc = NULL;
864 DPRINTFP((2,
"_sion_parclose_generic", rank,
"leave parallel close sid=%d\n", sid));
875 sion_int64 chunksize,
881 int rc = SION_SUCCESS;
883 sion_int64 lchunksize, lstartpointer, lglobalrank;
884 void *comm_group=NULL;
887 return(_sion_errorprint(SION_NOT_SUCCESS,_SION_ERROR_RETURN,
"_sion_parreinit_generic: invalid sion_filedesc, aborting %d ...\n", sid));
891 return(_sion_errorprint(SION_NOT_SUCCESS,_SION_ERROR_RETURN,
"_sion_parreinit_generic: sion file with sid=%d was not opened by a sion_paropen\n", sid));
894 DPRINTFP((2,
"_sion_parreinit_generic", sion_filedesc->
rank,
"enter parallel reinit sid=%d\n", sid));
896 comm_group=sion_gendata->comm_data_local;
899 return(_sion_errorprint(SION_NOT_SUCCESS,_SION_ERROR_RETURN,
"_sion_parreinit_generic: sion file with sid=%d only allowed for files openend for write\n", sid));
902 DPRINTFTS(sion_filedesc->
rank,
"before alloc");
903 if (sion_filedesc->
rank == 0) {
907 DPRINTFTS(sion_filedesc->
rank,
"after alloc");
910 lchunksize = (sion_int64) chunksize;
911 lglobalrank = (sion_int64) sion_filedesc->
globalrank;
913 DPRINTFTS2(sion_filedesc->
rank,
"before gather");
914 sion_gendata->apidesc->gatherr_cb(&lchunksize, sion_filedesc->
all_chunksizes, comm_group, _SION_INT64, 1, 0);
915 sion_gendata->apidesc->gatherr_cb(&lglobalrank, sion_filedesc->
all_globalranks, comm_group, _SION_INT64, 1, 0);
917 DPRINTFTS2(sion_filedesc->
rank,
"after gather");
921 if (sion_filedesc->
rank == 0) {
922 DPRINTFTS(sion_filedesc->
rank,
"before calculate");
923 if (!sion_filedesc->
usecoll) _sion_calculate_startpointers(sion_filedesc);
924 else _sion_calculate_startpointers_collective(sion_filedesc);
925 DPRINTFTS(sion_filedesc->
rank,
"after calculate");
929 if (sion_filedesc->
rank == 0) {
932 _sion_apply_hints(sion_filedesc,SION_HINTS_ACCESS_TYPE_METADATABLOCK1);
938 DPRINTFTS(sion_filedesc->
rank,
"before writeh");
940 DPRINTFTS(sion_filedesc->
rank,
"after writeh");
949 DPRINTFTS(sion_filedesc->
rank,
"before setp(0)");
952 DPRINTFTS(sion_filedesc->
rank,
"after setp(0)");
957 DPRINTFTS(sion_filedesc->
rank,
"before scatter");
958 sion_gendata->apidesc->scatterr_cb(sion_filedesc->
all_startpointers, &sion_filedesc->
startpos, comm_group, _SION_INT64, 1, 0);
959 DPRINTFTS(sion_filedesc->
rank,
"after scatter");
963 sion_gendata->apidesc->scatterr_cb(sion_filedesc->
all_coll_collsize, &sion_filedesc->
collsize, comm_group, _SION_INT32, 1, 0);
970 sion_gendata->apidesc->bcastr_cb(&sion_filedesc->
globalskip, comm_group, _SION_INT64, 1, 0);
972 DPRINTFP((32,
"_sion_parreinit_generic", sion_filedesc->
rank,
" start position is %10lld %10.4f MB\n",
976 DPRINTFTS(sion_filedesc->
rank,
"before setp");
977 sion_gendata->apidesc->barrier_cb(comm_group);
981 sion_filedesc->
chunksize = (sion_int64) chunksize;
982 sion_gendata->apidesc->barrier_cb(comm_group);
985 _sion_apply_hints(sion_filedesc,SION_HINTS_ACCESS_TYPE_CHUNK);
987 if (sion_filedesc->
rank == 0) {
992 DPRINTFTS(sion_filedesc->
rank,
"after setp");
993 DPRINTFP((32,
"_sion_parreinit_generic", sion_filedesc->
rank,
" ending open for write #tasks=%d filepos=%lld\n",
996 DPRINTFP((2,
"_sion_parreinit_generic", sion_filedesc->
rank,
"leave parallel reinit of file %s in #tasks=%d\n",
1009 #define DFUNCTION "_sion_generic_collect_mapping" 1010 int _sion_generic_collect_mapping(
_sion_filedesc *sion_filedesc,
1012 sion_int32 **mapping ) {
1013 int rc=SION_SUCCESS;
1017 sion_int32 lpos[2], *receivemap=NULL, iamreceiver, receiver = -1;
1020 sion_gendata=sion_filedesc->
dataptr;
1021 sion_apidesc=sion_gendata->apidesc;
1023 *mapping = NULL; *mapping_size = 0;
1031 *mapping_size=sion_gendata->gsize;
1032 *mapping = (sion_int32 *) malloc(*mapping_size * 2 *
sizeof(sion_int32));
1033 if (*mapping == NULL) {
1034 return(_sion_errorprint(SION_ID_NOT_VALID,_SION_ERROR_RETURN,
"sion_generic_parclose: Cannot allocate memory for mapping"));
1039 if(sion_gendata->grank==0) {
1040 receivemap = (sion_int32 *) malloc(sion_gendata->gsize *
sizeof(sion_int32));
1041 if (receivemap == NULL) {
1042 return(_sion_errorprint(SION_ID_NOT_VALID,_SION_ERROR_RETURN,
"sion_generic_parclose: Cannot allocate memory for receivemap"));
1046 if((sion_filedesc->
filenumber==0) && (sion_filedesc->
rank==0)) iamreceiver=sion_gendata->grank;
1047 else iamreceiver=-1;
1048 sion_apidesc->gatherr_cb(&iamreceiver, receivemap, sion_gendata->comm_data_global, _SION_INT32, 1, 0);
1049 if(sion_gendata->grank==0) {
1050 for(t=0;t<sion_gendata->gsize;t++) {
1051 if(receivemap[t]>=0) {
1052 receiver=receivemap[t];
1056 DPRINTFP((1,
DFUNCTION, sion_gendata->grank,
"receiver of mapping grank=%d\n", receiver));
1058 sion_apidesc->bcastr_cb(&receiver, sion_gendata->comm_data_global, _SION_INT32, 1, 0);
1062 lpos[1] = sion_filedesc->
rank;
1063 sion_apidesc->gatherr_cb(&lpos, *mapping, sion_gendata->comm_data_global, _SION_INT32, 2, receiver);
1066 if(receivemap!=NULL) free(receivemap);
1082 int grank = comm->grank;
1083 int gsize = comm->gsize;
1085 DPRINTFP((2, __func__, grank,
"enter\n"));
1087 if (0 != strcmp(comm->apidesc->name,
"SIONlib_MPI_API")) {
1088 return _sion_errorprint(SION_NOT_SUCCESS, _SION_ERROR_RETURN,
"MSA Collectives: not supported for generic API \"%s\"\n", comm->apidesc->name);
1095 fd.
collsize = atoi(flags_entry->val);
1097 _sion_coll_check_env(&fd);
1100 return _sion_errorprint(SION_NOT_SUCCESS, _SION_ERROR_RETURN,
"MSA Collectives: size of collective groups should be 2 or more, but is %d\n", collsize);
1102 sion_int32 n_groups = gsize / collsize + ((gsize % collsize) ? 1 : 0);
1105 return _sion_errorprint(SION_NOT_SUCCESS, _SION_ERROR_RETURN,
"MSA Collectives: usecoll == false\n");
1108 int is_candidate = _sion_generic_is_candidate(comm);
1109 int n_candidates = 0;
1110 int candidates_before = 0;
1113 int *candidates = NULL;
1114 if (0 == comm->grank) {
1115 candidates = calloc(gsize,
sizeof(
int));
1117 _sion_errorprint(SION_NOT_SUCCESS, _SION_ERROR_ABORT,
"MSA Collectives: malloc returned NULL\n");
1121 comm->apidesc->gatherr_cb(&is_candidate, candidates, comm->comm_data_global, _SION_INT32, 1, 0);
1123 if (0 == comm->grank) {
1124 for (
size_t i = 0; i < gsize; i++) {
1125 int tmp = candidates[i];
1126 candidates[i] = n_candidates;
1127 n_candidates += tmp;
1131 comm->apidesc->bcastr_cb(&n_candidates, comm->comm_data_global, _SION_INT32, 1, 0);
1132 comm->apidesc->scatterr_cb(candidates, &candidates_before, comm->comm_data_global, _SION_INT32, 1, 0);
1134 if (0 == comm->grank) {
1139 if ((n_candidates < n_groups) || (n_candidates < comm->numfiles)) {
1140 return _sion_errorprint(SION_NOT_SUCCESS, _SION_ERROR_RETURN,
"MSA Collectives: insufficient number of candidates %d, number of groups %d, number of files %d\n", n_groups, comm->numfiles);
1144 int groups_per_file = n_groups / comm->numfiles;
1145 int is_collector = is_candidate && (candidates_before < n_groups);
1147 int rank_collector = candidates_before;
1148 comm->filenumber = rank_collector % comm->numfiles;
1149 comm->lrank = (rank_collector / comm->numfiles) * collsize;
1151 int collectors_before = (candidates_before < n_groups) ? candidates_before : n_groups;
1152 int rank_sender = grank - collectors_before;
1153 int group_number = rank_sender / (collsize - 1);
1154 comm->filenumber = group_number % comm->numfiles;
1155 comm->lrank = (group_number / comm->numfiles) * collsize + rank_sender % (collsize - 1) + 1;
1157 comm->lsize = collsize * (groups_per_file + ((comm->filenumber < n_groups % comm->numfiles) ? 1 : 0));
1158 if (comm->filenumber == comm->numfiles - 1) {
1159 comm->lsize += gsize - n_groups * collsize;
1162 DPRINTFP((32, __func__, grank,
"MSA Collectives: global rank %d of %d, is candidate %d, is collector %d, file no %d, local rank %d, local size %d\n", grank, gsize, is_candidate, is_collector, comm->filenumber, comm->lrank, comm->lsize));
1163 DPRINTFP((2, __func__, grank,
"exit\n"));
1164 return SION_SUCCESS;
1168 #if defined(_SION_MSA_DEEP_EST_SDV) 1169 char hostname[1024];
1170 if (0 == gethostname(hostname, 1024)) {
1171 if (0 == strncmp(
"knl", hostname, 3)) {
1179 #elif defined(_SION_MSA_TEST) 1180 return comm->grank %2;
long _sion_file_get_opt_blksize(_sion_fileptr *sion_fileptr)
Get optional file system block size for a file.
sion_int64 _sion_file_get_position(_sion_fileptr *sion_fileptr)
Get new position in file.
int _sion_buffer_flush(_sion_filedesc *sion_filedesc)
Flush buffer.
int _sion_flush_block(_sion_filedesc *sion_filedesc)
Update the internal data structure.
sion_int64 _sion_file_set_position(_sion_fileptr *sion_fileptr, sion_int64 startpointer)
Set new position in file.
int _sion_reassignvcd(int sid, void *data, int type)
_sion_fileptr * _sion_file_open(const char *fname, unsigned int flags, unsigned int addflags)
Create and open a new file for writing.
Sion File Descriptor Structure.
int _sion_paropen_generic_one_file(int sid, char *fname, _sion_flags_store *flags_store, char *prefix, int *numFiles, int *filenumber, sion_int64 *chunksize, sion_int32 *fsblksize, int rank, int ntasks, int *globalrank, int flag, FILE **fileptr, _sion_generic_gendata *sion_gendata, _sion_generic_buddy *buddy_data)
Generic parallel open of one direct access file.
int _sion_parclose_generic(int sid, int rank, int ntasks, int mapping_size, sion_int32 *mapping, int flag, _sion_generic_gendata *sion_gendata, _sion_generic_buddy *buddy_data)
Internal function to close parallel opened SION file.
#define SION_FILE_FLAG_WRITE
int _sion_file_purge(_sion_fileptr *sion_fileptr)
Purge data to file.
sion_int64 * all_globalranks
int _sion_alloc_filedesc_arrays(_sion_filedesc *sion_filedesc)
Allocate memory for the internal sion arrays.
int _sion_free_filedesc_arrays(_sion_filedesc *sion_filedesc)
free memory for the internal sion arrays
#define SION_FILE_FLAG_READ
int _sion_realloc_filedesc_blocklist(_sion_filedesc *sion_filedesc, sion_int32 maxchunks)
Increase the memory used by the internal sion structure for the blocklist.
int _sion_vcdtype(int sid)
sion_int32 * all_coll_collector
sion_int32 fileptr_exported
sion_int32 _sion_get_endianness_with_flags(sion_int64 flags)
Return endianness including possible choice via flags.
#define SION_FILE_FLAG_POSIX
int _sion_parreinit_generic(int sid, sion_int64 chunksize, int rank, int ntasks, _sion_generic_gendata *sion_gendata)
change chunksize for an already opened SION file (write)
sion_int32 * all_coll_capability
void * _sion_vcdtovcon(int sid)
#define SION_FILE_FLAG_ANSI
#define SION_FILE_FLAG_CREATE
#define SION_FILEMODE_READ
sion_int64 * all_startpointers
int _sion_file_close(_sion_fileptr *sion_fileptr)
Close file and destroys fileptr structure.
#define SION_FILESTATE_PAROPEN
int _sion_free_filedesc_coll_arrays(_sion_filedesc *sion_filedesc)
free memory for the internal sion arrays
int _sion_buffer_check_env(_sion_filedesc *sion_filedesc)
Checks if environment variables are set to use buffer.
sion_int32 coll_capability
sion_int32 currentblocknr
int _sion_cache_check_env(_sion_filedesc *sion_filedesc)
Check if environment variables are set to use cache.
#define SION_FILEDESCRIPTOR
sion_int64 * all_chunksizes
int _sion_print_filedesc(_sion_filedesc *sion_filedesc, int level, char *desc, int flag)
Print the initialized sion file description.
#define SION_FILESTATE_CLOSE
_sion_filedesc * _sion_alloc_filedesc(void)
Allocates memory for internal sion structure.
int _sion_alloc_filedesc_coll_arrays(_sion_filedesc *sion_filedesc)
Allocate memory for the internal sion arrays.
int _sion_init_filedesc(_sion_filedesc *sion_filedesc)
Initialize the sion file description.
#define SION_FILEMODE_WRITE
#define DFUNCTION
collect mapping information on rank 0 of first file, mapping=NULL for all others
sion_int32 * all_coll_collsize
sion_int64 start_of_varheader
int _sion_file_flush(_sion_fileptr *sion_fileptr)
Flush data to file.
int _sion_generic_renumber_collmsa(_sion_generic_gendata *comm, _sion_flags_store *flags)
Splits a Communicator in numfiles different communicators.