mirror of
https://github.com/microsoft/mscclpp.git
synced 2026-05-12 09:17:06 +00:00
828be48b21
This PR implements and closes #137. The new `Endpoint` and `Context` classes expose the connection establishing functionality from `Communicator`, which now is only responsible for tying together the bootstrapper with a context. The largest breaking change here is that `Communicator.connectOnSetup(...)` now returns the `Connection` wrapped inside a `NonblockingFuture`. This is because with the way `Context` is implemented a `Connection` is now fully initialized on construction. Some smaller breaking API changes from this change are that `RegisteredMemory` no longer has a `rank()` function (as there maybe no concept of rank), and similarly `Connection` has no `remoteRank()` and `tag()` functions. The latter are replaced by `remoteRankOf` and `tagOf` functions in `Communicator`. A new `EndpointConfig` class is introduced to avoid duplication of the IB configuration parameters in the APIs of `Context` and `Communicator`. The usual usage pattern of just passing in a `Transport` still works due to an implicit conversion into `EndpointConfig`. Miscellaneous changes: -Cleans up how the PIMPL pattern is applied by making both the `Impl` struct and the `pimpl_` pointers private for all relevant classes in the core API. -Enables ctest to be run from the build root directory.
60 lines
2.1 KiB
C++
60 lines
2.1 KiB
C++
// Copyright (c) Microsoft Corporation.
|
|
// Licensed under the MIT license.
|
|
|
|
#include "context.hpp"
|
|
|
|
#include "api.h"
|
|
#include "connection.hpp"
|
|
#include "debug.h"
|
|
#include "endpoint.hpp"
|
|
#include "registered_memory.hpp"
|
|
|
|
namespace mscclpp {
|
|
|
|
Context::Impl::Impl() : ipcStream_(cudaStreamNonBlocking) {}
|
|
|
|
IbCtx* Context::Impl::getIbContext(Transport ibTransport) {
|
|
// Find IB context or create it
|
|
auto it = ibContexts_.find(ibTransport);
|
|
if (it == ibContexts_.end()) {
|
|
auto ibDev = getIBDeviceName(ibTransport);
|
|
ibContexts_[ibTransport] = std::make_unique<IbCtx>(ibDev);
|
|
return ibContexts_[ibTransport].get();
|
|
} else {
|
|
return it->second.get();
|
|
}
|
|
}
|
|
|
|
MSCCLPP_API_CPP Context::Context() : pimpl_(std::make_unique<Impl>()) {}
|
|
|
|
MSCCLPP_API_CPP Context::~Context() = default;
|
|
|
|
MSCCLPP_API_CPP RegisteredMemory Context::registerMemory(void* ptr, size_t size, TransportFlags transports) {
|
|
return RegisteredMemory(std::make_shared<RegisteredMemory::Impl>(ptr, size, transports, *pimpl_));
|
|
}
|
|
|
|
MSCCLPP_API_CPP Endpoint Context::createEndpoint(EndpointConfig config) {
|
|
return Endpoint(std::make_shared<Endpoint::Impl>(config, *pimpl_));
|
|
}
|
|
|
|
MSCCLPP_API_CPP std::shared_ptr<Connection> Context::connect(Endpoint localEndpoint, Endpoint remoteEndpoint) {
|
|
std::shared_ptr<Connection> conn;
|
|
if (localEndpoint.transport() == Transport::CudaIpc) {
|
|
if (remoteEndpoint.transport() != Transport::CudaIpc) {
|
|
throw mscclpp::Error("Local transport is CudaIpc but remote is not", ErrorCode::InvalidUsage);
|
|
}
|
|
conn = std::make_shared<CudaIpcConnection>(localEndpoint, remoteEndpoint, pimpl_->ipcStream_);
|
|
} else if (AllIBTransports.has(localEndpoint.transport())) {
|
|
if (!AllIBTransports.has(remoteEndpoint.transport())) {
|
|
throw mscclpp::Error("Local transport is IB but remote is not", ErrorCode::InvalidUsage);
|
|
}
|
|
conn = std::make_shared<IBConnection>(localEndpoint, remoteEndpoint, *this);
|
|
} else {
|
|
throw mscclpp::Error("Unsupported transport", ErrorCode::InternalError);
|
|
}
|
|
pimpl_->connections_.push_back(conn);
|
|
return conn;
|
|
}
|
|
|
|
} // namespace mscclpp
|