distributionMapBaseTemplates.C
Go to the documentation of this file.
1 /*---------------------------------------------------------------------------*\
2  ========= |
3  \\ / F ield | OpenFOAM: The Open Source CFD Toolbox
4  \\ / O peration | Website: https://openfoam.org
5  \\ / A nd | Copyright (C) 2015-2022 OpenFOAM Foundation
6  \\/ M anipulation |
7 -------------------------------------------------------------------------------
8 License
9  This file is part of OpenFOAM.
10 
11  OpenFOAM is free software: you can redistribute it and/or modify it
12  under the terms of the GNU General Public License as published by
13  the Free Software Foundation, either version 3 of the License, or
14  (at your option) any later version.
15 
16  OpenFOAM is distributed in the hope that it will be useful, but WITHOUT
17  ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
18  FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
19  for more details.
20 
21  You should have received a copy of the GNU General Public License
22  along with OpenFOAM. If not, see <http://www.gnu.org/licenses/>.
23 
24 \*---------------------------------------------------------------------------*/
25 
26 #include "Pstream.H"
27 #include "PstreamBuffers.H"
28 #include "PstreamCombineReduceOps.H"
29 #include "flipOp.H"
30 
31 // * * * * * * * * * * * * * Private Member Functions * * * * * * * * * * * //
32 
33 template<class T, class CombineOp, class negateOp>
34 void Foam::distributionMapBase::flipAndCombine
35 (
36  const UList<label>& map,
37  const bool hasFlip,
38  const UList<T>& rhs,
39  const CombineOp& cop,
40  const negateOp& negOp,
41  List<T>& lhs
42 )
43 {
44  if (hasFlip)
45  {
46  forAll(map, i)
47  {
48  if (map[i] > 0)
49  {
50  label index = map[i]-1;
51  cop(lhs[index], rhs[i]);
52  }
53  else if (map[i] < 0)
54  {
55  label index = -map[i]-1;
56  cop(lhs[index], negOp(rhs[i]));
57  }
58  else
59  {
60  FatalErrorInFunction
61  << "At index " << i << " out of " << map.size()
62  << " have illegal index " << map[i]
63  << " for field " << rhs.size() << " with flipMap"
64  << exit(FatalError);
65  }
66  }
67  }
68  else
69  {
70  forAll(map, i)
71  {
72  cop(lhs[map[i]], rhs[i]);
73  }
74  }
75 }
76 
77 
78 template<class T, class negateOp>
79 T Foam::distributionMapBase::accessAndFlip
80 (
81  const UList<T>& fld,
82  const label index,
83  const bool hasFlip,
84  const negateOp& negOp
85 )
86 {
87  T t;
88  if (hasFlip)
89  {
90  if (index > 0)
91  {
92  t = fld[index-1];
93  }
94  else if (index < 0)
95  {
96  t = negOp(fld[-index-1]);
97  }
98  else
99  {
100  FatalErrorInFunction
101  << "Illegal index " << index
102  << " into field of size " << fld.size()
103  << " with face-flipping"
104  << exit(FatalError);
105  t = fld[index];
106  }
107  }
108  else
109  {
110  t = fld[index];
111  }
112  return t;
113 }
114 
115 
116 // Distribute list.
117 template<class T, class negateOp>
118 void Foam::distributionMapBase::distribute
119 (
120  const Pstream::commsTypes commsType,
121  const List<labelPair>& schedule,
122  const label constructSize,
123  const labelListList& subMap,
124  const bool subHasFlip,
125  const labelListList& constructMap,
126  const bool constructHasFlip,
127  List<T>& field,
128  const negateOp& negOp,
129  const int tag
130 )
131 {
132  if (!Pstream::parRun())
133  {
134  // Do only me to me.
135 
136  const labelList& mySubMap = subMap[Pstream::myProcNo()];
137 
138  List<T> subField(mySubMap.size());
139  forAll(mySubMap, i)
140  {
141  subField[i] = accessAndFlip(field, mySubMap[i], subHasFlip, negOp);
142  }
143 
144  // Receive sub field from myself (subField)
145  const labelList& map = constructMap[Pstream::myProcNo()];
146 
147  field.setSize(constructSize);
148 
149  flipAndCombine
150  (
151  map,
152  constructHasFlip,
153  subField,
154  eqOp<T>(),
155  negOp,
156  field
157  );
158 
159  return;
160  }
161 
162  if (commsType == Pstream::commsTypes::blocking)
163  {
164  // Since buffered sending can reuse the field to collect the
165  // received data.
166 
167  // Send sub field to neighbour
168  for (label domain = 0; domain < Pstream::nProcs(); domain++)
169  {
170  const labelList& map = subMap[domain];
171 
172  if (domain != Pstream::myProcNo() && map.size())
173  {
174  OPstream toNbr(Pstream::commsTypes::blocking, domain, 0, tag);
175 
176  List<T> subField(map.size());
177  forAll(subField, i)
178  {
179  subField[i] = accessAndFlip
180  (
181  field,
182  map[i],
183  subHasFlip,
184  negOp
185  );
186  }
187  toNbr << subField;
188  }
189  }
190 
191  // Subset myself
192  const labelList& mySubMap = subMap[Pstream::myProcNo()];
193 
194  List<T> subField(mySubMap.size());
195  forAll(mySubMap, i)
196  {
197  subField[i] = accessAndFlip(field, mySubMap[i], subHasFlip, negOp);
198  }
199 
200  // Receive sub field from myself (subField)
201  const labelList& map = constructMap[Pstream::myProcNo()];
202 
203  field.setSize(constructSize);
204 
205  flipAndCombine
206  (
207  map,
208  constructHasFlip,
209  subField,
210  eqOp<T>(),
211  negOp,
212  field
213  );
214 
215  // Receive sub field from neighbour
216  for (label domain = 0; domain < Pstream::nProcs(); domain++)
217  {
218  const labelList& map = constructMap[domain];
219 
220  if (domain != Pstream::myProcNo() && map.size())
221  {
222  IPstream fromNbr(Pstream::commsTypes::blocking, domain, 0, tag);
223  List<T> subField(fromNbr);
224 
225  checkReceivedSize(domain, map.size(), subField.size());
226 
227  flipAndCombine
228  (
229  map,
230  constructHasFlip,
231  subField,
232  eqOp<T>(),
233  negOp,
234  field
235  );
236  }
237  }
238  }
239  else if (commsType == Pstream::commsTypes::scheduled)
240  {
241  // Need to make sure I don't overwrite field with received data
242  // since the data might need to be sent to another processor. So
243  // allocate a new field for the results.
244  List<T> newField(constructSize);
245 
246  // Receive sub field from myself
247  {
248  const labelList& mySubMap = subMap[Pstream::myProcNo()];
249 
250  List<T> subField(mySubMap.size());
251  forAll(subField, i)
252  {
253  subField[i] = accessAndFlip
254  (
255  field,
256  mySubMap[i],
257  subHasFlip,
258  negOp
259  );
260  }
261 
262  // Receive sub field from myself (subField)
263  flipAndCombine
264  (
265  constructMap[Pstream::myProcNo()],
266  constructHasFlip,
267  subField,
268  eqOp<T>(),
269  negOp,
270  newField
271  );
272  }
273 
274  // Schedule will already have pruned 0-sized comms
275  forAll(schedule, i)
276  {
277  const labelPair& twoProcs = schedule[i];
278  // twoProcs is a swap pair of processors. The first one is the
279  // one that needs to send first and then receive.
280 
281  label sendProc = twoProcs[0];
282  label recvProc = twoProcs[1];
283 
284  if (Pstream::myProcNo() == sendProc)
285  {
286  // I am send first, receive next
287  {
288  OPstream toNbr
289  (
290  Pstream::commsTypes::scheduled,
291  recvProc,
292  0,
293  tag
294  );
295 
296  const labelList& map = subMap[recvProc];
297  List<T> subField(map.size());
298  forAll(subField, i)
299  {
300  subField[i] = accessAndFlip
301  (
302  field,
303  map[i],
304  subHasFlip,
305  negOp
306  );
307  }
308  toNbr << subField;
309  }
310  {
311  IPstream fromNbr
312  (
313  Pstream::commsTypes::scheduled,
314  recvProc,
315  0,
316  tag
317  );
318  List<T> subField(fromNbr);
319 
320  const labelList& map = constructMap[recvProc];
321 
322  checkReceivedSize(recvProc, map.size(), subField.size());
323 
324  flipAndCombine
325  (
326  map,
327  constructHasFlip,
328  subField,
329  eqOp<T>(),
330  negOp,
331  newField
332  );
333  }
334  }
335  else
336  {
337  // I am receive first, send next
338  {
339  IPstream fromNbr
340  (
341  Pstream::commsTypes::scheduled,
342  sendProc,
343  0,
344  tag
345  );
346  List<T> subField(fromNbr);
347 
348  const labelList& map = constructMap[sendProc];
349 
350  checkReceivedSize(sendProc, map.size(), subField.size());
351 
352  flipAndCombine
353  (
354  map,
355  constructHasFlip,
356  subField,
357  eqOp<T>(),
358  negOp,
359  newField
360  );
361  }
362  {
363  OPstream toNbr
364  (
365  Pstream::commsTypes::scheduled,
366  sendProc,
367  0,
368  tag
369  );
370 
371  const labelList& map = subMap[sendProc];
372  List<T> subField(map.size());
373  forAll(subField, i)
374  {
375  subField[i] = accessAndFlip
376  (
377  field,
378  map[i],
379  subHasFlip,
380  negOp
381  );
382  }
383  toNbr << subField;
384  }
385  }
386  }
387  field.transfer(newField);
388  }
389  else if (commsType == Pstream::commsTypes::nonBlocking)
390  {
391  label nOutstanding = Pstream::nRequests();
392 
393  if (!contiguous<T>())
394  {
395  PstreamBuffers pBufs(Pstream::commsTypes::nonBlocking, tag);
396 
397  // Stream data into buffer
398  for (label domain = 0; domain < Pstream::nProcs(); domain++)
399  {
400  const labelList& map = subMap[domain];
401 
402  if (domain != Pstream::myProcNo() && map.size())
403  {
404  // Put data into send buffer
405  UOPstream toDomain(domain, pBufs);
406 
407  List<T> subField(map.size());
408  forAll(subField, i)
409  {
410  subField[i] = accessAndFlip
411  (
412  field,
413  map[i],
414  subHasFlip,
415  negOp
416  );
417  }
418  toDomain << subField;
419  }
420  }
421 
422  // Start receiving. Do not block.
423  pBufs.finishedSends(false);
424 
425  {
426  // Set up 'send' to myself
427  const labelList& mySub = subMap[Pstream::myProcNo()];
428  List<T> mySubField(mySub.size());
429  forAll(mySub, i)
430  {
431  mySubField[i] = accessAndFlip
432  (
433  field,
434  mySub[i],
435  subHasFlip,
436  negOp
437  );
438  }
439  // Combine bits. Note that can reuse field storage
440  field.setSize(constructSize);
441  // Receive sub field from myself
442  {
443  const labelList& map = constructMap[Pstream::myProcNo()];
444 
445  flipAndCombine
446  (
447  map,
448  constructHasFlip,
449  mySubField,
450  eqOp<T>(),
451  negOp,
452  field
453  );
454  }
455  }
456 
457  // Block ourselves, waiting only for the current comms
458  Pstream::waitRequests(nOutstanding);
459 
460  // Consume
461  for (label domain = 0; domain < Pstream::nProcs(); domain++)
462  {
463  const labelList& map = constructMap[domain];
464 
465  if (domain != Pstream::myProcNo() && map.size())
466  {
467  UIPstream str(domain, pBufs);
468  List<T> recvField(str);
469 
470  checkReceivedSize(domain, map.size(), recvField.size());
471 
472  flipAndCombine
473  (
474  map,
475  constructHasFlip,
476  recvField,
477  eqOp<T>(),
478  negOp,
479  field
480  );
481  }
482  }
483  }
484  else
485  {
486  // Set up sends to neighbours
487 
488  List<List<T>> sendFields(Pstream::nProcs());
489 
490  for (label domain = 0; domain < Pstream::nProcs(); domain++)
491  {
492  const labelList& map = subMap[domain];
493 
494  if (domain != Pstream::myProcNo() && map.size())
495  {
496  List<T>& subField = sendFields[domain];
497  subField.setSize(map.size());
498  forAll(map, i)
499  {
500  subField[i] = accessAndFlip
501  (
502  field,
503  map[i],
504  subHasFlip,
505  negOp
506  );
507  }
508 
509  OPstream::write
510  (
511  Pstream::commsTypes::nonBlocking,
512  domain,
513  reinterpret_cast<const char*>(subField.begin()),
514  subField.byteSize(),
515  tag
516  );
517  }
518  }
519 
520  // Set up receives from neighbours
521 
522  List<List<T>> recvFields(Pstream::nProcs());
523 
524  for (label domain = 0; domain < Pstream::nProcs(); domain++)
525  {
526  const labelList& map = constructMap[domain];
527 
528  if (domain != Pstream::myProcNo() && map.size())
529  {
530  recvFields[domain].setSize(map.size());
531  IPstream::read
532  (
533  Pstream::commsTypes::nonBlocking,
534  domain,
535  reinterpret_cast<char*>(recvFields[domain].begin()),
536  recvFields[domain].byteSize(),
537  tag
538  );
539  }
540  }
541 
542 
543  // Set up 'send' to myself
544 
545  {
546  const labelList& map = subMap[Pstream::myProcNo()];
547 
548  List<T>& subField = sendFields[Pstream::myProcNo()];
549  subField.setSize(map.size());
550  forAll(map, i)
551  {
552  subField[i] = accessAndFlip
553  (
554  field,
555  map[i],
556  subHasFlip,
557  negOp
558  );
559  }
560  }
561 
562 
563  // Combine bits. Note that can reuse field storage
564 
565  field.setSize(constructSize);
566 
567 
568  // Receive sub field from myself (sendFields[Pstream::myProcNo()])
569  {
570  const labelList& map = constructMap[Pstream::myProcNo()];
571  const List<T>& subField = sendFields[Pstream::myProcNo()];
572 
573  flipAndCombine
574  (
575  map,
576  constructHasFlip,
577  subField,
578  eqOp<T>(),
579  negOp,
580  field
581  );
582  }
583 
584 
585  // Wait for all to finish
586 
587  Pstream::waitRequests(nOutstanding);
588 
589 
590  // Collect neighbour fields
591 
592  for (label domain = 0; domain < Pstream::nProcs(); domain++)
593  {
594  const labelList& map = constructMap[domain];
595 
596  if (domain != Pstream::myProcNo() && map.size())
597  {
598  const List<T>& subField = recvFields[domain];
599 
600  checkReceivedSize(domain, map.size(), subField.size());
601 
602  flipAndCombine
603  (
604  map,
605  constructHasFlip,
606  subField,
607  eqOp<T>(),
608  negOp,
609  field
610  );
611  }
612  }
613  }
614  }
615  else
616  {
617  FatalErrorInFunction
618  << "Unknown communication schedule " << int(commsType)
619  << abort(FatalError);
620  }
621 }
622 
623 
624 // Distribute list.
625 template<class T, class CombineOp, class negateOp>
626 void Foam::distributionMapBase::distribute
627 (
628  const Pstream::commsTypes commsType,
629  const List<labelPair>& schedule,
630  const label constructSize,
631  const labelListList& subMap,
632  const bool subHasFlip,
633  const labelListList& constructMap,
634  const bool constructHasFlip,
635  List<T>& field,
636  const CombineOp& cop,
637  const negateOp& negOp,
638  const T& nullValue,
639  const int tag
640 )
641 {
642  if (!Pstream::parRun())
643  {
644  // Do only me to me.
645 
646  const labelList& mySubMap = subMap[Pstream::myProcNo()];
647 
648  List<T> subField(mySubMap.size());
649  forAll(mySubMap, i)
650  {
651  subField[i] = accessAndFlip(field, mySubMap[i], subHasFlip, negOp);
652  }
653 
654  // Receive sub field from myself (subField)
655  const labelList& map = constructMap[Pstream::myProcNo()];
656 
657  field.setSize(constructSize);
658  field = nullValue;
659 
660  flipAndCombine(map, constructHasFlip, subField, cop, negOp, field);
661 
662  return;
663  }
664 
665  if (commsType == Pstream::commsTypes::blocking)
666  {
667  // Since buffered sending can reuse the field to collect the
668  // received data.
669 
670  // Send sub field to neighbour
671  for (label domain = 0; domain < Pstream::nProcs(); domain++)
672  {
673  const labelList& map = subMap[domain];
674 
675  if (domain != Pstream::myProcNo() && map.size())
676  {
677  OPstream toNbr(Pstream::commsTypes::blocking, domain, 0, tag);
678  List<T> subField(map.size());
679  forAll(subField, i)
680  {
681  subField[i] = accessAndFlip
682  (
683  field,
684  map[i],
685  subHasFlip,
686  negOp
687  );
688  }
689  toNbr << subField;
690  }
691  }
692 
693  // Subset myself
694  const labelList& mySubMap = subMap[Pstream::myProcNo()];
695 
696  List<T> subField(mySubMap.size());
697  forAll(mySubMap, i)
698  {
699  subField[i] = accessAndFlip(field, mySubMap[i], subHasFlip, negOp);
700  }
701 
702  // Receive sub field from myself (subField)
703  const labelList& map = constructMap[Pstream::myProcNo()];
704 
705  field.setSize(constructSize);
706  field = nullValue;
707 
708  flipAndCombine(map, constructHasFlip, subField, cop, negOp, field);
709 
710  // Receive sub field from neighbour
711  for (label domain = 0; domain < Pstream::nProcs(); domain++)
712  {
713  const labelList& map = constructMap[domain];
714 
715  if (domain != Pstream::myProcNo() && map.size())
716  {
717  IPstream fromNbr(Pstream::commsTypes::blocking, domain, 0, tag);
718  List<T> subField(fromNbr);
719 
720  checkReceivedSize(domain, map.size(), subField.size());
721 
722  flipAndCombine
723  (
724  map,
725  constructHasFlip,
726  subField,
727  cop,
728  negOp,
729  field
730  );
731  }
732  }
733  }
734  else if (commsType == Pstream::commsTypes::scheduled)
735  {
736  // Need to make sure I don't overwrite field with received data
737  // since the data might need to be sent to another processor. So
738  // allocate a new field for the results.
739  List<T> newField(constructSize, nullValue);
740 
741  {
742  const labelList& mySubMap = subMap[Pstream::myProcNo()];
743 
744  // Subset myself
745  List<T> subField(mySubMap.size());
746  forAll(subField, i)
747  {
748  subField[i] = accessAndFlip
749  (
750  field,
751  mySubMap[i],
752  subHasFlip,
753  negOp
754  );
755  }
756 
757  // Receive sub field from myself (subField)
758  const labelList& map = constructMap[Pstream::myProcNo()];
759 
760  flipAndCombine
761  (
762  map,
763  constructHasFlip,
764  subField,
765  cop,
766  negOp,
767  newField
768  );
769  }
770 
771 
772  // Schedule will already have pruned 0-sized comms
773  forAll(schedule, i)
774  {
775  const labelPair& twoProcs = schedule[i];
776  // twoProcs is a swap pair of processors. The first one is the
777  // one that needs to send first and then receive.
778 
779  label sendProc = twoProcs[0];
780  label recvProc = twoProcs[1];
781 
782  if (Pstream::myProcNo() == sendProc)
783  {
784  // I am send first, receive next
785  {
786  OPstream toNbr
787  (
788  Pstream::commsTypes::scheduled,
789  recvProc,
790  0,
791  tag
792  );
793 
794  const labelList& map = subMap[recvProc];
795 
796  List<T> subField(map.size());
797  forAll(subField, i)
798  {
799  subField[i] = accessAndFlip
800  (
801  field,
802  map[i],
803  subHasFlip,
804  negOp
805  );
806  }
807  toNbr << subField;
808  }
809  {
810  IPstream fromNbr
811  (
812  Pstream::commsTypes::scheduled,
813  recvProc,
814  0,
815  tag
816  );
817  List<T> subField(fromNbr);
818  const labelList& map = constructMap[recvProc];
819 
820  checkReceivedSize(recvProc, map.size(), subField.size());
821 
822  flipAndCombine
823  (
824  map,
825  constructHasFlip,
826  subField,
827  cop,
828  negOp,
829  newField
830  );
831  }
832  }
833  else
834  {
835  // I am receive first, send next
836  {
837  IPstream fromNbr
838  (
839  Pstream::commsTypes::scheduled,
840  sendProc,
841  0,
842  tag
843  );
844  List<T> subField(fromNbr);
845  const labelList& map = constructMap[sendProc];
846 
847  checkReceivedSize(sendProc, map.size(), subField.size());
848 
849  flipAndCombine
850  (
851  map,
852  constructHasFlip,
853  subField,
854  cop,
855  negOp,
856  newField
857  );
858  }
859  {
860  OPstream toNbr
861  (
862  Pstream::commsTypes::scheduled,
863  sendProc,
864  0,
865  tag
866  );
867 
868  const labelList& map = subMap[sendProc];
869 
870  List<T> subField(map.size());
871  forAll(subField, i)
872  {
873  subField[i] = accessAndFlip
874  (
875  field,
876  map[i],
877  subHasFlip,
878  negOp
879  );
880  }
881  toNbr << subField;
882  }
883  }
884  }
885  field.transfer(newField);
886  }
887  else if (commsType == Pstream::commsTypes::nonBlocking)
888  {
889  label nOutstanding = Pstream::nRequests();
890 
891  if (!contiguous<T>())
892  {
893  PstreamBuffers pBufs(Pstream::commsTypes::nonBlocking, tag);
894 
895  // Stream data into buffer
896  for (label domain = 0; domain < Pstream::nProcs(); domain++)
897  {
898  const labelList& map = subMap[domain];
899 
900  if (domain != Pstream::myProcNo() && map.size())
901  {
902  // Put data into send buffer
903  UOPstream toDomain(domain, pBufs);
904 
905  List<T> subField(map.size());
906  forAll(subField, i)
907  {
908  subField[i] = accessAndFlip
909  (
910  field,
911  map[i],
912  subHasFlip,
913  negOp
914  );
915  }
916  toDomain << subField;
917  }
918  }
919 
920  // Start receiving. Do not block.
921  pBufs.finishedSends(false);
922 
923  {
924  // Set up 'send' to myself
925  const labelList& myMap = subMap[Pstream::myProcNo()];
926 
927  List<T> mySubField(myMap.size());
928  forAll(myMap, i)
929  {
930  mySubField[i] = accessAndFlip
931  (
932  field,
933  myMap[i],
934  subHasFlip,
935  negOp
936  );
937  }
938 
939  // Combine bits. Note that can reuse field storage
940  field.setSize(constructSize);
941  field = nullValue;
942  // Receive sub field from myself
943  {
944  const labelList& map = constructMap[Pstream::myProcNo()];
945 
946  flipAndCombine
947  (
948  map,
949  constructHasFlip,
950  mySubField,
951  cop,
952  negOp,
953  field
954  );
955  }
956  }
957 
958  // Block ourselves, waiting only for the current comms
959  Pstream::waitRequests(nOutstanding);
960 
961  // Consume
962  for (label domain = 0; domain < Pstream::nProcs(); domain++)
963  {
964  const labelList& map = constructMap[domain];
965 
966  if (domain != Pstream::myProcNo() && map.size())
967  {
968  UIPstream str(domain, pBufs);
969  List<T> recvField(str);
970 
971  checkReceivedSize(domain, map.size(), recvField.size());
972 
973  flipAndCombine
974  (
975  map,
976  constructHasFlip,
977  recvField,
978  cop,
979  negOp,
980  field
981  );
982  }
983  }
984  }
985  else
986  {
987  // Set up sends to neighbours
988 
989  List<List<T>> sendFields(Pstream::nProcs());
990 
991  for (label domain = 0; domain < Pstream::nProcs(); domain++)
992  {
993  const labelList& map = subMap[domain];
994 
995  if (domain != Pstream::myProcNo() && map.size())
996  {
997  List<T>& subField = sendFields[domain];
998  subField.setSize(map.size());
999  forAll(map, i)
1000  {
1001  subField[i] = accessAndFlip
1002  (
1003  field,
1004  map[i],
1005  subHasFlip,
1006  negOp
1007  );
1008  }
1009 
1010  OPstream::write
1011  (
1012  Pstream::commsTypes::nonBlocking,
1013  domain,
1014  reinterpret_cast<const char*>(subField.begin()),
1015  subField.size()*sizeof(T),
1016  tag
1017  );
1018  }
1019  }
1020 
1021  // Set up receives from neighbours
1022 
1023  List<List<T>> recvFields(Pstream::nProcs());
1024 
1025  for (label domain = 0; domain < Pstream::nProcs(); domain++)
1026  {
1027  const labelList& map = constructMap[domain];
1028 
1029  if (domain != Pstream::myProcNo() && map.size())
1030  {
1031  recvFields[domain].setSize(map.size());
1032  UIPstream::read
1033  (
1034  Pstream::commsTypes::nonBlocking,
1035  domain,
1036  reinterpret_cast<char*>(recvFields[domain].begin()),
1037  recvFields[domain].size()*sizeof(T),
1038  tag
1039  );
1040  }
1041  }
1042 
1043  // Set up 'send' to myself
1044 
1045  {
1046  const labelList& map = subMap[Pstream::myProcNo()];
1047 
1048  List<T>& subField = sendFields[Pstream::myProcNo()];
1049  subField.setSize(map.size());
1050  forAll(map, i)
1051  {
1052  subField[i] = accessAndFlip
1053  (
1054  field,
1055  map[i],
1056  subHasFlip,
1057  negOp
1058  );
1059  }
1060  }
1061 
1062 
1063  // Combine bits. Note that can reuse field storage
1064 
1065  field.setSize(constructSize);
1066  field = nullValue;
1067 
1068  // Receive sub field from myself (subField)
1069  {
1070  const labelList& map = constructMap[Pstream::myProcNo()];
1071  const List<T>& subField = sendFields[Pstream::myProcNo()];
1072 
1073  flipAndCombine
1074  (
1075  map,
1076  constructHasFlip,
1077  subField,
1078  cop,
1079  negOp,
1080  field
1081  );
1082  }
1083 
1084 
1085  // Wait for all to finish
1086 
1087  Pstream::waitRequests(nOutstanding);
1088 
1089 
1090  // Collect neighbour fields
1091 
1092  for (label domain = 0; domain < Pstream::nProcs(); domain++)
1093  {
1094  const labelList& map = constructMap[domain];
1095 
1096  if (domain != Pstream::myProcNo() && map.size())
1097  {
1098  const List<T>& subField = recvFields[domain];
1099 
1100  checkReceivedSize(domain, map.size(), subField.size());
1101 
1102  flipAndCombine
1103  (
1104  map,
1105  constructHasFlip,
1106  subField,
1107  cop,
1108  negOp,
1109  field
1110  );
1111  }
1112  }
1113  }
1114  }
1115  else
1116  {
1117  FatalErrorInFunction
1118  << "Unknown communication schedule " << int(commsType)
1119  << abort(FatalError);
1120  }
1121 }
1122 
1123 
1124 template<class T>
1125 void Foam::distributionMapBase::send
1126 (
1127  PstreamBuffers& pBufs,
1128  const List<T>& field
1129 )
1130 const
1131 {
1132  // Stream data into buffer
1133  for (label domain = 0; domain < Pstream::nProcs(); domain++)
1134  {
1135  const labelList& map = subMap_[domain];
1136 
1137  if (map.size())
1138  {
1139  // Put data into send buffer
1140  UOPstream toDomain(domain, pBufs);
1141 
1142  List<T> subField(map.size());
1143  forAll(subField, i)
1144  {
1145  subField[i] = accessAndFlip
1146  (
1147  field,
1148  map[i],
1149  subHasFlip_,
1150  flipOp()
1151  );
1152  }
1153  toDomain << subField;
1154  }
1155  }
1156 
1157  // Start sending and receiving but do not block.
1158  pBufs.finishedSends(false);
1159 }
1160 
1161 
1162 template<class T>
1163 void Foam::distributionMapBase::receive(PstreamBuffers& pBufs, List<T>& field)
1164 const
1165 {
1166  // Consume
1167  field.setSize(constructSize_);
1168 
1169  for (label domain = 0; domain < Pstream::nProcs(); domain++)
1170  {
1171  const labelList& map = constructMap_[domain];
1172 
1173  if (map.size())
1174  {
1175  UIPstream str(domain, pBufs);
1176  List<T> recvField(str);
1177 
1178  if (recvField.size() != map.size())
1179  {
1180  FatalErrorInFunction
1181  << "Expected from processor " << domain
1182  << " " << map.size() << " but received "
1183  << recvField.size() << " elements."
1184  << abort(FatalError);
1185  }
1186 
1187  flipAndCombine
1188  (
1189  map,
1190  constructHasFlip_,
1191  recvField,
1192  eqOp<T>(),
1193  flipOp(),
1194  field
1195  );
1196  }
1197  }
1198 }
1199 
1200 
1201 //- Distribute data using default commsType.
1202 template<class T, class negateOp>
1203 void Foam::distributionMapBase::distribute
1204 (
1205  List<T>& fld,
1206  const negateOp& negOp,
1207  const int tag
1208 ) const
1209 {
1210  if (Pstream::defaultCommsType == Pstream::commsTypes::nonBlocking)
1211  {
1212  distribute
1213  (
1214  Pstream::commsTypes::nonBlocking,
1215  List<labelPair>(),
1216  constructSize_,
1217  subMap_,
1218  subHasFlip_,
1219  constructMap_,
1220  constructHasFlip_,
1221  fld,
1222  negOp,
1223  tag
1224  );
1225  }
1226  else if (Pstream::defaultCommsType == Pstream::commsTypes::scheduled)
1227  {
1228  distribute
1229  (
1230  Pstream::commsTypes::scheduled,
1231  schedule(),
1232  constructSize_,
1233  subMap_,
1234  subHasFlip_,
1235  constructMap_,
1236  constructHasFlip_,
1237  fld,
1238  negOp,
1239  tag
1240  );
1241  }
1242  else
1243  {
1244  distribute
1245  (
1246  Pstream::commsTypes::blocking,
1247  List<labelPair>(),
1248  constructSize_,
1249  subMap_,
1250  subHasFlip_,
1251  constructMap_,
1252  constructHasFlip_,
1253  fld,
1254  negOp,
1255  tag
1256  );
1257  }
1258 }
1259 
1260 
1261 //- Distribute data using default commsType.
1262 template<class T>
1263 void Foam::distributionMapBase::distribute
1264 (
1265  List<T>& fld,
1266  const int tag
1267 ) const
1268 {
1269  distribute(fld, flipOp(), tag);
1270 }
1271 
1272 
1273 //- Distribute data using default commsType.
1274 template<class T>
1275 void Foam::distributionMapBase::distribute
1276 (
1277  DynamicList<T>& fld,
1278  const int tag
1279 ) const
1280 {
1281  fld.shrink();
1282 
1283  List<T>& fldList = static_cast<List<T>& >(fld);
1284 
1285  distribute(fldList, tag);
1286 
1287  fld.setCapacity(fldList.size());
1288 }
1289 
1290 
1291 //- Reverse distribute data using default commsType.
1292 template<class T>
1293 void Foam::distributionMapBase::reverseDistribute
1294 (
1295  const label constructSize,
1296  List<T>& fld,
1297  const int tag
1298 ) const
1299 {
1300  if (Pstream::defaultCommsType == Pstream::commsTypes::nonBlocking)
1301  {
1302  distribute
1303  (
1304  Pstream::commsTypes::nonBlocking,
1305  List<labelPair>(),
1306  constructSize,
1307  constructMap_,
1308  constructHasFlip_,
1309  subMap_,
1310  subHasFlip_,
1311  fld,
1312  flipOp(),
1313  tag
1314  );
1315  }
1316  else if (Pstream::defaultCommsType == Pstream::commsTypes::scheduled)
1317  {
1318  distribute
1319  (
1320  Pstream::commsTypes::scheduled,
1321  schedule(),
1322  constructSize,
1323  constructMap_,
1324  constructHasFlip_,
1325  subMap_,
1326  subHasFlip_,
1327  fld,
1328  flipOp(),
1329  tag
1330  );
1331  }
1332  else
1333  {
1334  distribute
1335  (
1336  Pstream::commsTypes::blocking,
1337  List<labelPair>(),
1338  constructSize,
1339  constructMap_,
1340  constructHasFlip_,
1341  subMap_,
1342  subHasFlip_,
1343  fld,
1344  flipOp(),
1345  tag
1346  );
1347  }
1348 }
1349 
1350 
1351 //- Reverse distribute data using default commsType.
1352 // Since constructSize might be larger than supplied size supply
1353 // a nullValue
1354 template<class T>
1355 void Foam::distributionMapBase::reverseDistribute
1356 (
1357  const label constructSize,
1358  const T& nullValue,
1359  List<T>& fld,
1360  const int tag
1361 ) const
1362 {
1363  if (Pstream::defaultCommsType == Pstream::commsTypes::nonBlocking)
1364  {
1365  distribute
1366  (
1367  Pstream::commsTypes::nonBlocking,
1368  List<labelPair>(),
1369  constructSize,
1370  constructMap_,
1371  constructHasFlip_,
1372  subMap_,
1373  subHasFlip_,
1374  fld,
1375  eqOp<T>(),
1376  flipOp(),
1377  nullValue,
1378  tag
1379  );
1380  }
1381  else if (Pstream::defaultCommsType == Pstream::commsTypes::scheduled)
1382  {
1383  distribute
1384  (
1385  Pstream::commsTypes::scheduled,
1386  schedule(),
1387  constructSize,
1388  constructMap_,
1389  constructHasFlip_,
1390  subMap_,
1391  subHasFlip_,
1392  fld,
1393  eqOp<T>(),
1394  flipOp(),
1395  nullValue,
1396  tag
1397  );
1398  }
1399  else
1400  {
1401  distribute
1402  (
1403  Pstream::commsTypes::blocking,
1404  List<labelPair>(),
1405  constructSize,
1406  constructMap_,
1407  constructHasFlip_,
1408  subMap_,
1409  subHasFlip_,
1410  fld,
1411  eqOp<T>(),
1412  flipOp(),
1413  nullValue,
1414  tag
1415  );
1416  }
1417 }
1418 
1419 
1420 // ************************************************************************* //
PstreamCombineReduceOps.H
Combination-Reduction operation for a parallel run. The information from all nodes is collected on th...
forAll
#define forAll(list, i)
Loop across all elements in list.
Definition: UList.H:434
Foam::DynamicList
A 1D vector of objects of type <T> that resizes itself as necessary to accept the new objects.
Definition: DynamicList.H:78
Foam::IPstream
Input inter-processor communications stream.
Definition: IPstream.H:54
Foam::List
A 1D array of objects of type <T>, where the size of the vector is known and used for subscript bound...
Definition: List.H:91
Foam::List::transfer
void transfer(List< T > &)
Transfer the contents of the argument List into this list.
Definition: List.C:342
Foam::List::size
void size(const label)
Override size to be inconsistent with allocated storage.
Definition: ListI.H:164
Foam::List::setSize
void setSize(const label)
Reset size of List.
Definition: List.C:281
Foam::OPstream
Output inter-processor communications stream.
Definition: OPstream.H:54
Foam::PstreamBuffers
Buffers for inter-processor communications streams (UOPstream, UIPstream).
Definition: PstreamBuffers.H:86
Foam::PstreamBuffers::finishedSends
void finishedSends(const bool block=true)
Mark all sends as having been done. This will start receives.
Definition: PstreamBuffers.C:82
Foam::UIPstream
Input inter-processor communications stream operating on external buffer.
Definition: UIPstream.H:57
Foam::UIPstream::read
static label read(const commsTypes commsType, const int fromProcNo, char *buf, const std::streamsize bufSize, const int tag=UPstream::msgType(), const label communicator=0)
Read into given buffer from given processor and return the.
Definition: UIPread.C:79
Foam::UList::size
label size() const
Return the number of elements in the UList.
Definition: UListI.H:311
Foam::UList::begin
iterator begin()
Return an iterator to begin traversing the UList.
Definition: UListI.H:216
Foam::UList::byteSize
std::streamsize byteSize() const
Return the binary size in number of characters of the UList.
Definition: UList.C:100
Foam::UOPstream
Output inter-processor communications stream operating on external buffer.
Definition: UOPstream.H:58
Foam::UOPstream::write
static bool write(const commsTypes commsType, const int toProcNo, const char *buf, const std::streamsize bufSize, const int tag=UPstream::msgType(), const label communicator=0)
Write given buffer to given processor.
Definition: UOPwrite.C:34
Foam::UPstream::commsTypes
commsTypes
Types of communications.
Definition: UPstream.H:65
Foam::UPstream::nRequests
static label nRequests()
Get number of outstanding requests.
Definition: UPstream.C:137
Foam::UPstream::nProcs
static label nProcs(const label communicator=0)
Number of processes in parallel run.
Definition: UPstream.H:411
Foam::UPstream::waitRequests
static void waitRequests(const label start=0)
Wait until all requests (from start onwards) have finished.
Definition: UPstream.C:147
Foam::UPstream::parRun
static bool & parRun()
Is this a parallel run?
Definition: UPstream.H:399
Foam::UPstream::defaultCommsType
static commsTypes defaultCommsType
Default commsType.
Definition: UPstream.H:272
Foam::UPstream::myProcNo
static int myProcNo(const label communicator=0)
Number of this process (starting from masterNo() = 0)
Definition: UPstream.H:429
Foam::distributionMapBase::reverseDistribute
void reverseDistribute(const label constructSize, List< T > &, const int tag=UPstream::msgType()) const
Reverse distribute data using default commsType.
Definition: distributionMapBaseTemplates.C:1294
Foam::distributionMapBase::distribute
static void distribute(const Pstream::commsTypes commsType, const List< labelPair > &schedule, const label constructSize, const labelListList &subMap, const bool subHasFlip, const labelListList &constructMap, const bool constructHasFlip, List< T > &, const negateOp &negOp, const int tag=UPstream::msgType())
Distribute data. Note:schedule only used for.
Definition: distributionMapBaseTemplates.C:119
Foam::distributionMapBase::accessAndFlip
static T accessAndFlip(const UList< T > &fld, const label index, const bool hasFlip, const negateOp &negOp)
Definition: distributionMapBaseTemplates.C:80
Foam::distributionMapBase::send
void send(PstreamBuffers &, const List< T > &) const
Do all sends using PstreamBuffers.
Definition: distributionMapBaseTemplates.C:1126
Foam::distributionMapBase::flipAndCombine
static void flipAndCombine(const UList< label > &map, const bool hasFlip, const UList< T > &rhs, const CombineOp &cop, const negateOp &negOp, List< T > &lhs)
Definition: distributionMapBaseTemplates.C:35
Foam::distributionMapBase::receive
void receive(PstreamBuffers &, List< T > &) const
Do all receives using PstreamBuffers.
Definition: distributionMapBaseTemplates.C:1163
Foam::eqOp
Definition: ops.H:70
Foam::flipOp
Class containing functor to negate primitives. Dummy for all other types.
Definition: flipOp.H:51
FatalErrorInFunction
#define FatalErrorInFunction
Report an error message using Foam::FatalError.
Definition: error.H:334
fld
gmvFile<< "tracers "<< particles.size()<< nl;{ pointField positions(particles.size());label particlei=0;forAllConstIter(Cloud< passiveParticle >, particles, iter) { positions[particlei++]=iter().position(mesh);} for(i=0;i< pTraits< point >::nComponents;i++) { forAll(positions, particlei) { gmvFile<< component(positions[particlei], i)<< ' ';} gmvFile<< nl;}}forAll(lagrangianScalarNames, i){ const word &name=lagrangianScalarNames[i];IOField< scalar > fld(IOobject(name, runTime.name(), cloud::prefix, mesh, IOobject::MUST_READ, IOobject::NO_WRITE))
Foam::exit
errorManipArg< error, int > exit(error &err, const int errNo=1)
Definition: errorManip.H:124
Foam::label
intWM_LABEL_SIZE_t label
A label is an int32_t or int64_t as specified by the pre-processor macro WM_LABEL_SIZE.
Definition: label.H:59
Foam::abort
errorManip< error > abort(error &err)
Definition: errorManip.H:131
Foam::FatalError
error FatalError
Foam::T
void T(FieldField< Field, Type > &f1, const FieldField< Field, Type > &f2)
Definition: FieldFieldFunctions.C:55