Add Active Messages support #48
Conversation
There was a problem hiding this comment.
Thanks @wence- for looking at this, apologies for not addressing it earlier but somehow I completely missed the notification for this PR. I've addressed/responded to your comments now, please have a look when you've got the chance.
|
|
||
| void RecvAmMessage::callback(void* request, ucs_status_t status) | ||
| { | ||
| _request->_buffer = _buffer; |
There was a problem hiding this comment.
Can
_requestbe used for something else during the lifetime ofthis? if not, why not set up the_request->_bufferpointer in the constructor ofRecvAmMessage, if yes, then this seems potentially racy.
It shouldn't be used for anything else. The reason I wrote the code in this way is to prevent the user from calling getRecvBuffer before isCompleted() == true && getStatus() == UCS_OK and get a proper Buffer that has invalid data. Do you think it would be better to handle it differently?
|
|
||
| void RecvAmMessage::callback(void* request, ucs_status_t status) | ||
| { | ||
| _request->_buffer = _buffer; |
There was a problem hiding this comment.
That's a good idea, done in 6eb5f95.
|
Thanks, not sure what went on in the tests though |
For some reason the `posix` transport isn't available in CI, thus replace it with `self`, given the exact transport is not the object of `test_check_transport`.
This reverts commit d7dcea0.
| // delayed to allow the worker progress thread to set its status, and more | ||
| // importantly the Python future later on, so that we don't need the GIL here. | ||
| req->_worker->registerDelayedSubmission( | ||
| req, std::bind(std::mem_fn(&Request::populateDelayedSubmission), req.get())); |
There was a problem hiding this comment.
Ah, this needs to be moved from the constructor because we need the shared_ptr to this which isn't available.
There was a problem hiding this comment.
Yes, it's not the friendliest design, but it is the best we can do here (I think).
| @@ -35,26 +35,25 @@ void DelayedSubmissionCollection::process() | |||
| toProcess = std::move(_collection); | |||
There was a problem hiding this comment.
What does std::move do to _collection? Does it just drain the vector, leaving it "as-if" it were empty again? I guess so.
There was a problem hiding this comment.
Exactly, that is the standard behavior for the move constructor and operator.
Required until progressing the worker is not necessary within the call anymore.
Co-authored-by: Lawrence Mitchell <[email protected]>
Follow the default of other tests for now, relying on the default progress mode or that specified via environment variables.
| @@ -35,26 +35,25 @@ void DelayedSubmissionCollection::process() | |||
| toProcess = std::move(_collection); | |||
There was a problem hiding this comment.
Exactly, that is the standard behavior for the move constructor and operator.
| // delayed to allow the worker progress thread to set its status, and more | ||
| // importantly the Python future later on, so that we don't need the GIL here. | ||
| req->_worker->registerDelayedSubmission( | ||
| req, std::bind(std::mem_fn(&Request::populateDelayedSubmission), req.get())); |
There was a problem hiding this comment.
Yes, it's not the friendliest design, but it is the best we can do here (I think).
|
All tests pass now, I'll leave this to be merged Monday in case you have any last comments @wence- , thanks again for reviewing! |
|
Thanks @wence- and @robertmaynard for reviews and approvals. |
|
/merge |
Introduce new
ucxx::RequestAmclass to allow transferring via the Active Messages UCX API.