Skip to content

Commit f977420

Browse files
cmsbuildcmsbuild
authored
Merge pull request #50817 from fwyzard/MPI_developments_backport_161x
Backport MPI-related developments [16.1.x]
2 parents 27d9e49 + 1f73224 commit f977420

28 files changed

Lines changed: 877 additions & 529 deletions

‎HeterogeneousCore/MPICore/plugins/MPIController.cc‎

Lines changed: 33 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@
1818
#include "FWCore/Framework/interface/MakerMacros.h"
1919
#include "FWCore/Framework/interface/Run.h"
2020
#include "FWCore/Framework/interface/one/EDProducer.h"
21+
#include "FWCore/Framework/interface/TriggerNamesService.h"
2122
#include "FWCore/MessageLogger/interface/MessageLogger.h"
2223
#include "FWCore/ParameterSet/interface/ConfigurationDescriptions.h"
2324
#include "FWCore/ParameterSet/interface/EmptyGroupDescription.h"
@@ -99,34 +100,34 @@ MPIController::MPIController(edm::ParameterSet const& config)
99100
int rank;
100101
MPI_Comm_rank(MPI_COMM_WORLD, &rank);
101102

102-
// Determine the rank of the other process.
103-
auto followers = config.getUntrackedParameter<std::vector<int32_t>>("followers");
104-
if (followers.empty()) {
105-
// When there are only two proccesses, we can assume the ranks to be 0 and 1,
106-
// and we can infer the other process rank from our own.
107-
if (size == 2) {
108-
followers = {1 - rank};
109-
} else {
110-
throw edm::Exception(edm::errors::Configuration)
111-
<< "An empty list of remote processes is valid only where there are exactly two processes.";
112-
}
103+
edm::LogInfo("MPI") << "MPIController sees world size " << size;
104+
105+
// Determine the ranks of the follower processes.
106+
auto follower_name = config.getParameter<std::string>("followerProcessName");
107+
if (follower_name.empty()) {
108+
throw edm::Exception(edm::errors::Configuration)
109+
<< "ERROR: Follower process name cannot be empty. Aborting MPIController...";
113110
}
114-
if (followers.size() >= static_cast<size_t>(size)) {
111+
112+
edm::Service<edm::service::TriggerNamesService> tns;
113+
std::string const& this_process_name = tns->getProcessName();
114+
if (follower_name == this_process_name) {
115115
throw edm::Exception(edm::errors::Configuration)
116-
<< "The number of remote processes is invalid. Please specify at most " << size - 1 << "remote processes.";
116+
<< "ERROR: controller and follower processes cannot have the same name. Aborting MPIController...";
117117
}
118-
std::vector<int32_t> invalid;
119-
for (int follower : followers) {
120-
if (follower < 0 or follower >= size) {
121-
invalid.push_back(follower);
122-
}
118+
119+
edm::Service<MPIService> mpiservice;
120+
auto followers = mpiservice->getRanksByProcessName(follower_name);
121+
if (followers.empty()) {
122+
throw edm::Exception(edm::errors::Configuration)
123+
<< "ERROR: No follower process with name " << follower_name << " found. Aborting...";
123124
}
124-
if (invalid.size() == 1) {
125+
126+
if (followers.size() == static_cast<size_t>(size)) {
125127
throw edm::Exception(edm::errors::Configuration)
126-
<< fmt::format("The remote process {} is invalid. Valid ranks are 0 to {}.", invalid.front(), size - 1);
127-
} else if (invalid.size() > 1) {
128-
throw edm::Exception(edm::errors::Configuration) << fmt::format(
129-
"The remote processes {} are invalid. Valid ranks are 0 to {}.", fmt::join(invalid, ", "), size - 1);
128+
<< "The number of found followers equals to the world size. "
129+
<< "Possible reason could be process names' hash collision. "
130+
<< "Please check process names in follower and controller. Aborting...";
130131
}
131132

132133
for (int follower : followers) {
@@ -185,7 +186,9 @@ MPIController::~MPIController() {
185186
// Disconnect the per-stream communicators.
186187
for (auto& stream : streams_) {
187188
// TODO move this to end stream
188-
stream->reset();
189+
if (stream) {
190+
stream->reset();
191+
}
189192
}
190193

191194
// Close the intercommunicator.
@@ -352,15 +355,12 @@ void MPIController::fillDescriptions(edm::ConfigurationDescriptions& description
352355
desc.ifValue(
353356
edm::ParameterDescription<std::string>("mode", "CommWorld", false),
354357
ModeDescription[kCommWorld] >>
355-
edm::ParameterDescription<std::vector<int32_t>>(
356-
"followers",
357-
{},
358-
false,
359-
edm::Comment("Ranks of the remote \"follower\" processes.\n"
360-
"When there are two or more follower processes, framework streams are associated to "
361-
"each follower in a round-robin fashion.\n"
362-
"When there is only one remote process, pass an empty list to autodetect its rank "
363-
"based on the rank of the current process.")) or
358+
edm::ParameterDescription<std::string>(
359+
"followerProcessName",
360+
"",
361+
true,
362+
edm::Comment("All processes with this process name should act as followers, "
363+
"and should be configured with an MPISource that follows this controller.")) or
364364
ModeDescription[kIntercommunicator] >> edm::ParameterDescription<std::string>("name", "server", false))
365365
->setComment(
366366
"Valid modes are CommWorld (use MPI_COMM_WORLD) and Intercommunicator (use an MPI name server to setup an "

‎HeterogeneousCore/MPICore/plugins/MPIReceiver.cc‎

Lines changed: 83 additions & 64 deletions
Original file line numberDiff line numberDiff line change
@@ -64,6 +64,8 @@ class MPIReceiver : public edm::stream::EDProducer<edm::ExternalWork> {
6464

6565
products_.emplace_back(std::move(entry));
6666
}
67+
68+
received_wrappers_.resize(products_.size());
6769
}
6870

6971
void acquire(edm::Event const& event, edm::EventSetup const&, edm::WaitingTaskWithArenaHolder holder) final {
@@ -75,17 +77,87 @@ class MPIReceiver : public edm::stream::EDProducer<edm::ExternalWork> {
7577
edm::Service<edm::Async> as;
7678
as->runAsync(
7779
std::move(holder),
78-
[this, token]() { token.channel()->receiveMetadata(instance_, received_meta_); },
80+
[this, token]() {
81+
token.channel()->receiveMetadata(instance_, received_meta_);
82+
#ifdef EDM_ML_DEBUG
83+
// dump the summary of metadata
84+
received_meta_->debugPrintMetadataSummary();
85+
#endif
86+
87+
// if filter was false before the sender, receive nothing
88+
if (received_meta_->productCount() == -1) {
89+
return;
90+
}
91+
92+
std::unique_ptr<TBufferFile> serialized_buffer;
93+
if (received_meta_->hasSerialized()) {
94+
serialized_buffer =
95+
token.channel()->receiveSerializedBuffer(instance_, received_meta_->serializedBufferSize());
96+
#ifdef EDM_ML_DEBUG
97+
{
98+
edm::LogSystem msg("MPIReceiver");
99+
msg << "Received serialised product:\n";
100+
for (int i = 0; i < received_meta_->serializedBufferSize(); ++i) {
101+
msg << "0x" << std::hex << std::setw(2) << std::setfill('0')
102+
<< (unsigned int)(unsigned char)serialized_buffer->Buffer()[i] << (i % 16 == 15 ? '\n' : ' ');
103+
}
104+
}
105+
#endif
106+
}
107+
108+
for (size_t i = 0; i < products_.size(); ++i) {
109+
auto product_meta = received_meta_->getNext();
110+
if (product_meta.kind == ProductMetadata::Kind::Missing) {
111+
continue;
112+
}
113+
114+
auto const& entry = products_[i];
115+
116+
if (product_meta.kind == ProductMetadata::Kind::Serialized) {
117+
std::unique_ptr<edm::WrapperBase> wrapper(
118+
reinterpret_cast<edm::WrapperBase*>(entry.wrappedType.getClass()->New()));
119+
assert(static_cast<int32_t>(serialized_buffer->Length() + product_meta.sizeMeta) <=
120+
received_meta_->serializedBufferSize() &&
121+
"serialized data buffer is shorter than expected");
122+
entry.wrappedType.getClass()->Streamer(wrapper.get(), *serialized_buffer);
123+
received_wrappers_[i] = std::move(wrapper);
124+
}
125+
126+
else if (product_meta.kind == ProductMetadata::Kind::TrivialCopy) {
127+
if (not enableTrivialSerialisation_) {
128+
edm::LogError("MPIReceiver")
129+
<< "Received a trivially-serialised product, but enableTrivialSerialisation is set to false in "
130+
"this MPIReceiver. Please check that the MPISender and MPIReceiver have consistent settings.";
131+
MPI_Abort(MPI_COMM_WORLD, EXIT_FAILURE);
132+
}
133+
std::unique_ptr<ngt::SerialiserBase> serialiser =
134+
ngt::SerialiserFactory::get()->tryToCreate(entry.type.typeInfo().name());
135+
if (not serialiser) {
136+
throw cms::Exception("SerializerError")
137+
<< "Receiver could not retrieve its serializer when it was expected";
138+
}
139+
auto writer = serialiser->writer();
140+
ngt::AnyBuffer buffer = writer->uninitialized_parameters(); // constructs buffer with typeid
141+
assert(buffer.size_bytes() == product_meta.sizeMeta);
142+
std::memcpy(buffer.data(), product_meta.trivialCopyOffset, product_meta.sizeMeta);
143+
writer->initialize(buffer);
144+
token.channel()->receiveInitializedTrivialCopy(instance_, *writer);
145+
writer->finalize();
146+
received_wrappers_[i] = writer->get();
147+
}
148+
}
149+
150+
if (received_meta_->hasSerialized()) {
151+
assert(serialized_buffer->Length() == received_meta_->serializedBufferSize() &&
152+
"serialized data buffer is not equal to the expected length");
153+
}
154+
},
79155
[]() { return "Calling MPIReceiver::acquire()"; });
80156
}
81157

82158
void produce(edm::Event& event, edm::EventSetup const&) final {
83159
// read the MPIToken used to establish the communication channel
84160
MPIToken token = event.get(upstream_);
85-
#ifdef EDM_ML_DEBUG
86-
// dump the summary of metadata
87-
received_meta_->debugPrintMetadataSummary();
88-
#endif
89161

90162
// if filter was false before the sender, receive nothing
91163
if (received_meta_->productCount() == -1) {
@@ -96,68 +168,14 @@ class MPIReceiver : public edm::stream::EDProducer<edm::ExternalWork> {
96168
event.emplace(pathStateToken_);
97169
}
98170

99-
std::unique_ptr<TBufferFile> serialized_buffer;
100-
if (received_meta_->hasSerialized()) {
101-
serialized_buffer = token.channel()->receiveSerializedBuffer(instance_, received_meta_->serializedBufferSize());
102-
#ifdef EDM_ML_DEBUG
103-
{
104-
edm::LogSystem msg("MPISender");
105-
msg << "Received serialised product:\n";
106-
for (int i = 0; i < received_meta_->serializedBufferSize(); ++i) {
107-
msg << "0x" << std::hex << std::setw(2) << std::setfill('0')
108-
<< (unsigned int)(unsigned char)serialized_buffer->Buffer()[i] << (i % 16 == 15 ? '\n' : ' ');
109-
}
110-
}
111-
#endif
112-
}
113-
114-
for (auto const& entry : products_) {
115-
auto product_meta = received_meta_->getNext();
116-
if (product_meta.kind == ProductMetadata::Kind::Missing) {
117-
edm::LogWarning("MPIReceiver") << "Product " << entry.type.name() << " was not received.";
118-
continue; // Skip products that weren't sent
119-
}
120-
121-
else if (product_meta.kind == ProductMetadata::Kind::Serialized) {
122-
std::unique_ptr<edm::WrapperBase> wrapper(
123-
reinterpret_cast<edm::WrapperBase*>(entry.wrappedType.getClass()->New()));
124-
assert(static_cast<int32_t>(serialized_buffer->Length() + product_meta.sizeMeta) <=
125-
received_meta_->serializedBufferSize() &&
126-
"serialized data buffer is shorter than expected");
127-
entry.wrappedType.getClass()->Streamer(wrapper.get(), *serialized_buffer);
128-
// put the data into the Event
129-
event.put(entry.token, std::move(wrapper));
130-
}
131-
132-
else if (product_meta.kind == ProductMetadata::Kind::TrivialCopy) {
133-
if (not enableTrivialSerialisation_) {
134-
edm::LogError("MPIReceiver")
135-
<< "Received a trivially-serialised product, but enableTrivialSerialisation is set to false in this "
136-
"MPIReceiver. Please check that the MPISender and MPIReceiver have consistent settings.";
137-
MPI_Abort(MPI_COMM_WORLD, EXIT_FAILURE);
138-
}
139-
std::unique_ptr<ngt::SerialiserBase> serialiser =
140-
ngt::SerialiserFactory::get()->tryToCreate(entry.type.typeInfo().name());
141-
if (not serialiser) {
142-
throw cms::Exception("SerializerError") << "Receiver could not retrieve its serializer when it was expected";
143-
}
144-
auto writer = serialiser->writer();
145-
ngt::AnyBuffer buffer = writer->uninitialized_parameters(); // constructs buffer with typeid
146-
assert(buffer.size_bytes() == product_meta.sizeMeta);
147-
std::memcpy(buffer.data(), product_meta.trivialCopyOffset, product_meta.sizeMeta);
148-
writer->initialize(buffer);
149-
token.channel()->receiveInitializedTrivialCopy(instance_, *writer);
150-
writer->finalize();
151-
// put the data into the Event
152-
event.put(entry.token, writer->get());
171+
for (size_t i = 0; i < products_.size(); ++i) {
172+
if (received_wrappers_[i]) {
173+
event.put(products_[i].token, std::move(received_wrappers_[i]));
174+
} else {
175+
edm::LogWarning("MPIReceiver") << "Product " << products_[i].type.name() << " was not received.";
153176
}
154177
}
155178

156-
if (received_meta_->hasSerialized()) {
157-
assert(serialized_buffer->Length() == received_meta_->serializedBufferSize() &&
158-
"serialized data buffer is not equal to the expected length");
159-
}
160-
161179
// write a shallow copy of the channel to the output, so other modules can consume it
162180
// to indicate that they should run after this
163181
event.emplace(token_, token);
@@ -207,6 +225,7 @@ class MPIReceiver : public edm::stream::EDProducer<edm::ExternalWork> {
207225
bool activity_; // indicator whether the PathStateToken will be received by the module
208226
edm::EDPutTokenT<edm::PathStateToken> pathStateToken_;
209227
std::shared_ptr<ProductMetadataBuilder> received_meta_;
228+
std::vector<std::unique_ptr<edm::WrapperBase>> received_wrappers_;
210229
bool enableTrivialSerialisation_ = true;
211230
};
212231

‎HeterogeneousCore/MPICore/plugins/MPISender.cc‎

Lines changed: 28 additions & 38 deletions
Original file line numberDiff line numberDiff line change
@@ -112,6 +112,13 @@ class MPISender : public edm::stream::EDProducer<edm::ExternalWork> {
112112
const MPIToken& token = event.get(upstream_);
113113
// pass the number of products to estimate the right size for the metadata buffer
114114
auto meta = std::make_shared<ProductMetadataBuilder>(products_.size());
115+
116+
// We use std::shared_ptr, instead of std::unique_ptr, so that readers can
117+
// be captured by move by runAsync's lamnda. This is ultimately because this
118+
// lambda is used to construct an std::function, which requires its callable
119+
// to be copy-constructible.
120+
std::vector<std::shared_ptr<const ngt::ReaderBase>> readers;
121+
readers.reserve(products_.size());
115122
size_t index = 0;
116123
buffer_->Reset();
117124
has_serialized_ = false;
@@ -145,6 +152,7 @@ class MPISender : public edm::stream::EDProducer<edm::ExternalWork> {
145152
auto reader = serialiser->reader(*wrapper);
146153
ngt::AnyBuffer buffer = reader->parameters();
147154
meta->addTrivialCopy(buffer.data(), buffer.size_bytes());
155+
readers.push_back(std::move(reader));
148156
} else {
149157
LogDebug("MPISender") << "No serializer for type \"" << entry.type.name() << "\" ("
150158
<< entry.type.typeInfo().name() << "), using ROOT serialization";
@@ -169,50 +177,32 @@ class MPISender : public edm::stream::EDProducer<edm::ExternalWork> {
169177
edm::Service<edm::Async> as;
170178
as->runAsync(
171179
std::move(holder),
172-
[this, token, meta = std::move(meta)]() { token.channel()->sendMetadata(instance_, meta); },
180+
[this, token, meta = std::move(meta), readers = std::move(readers)]() {
181+
token.channel()->sendMetadata(instance_, meta);
182+
if (has_serialized_) {
183+
#ifdef EDM_ML_DEBUG
184+
{
185+
edm::LogSystem msg("MPISender");
186+
msg << "Sending serialised product:\n";
187+
for (int i = 0; i < buffer_->Length(); ++i) {
188+
msg << "0x" << std::hex << std::setw(2) << std::setfill('0')
189+
<< (unsigned int)(unsigned char)buffer_->Buffer()[i] << (i % 16 == 15 ? '\n' : ' ');
190+
}
191+
}
192+
#endif
193+
token.channel()->sendBuffer(buffer_->Buffer(), buffer_->Length(), instance_, EDM_MPI_SendSerializedProduct);
194+
}
195+
for (auto const& reader : readers) {
196+
token.channel()->sendTrivialCopyProduct(instance_, *reader);
197+
}
198+
},
173199
[]() { return "Calling MPISender::acquire()"; });
174200
}
175201

176202
void produce(edm::Event& event, edm::EventSetup const&) final {
177-
MPIToken token = event.get(upstream_);
178-
179-
if (!is_active_) {
180-
event.emplace(token_, token);
181-
return;
182-
}
183-
184-
if (has_serialized_) {
185-
#ifdef EDM_ML_DEBUG
186-
{
187-
edm::LogSystem msg("MPISender");
188-
msg << "Sending serialised product:\n";
189-
for (int i = 0; i < buffer_->Length(); ++i) {
190-
msg << "0x" << std::hex << std::setw(2) << std::setfill('0')
191-
<< (unsigned int)(unsigned char)buffer_->Buffer()[i] << (i % 16 == 15 ? '\n' : ' ');
192-
}
193-
}
194-
#endif
195-
token.channel()->sendBuffer(buffer_->Buffer(), buffer_->Length(), instance_, EDM_MPI_SendSerializedProduct);
196-
}
197-
198-
for (auto const& entry : products_) {
199-
edm::Handle<edm::WrapperBase> handle(entry.type.typeInfo());
200-
event.getByToken(entry.token, handle);
201-
edm::WrapperBase const* wrapper = handle.product();
202-
// we don't send missing products
203-
if (handle.isValid()) {
204-
std::unique_ptr<ngt::SerialiserBase> serialiser;
205-
if (enableTrivialSerialisation_) {
206-
serialiser = ngt::SerialiserFactory::get()->tryToCreate(entry.type.typeInfo().name());
207-
}
208-
if (serialiser) {
209-
auto reader = serialiser->reader(*wrapper);
210-
token.channel()->sendTrivialCopyProduct(instance_, *reader);
211-
}
212-
}
213-
}
214203
// write a shallow copy of the channel to the output, so other modules can consume it
215204
// to indicate that they should run after this
205+
MPIToken token = event.get(upstream_);
216206
event.emplace(token_, token);
217207
}
218208

0 commit comments

Comments
 (0)