Namespace faabric::mpi

namespace mpi

Typedefs

typedef faabric::util::FixedCapacityQueue<MpiMessage> InMemoryMpiQueue

Enums

enum MpiMessageType

Values:

enumerator NORMAL

enumerator BARRIER_JOIN

enumerator BARRIER_DONE

enumerator SCATTER

enumerator GATHER

enumerator ALLGATHER

enumerator REDUCE

enumerator SCAN

enumerator ALLREDUCE

enumerator ALLTOALL

enumerator ALLTOALL_PACKED

enumerator SENDRECV

enumerator BROADCAST

enumerator UNACKED_MPI_MESSAGE

enumerator HANDSHAKE

Functions

inline size_t payloadSize(const MpiMessage &msg)

inline size_t msgSize(const MpiMessage &msg)

void serializeMpiMsg(std::vector<uint8_t> &buffer, const MpiMessage &msg)

void parseMpiMsg(const std::vector<uint8_t> &bytes, MpiMessage *msg)

std::vector<MpiMessage> getMpiMockedMessages(int sendRank)

MpiWorldRegistry &getMpiWorldRegistry()

static int32_t getAsyncRequestId(int sendRank, int recvRank, bool isSend)

static std::tuple<int, int, bool> getRanksFromRequestId(int32_t requestId)

void checkSendRecvMatch(faabric_datatype_t *sendType, int sendCount, faabric_datatype_t *recvType, int recvCount)

Variables

static std::mutex mockMutex

static std::map<int, std::vector<MpiMessage>> mpiMockedMessages

static thread_local MpiRankState rankState

const uint8_t iSendMagic = 0xFF

const uint8_t iRecvMagic = 0x00

class MpiContext

Public Functions

MpiContext()

int createWorld(faabric::Message &msg)

void joinWorld(faabric::Message &msg)

bool getIsMpi() const

int getRank() const

int getWorldId() const

struct MpiMessage

Public Members

int32_t id

int32_t worldId

int32_t sendRank

int32_t recvRank

int32_t typeSize

int32_t count

int32_t requestId

MpiMessageType messageType

void *buffer

struct MpiRankState

Public Functions

inline void reset()

Public Members

int msgCount = 1

faabric::Message *msg = nullptr

std::vector<std::unique_ptr<std::list<MpiMessage>>> unackedMessageBuffers

std::unique_ptr<faabric::util::FaabricCpuSet> pinnedCpu

std::vector<std::unique_ptr<faabric::transport::tcp::SendSocket>> sendSockets

std::unique_ptr<faabric::transport::tcp::RecvSocket> recvSocket

std::vector<int> recvConnPool

class MpiWorld

Public Functions

MpiWorld()

void create(faabric::Message &call, int newId, int newSize)

void broadcastHostsToRanks()

void initialiseFromMsg(faabric::Message &msg)

void initialiseRankFromMsg(faabric::Message &msg)

std::string getHostForRank(int rank)

std::string getUser()

std::string getFunction()

int getId() const

int getSize() const

bool destroy()

void getCartesianRank(int rank, int maxDims, const int *dims, int *periods, int *coords)

void getRankFromCoords(int *rank, int *coords)

void shiftCartesianCoords(int rank, int direction, int disp, int *source, int *destination)

void send(int sendRank, int recvRank, const uint8_t *buffer, faabric_datatype_t *dataType, int count, MpiMessageType messageType = MpiMessageType::NORMAL)

int isend(int sendRank, int recvRank, const uint8_t *buffer, faabric_datatype_t *dataType, int count, MpiMessageType messageType = MpiMessageType::NORMAL)

void broadcast(int rootRank, int thisRank, uint8_t *buffer, faabric_datatype_t *dataType, int count, MpiMessageType messageType = MpiMessageType::NORMAL)

void recv(int sendRank, int recvRank, uint8_t *buffer, faabric_datatype_t *dataType, int count, MPI_Status *status, MpiMessageType messageType = MpiMessageType::NORMAL)

int irecv(int sendRank, int recvRank, uint8_t *buffer, faabric_datatype_t *dataType, int count, MpiMessageType messageType = MpiMessageType::NORMAL)

void awaitAsyncRequest(int requestId)

void sendRecv(uint8_t *sendBuffer, int sendcount, faabric_datatype_t *sendDataType, int sendRank, uint8_t *recvBuffer, int recvCount, faabric_datatype_t *recvDataType, int recvRank, int myRank, MPI_Status *status)

void scatter(int sendRank, int recvRank, const uint8_t *sendBuffer, faabric_datatype_t *sendType, int sendCount, uint8_t *recvBuffer, faabric_datatype_t *recvType, int recvCount)

void gather(int sendRank, int recvRank, const uint8_t *sendBuffer, faabric_datatype_t *sendType, int sendCount, uint8_t *recvBuffer, faabric_datatype_t *recvType, int recvCount)

void allGather(int rank, const uint8_t *sendBuffer, faabric_datatype_t *sendType, int sendCount, uint8_t *recvBuffer, faabric_datatype_t *recvType, int recvCount)

void reduce(int sendRank, int recvRank, uint8_t *sendBuffer, uint8_t *recvBuffer, faabric_datatype_t *datatype, int count, faabric_op_t *operation)

void allReduce(int rank, uint8_t *sendBuffer, uint8_t *recvBuffer, faabric_datatype_t *datatype, int count, faabric_op_t *operation)

void op_reduce(faabric_op_t *operation, faabric_datatype_t *datatype, int count, uint8_t *inBuffer, uint8_t *resultBuffer)

void scan(int rank, uint8_t *sendBuffer, uint8_t *recvBuffer, faabric_datatype_t *datatype, int count, faabric_op_t *operation)

void allToAll(int rank, uint8_t *sendBuffer, faabric_datatype_t *sendType, int sendCount, uint8_t *recvBuffer, faabric_datatype_t *recvType, int recvCount)

void probe(int sendRank, int recvRank, MPI_Status *status)

void barrier(int thisRank)

std::shared_ptr<InMemoryMpiQueue> getLocalQueue(int sendRank, int recvRank)

long getLocalQueueSize(int sendRank, int recvRank)

void overrideHost(const std::string &newHost)

double getWTime()

void prepareMigration(int newGroupId, int thisRank, bool thisRankMustMigrate)

class MpiWorldRegistry

Public Functions

MpiWorldRegistry() = default

MpiWorld &createWorld(faabric::Message &msg, int worldId, std::string hostOverride = "")

MpiWorld &getOrInitialiseWorld(faabric::Message &msg)

MpiWorld &getWorld(int worldId)

bool worldExists(int worldId)

void clearWorld(int worldId)

void clear()