14 #define _XOPEN_SOURCE 700 23 #include "sion_error_handler.h" 39 #include "sion_generic_buddy.h" 63 #define DFUNCTION "_sion_paropen_generic_buddy" 64 int _sion_paropen_generic_buddy(
71 sion_int64 *chunksize,
72 sion_int32 *fsblksize,
79 int rc = SION_SUCCESS;
80 int b, capability, pass, buddylevel;
81 sion_int32 file_globalrank;
82 sion_int64 file_chunksize;
83 char *buddy_fn, *nfname=NULL;
88 buddylevel = atoi(_sion_flags_get(flags_store,
"buddy")->val);
89 if (buddylevel==0) buddylevel=1;
91 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"allocate memory for %d buddy levels\n", buddylevel));
94 if (buddies == NULL) {
95 return(_sion_errorprint(SION_NOT_SUCCESS,_SION_ERROR_RETURN,
"cannot allocate buddies structure of size %lu (_sion_generic_buddy), aborting ...\n",
100 buddy_fn = calloc(SION_FILENAME_LENGTH+10,1);
101 if (buddy_fn == NULL) {
103 return(_sion_errorprint(SION_ID_NOT_VALID,_SION_ERROR_RETURN,
"_sion_paropen_generic_buddy: cannot allocate temporary memory of size %lu (buddy_fn), aborting ...\n",
104 (
unsigned long) SION_FILENAME_LENGTH+10));
108 sion_apidesc=sion_gendata->apidesc;
110 if (flags_store->mask&_SION_FMODE_WRITE) {
114 (fname,sion_gendata->filenumber);
115 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"call parallel open of %d files (current name %s)\n",
116 sion_gendata->numfiles, nfname));
118 &sion_gendata->numfiles, &sion_gendata->filenumber,
119 chunksize, fsblksize,
120 sion_gendata->lrank, sion_gendata->lsize, globalrank,
121 _SION_INTERNAL_FLAG_BUDDY_NORMAL, fileptr, sion_gendata, NULL);
122 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"leave parallel open of %d files in #tasks=%d sid=%d globalrank=%d\n", sion_gendata->numfiles,
123 sion_gendata->lsize, sid, sion_gendata->grank));
130 return(_sion_errorprint(SION_ID_NOT_VALID,_SION_ERROR_RETURN,
"_sion_paropen_generic_buddy: invalid sion_filedesc %d", sid));
134 sion_filedesc->
buddies=buddies;
136 capability=sion_apidesc->get_capability_cb(sion_gendata->comm_data_global);
140 buddyptr=&buddies[b];
143 buddyptr->buddy_send.commgroup=NULL; buddyptr->buddy_coll.commgroup=NULL;
144 rc=_sion_buddy_map(sion_gendata,capability,b+1,&buddyptr->buddy_send,&buddyptr->buddy_coll);
147 sprintf(buddy_fn,
"%s_BUDDY_%02d", fname,b);
150 for(pass=1;pass<=_SION_BW_SCHED_NUM_PASSES;pass++) {
152 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"BUDDYO pass=%d [grank=%2d] op=%d\n",pass,*globalrank,
153 _sion_buddy_bwsched(buddyptr->buddy_send.groupnum, sion_gendata->numfiles, pass)));
156 switch (_sion_buddy_bwsched(buddyptr->buddy_send.groupnum, sion_gendata->numfiles, pass)) {
157 case _SION_BW_SCHED_ACTIONA:
159 file_globalrank=-1*(*globalrank+1);
162 (buddy_fn,buddyptr->buddy_coll.filenum);
163 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"BUDDYO call open_one_file COLL [gendata: b=%d grank=%2d, group=%d of %d, lrank=%d of %d] fn=%s (%d)\n",
164 b,*globalrank, buddyptr->buddy_coll.groupnum,sion_gendata->numfiles,
165 buddyptr->buddy_coll.rank, buddyptr->buddy_coll.size,nfname,buddyptr->buddy_coll.filenum));
168 &sion_gendata->numfiles, &buddyptr->buddy_coll.filenum,
169 &file_chunksize, fsblksize,
170 buddyptr->buddy_coll.rank,buddyptr->buddy_coll.size, &file_globalrank,
171 _SION_INTERNAL_FLAG_BUDDY_COLL, NULL, sion_gendata,
175 case _SION_BW_SCHED_ACTIONB:
177 file_globalrank=*globalrank;
178 file_chunksize=*chunksize;
180 (buddy_fn,buddyptr->buddy_send.filenum);
181 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"BUDDYO call open_one_file SEND [gendata: b=%d grank=%2d, group=%d of %d, lrank=%d of %d] fn=%s \n",
182 b,*globalrank, buddyptr->buddy_send.groupnum,sion_gendata->numfiles,
183 buddyptr->buddy_send.rank, buddyptr->buddy_send.size,nfname));
186 &sion_gendata->numfiles, &buddyptr->buddy_send.filenum,
187 &file_chunksize, fsblksize,
188 buddyptr->buddy_send.rank,buddyptr->buddy_send.size, &file_globalrank,
189 _SION_INTERNAL_FLAG_BUDDY_SEND, NULL, sion_gendata,
194 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"BUDDYO no-op [grank=%2d]\n",*globalrank));
200 }
else if (flags_store->mask&_SION_FMODE_READ) {
201 int filefound, lgroup=*filenumber, root=0, task, tmpsize, group, masterfile_found;
202 sion_int32 filenumber, numfiles, lrank, lsize, file, file_lgroup, myrole, steprank;
203 sion_int32 helpint32=0, numsteps=0, numgroups=0, numreaderinthisfile;
204 sion_int32 *sion_tmpintfield1 = NULL;
207 int *stepvectorlist[MAXREADSTEPS];
208 int *stepvector, readercount=0, readyflag=0, datafound=0;
209 int stepcount_coll,stepcount_collreader,stepcount_reader,stepcount_reader_rank,stepcount_noreader,stepcount_noreader_rank, groupcounter;
210 int step,fit,newrank;
212 sion_int32 helpint7[7];
214 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"enter buddy open in read mode\n"));
216 if(sion_gendata->grank==0) {
217 tmpsize=1*sion_gendata->gsize;
218 sion_tmpintfield1 = (sion_int32 *) malloc(tmpsize *
sizeof(sion_int32));
219 if (sion_tmpintfield1 == NULL) {
222 return(_sion_errorprint(SION_ID_NOT_VALID,_SION_ERROR_RETURN,
"_sion_paropen_generic_buddy: cannot allocate temporary memory of size %lu (sion_tmpintfield), aborting ...\n",
223 (
unsigned long) tmpsize *
sizeof(sion_int32)));
229 for(b=-1; ( (b<buddylevel) && (!readyflag));b++) {
232 if(b==-1) sprintf(buddy_fn,
"%s", fname);
233 else sprintf(buddy_fn,
"%s_BUDDY_%02d", fname,b);
234 nfname=(sion_apidesc->get_multi_filename_cb?sion_apidesc->get_multi_filename_cb:
_sion_get_multi_filename)(buddy_fn,0);
235 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"BUDDY[%2d] start checking buddy level (file=%s)\n",b,nfname));
238 if(sion_gendata->lrank==0) {
241 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"check for existence of file %s --> file found\n",nfname));
247 sion_gendata->apidesc->gatherr_cb(&helpint32, sion_tmpintfield1, sion_gendata->comm_data_global, _SION_INT32, 1, 0);
248 if (sion_gendata->grank == 0) {
250 if (sion_tmpintfield1 == NULL) {
253 return(_sion_errorprint(SION_NOT_SUCCESS,_SION_ERROR_ABORT, DFUNCTION
": gatherr_cb returned sion_tmpintfield1 == NULL"));
255 for(task=0;task<sion_gendata->gsize;task++) {
256 if(sion_tmpintfield1[task]==1) root=task;
259 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"after scan master: root for this file is %d #tasks=%d\n",root,sion_gendata->gsize));
261 sion_gendata->apidesc->bcastr_cb(&helpint32, sion_gendata->comm_data_global, _SION_INT32, 1, 0); root=helpint32;
262 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"root for this file is %d\n",root));
266 _sion_generic_buddy_get_and_distribute_info_from_file(sion_gendata,nfname,root, &filenumber, &numfiles, &lrank, &lsize);
267 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"BUDDY[%d] after read mapping file=%d of %d, lrank=%d of %d\n",b,filenumber, numfiles, lrank, lsize));
270 filenumber=lrank=lsize=-1;
277 DPRINTFP((32, DFUNCTION, sion_gendata->grank,
"Info avail: b=%d numfiles=%d root=%d lgroup=%d\n",b,numfiles,root,lgroup));
280 for(file=0;( (file<numfiles) && (!readyflag) );file++) {
283 nfname=(sion_apidesc->get_multi_filename_cb?sion_apidesc->get_multi_filename_cb:
_sion_get_multi_filename)(buddy_fn,file);
285 if(sion_gendata->lrank==0) {
288 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"check for existence of file %s --> file found\n",nfname));
294 sion_gendata->apidesc->gatherr_cb(&helpint32, sion_tmpintfield1, sion_gendata->comm_data_global, _SION_INT32, 1, 0);
295 if (sion_gendata->grank == 0) {
297 for(task=0;task<sion_gendata->gsize;task++) {
298 if(sion_tmpintfield1[task]>-1) {
300 file_lgroup=sion_tmpintfield1[task];
303 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"after scan one file: root for this file is %d file_lgroup=%d\n",root,file_lgroup));
305 if (sion_gendata->grank == 0) helpint32=root;
306 sion_gendata->apidesc->bcastr_cb(&helpint32, sion_gendata->comm_data_global, _SION_INT32, 1, 0); root=helpint32;
307 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"root for this file is %d\n",root));
312 if(!masterfile_found) {
314 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"BUDDY[%d] befre read globalranks file=%d of %d, lrank=%d of %d\n",
315 b, filenumber, numfiles, lrank, lsize));
316 _sion_generic_buddy_get_and_distribute_info_from_one_file(sion_gendata, nfname, root, &filenumber, &numfiles, &lrank, &lsize);
317 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"BUDDY[%d] after read globalranks file=%d of %d, lrank=%d of %d\n",
318 b, filenumber, numfiles, lrank, lsize));
322 if (sion_gendata->grank == 0) helpint32=file_lgroup;
323 sion_gendata->apidesc->bcastr_cb(&helpint32, sion_gendata->comm_data_global, _SION_INT32, 1, 0); file_lgroup=helpint32;
324 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"file_lgroup for this file is %d\n",file_lgroup));
327 myrole=SION_ROLE_NONE;
328 if((filenumber==file) && (!datafound)) {
330 if(sion_gendata->grank==root) myrole=SION_ROLE_COLLECTOR_READER;
331 else myrole=SION_ROLE_READER;
333 if(sion_gendata->grank==root) myrole=SION_ROLE_COLLECTOR;
334 else if(file_lgroup==lgroup) myrole=SION_ROLE_NOREADER;
339 sion_gendata->apidesc->gatherr_cb(&helpint32, sion_tmpintfield1, sion_gendata->comm_data_global, _SION_INT32, 1, 0);
342 if(sion_gendata->grank == 0) {
343 numreaderinthisfile=0;
344 if (sion_tmpintfield1 == NULL) {
345 return(_sion_errorprint(SION_NOT_SUCCESS,_SION_ERROR_ABORT, DFUNCTION
": gatherr_cb returned sion_tmpintfield1 == NULL"));
347 for(task=0; ( (task<sion_gendata->gsize) ) ;task++)
348 if ( (sion_tmpintfield1[task]==SION_ROLE_COLLECTOR_READER) || (sion_tmpintfield1[task]==SION_ROLE_READER) ) numreaderinthisfile++;
350 sion_gendata->apidesc->bcastr_cb(&numreaderinthisfile, sion_gendata->comm_data_global, _SION_INT32, 1, 0);
352 if (numreaderinthisfile>0) {
354 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"myrole for file %s is %s\n",nfname,_sion_buddy_role_to_str(myrole)));
357 if (sion_gendata->grank == 0) {
360 for(step=0; ( (step<numsteps) && (fit==0) ) ;step++) {
361 stepvector=stepvectorlist[step];
363 for(task=0; ( (task<sion_gendata->gsize) && (fit==1) ) ;task++) {
364 if(sion_tmpintfield1[task]!=SION_ROLE_NONE) {
365 if(stepvector[task]!=SION_ROLE_NONE) fit=0;
369 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"BUDDY STEP assign: after scanning %d steps fit=%d\n",numsteps,fit));
372 stepvector = (
int *) malloc(sion_gendata->gsize *
sizeof(
int));
373 if (stepvector == NULL) {
374 return(_sion_errorprint(SION_ID_NOT_VALID,_SION_ERROR_RETURN,
"_sion_paropen_generic_buddy: cannot allocate temporary memory of size %lu (stepvector), aborting ...\n",
375 (
unsigned long) sion_gendata->gsize *
sizeof(
int)));
377 for(task=0; (task<sion_gendata->gsize);task++) stepvector[task]=SION_ROLE_NONE;
379 stepvectorlist[numsteps-1]=stepvector;
380 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"BUDDY STEP assign: allocating new step %d\n",numsteps-1));
383 stepcount_coll=stepcount_collreader=stepcount_reader=stepcount_noreader=0;
384 for(task=0; (task<sion_gendata->gsize);task++) {
385 if(sion_tmpintfield1[task]!=SION_ROLE_NONE) {
386 stepvector[task]=sion_tmpintfield1[task];
387 if (sion_tmpintfield1[task]==SION_ROLE_COLLECTOR) stepcount_coll++;
388 if (sion_tmpintfield1[task]==SION_ROLE_READER) {readercount++;stepcount_reader++;}
389 if (sion_tmpintfield1[task]==SION_ROLE_COLLECTOR_READER) {readercount++;stepcount_collreader++;}
390 if (sion_tmpintfield1[task]==SION_ROLE_NOREADER) stepcount_noreader++;
395 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"BUDDY STEP assign: group assigned to step %d (tasks with no data: %d)\n",numsteps-1,sion_gendata->gsize-readercount));
396 if(sion_gendata->gsize-readercount==0) readyflag=1;
399 stepcount_noreader_rank = stepcount_coll;
400 stepcount_reader_rank = stepcount_noreader_rank+stepcount_noreader;
401 for(task=0; (task<sion_gendata->gsize);task++) {
404 if(sion_tmpintfield1[task]==SION_ROLE_NONE) newrank=-1;
405 if(sion_tmpintfield1[task]==SION_ROLE_COLLECTOR) newrank=0;
406 if(sion_tmpintfield1[task]==SION_ROLE_COLLECTOR_READER) newrank=stepcount_reader_rank++;
407 if(sion_tmpintfield1[task]==SION_ROLE_NOREADER) newrank=stepcount_noreader_rank++;
408 if(sion_tmpintfield1[task]==SION_ROLE_READER) newrank=stepcount_reader_rank++;
409 sion_tmpintfield1[task]=newrank;
416 sion_gendata->apidesc->scatterr_cb(sion_tmpintfield1, &steprank, sion_gendata->comm_data_global, _SION_INT32, 1, 0);
420 if (sion_gendata->grank == 0) {
423 helpint7[1]=readyflag;
424 helpint7[2]=stepcount_coll+stepcount_collreader+stepcount_reader+stepcount_noreader;
425 helpint7[3]=stepcount_collreader+stepcount_reader;
426 helpint7[4]=stepcount_coll+stepcount_noreader;
427 helpint7[5]=stepcount_coll+stepcount_noreader+stepcount_collreader+stepcount_reader-1;
428 helpint7[6]=groupcounter++;
429 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"define group #%d (step=%d, readyflag=%d, size=%d, collsize=%d, from=%d, to=%d) \n",
430 helpint7[6], helpint7[0], helpint7[1], helpint7[2], helpint7[3], helpint7[4], helpint7[5]));
432 sion_gendata->apidesc->bcastr_cb(helpint7, sion_gendata->comm_data_global, _SION_INT32, 7, 0);
433 readyflag=helpint7[1];
436 if(myrole!=SION_ROLE_NONE) {
438 if (buddy_info == NULL) {
440 return(_sion_errorprint(SION_ID_NOT_VALID,_SION_ERROR_RETURN,
"_sion_paropen_generic_buddy: cannot allocate temporary memory of size %lu (buddy_info), aborting ...\n",
443 buddies->groups[buddies->numgroups]=buddy_info; buddies->numgroups++;
444 buddy_info->groupid=helpint7[6];
445 buddy_info->stepnum=helpint7[0];
446 buddy_info->rank=steprank;
447 buddy_info->size=helpint7[2];
448 buddy_info->commgroup=NULL;
449 buddy_info->collsize=helpint7[3];
450 buddy_info->from_index=helpint7[4];
451 buddy_info->to_index=helpint7[5];
452 buddy_info->myrole=myrole;
454 if( (myrole==SION_ROLE_READER) || (myrole==SION_ROLE_COLLECTOR_READER)) {
455 buddy_info->filelrank=lrank;
457 buddy_info->filelrank=-1;
459 buddy_info->filenum=file;
460 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"create group #%d (step=%d, rank=%d, size=%d, collsize=%d, from=%2d, to=%2d) (bnum=%2d file# %2d) with role %s \n",
461 buddy_info->groupid, buddy_info->stepnum, buddy_info->rank, buddy_info->size, buddy_info->collsize,
462 buddy_info->from_index, buddy_info->to_index, buddy_info->bnum, buddy_info->filenum, _sion_buddy_role_to_str(buddy_info->myrole)));
474 return(_sion_errorprint(SION_ID_NOT_VALID,_SION_ERROR_RETURN,
"_sion_paropen_generic_buddy: cannot open file, not all files available"));
478 sion_gendata->apidesc->bcastr_cb(&numsteps, sion_gendata->comm_data_global, _SION_INT32, 1, 0);
479 buddies->numsteps=numsteps;
482 if (sion_gendata->grank == 0) numgroups=groupcounter;
483 sion_gendata->apidesc->bcastr_cb(&numgroups, sion_gendata->comm_data_global, _SION_INT32, 1, 0);
486 for(step=0;step<numsteps;step++) {
489 for(b=0; ( (b<buddies->numgroups) && (group==-1) ); b++ ) {
491 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"check group #%d of %d in stepnum=%d step=%d\n",b, buddies->numgroups, buddies->groups[b]->stepnum,step));
492 if (buddies->groups[b]->stepnum==step) group=b;
497 buddy_info=buddies->groups[group];
498 rc=sion_gendata->apidesc->create_lcg_cb(&buddy_info->commgroup,sion_gendata->comm_data_global,
499 sion_gendata->grank,sion_gendata->gsize,
500 buddy_info->rank,buddy_info->size,
501 buddy_info->groupid,numgroups);
502 DPRINTFP((256, DFUNCTION, _SION_DEFAULT_RANK,
"COMM_created active [grank=%2d of %2d, group=%2d of %2d, lrank=%2d of %2d]\n",
503 sion_gendata->grank,sion_gendata->gsize, buddy_info->groupid,numgroups, buddy_info->rank,buddy_info->size));
506 void *dummycommgroup=NULL;
507 rc=sion_gendata->apidesc->create_lcg_cb(&dummycommgroup,sion_gendata->comm_data_global,
508 sion_gendata->grank,sion_gendata->gsize,
511 DPRINTFP((256, DFUNCTION, _SION_DEFAULT_RANK,
"COMM_created dummy [grank=%2d of %2d, group=%2d of %2d, lrank=%2d of %2d]\n",
512 sion_gendata->grank,sion_gendata->gsize,
520 buddy_info=buddies->groups[group];
521 buddies->currentgroup=group;
523 if(buddy_info->bnum==-1) sprintf(buddy_fn,
"%s", fname);
524 else sprintf(buddy_fn,
"%s_BUDDY_%02d", fname,buddy_info->bnum);
525 nfname=(sion_apidesc->get_multi_filename_cb?sion_apidesc->get_multi_filename_cb:
_sion_get_multi_filename)(buddy_fn,buddy_info->filenum);
527 DPRINTFP((256, DFUNCTION, _SION_DEFAULT_RANK,
"open file for group #%d %s [step=%d grank=%2d of %2d, lrank=%2d of %2d filelrank=%2d]\n",
528 buddy_info->groupid,nfname,buddy_info->stepnum,
529 sion_gendata->grank,sion_gendata->gsize, buddy_info->rank,buddy_info->size,buddy_info->filelrank));
532 if( (buddy_info->myrole == SION_ROLE_READER)
533 || (buddy_info->myrole == SION_ROLE_COLLECTOR_READER)) {
536 &sion_gendata->numfiles, &buddy_info->filenum,
537 chunksize, fsblksize,
538 buddy_info->rank, buddy_info->size, &sion_gendata->grank,
539 _SION_INTERNAL_FLAG_BUDDY_READ, NULL, sion_gendata,
546 return(_sion_errorprint(SION_ID_NOT_VALID,_SION_ERROR_RETURN,
"_sion_paropen_generic_buddy: invalid sion_filedesc %d", sid));
550 sion_filedesc->
buddies=buddies;
553 sion_int64 dummy_chunksize;
554 sion_int32 dummy_fsblksize;
557 &sion_gendata->numfiles, &buddy_info->filenum,
558 &dummy_chunksize, &dummy_fsblksize,
559 buddy_info->rank, buddy_info->size, &sion_gendata->grank,
560 _SION_INTERNAL_FLAG_BUDDY_READ, NULL, sion_gendata,
565 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"BUDDYO afteropen [grank=%2d step=%d] rc=%d\n",sion_gendata->grank,step,rc));
570 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"BUDDYO no-op [grank=%2d]\n",sion_gendata->grank));
574 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"BUDDYO beforegbarrier [grank=%2d]\n",sion_gendata->grank));
575 sion_apidesc->barrier_cb(sion_gendata->comm_data_global);
576 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"BUDDYO aftergbarrier [grank=%2d]\n",sion_gendata->grank));
585 return(_sion_errorprint(SION_ID_NOT_VALID,_SION_ERROR_RETURN,
"_sion_paropen_generic_buddy: unknown file mode"));
590 sion_apidesc->barrier_cb(sion_gendata->comm_data_global);
609 #define DFUNCTION "_sion_parclose_generic_buddy" 610 int _sion_parclose_generic_buddy(
int sid,
620 int b_mapping_size=0,group,step;
621 sion_int32 *b_mapping=NULL;
628 return(_sion_errorprint(SION_ID_NOT_VALID,_SION_ERROR_RETURN,
"_sion_parclose_generic_buddy: invalid sion_filedesc %d", sid));
631 buddies=sion_filedesc->
buddies;
638 buddyptr=&buddies[b];
642 _sion_generic_collect_mapping_buddy(buddyptr, sion_gendata, &b_mapping_size, &b_mapping);
646 for(pass=1;pass<=_SION_BW_SCHED_NUM_PASSES;pass++) {
648 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"BUDDYC pass=%d [grank=%2d] op=%d\n",pass,sion_gendata->grank,
649 _sion_buddy_bwsched(buddyptr->buddy_send.groupnum, sion_gendata->numfiles, pass)));
651 switch (_sion_buddy_bwsched(buddyptr->buddy_send.groupnum, sion_gendata->numfiles, pass)) {
652 case _SION_BW_SCHED_ACTIONA:
653 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"BUDDYc call parallel close COLL [gendata: grank=%2d, group=%d of %d, lrank=%d of %d]\n",
654 sion_gendata->grank, buddyptr->buddy_coll.groupnum,sion_gendata->numfiles,
655 buddyptr->buddy_coll.rank, buddyptr->buddy_coll.size));
657 b_mapping_size, b_mapping, _SION_INTERNAL_FLAG_BUDDY_COLL, sion_gendata, buddyptr );
659 case _SION_BW_SCHED_ACTIONB:
660 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"BUDDYc call parallel close SEND [gendata: grank=%2d, group=%d of %d, lrank=%d of %d]\n",
661 sion_gendata->grank, buddyptr->buddy_send.groupnum,sion_gendata->numfiles,
662 buddyptr->buddy_send.rank, buddyptr->buddy_send.size));
664 0, NULL, _SION_INTERNAL_FLAG_BUDDY_SEND, sion_gendata, buddyptr );
667 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"BUDDYC no-op [grank=%2d]\n",sion_gendata->grank));
674 _SION_SAFE_FREE(b_mapping, NULL);
675 _SION_SAFE_FREE(buddies, NULL);
679 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"call parallel close of %d files, sid=%d\n",
680 sion_gendata->numfiles, sid));
682 sion_gendata, NULL );
683 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"leave parallel close of %d files in #tasks=%d globalrank=%d\n", sion_gendata->numfiles,
684 sion_gendata->lsize, sion_gendata->grank));
689 for(step=0;step<buddies->numsteps;step++) {
692 for(b=0; ( (b<buddies->numgroups) && (group==-1) ); b++ ) {
694 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"check group #%d of %d %d==%d %x\n",b, buddies->numgroups, buddies->groups[b]->stepnum,step, buddies->groups[b]));
695 if (buddies->groups[b]->stepnum==step) group=b;
697 buddy_info=buddies->groups[group];
701 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"call parallel close of %d files, sid=%d\n",
702 sion_gendata->numfiles, sid));
703 buddies->currentgroup=group;
704 rc =
_sion_parclose_generic( buddy_info->sid, buddy_info->rank,buddy_info->size, -1, NULL, _SION_INTERNAL_FLAG_BUDDY_READ,
705 sion_gendata, buddies );
706 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"leave parallel close of %d files in #tasks=%d globalrank=%d\n", sion_gendata->numfiles,
707 sion_gendata->lsize, sion_gendata->grank));
713 return(_sion_errorprint(SION_ID_NOT_VALID,_SION_ERROR_RETURN,
"_sion_parclose_generic_buddy: unknown file mode"));
730 #define DFUNCTION "_sion_coll_fwrite_buddy" 731 int _sion_coll_fwrite_buddy(
const void *data,
737 _sion_filedesc *sion_filedesc,*sion_filedesc_coll,*sion_filedesc_send;
740 sion_int64 spec[2], ownnewposition;
741 int b, pass, collector=0;
743 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"enter parallel write buddy\n"));
747 return(_sion_errorprint(SION_ID_NOT_VALID,_SION_ERROR_RETURN,
"_sion_coll_fwrite_buddy: invalid sion_filedesc %d", sid));
749 sion_apidesc=sion_gendata->apidesc;
750 buddies=sion_filedesc->
buddies;
754 buddyptr=&buddies[b];
758 return(_sion_errorprint(SION_ID_NOT_VALID,_SION_ERROR_RETURN,
"_sion_coll_fwrite_buddy: invalid sion_filedesc %d", buddyptr->buddy_coll.sid));
761 return(_sion_errorprint(SION_ID_NOT_VALID,_SION_ERROR_RETURN,
"_sion_coll_fwrite_buddy: invalid sion_filedesc %d", buddyptr->buddy_send.sid));
766 for(pass=1;pass<=_SION_BW_SCHED_NUM_PASSES;pass++) {
768 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"BUDDYw pass=%d [grank=%2d] op=%d\n",pass,sion_gendata->grank,
769 _sion_buddy_bwsched(buddyptr->buddy_send.groupnum, sion_gendata->numfiles, pass)));
771 switch (_sion_buddy_bwsched(buddyptr->buddy_send.groupnum, sion_gendata->numfiles, pass)) {
772 case _SION_BW_SCHED_ACTIONA:
773 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"BUDDYw call parallel write COLL [gendata: grank=%2d, group=%d of %d, lrank=%d of %d] %d..%d\n",
774 sion_gendata->grank, buddyptr->buddy_coll.groupnum,sion_gendata->numfiles,
775 buddyptr->buddy_coll.rank, buddyptr->buddy_coll.size,buddyptr->buddy_coll.from_index,buddyptr->buddy_coll.to_index));
777 if(sion_filedesc_coll->
rank == collector) {
778 ownnewposition=sion_filedesc_coll->
currentpos;
779 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"BUDDYw call parallel write COLL currentpos=%d\n",(
int) sion_filedesc_coll->
currentpos));
781 rc = sion_apidesc->gather_execute_cb(data,spec,2, sion_filedesc_coll->
fsblksize,
782 buddyptr->buddy_coll.commgroup,collector,buddyptr->buddy_coll.from_index,buddyptr->buddy_coll.to_index,
783 buddyptr->buddy_coll.sid, _sion_generic_collective_process_write);
784 if(sion_filedesc_coll->
rank == collector) {
787 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"BUDDYw call parallel write COLL currentpos=%d\n",(
int) sion_filedesc_coll->
currentpos));
790 case _SION_BW_SCHED_ACTIONB:
791 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"BUDDYw call parallel write SEND [gendata: grank=%2d, group=%d of %d, lrank=%d of %d] %d..%d\n",
792 sion_gendata->grank, buddyptr->buddy_send.groupnum,sion_gendata->numfiles,
793 buddyptr->buddy_send.rank, buddyptr->buddy_send.size,buddyptr->buddy_send.from_index,buddyptr->buddy_send.to_index));
797 _sion_errorprint(SION_SIZE_NOT_VALID,_SION_ERROR_RETURN,
"could not ensure free space for this block, returning %d ...\n", sid);
803 rc = sion_apidesc->gather_execute_cb(data,spec,2, sion_filedesc_send->
fsblksize,
804 buddyptr->buddy_send.commgroup,collector,buddyptr->buddy_send.from_index,buddyptr->buddy_send.to_index,
805 buddyptr->buddy_send.sid, _sion_generic_collective_process_write);
809 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"BUDDYw no-op [grank=%2d]\n",sion_gendata->grank));
818 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"leave parallel write buddy\n"));
835 #define DFUNCTION "_sion_coll_fread_buddy" 836 int _sion_coll_fread_buddy(
void *data,
size_t size,
size_t nitems,
int sid) {
840 sion_int64 bread=-1, spec[2], ownnewposition, items_read = 0;
841 int rc_own=SION_STD_SUCCESS,rc_cb=SION_STD_SUCCESS;
842 int collector, firstsender, lastsender, step, b, group;
848 return(_sion_errorprint(SION_SIZE_NOT_VALID,_SION_ERROR_RETURN,
"_sion_coll_fread_buddy: invalid sion_filedesc %d", sid));
850 DPRINTFP((4, DFUNCTION, _SION_DEFAULT_RANK,
"enter usecoll=%d collector=%d collsize=%d (%d tasks, %d files)\n",
853 sion_gendata=sion_filedesc->
dataptr;
854 sion_apidesc=sion_gendata->apidesc;
855 buddies=sion_filedesc->
buddies;
862 return(_sion_errorprint(SION_SIZE_NOT_VALID,_SION_ERROR_RETURN,
"_sion_coll_fread_buddy: collsize=%d <= 0, returning ...\n",
868 for(step=0;step<buddies->numsteps;step++) {
871 for(b=0; ( (b<buddies->numgroups) && (group==-1) ); b++ ) {
873 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"check group #%d of %d %d==%d %x\n",b, buddies->numgroups, buddies->groups[b]->stepnum,step, buddies->groups[b]));
874 if (buddies->groups[b]->stepnum==step) group=b;
880 buddy_info=buddies->groups[group];
883 return(_sion_errorprint(SION_SIZE_NOT_VALID,_SION_ERROR_RETURN,
"_sion_coll_fread_buddy: invalid sion_filedesc %d", buddy_info->sid));
888 firstsender=buddy_info->from_index;
889 lastsender=buddy_info->to_index;
893 if( (buddy_info->myrole == SION_ROLE_READER)
894 || (buddy_info->myrole == SION_ROLE_COLLECTOR_READER)) {
899 _sion_errorprint(SION_SIZE_NOT_VALID,_SION_ERROR_RETURN,
"early eof found for this block, returning %d ...\n", buddy_info->sid);
916 if( (buddy_info->myrole == SION_ROLE_COLLECTOR_READER) ) {
917 rc_own=_sion_generic_collective_process_read(data,spec,buddy_info->sid);
921 if( (buddy_info->myrole == SION_ROLE_COLLECTOR)
922 || (buddy_info->myrole == SION_ROLE_COLLECTOR_READER)) {
928 if(!sion_apidesc->execute_scatter_cb ) {
929 return(_sion_errorprint(SION_SIZE_NOT_VALID,_SION_ERROR_RETURN,
930 "_sion_coll_fread_buddy: API %s not correctly initalized, collective I/O calls missing, aborting",sion_apidesc->name));
932 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"call execute_gather #%d (step=%d, rank=%d, size=%d, collsize=%d, from=%2d, to=%2d) (bnum=%2d file# %2d) with role %s \n",
933 buddy_info->groupid, buddy_info->stepnum, buddy_info->rank, buddy_info->size, buddy_info->collsize,
934 buddy_info->from_index, buddy_info->to_index, buddy_info->bnum, buddy_info->filenum, _sion_buddy_role_to_str(buddy_info->myrole)));
936 rc_cb=sion_apidesc->execute_scatter_cb(data,spec,2, b_sion_filedesc->
fsblksize,
937 buddy_info->commgroup,collector,firstsender,lastsender,buddy_info->sid,
938 _sion_generic_collective_process_read);
943 if( (buddy_info->myrole == SION_ROLE_COLLECTOR)
944 || (buddy_info->myrole == SION_ROLE_COLLECTOR_READER)) {
950 if(buddy_info->myrole == SION_ROLE_READER) {
957 if( (buddy_info->myrole == SION_ROLE_READER)
958 || (buddy_info->myrole == SION_ROLE_COLLECTOR_READER)) {
959 if( (rc_own == SION_STD_SUCCESS) && (rc_cb == SION_STD_SUCCESS) ) {
964 items_read = size ? bread / size : 0;
970 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"BUDDYO beforegbarrier [grank=%2d]\n",sion_gendata->grank));
971 sion_apidesc->barrier_cb(sion_gendata->comm_data_global);
972 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"BUDDYO aftergbarrier [grank=%2d]\n",sion_gendata->grank));
976 DPRINTFP((4, DFUNCTION, _SION_DEFAULT_RANK,
"leave usecoll=%d collector=%d collsize=%d (%d tasks, %d files) rc=%d\n",
984 #define DFUNCTION "_sion_buddy_map" 992 int tmpsize,t,g, g_map, p, p_map, n, n_map, orig, distance,distance_1step,pass;
993 int *size_per_group = NULL, *group_map_orig_to_new, *group_map_new_to_orig, mynewgroupnr;
994 sion_int32 helpint2[2];
995 sion_int32 *tasktogroup = NULL;
996 sion_int32 *tmpintfield1 = NULL;
997 int comm_send_rank, comm_send_size, comm_send_grpnum;
998 int comm_coll_rank, comm_coll_size, comm_coll_grpnum;
999 void *dummycommgroup=NULL;
1001 DPRINTFP((256, DFUNCTION, _SION_DEFAULT_RANK,
"buddy_idx=%d capability=%d [gendata: grank=%d of %d, group=%d of %d, lrank=%d of %d]\n",
1002 buddy_idx, capability,
1003 sion_gendata->grank, sion_gendata->gsize,
1004 sion_gendata->filenumber, sion_gendata->numfiles,
1005 sion_gendata->lrank, sion_gendata->lsize));
1011 if(sion_gendata->grank==0) {
1013 tmpsize=1*sion_gendata->gsize;
1014 tasktogroup = (sion_int32 *) malloc(tmpsize *
sizeof(sion_int32));
1015 if (tasktogroup == NULL) {
1016 return(_sion_errorprint(SION_ID_NOT_VALID,_SION_ERROR_RETURN,
"_sion_buddy_map: cannot allocate temporary memory of size %lu (tasktogroup), aborting ...\n",
1017 (
unsigned long) tmpsize *
sizeof(sion_int32)));
1020 tmpsize=2*sion_gendata->gsize;
1021 tmpintfield1 = (sion_int32 *) malloc(tmpsize *
sizeof(sion_int32));
1022 if (tmpintfield1 == NULL) {
1024 return(_sion_errorprint(SION_ID_NOT_VALID,_SION_ERROR_RETURN,
"_sion_buddy_map: cannot allocate temporary memory of size %lu (tmpintfield1), aborting ...\n",
1025 (
unsigned long) tmpsize *
sizeof(sion_int32)));
1028 size_per_group = (
int *) malloc(sion_gendata->numfiles *
sizeof(
int));
1029 if (size_per_group == NULL) {
1032 return(_sion_errorprint(SION_ID_NOT_VALID,_SION_ERROR_RETURN,
"_sion_buddy_map: cannot allocate temporary memory of size %lu (size_per_group), aborting ...\n",
1033 (
unsigned long) sion_gendata->numfiles *
sizeof(
int)));
1039 group_map_orig_to_new = (
int *) malloc(sion_gendata->numfiles *
sizeof(
int));
1040 if (group_map_orig_to_new == NULL) {
1042 _SION_SAFE_FREE(size_per_group, NULL);
1044 return(_sion_errorprint(SION_ID_NOT_VALID,_SION_ERROR_RETURN,
"_sion_buddy_map: cannot allocate temporary memory of size %lu (group_map_orig_to_new), aborting ...\n",
1045 (
unsigned long) sion_gendata->numfiles *
sizeof(
int)));
1048 group_map_new_to_orig = (
int *) malloc(sion_gendata->numfiles *
sizeof(
int));
1049 if (group_map_new_to_orig == NULL) {
1050 _SION_SAFE_FREE(size_per_group, NULL);
1051 free(group_map_orig_to_new);
1054 return(_sion_errorprint(SION_ID_NOT_VALID,_SION_ERROR_RETURN,
"_sion_buddy_map: cannot allocate temporary memory of size %lu (group_map_new_to_orig), aborting ...\n",
1055 (
unsigned long) sion_gendata->numfiles *
sizeof(
int)));
1059 helpint2[0]=sion_gendata->filenumber;
1060 sion_gendata->apidesc->gatherr_cb(helpint2, tasktogroup, sion_gendata->comm_data_global, _SION_INT32,1,0);
1064 helpint2[0]=sion_gendata->lsize;
1065 helpint2[1]=capability;
1066 sion_gendata->apidesc->gatherr_cb(helpint2, tmpintfield1, sion_gendata->comm_data_global, _SION_INT32,2,0);
1067 if(sion_gendata->grank==0) {
1068 for(t=0;t<sion_gendata->gsize;t++) {
1069 DPRINTFP((256, DFUNCTION, _SION_DEFAULT_RANK,
"task-map --> t=%2d grpnr=%2d lsize=%2d capability=%d\n",
1070 t,tasktogroup[t],tmpintfield1[2*t+0],tmpintfield1[2*t+1]));
1076 if(sion_gendata->grank==0) {
1077 for(t=0;t<sion_gendata->gsize;t++) {
1078 size_per_group[tasktogroup[t]] = tmpintfield1[2*t+0];
1082 for(g=0;g<sion_gendata->numfiles;g++) {
1083 orig=(g*distance) % sion_gendata->numfiles;
1084 group_map_orig_to_new[orig]=g;
1085 group_map_new_to_orig[g]=orig;
1088 for(g=0;g<sion_gendata->numfiles;g++) {
1089 DPRINTFP((256, DFUNCTION, _SION_DEFAULT_RANK,
"grp-map --> g=%2d->%2d\n",g,group_map_orig_to_new[g]));
1094 if(sion_gendata->grank==0) {
1095 for(t=0;t<sion_gendata->gsize;t++) {
1097 g_map=group_map_orig_to_new[g];
1098 p_map=(g_map-distance_1step + sion_gendata->numfiles) % sion_gendata->numfiles;
1099 p=group_map_new_to_orig[p_map];
1100 tmpintfield1[2*t+0] = g_map;
1101 tmpintfield1[2*t+1] = size_per_group[p];
1102 DPRINTFP((256, DFUNCTION, _SION_DEFAULT_RANK,
"comm-send --> t=%2d grpnr=%2d grpmap=%2d prevgr=%2d prevgrmap=%2d prev_lsize=%2d\n",
1103 t, g, tmpintfield1[2*t+0], p, p_map, tmpintfield1[2*t+1]));
1109 sion_gendata->apidesc->scatterr_cb(tmpintfield1, helpint2, sion_gendata->comm_data_global, _SION_INT32,2,0);
1112 mynewgroupnr=comm_send_grpnum=helpint2[0];
1113 buddy_send->rank=comm_send_rank=helpint2[1]+sion_gendata->lrank;
1114 buddy_send->size=comm_send_size=helpint2[1]+sion_gendata->lsize;
1115 buddy_send->collsize=sion_gendata->lsize;
1116 buddy_send->groupnum=mynewgroupnr;
1117 buddy_send->filenum=group_map_new_to_orig[mynewgroupnr];
1118 buddy_send->from_index=helpint2[1];
1119 buddy_send->to_index=helpint2[1]+sion_gendata->lsize-1;
1120 DPRINTFP((256, DFUNCTION, _SION_DEFAULT_RANK,
"COMM_SEND d=%d grpnum=%2d lrank=%2d lsize=%2d mynewgroupnr=%2d collsize=%2d %d..%d fnum=%d\n",
1121 distance,comm_send_grpnum,comm_send_rank,comm_send_size,mynewgroupnr,
1122 buddy_send->collsize,buddy_send->from_index,buddy_send->to_index,
1123 buddy_send->filenum));
1126 if(sion_gendata->grank==0) {
1127 for(t=0;t<sion_gendata->gsize;t++) {
1129 g_map=group_map_orig_to_new[g];
1130 n_map=(g_map+distance_1step) % sion_gendata->numfiles;
1131 n=group_map_new_to_orig[n_map];
1132 tmpintfield1[2*t+0] = n_map;
1133 tmpintfield1[2*t+1] = size_per_group[n];
1134 DPRINTFP((256, DFUNCTION, _SION_DEFAULT_RANK,
"comm-coll --> t=%2d grpnr=%2d grpmap=%2d nextgr=%2d nextgrmap=%2d next_lsize=%2d\n",
1135 t, g, g_map, n, tmpintfield1[2*t+0], tmpintfield1[2*t+1]));
1141 sion_gendata->apidesc->scatterr_cb(tmpintfield1, helpint2, sion_gendata->comm_data_global, _SION_INT32,2,0);
1144 comm_coll_grpnum=helpint2[0];
1145 buddy_coll->rank=comm_coll_rank=sion_gendata->lrank;
1146 buddy_coll->size=comm_coll_size=sion_gendata->lsize+helpint2[1];
1147 buddy_coll->collsize=helpint2[1];
1148 buddy_coll->groupnum=comm_coll_grpnum;
1149 buddy_coll->filenum=group_map_new_to_orig[comm_coll_grpnum];
1150 buddy_coll->from_index=sion_gendata->lsize;
1151 buddy_coll->to_index=sion_gendata->lsize+helpint2[1]-1;
1152 DPRINTFP((256, DFUNCTION, _SION_DEFAULT_RANK,
"COMM_COLL d=%d grpnum=%2d lrank=%2d lsize=%2d collsize=%2d %d..%d fnum=%d\n",
1153 distance,comm_coll_grpnum,comm_coll_rank,comm_coll_size,
1154 buddy_coll->collsize,buddy_coll->from_index,buddy_coll->to_index,
1155 buddy_coll->filenum));
1159 for(pass=1;pass<=_SION_BW_SCHED_NUM_PASSES;pass++) {
1160 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"pass=%d [grank=%2d] op=%d\n",pass,sion_gendata->grank,
1161 _sion_buddy_bwsched(mynewgroupnr, sion_gendata->numfiles, pass)));
1163 switch (_sion_buddy_bwsched(mynewgroupnr, sion_gendata->numfiles, pass)) {
1164 case _SION_BW_SCHED_ACTIONA:
1165 rc=sion_gendata->apidesc->create_lcg_cb(&buddy_coll->commgroup,sion_gendata->comm_data_global,
1166 sion_gendata->grank,sion_gendata->gsize,
1167 comm_coll_rank,comm_coll_size,
1168 comm_coll_grpnum,sion_gendata->numfiles);
1169 DPRINTFP((256, DFUNCTION, _SION_DEFAULT_RANK,
"COMM_created comm_coll (d=%1d, p=%1d) [gendata: mygrp=%2d grank=%2d of %2d, group=%2d of %2d, lrank=%2d of %2d]\n",
1170 distance,pass,mynewgroupnr,sion_gendata->grank,sion_gendata->gsize,
1171 comm_coll_grpnum,sion_gendata->numfiles,
1172 comm_coll_rank,comm_coll_size ));
1174 case _SION_BW_SCHED_ACTIONB:
1175 rc=sion_gendata->apidesc->create_lcg_cb(&buddy_send->commgroup,sion_gendata->comm_data_global,
1176 sion_gendata->grank,sion_gendata->gsize,
1177 comm_send_rank,comm_send_size,
1178 comm_send_grpnum,sion_gendata->numfiles);
1179 DPRINTFP((256, DFUNCTION, _SION_DEFAULT_RANK,
"COMM_created comm_send (d=%1d, p=%1d) [gendata: mygrp=%2d grank=%2d of %2d, group=%2d of %2d, lrank=%2d of %2d]\n",
1180 distance,pass,mynewgroupnr,sion_gendata->grank,sion_gendata->gsize,
1181 comm_send_grpnum,sion_gendata->numfiles,
1182 comm_send_rank,comm_send_size ));
1184 case _SION_BW_SCHED_NOACTION:
1185 rc=sion_gendata->apidesc->create_lcg_cb(&dummycommgroup,sion_gendata->comm_data_global,
1186 sion_gendata->grank,sion_gendata->gsize,
1188 -1,sion_gendata->numfiles);
1189 DPRINTFP((256, DFUNCTION, _SION_DEFAULT_RANK,
"COMM_created dummy (d=%1d, p=%1d) [gendata: mygrp=%2d grank=%2d of %2d, group=%2d of %2d, lrank=%2d of %2d]\n",
1190 distance,pass,mynewgroupnr,sion_gendata->grank,sion_gendata->gsize,
1191 -1,sion_gendata->numfiles,0,1 ));
1199 if(sion_gendata->grank==0) {
1202 free(size_per_group);
1204 free(group_map_new_to_orig);
1205 free(group_map_orig_to_new);
1212 #define DFUNCTION "_sion_buddy_bwsched" 1213 int _sion_buddy_bwsched(
int groupnr,
int numgroups,
int pass) {
1214 int res=_SION_BW_SCHED_NOACTION;
1215 if(numgroups%2==0) {
1217 if(pass==1) res=_SION_BW_SCHED_ACTIONA;
1218 if(pass==2) res=_SION_BW_SCHED_ACTIONB;
1220 if(pass==1) res=_SION_BW_SCHED_ACTIONB;
1221 if(pass==2) res=_SION_BW_SCHED_ACTIONA;
1225 if((pass==1) && (groupnr != (numgroups-1))) res=_SION_BW_SCHED_ACTIONA;
1226 if((pass==2) && (groupnr != 0)) res=_SION_BW_SCHED_ACTIONB;
1227 if((pass==3) && (groupnr == 0)) res=_SION_BW_SCHED_ACTIONB;
1228 if((pass==3) && (groupnr == (numgroups-1))) res=_SION_BW_SCHED_ACTIONA;
1230 if(pass==1) res=_SION_BW_SCHED_ACTIONB;
1231 if(pass==2) res=_SION_BW_SCHED_ACTIONA;
1242 #define DFUNCTION "_sion_generic_collect_mapping" 1246 sion_int32 **mapping) {
1247 int rc=SION_SUCCESS;
1250 sion_int32 lpos[2], *receivemap=NULL, iamreceiver, receiver = -1;
1252 sion_apidesc=sion_gendata->apidesc;
1254 *mapping = NULL; *mapping_size = 0;
1258 if((buddyptr->buddy_coll.groupnum==0) && (buddyptr->buddy_coll.rank==0)) {
1260 *mapping_size=sion_gendata->gsize;
1261 *mapping = (sion_int32 *) malloc(*mapping_size * 2 *
sizeof(sion_int32));
1262 if (*mapping == NULL) {
1263 return(_sion_errorprint(SION_ID_NOT_VALID,_SION_ERROR_RETURN,
"_sion_generic_collect_mapping_buddy: Cannot allocate memory for mapping"));
1268 if(sion_gendata->grank==0) {
1269 receivemap = (sion_int32 *) malloc(sion_gendata->gsize *
sizeof(sion_int32));
1270 if (receivemap == NULL) {
1271 return(_sion_errorprint(SION_ID_NOT_VALID,_SION_ERROR_RETURN,
"_sion_generic_collect_mapping_buddy: Cannot allocate memory for receivemap"));
1275 if((buddyptr->buddy_coll.filenum==0) && (buddyptr->buddy_coll.rank==0)) iamreceiver=sion_gendata->grank;
1276 else iamreceiver=-1;
1277 sion_apidesc->gatherr_cb(&iamreceiver, receivemap, sion_gendata->comm_data_global, _SION_INT32, 1, 0);
1278 if(sion_gendata->grank==0) {
1279 for(t=0;t<sion_gendata->gsize;t++) {
1280 if(receivemap[t]>=0) {
1281 receiver=receivemap[t];
1285 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"receiver of mapping grank=%d\n", receiver));
1287 sion_apidesc->bcastr_cb(&receiver, sion_gendata->comm_data_global, _SION_INT32, 1, 0);
1290 lpos[0] = buddyptr->buddy_send.filenum;
1291 lpos[1] = buddyptr->buddy_send.rank;
1292 sion_apidesc->gatherr_cb(&lpos, *mapping, sion_gendata->comm_data_global, _SION_INT32, 2, receiver);
1294 if(receivemap!=NULL) free(receivemap);
1301 #define DFUNCTION "_sion_generic_buddy_get_and_distribute_info_from_file" 1302 int _sion_generic_buddy_get_and_distribute_info_from_file(
_sion_generic_gendata *sion_gendata,
char *fname,
int root,
1303 sion_int32 *filenumber, sion_int32 *numfiles,
1304 sion_int32 *lrank, sion_int32 *lsize) {
1306 int sid = -1, ntasks, nfiles, t;
1307 int rc = SION_SUCCESS;
1309 sion_int32 fsblksize;
1310 int *tasksinfile = NULL;
1311 int mapping_size = -1;
1312 sion_int32 *mapping = NULL;
1316 sion_apidesc=sion_gendata->apidesc;
1318 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"entering function sion_gendata->grank=%d\n",sion_gendata->grank));
1320 if(sion_gendata->grank == root) {
1322 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"before open\n"));
1323 sid=
_sion_open_read(fname,_SION_FMODE_READ|_SION_FMODE_ANSI,_SION_READ_MASTER_ONLY_OF_MULTI_FILES,
1324 &ntasks,&nfiles,NULL,&fsblksize,NULL,&fileptr);
1326 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"after open\n"));
1328 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"sion file %d files\n", *numfiles));
1336 sion_apidesc->bcastr_cb(numfiles, sion_gendata->comm_data_global, _SION_INT32, 1, root);
1337 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"%s: numfiles=%d\n",fname,*numfiles));
1339 if((sion_gendata->grank == root) && (*numfiles>1)) {
1340 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"%s: mapping_size=%d sion_gendata->gsize=%d\n",fname,*numfiles,mapping_size,sion_gendata->gsize));
1341 if(mapping_size!=sion_gendata->gsize) {
1342 return(_sion_errorprint(SION_NOT_SUCCESS,_SION_ERROR_RETURN,
1343 "_sion_generic_buddy_get_and_distribute_info_from_file: Incorrect sum of ntasks of files %d <> %d\n",
1344 mapping_size, sion_gendata->gsize));
1349 return(_sion_errorprint(SION_NOT_SUCCESS,_SION_ERROR_RETURN,
1350 "_sion_generic_buddy_get_and_distribute_info_from_file: could not get numfiles from sion file\n"));
1354 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"before scatter\n"));
1355 if(sion_gendata->grank==root) {
1356 for(t=0;t<mapping_size;t++) {
1357 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
" %d -> (%d,%d)\n",t,mapping[t*2],mapping[t*2+1]));
1362 sion_apidesc->scatterr_cb(mapping, lpos, sion_gendata->comm_data_global, _SION_INT32, 2, root);
1363 *filenumber=lpos[0];
1365 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"after scatter filenum+lrank (%d,%d)\n",*filenumber,*lrank));
1368 if(sion_gendata->grank==root) {
1369 tasksinfile = (
int *) malloc(*numfiles *
sizeof(
int));
1370 if (tasksinfile == NULL) {
1371 return(_sion_errorprint(SION_ID_NOT_VALID,_SION_ERROR_RETURN,
"_sion_generic_get_and_distribute_info_from_file: Cannot allocate memory for tasksinfile counter vector"));
1373 for(t=0;t<*numfiles;t++) tasksinfile[t]=0;
1374 for(t=0;t<mapping_size;t++) tasksinfile[ mapping[t*2] ]++;
1375 for(t=0;t<mapping_size;t++) mapping[t*2+1]=tasksinfile[ mapping[t*2] ];
1377 sion_apidesc->scatterr_cb(mapping, lpos, sion_gendata->comm_data_global, _SION_INT32, 2, root);
1379 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"after scatter lsize (%d, %d of %d)\n",*filenumber, *lrank, *lsize));
1381 if(sion_gendata->grank==root) {
1382 if(tasksinfile) free(tasksinfile);
1388 *lrank = sion_gendata->grank;
1389 *lsize = sion_gendata->gsize;
1390 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"only one file -> filenumber=%d lRank=%d\n",*filenumber,*lrank));
1393 if(sion_gendata->grank == root) {
1395 if (sid>=0) _sion_close_sid(sid);
1403 #define DFUNCTION "_sion_generic_buddy_get_and_distribute_info_from_one_file" 1404 int _sion_generic_buddy_get_and_distribute_info_from_one_file(
_sion_generic_gendata *sion_gendata,
char *fname,
int root,
1405 sion_int32 *filenumber, sion_int32 *numfiles,
1406 sion_int32 *lrank, sion_int32 *lsize) {
1408 int rc = SION_SUCCESS;
1409 int t, mapping_size = 0, grank, *mapping = NULL;
1410 sion_int32 file_filenumber,file_numfiles,file_lrank,file_lsize;
1415 sion_apidesc=sion_gendata->apidesc;
1417 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"entering function sion_gendata->grank=%d\n",sion_gendata->grank));
1419 if(sion_gendata->grank == root) {
1421 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"before open\n"));
1424 if (sion_filedesc == NULL) {
1425 return(_sion_errorprint(SION_ID_UNDEF,SION_ID_UNDEF,
1426 "sion_open: cannot allocate filedescriptor structure of size %lu (sion_filedesc), aborting ...\n",
1427 (
unsigned long)
sizeof(sion_filedesc)));
1433 if (!sion_fileptr) {
1434 return(_sion_errorprint(SION_ID_UNDEF,_SION_ERROR_RETURN,
"sion_open: cannot open %s for reading, aborting ...\n", fname));
1436 sion_filedesc->
fileptr = sion_fileptr;
1440 if (rc!=SION_SUCCESS) {
1441 return(_sion_errorprint(SION_ID_UNDEF,_SION_ERROR_RETURN,
"sion_open: cannot read header from file %s, aborting ...\n", fname));
1443 sion_filedesc->
rank = 0;
1452 if (rc!=SION_SUCCESS) {
1453 return(_sion_errorprint(SION_ID_UNDEF,_SION_ERROR_RETURN,
"sion_open: cannot read header (var part) from file %s, aborting ...\n", fname));
1456 file_numfiles=sion_filedesc->
nfiles;
1458 file_lsize=sion_filedesc->
ntasks;
1461 mapping_size=sion_gendata->gsize;
1462 mapping = (sion_int32 *) malloc(mapping_size * 2 *
sizeof(sion_int32));
1463 if (mapping == NULL) {
1464 return(_sion_errorprint(SION_ID_NOT_VALID,_SION_ERROR_RETURN,
"_sion_generic_buddy_get_and_distribute_info_from_one_file: cannot allocate temporary memory of size %lu (mapping), aborting ...\n",
1465 (
unsigned long) mapping_size * 2 *
sizeof(sion_int32)));
1472 sion_apidesc->bcastr_cb(&file_numfiles, sion_gendata->comm_data_global, _SION_INT32, 1, root);
1473 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"%s: numfiles=%d\n",fname,file_numfiles));
1474 sion_apidesc->bcastr_cb(&file_filenumber, sion_gendata->comm_data_global, _SION_INT32, 1, root);
1475 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"%s: filenumber=%d\n",fname,file_filenumber));
1476 sion_apidesc->bcastr_cb(&file_lsize, sion_gendata->comm_data_global, _SION_INT32, 1, root);
1477 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"%s: file_lsize=%d\n",fname,file_lsize));
1480 if(file_numfiles!=*numfiles) {
1481 _SION_SAFE_FREE(mapping, NULL);
1482 return(_sion_errorprint(SION_NOT_SUCCESS,_SION_ERROR_RETURN,
1483 "_sion_generic_buddy_get_and_distribute_info_from_one_file: Incorrect number of files %d <> %d\n",
1484 file_numfiles,*numfiles));
1488 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"before scatter\n"));
1489 if(sion_gendata->grank==root) {
1490 for(t=0;t<mapping_size;t++) mapping[t]=-1;
1491 for(t=0;t<file_lsize;t++) {
1495 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
" file=%s mapping[%d] -> (%d)\n",fname,grank,mapping[grank]));
1501 sion_apidesc->scatterr_cb(mapping, &file_lrank, sion_gendata->comm_data_global, _SION_INT32, 1, root);
1502 if(file_lrank!=-1) {
1503 *filenumber=file_filenumber;
1506 DPRINTFP((1, DFUNCTION, sion_gendata->grank,
"after scatter filenum=%d lrank=%d lsize=%d\n",*filenumber,*lrank,*lsize));
1509 if(sion_gendata->grank == root) {
int sion_feof(int sid)
Function that indicates whether the end of file is reached for this task.
sion_int64 _sion_file_set_position(_sion_fileptr *sion_fileptr, sion_int64 startpointer)
Set new position in file.
_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_ensure_free_space(int sid, sion_int64 bytes)
Funtion to ensure that enough space is available for writing.
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_FILESTATE_SEROPEN
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_vcdtype(int sid)
sion_int32 fileptr_exported
#define SION_FILE_FLAG_POSIX
int sion_get_mapping(int sid, int *mapping_size, sion_int32 **mapping, int *numfiles)
Returns pointers to the internal field mapping.
void * _sion_vcdtovcon(int sid)
#define SION_FILEMODE_READ
int _sion_file_close(_sion_fileptr *sion_fileptr)
Close file and destroys fileptr structure.
int _sion_newvcd(void *data, int type)
char * _sion_get_multi_filename(const char *fname, int filenumber)
generates the multi filename
#define SION_FILEDESCRIPTOR
_sion_filedesc * _sion_alloc_filedesc(void)
Allocates memory for internal sion structure.
int _sion_init_filedesc(_sion_filedesc *sion_filedesc)
Initialize the sion file description.
#define SION_FILEMODE_WRITE
int _sion_file_stat_file(const char *fname)
Check if file exists (LARGE_FILE support on BlueGene)
int _sion_file_flush(_sion_fileptr *sion_fileptr)
Flush data to file.
int _sion_open_read(const char *fname, sion_int64 file_mode_flags, int read_all, int *ntasks, int *nfiles, sion_int64 **chunksizes, sion_int32 *fsblksize, int **globalranks, FILE **fileptr)
internal sion serial open function for reading on one or more files