mpi_aggregator: don't pass a local variable to MPI_Isend

[germasch]

The data passed to MPI_Isend needs to remain available until after
MPI_Wait, so use a class member variable instead.

This also means the member variable can't be const, so the workaround for some MPI_Isend implementations isn't needed anymore.

[chuckatkins]

Please rebase off current master as the version of clang-format was just bumped from 3.8.1 to 7.0.1

[eisenhauer]

The original code is clearly erroneous because it passes the address of a stack-allocated variable to a routine that was allowed to grab that buffer asynchronously at any time prior to the MPI_Wait(), which occurs outside the scope of the passed variable. It likely worked because any MPI implementation worth its salt can (and likely would) buffer an 8 byte send, but it wasn't guaranteed to work by MPI semantics.

This PR fixes the erroneous nature of the code by creating a class member which will persist until the MPI_Wait() happens, in the WaitAbsolutePosition method. Still not entirely thrilled with this abstraction, because it is really customized for the particular usage that appears elsewhere. The IExchangeAbsolutePosition and WaitAbsolutePosition methods are semantically linked by some state which is exposed and returned to the calling environment (the requests vector), and other state that is hidden inside the object. So, one couldn't, for example, do two calls to IExchangeAbsolutePosition() before WaitAbsolutePosition() without having hidden potentially erroneous behavior (that would probably never happen because Isend of 8 bytes would probably always buffer, but still erroneous). I considered the utility of using Send here instead of ISend, relying upon the usual equivalence of Send and ISend for tiny buffers, but that would just be replacing one bit of strictly erroneous code with another. Alternatively, the code could be refactored to export the send buffer as well as the requests vector to the calling environment, where it could be deallocated after (or in) the Wait call. But this code is non-generic (so highly specialized to the needs of its caller that it is unlikely to be used anywhere else, in any other manner) and even if it was called differently, the same factor that made the original erroneous implementation work (MPI always buffering tiny sends) means that actual problems might be unlikely to occur. Given those factors, I'm tempted to say that a more complex, "export all the state to the caller" sort of fix is likely more trouble than it's worth. @germasch @pnorbert do you concur? Maybe at most add a comment about the situation, warning future developers who might be tempted to use this as an example?

[germasch]

I was actually going to suggest that this communication piece of code could be cleaned up in general. It's quite opaque in what it does, and also not necessarily particularly efficient in the communication pattern it uses. But I figured I'd start with what's obviously just a small self-contained fix.

On the particular topic of maintaining state, I actually changed things in my local version in a way that the MPI_Request's are not returned to the user but maintained in the class, because as @eisenhauer says, the code maintains internal state anyway that prevents it from being used more generally, so one might as well make the interface simpler.

Let me describe what MPIAggregator actually does (more specifically, what it's derived class MPIChain does):

• It performs essentially an exclusive scan (in MPI terms, a.k.a. prefix sum) of m_AbsolutePosition. (On rank 0, however, it finds the total sum)
• It performs a Gatherv-like gathering of all data buffers onto rank 0, where they get written.

So my suggestion for the absolute position would be to just use MPI_Exscan (or, ideally, MPI_Iexscan if it is okay to require MPI >= version 3). No loop over rank required to achieve this.

For the actual data communication, as implemented by MPIChain, it actually shifts buffers down one rank at a time, ie., if you're aggregating over 4 processes, the buffer on rank 3 will be communicated rank 3 -> rank 2 -> rank 1 -> rank 0. Obviously, less data would go over the network if it were sent rank 3 -> rank 0 directly.

While MPI_Gatherv could be used directly, I think that's dangerous to do in general because it would result in rank 0 having to buffer the entire data from all processes in the aggregation group at once, which it might not have enough memory for. So my suggestion would be to keep the current interface, where rank 0 receives one data buffer at a time and then writes it as it goes. One could eventually go fancy and adapt behavior depending on available buffer space (ie, if it's all small buffers to be collected, use Gatherv, if they're big, do one a time), but I don't think there's a good reason to do so right now. But even when receiving / handling buffers one at a time, I think they should be sent directly to rank 0, not passed down one rank at a time.

[pnorbert]

The chain communication (a.k.a. Brigade method) was implemented in adios 1.x by Qing Liu, which proved to be better performing on Jaguar/Titan for our apps then directly sending buffers to rank 0. However counter-intuitive and strange it seems, this was the case, because it always kept the next "receive next buffer on rank 0" operation faster than the "write current buffer to disk" and hence avoided gaps between write operations. This may or may not be true for Summit anymore but the approach would be to have various aggregations strategies available. I guess the infrastructure is not in place to allow for multiple strategies right now.

…

On Sun, Mar 31, 2019 at 11:11 AM Kai Germaschewski ***@***.***> wrote: I was actually going to suggest that this communication piece of code could be cleaned up in general. It's quite opaque in what it does, and also not necessarily particularly efficient in the communication pattern it uses. But I figured I'd start with what's obviously just a small self-contained fix. On the particular topic of maintaining state, I actually changed things in my local version in a way that the MPI_Request's are not returned to the user but maintained in the class, because as @eisenhauer <https://github.com/eisenhauer> says, the code maintains internal state anyway that prevents it from being used more generally, so one might as well make the interface simpler. Let me describe what MPIAggregator actually does (more specifically, what it's derived class MPIChain does): - It performs essentially an exclusive scan (in MPI terms, a.k.a. prefix sum) of m_AbsolutePosition. (On rank 0, however, it finds the total sum) - It performs a Gatherv-like gathering of all data buffers onto rank 0, where they get written. So my suggestion for the absolute position would be to just use MPI_Exscan (or, ideally, MPI_Iexscan if it is okay to require MPI >= version 3). No loop over rank required to achieve this. For the actual data communication, as implemented by MPIChain, it actually shifts buffers down one rank at a time, ie., if you're aggregating over 4 processes, the buffer on rank 3 will be communicated rank 3 -> rank 2 -> rank 1 -> rank 0. Obviously, less data would go over the network if it were sent rank 3 -> rank 0 directly. While MPI_Gatherv could be used directly, I think that's dangerous to do in general because it would result in rank 0 having to buffer the entire data from all processes in the aggregation group at once, which it might not have enough memory for. So my suggestion would be to keep the current interface, where rank 0 receives one data buffer at a time and then writes it as it goes. One could eventually go fancy and adapt behavior depending on available buffer space (ie, if it's all small buffers to be collected, use Gatherv, if they're big, do one a time), but I don't think there's a good reason to do so right now. But even when receiving / handling buffers one at a time, I think they should be sent directly to rank 0, not passed down one rank at a time. — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#1326 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/ADGMLZP_cIUb-E_zeD3_HaSZY8wOY5y3ks5vcNAcgaJpZM4cLnS9> .

[germasch]

I agree that there is a point in having multiple strategies available, so they can be benchmarked and the best option chosen. The current code has laid the groundwork for that in that MPIAggregator can provide different implementations by way of a derived class. Right now MPIChain is the only (and hardcoded) option, but turning m_Aggregator into a pointer to the base class and making it configurable should be quite straightforward.

[germasch]

By the way, I think the current (and guess having a longer history method) of chaining the buffers has the following (possibly advantageous) properties:

• It "double buffers", ie, while one buffer is written, the data from the next rank is being received into a second buffer using non-blocking MPI. I would think this kind of overlap is a good idea for performance.
• It kinda throttles messages onto rank 0. That is, instead of having all ranks send the buffers to rank 0 at the same time, while rank 0 is only ready to receive one at a time, the current scheme basically makes sure that every process is at most sending one and receiving one buffer at the same time. For this one, I'm less sure what performance implications might be, because in general, I believe MPI may communicate small messages even while there is no receiver yet (so it needs to buffer them), while large messages will be kept on the sender side until the receiver is ready, so that'd do similar throttling. But I guess I also know that reality is yet more complicated, and beyond what I really know details about.

[eisenhauer]

So, action plan? @germasch, maybe amend this with your changes that maintain the requests internally to simplify the interface? Then alternative implementations of aggregate left for future work?

[germasch]

I rebased and added commits to maintain the requests internally.
While I was at it, I also added some simplification of MPI_Irecv + immediately following MPI_Wait to MPI_Recv, and used MPI_STATUS_IGNORE. Those additional commits are at the end, so they can be easily removed if considered objectionable.

There's actually more simple maintenance that could be done, but I kept it at the trivial changes:

• use MPI_Waitall to combine MPI_Wait calls.
• make MPIAggregator an abstract base class, because right now you can use MPIAggregator directly, but it won't actually communicate your data buffers, which is highly likely not what one wants/expects.

[williamfgc]

I'd be conservative in changing the current aggregation strategy (MPIChain) if no performance gain is achieved (fixing bugs is fine as in the first commit). Changing internal behavior means rerunning current application benchmarks at scale as that's the only way we'll know (the CI is pointless here, and I personally prefer non-blocking calls), which will involve other people's time and burning core-hours.

As for virtualizing calls to MPIAggregator, that's in the plans (hence the OO boilerplate, MPIChain predates this), so please go ahead as this brings the possibilities to add more strategies for upcoming systems other than MPIChain.

[germasch]

I wouldn't want to change the current MPIChain implementation (at least not in fundamental ways), if for no other reason that it wouldn't even be possible to tell whether an updated strategy performs better without having them both available to run benchmarks.

This PR doesn't change anything about the MPIChain strategy -- initially it was just a simple bugfix, now it also includes some cleanup, but doesn't change behavior. (Unless you consider changing MPI_Irecv + immediate MPI_Wait -> MPI_Recv a change of behavior, but your MPI library would have to be rather broken for that to cause troubles.)

[williamfgc]

Can we justify why we need the architectural change? (New use-case? New strategy? Performance gain? New engine? Reusability?) While the proposed interface makes everything private, takes away control for Engine developers to play with different Wait options, as performance is driven by potential overlap. Also, no need to add blocking calls as we don't know how each system handles network blocking without measuring (the standard is ambiguous). Rather be safe and keep everything non-blocking.

[williamfgc]

I would make a separate MPI aggregation strategy as it is allowed by the object oriented design. We can always overload MPIAggregator calls if needed. Let's not forget this is working at scale.

[germasch]

Can we justify why we need the architectural change?

@eisenhauer justified why not to return the MPI_Request vector above: Returning this state implies to the user that the function is designed to be called multiple times -- but it is not, because it also maintains internal state (m_SizeSend was already there, and with the bug fix m_AbsolutePositionSend). So instead of mixing internal and external state, keep it all internal to make it consistent / less prone to misuse.

While the proposed interface makes everything private, takes away control for Engine developers to play with different Wait options, as performance is driven by potential overlap.

Engine developers supposedly need to use WaitAbsolutePosition to do the waiting, so they don't have a choice anyway. If you mean that people developing different implementations of MPIAggregator don't have a choice, I don't think that's true: They can store as many MPI_Requests or whatever they need within their derived class, and external users will see no difference (that's a good thing).

Also, no need to add blocking calls as we don't know how each system handles networking blocking without measuring. Rather be safe and keep everything non-blocking.

MPI_Irecv + MPI_Wait is already blocking, just like MPI_Recv. The only difference is that the former allows for overlap in between the two calls, but in the cases I changed, there was nothing there in between. Nevertheless, I can drop that commit if that's what you want.

[pnorbert]

Let's put this topic on the dev-call agenda for tomorrow. We need to pick this topic apart and see what are bug fixes, what are changes to the (internal) interface to provide multiple aggregation strategies, and what are new strategies.
Kai, the problem with any direct changes to the aggregation is that we only have one and it requires testing apps to make sure we don't break anything. Well, the problem is manpower, not the testing itself, which will be needed anyway when the internal interface changes.
We need to discuss if we accept the fixes in this PR and than figure out who is going to test it with apps. Those steps cannot be too far apart.

[williamfgc]

That's exactly my point, the proposed architecture changes is constraining things to how the aggregator's Wait is being used today. That's not how object oriented systems are designed, but rather as flexible and extendable interfaces.

If a new performance strategy requires splitting those Wait calls, we'll have to go back to the current architecture. I wouldn't compromise flexibility of the current OO design in choosing to partition Wait calls if a new strategy requires for performance.

Yes, please drop the commit, rather make a separate strategy under MPIAggregator.

[germasch]

@williamfgc I'm sorry, but I'm a bit lost. What are we talking about: (1) stop returning a vector of MPI_Requests from the IExchange* functions and handing it back to the corresponding Wait* function or (2) Changing the internal MPI_IRecv + immediate MPI_Wait calls to MPI_Recv?

[williamfgc]

@germasch removing (1) will take away control of when to call each Wait for particular Requests as we are reusing the same interface, (2) let's just make everything non-blocking. These are deeper APIs for Engine developers (not users), as @pnorbert said, we want to be open to new strategies, so the MPIAggregator API must be very flexible (not tailored to a single use-case). FYI, getting to the point of making this work at scale was an iterative process that took a lot of work, but it's at least reusable.

[williamfgc]

@germasch BTW, glad you reached this point, as aggregation strategies is another optimization technique (@pnorbert described the history) to achieve performance. We need help understanding potential new strategies for new systems, the cost is higher as testing at scale is never cheap.

[germasch]

Let's put this topic on the dev-call agenda for tomorrow. We need to pick this topic apart and see what are bug fixes, what are changes to the (internal) interface to provide multiple aggregation strategies, and what are new strategies.

The first commit is a bug fix (temporary passed to MPI_Isend).
This triggered the following two commits which keep state entirely internal following comments from @eisenhauer (which I agree with). Note that this is also can be argued to be "object-oriented" (encapsulated) way. If some future generation machine for whatever reason wanted to use RDMA directly in the aggregation and would need to keep track of RDMA handles while overlapping, that change could be handled internally, transparent to the user (ie., the engines using aggregation). Whereas the current interface does not abstract the internal state between the I and Wait calls, but explicitly relies on std::vector<MPI_Request>. (This second rationale is not why me/Greg proposed this change, it's just about don't split state between internal and external. But it shows that it's generally a useful way to handle separate interface and implementation details.)

Kai, the problem with any direct changes to the aggregation is that we only have one and it requires testing apps to make sure we don't break anything. Well, the problem is manpower, not the testing itself, which will be needed anyway when the internal interface changes.

There is a test which exercises this functionality in a bunch of cases: TestBPWriteAggregateRead I'd consider that good enough to test the aggregation for correctness (a.k.a., stupid bugs I might have introduced by a typo or something). In terms of performance, there's really nothing in this PR which changes which messages are sent and when, so there will be no measurable difference. (It avoids a bunch of dynamic memory allocations for std::vector<MPI_Request>, but that also won't be measurable except for in a micro-benchmark).

We need to discuss if we accept the fixes in this PR and than figure out who is going to test it with apps. Those steps cannot be too far apart.

I certainly appreciate the strong push for stability. But I think the validation effort should be proportional to the kind of changes. If I were to add a new implementation of MPIAggregator with a different communication strategy, and the interface changes that are needed to select that at runtime, and if that new strategy became default, I most definitely agree that the impact on apps needs to be evaluated. But this PR is not that kind of a change, it's just a bunch of minor changes that don't change behavior in any noticeable way.

[germasch]

@germasch removing (1) will take away control of when to call each Wait for particular Requests as we are reusing the same interface,

The current API provides two calls to Wait, one for the data buffer, one for the absolute position. The user (ie, engine) can place those two calls according to their needs:

    void WaitAbsolutePosition(size_t &absolutePosition, const int step);
    void Wait(std::vector<MPI_Request> &requests, const int step);

After the change, the API still has those exact two calls:

    void WaitAbsolutePosition(const int step);
    void Wait(const int step);

So the user still has the two Wait calls that they can place according to their needs. I don't see how the second way takes away anything that was possible previously?

(2) let's just make everything non-blocking. These are deeper APIs for Engine developers (not users), as @pnorbert said, we want to be open to new strategies, so the MPIAggregator API must be very flexible. FYI, getting to the point of making this work at scale was an iterative process that took a lot of work.

No-one suggested to make everything non-blocking, as that's clearly a bad idea. Everything that's currently achieving overlap by non-blocking calls in MPIChain still does so (in particular, rank 0 writes out the last-received data buffer at the same time that the next buffer is in flight).

The change we're talking about looks like this:

@@ -64,19 +64,12 @@ void MPIChain::IExchange(BufferSTL &bufferSTL, const int step)
     if (receiver)
     {
         size_t bufferSize = 0;
-        MPI_Request receiveSizeRequest;
-        helper::CheckMPIReturn(MPI_Irecv(&bufferSize, 1, ADIOS2_MPI_SIZE_T,
-                                         m_Rank + 1, 0, m_Comm,
-                                         &receiveSizeRequest),
+        helper::CheckMPIReturn(MPI_Recv(&bufferSize, 1, ADIOS2_MPI_SIZE_T,
+                                        m_Rank + 1, 0, m_Comm,
+                                        MPI_STATUS_IGNORE),
                                ", aggregation Recv size at iteration " +
                                    std::to_string(step) + "\n");

-        MPI_Status receiveStatus;
-        helper::CheckMPIReturn(
-            MPI_Wait(&receiveSizeRequest, &receiveStatus),
-            ", aggregation waiting for receiver size at iteration " +
-                std::to_string(step) + "\n");
-
         BufferSTL &receiveBuffer = GetReceiver(bufferSTL);
         ResizeUpdateBufferSTL(
             bufferSize, receiveBuffer,

Note that it uses a local variable MPI_Request receiveSizeRequest. That's never exposed as part of the interface. None of this change has anything to do with the API of MPIAggregator, no matter how flexible or not you want it to be.

[williamfgc]

Current: void Wait(std::vector<MPI_Request> &requests, const int step);
Proposed: void Wait(const int step);

The current call can be reused by any group of Requests as decided by the Engine developer, the proposed one can't.

[germasch]

Can you provide an example of how you imagine reusing the call by passing "any group of Requests"? Where do those requests come from? Is the caller now doing their own MPI_Isend / MPI_Irecv? How many requests are in "any group of Requests"?

[germasch]

I took out the commit that changes MPI_Irecv + MPI_Wait to MPI_Recv. It's the only change in this PR that actually changed behavior (other than the original bug fix), and it's now gone. I don't think it's worth the time arguing over.

It still moves the MPI_Request vector internal to MPIAggregator, as suggested by @eisenhauer. I have trouble imagining an actual use case where this reduces needed flexibility, and no example has been provided to show otherwise.

[germasch]

Also, as I generally try to do, it reduces lines of code:

 8 files changed, 55 insertions(+), 77 deletions(-)

[chuckatkins]

Indeed this is a bug, sending a local stack variable via MPI_Isend, so thank you for identifying it. So, we just went through this in a fair amount of detail on our weekly dev call and the resolution is more complicated and nuanced. Your fix here, moving both both the absoluteposition and request to local state, resolves this but ends up breaking the interface since both of them are derived quantities based on the input buffer. So if IExchangeAbsolutePosition is called repeatedly with separate buffers then that internal state gets clobbered.

W.R.T how to fix the bug though, to do it via tracking an internal state we'd need to maintain an arbitrary length of requests and absolutesends returning some way to reference them on subsequent wait calls, or not keep it as internal state at all and instead return something akin to a pair<request, position> of sorts that can be referenced in subsequent wait calls; really just something that ensures the data sent lives as long as the request does. Since they are both derived quantities on the input parameter and either way would be returning something that needs to be passed back and referenced later, we're leaning towards the second to avoid adding additional internal state. Since it's more in depth than appears on the surface and goes in part to the intended use cases for the MPIAggregator class, it's been added to @williamfgc's list to address in the next week or two so I'm going to close this for now.

@williamfgc we discussed today about about you implementing the second solution. However, it may make sense to do the first if we can ensure that the buffer passed in that get's operated on is guaranteed to be one of m_Buffer and somehow ties to that.

[chuckatkins]

So, after more digging on this, rather than "fix" the underlying state corruption in various places in the class, I think we'll end up taking the approach you have here with adding an additional check to ensure the repeated call sequence that corrupts the state can't happen. I'm going to re-open this just add the 1 extra commit for said check. It should be good to go then.

[germasch]

@chuckatkins I can try fortifying it, but since I've thought about it for a while now, let me throw an alternative idea for the interface out there:

@@ -328,30 +348,14 @@ void BP3Writer::AggregateWriteData(const bool isFinal, const int transportIndex)
     m_BP3Serializer.CloseStream(m_IO, false);

     // async?
-    for (int r = 0; r < m_BP3Serializer.m_Aggregator.m_Size; ++r)
-    {
-        m_BP3Serializer.m_Aggregator.IExchange(m_BP3Serializer.m_Data, r);
-
-        m_BP3Serializer.m_Aggregator.IExchangeAbsolutePosition(
-            m_BP3Serializer.m_Data, r);
-
-        if (m_BP3Serializer.m_Aggregator.m_IsConsumer)
-        {
-            const BufferSTL &bufferSTL =
-                m_BP3Serializer.m_Aggregator.GetConsumerBuffer(
-                    m_BP3Serializer.m_Data);
-
-            m_FileDataManager.WriteFiles(bufferSTL.m_Buffer.data(),
-                                         bufferSTL.m_Position, transportIndex);
-
-            m_FileDataManager.FlushFiles(transportIndex);
-        }
-
-        m_BP3Serializer.m_Aggregator.WaitAbsolutePosition(r);
-
-        m_BP3Serializer.m_Aggregator.Wait(r);
-        m_BP3Serializer.m_Aggregator.SwapBuffers(r);
-    }
+    m_BP3Serializer.m_Aggregator(m_BP3Serializer.m_Data,
+              [&](const BufferSTL &buffer) {
+                  m_FileDataManager.WriteFiles(buffer.m_Buffer.data(),
+                                               buffer.m_Position,
+                                               transportIndex);
+
+                  m_FileDataManager.FlushFiles(transportIndex);
+              });

     m_BP3Serializer.UpdateOffsetsInMetadata();

This is clearly much simpler and less error-prone to use. It should be a straightforward change, basically just moving the loop over ranks into MPIAggregator::operator() and calling back to a lambda function that will be passed the aggregated buffers one-by-one.

It's still very specialized (really only useful for this one particular use case). But with this interface, it'd be hard to get issues with maintaining internal state, since it's one single function call one can not really jump into the middle of to break things, so the only way to get that issue would be calling it from multiple threads at the same time. And even that could then actually be handled nicely by moving the state to local variables in the implementation that performs the loop.

It's still not general enough to support all kinds of possible communication strategies easily (e.g., MPI_Gatherv wouldn't quite work), but it does at least hide the fact that double buffering is used.

[chuckatkins]

Supersceded by #1369, which is based on this implementation with the member variable plus a few changes.

[chuckatkins]

I don't disagree with your ideas here, but I want to limit the scope to fixing the bug in the most limited way possible given the critical path this code lives in.