communicationManager.hpp

Go to the documentation of this file.

/*
 *   Copyright 2025 Huawei Technologies Co., Ltd.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
 
/*
 * Copyright Huawei Technologies Switzerland AG
 * All rights reserved.
 */
 
#pragma once
 
#include <map>
#include <memory>
#include <mutex>
#include <utility>
#include <vector>
#include <hicr/core/localMemorySlot.hpp>
#include <hicr/core/globalMemorySlot.hpp>
#include <hicr/core/memorySpace.hpp>
#include <hicr/core/definitions.hpp>
#include <hicr/core/exceptions.hpp>
 
namespace HiCR
{
 
class CommunicationManager
{
  public:
 
  using globalKeyMemorySlotPair_t = std::pair<GlobalMemorySlot::globalKey_t, std::shared_ptr<LocalMemorySlot>>;
 
  using globalKeyToMemorySlotMap_t = std::map<GlobalMemorySlot::globalKey_t, std::shared_ptr<GlobalMemorySlot>>;
 
  using globalMemorySlotTagKeyMap_t = std::map<GlobalMemorySlot::tag_t, globalKeyToMemorySlotMap_t>;
 
  virtual ~CommunicationManager() = default;
 
  virtual void lock() { _mutex.lock(); };
 
  virtual void unlock() { _mutex.unlock(); };
 
  __INLINE__ void exchangeGlobalMemorySlots(GlobalMemorySlot::tag_t tag, const std::vector<globalKeyMemorySlotPair_t> &memorySlots)
  {
    // Calling internal implementation of this function
    exchangeGlobalMemorySlotsImpl(tag, memorySlots);
  }
 
  __INLINE__ std::shared_ptr<GlobalMemorySlot> getGlobalMemorySlot(GlobalMemorySlot::tag_t tag, GlobalMemorySlot::globalKey_t globalKey)
  {
    auto globalSlotImpl = getGlobalMemorySlotImpl(tag, globalKey);
 
    if (globalSlotImpl != nullptr) { return globalSlotImpl; }
    else
    {
      // If the requested tag and key are not found, return empty storage
      if (_globalMemorySlotTagKeyMap.contains(tag) == false) { HICR_THROW_LOGIC("Requesting a global memory slot for a tag (%lu) that has not been registered.", tag); }
      if (_globalMemorySlotTagKeyMap.at(tag).contains(globalKey) == false)
      {
        for (const auto &elem : _globalMemorySlotTagKeyMap.at(tag)) { printf("For Tag %lu: Key %lu\n", tag, elem.first); }
        printf("But tag %lu does not contain globalKey = %lu\n", tag, globalKey);
        HICR_THROW_LOGIC("Requesting a global memory slot for a  global key (%lu) not registered within the tag (%lu).", globalKey, tag);
      }
 
      // Getting requested memory slot
      auto value = _globalMemorySlotTagKeyMap.at(tag).at(globalKey);
 
      // Returning value
      return value;
    }
  }
 
  virtual uint8_t *serializeGlobalMemorySlot(const std::shared_ptr<HiCR::GlobalMemorySlot> &globalSlot) const
  {
    HICR_THROW_LOGIC("Trying to serialize a global memory slot; this is not supported in this backend\n");
    return nullptr;
  }
 
  virtual std::shared_ptr<GlobalMemorySlot> deserializeGlobalMemorySlot(uint8_t *buffer, GlobalMemorySlot::tag_t tag)
  {
    HICR_THROW_LOGIC("Trying to deserialize a global memory slot; this is not supported in this backend\n");
    return nullptr;
  }
 
  virtual std::shared_ptr<GlobalMemorySlot> promoteLocalMemorySlot(const std::shared_ptr<LocalMemorySlot> &localMemorySlot, GlobalMemorySlot::tag_t tag)
  {
    HICR_THROW_LOGIC("This backend does not support one-sided promotion of local memory slots to global");
    return nullptr;
  }
 
  __INLINE__ void deregisterGlobalMemorySlot(const std::shared_ptr<GlobalMemorySlot> &memorySlot)
  {
    // Getting memory slot global information
    const auto memorySlotTag       = memorySlot->getGlobalTag();
    const auto memorySlotGlobalKey = memorySlot->getGlobalKey();
 
    // Checking whether the memory slot is correctly registered as global
    if (_globalMemorySlotTagKeyMap.contains(memorySlotTag) == false)
    {
      HICR_THROW_LOGIC("Attempting to de-register a global memory slot but its tag/key pair is not registered in this backend");
    }
 
    // Removing memory slot from the global memory slot map
    _globalMemorySlotTagKeyMap.at(memorySlotTag).erase(memorySlotGlobalKey);
    deregisterGlobalMemorySlotImpl(memorySlot);
  }
 
  __INLINE__ void destroyGlobalMemorySlot(const std::shared_ptr<GlobalMemorySlot> &memorySlot)
  {
    auto tag = memorySlot->getGlobalTag();
    // Implicit creation of the tag entry if it doesn't exist is desired here
    _globalMemorySlotsToDestroyPerTag[tag].push_back(memorySlot);
  }
 
  virtual void destroyPromotedGlobalMemorySlot(const std::shared_ptr<GlobalMemorySlot> &memorySlot)
  {
    HICR_THROW_LOGIC("This backend does not support promoted global memory slots.");
  }
 
  __INLINE__ void queryMemorySlotUpdates(std::shared_ptr<LocalMemorySlot> memorySlot)
  {
    // Getting value by copy
    queryMemorySlotUpdatesImpl(std::move(memorySlot));
  }
 
  __INLINE__ void memcpy(const std::shared_ptr<LocalMemorySlot> &destination, size_t dst_offset, const std::shared_ptr<LocalMemorySlot> &source, size_t src_offset, size_t size)
  {
    // Getting slot sizes. This operation is thread-safe
    const auto srcSize = source->getSize();
    const auto dstSize = destination->getSize();
 
    // Making sure the memory slots exist and is not null
    const auto actualSrcSize = size + src_offset;
    const auto actualDstSize = size + dst_offset;
 
    // Checking size doesn't exceed slot size
    if (actualSrcSize > srcSize)
      HICR_THROW_RUNTIME("Memcpy size (%lu) + offset (%lu) = (%lu) exceeds source slot (%p) capacity (%lu).", size, src_offset, actualSrcSize, source->getPointer(), srcSize);
 
    // Checking size doesn't exceed slot size
    if (actualDstSize > dstSize)
      HICR_THROW_RUNTIME(
        "Memcpy size (%lu) + offset (%lu) = (%lu) exceeds destination slot (%p) capacity (%lu).", size, dst_offset, actualDstSize, destination->getPointer(), dstSize);
 
    // To enable concurrent memcpy operations, the implementation is executed outside the mutex zone
    // This means that the developer needs to make sure that the implementation is concurrency-safe,
    // and try not to access any of the internal Backend class fields without proper mutex locking
 
    // Now calling internal memcpy function to give us a function that satisfies the operation
    memcpyImpl(destination, dst_offset, source, src_offset, size);
  }
 
  __INLINE__ void memcpy(const std::shared_ptr<GlobalMemorySlot> &destination, size_t dst_offset, const std::shared_ptr<LocalMemorySlot> &source, size_t src_offset, size_t size)
  {
    // Getting slot sizes. This operation is thread-safe
    const auto srcSize = source->getSize();
 
    // Making sure the memory slots exist and is not null
    const auto actualSrcSize = size + src_offset;
 
    // Checking size doesn't exceed slot size
    if (actualSrcSize > srcSize)
      HICR_THROW_RUNTIME("Memcpy size (%lu) + offset (%lu) = (%lu) exceeds source slot (%p) capacity (%lu).", size, src_offset, actualSrcSize, source->getPointer(), srcSize);
 
    // Now calling internal memcpy function to give us a function that satisfies the operation
    memcpyImpl(destination, dst_offset, source, src_offset, size);
  }
 
  __INLINE__ void memcpy(const std::shared_ptr<LocalMemorySlot> &destination, size_t dst_offset, const std::shared_ptr<GlobalMemorySlot> &source, size_t src_offset, size_t size)
  {
    // Getting slot sizes. This operation is thread-safe
    const auto dstSize = destination->getSize();
 
    // Making sure the memory slots exist and is not null
    const auto actualDstSize = size + dst_offset;
 
    // Checking size doesn't exceed slot size
    if (actualDstSize > dstSize)
      HICR_THROW_RUNTIME(
        "Memcpy size (%lu) + offset (%lu) = (%lu) exceeds destination slot (%p) capacity (%lu).", size, dst_offset, actualDstSize, destination->getPointer(), dstSize);
 
    // Now calling internal memcpy function to give us a function that satisfies the operation
    memcpyImpl(destination, dst_offset, source, src_offset, size);
  }
 
  __INLINE__ void fence(GlobalMemorySlot::tag_t tag)
  {
    lock();
    // Now call the proper fence, as implemented by the backend
    fenceImpl(tag);
 
    // Clear the memory slots to destroy
    _globalMemorySlotsToDestroyPerTag[tag].clear();
    _globalMemorySlotsToDestroyPerTag.erase(tag);
    unlock();
  }
 
  __INLINE__ void fence(const std::shared_ptr<LocalMemorySlot> &slot, size_t expectedSent, size_t expectedRecvd)
  {
    // To enable concurrent fence operations, the implementation is executed outside the mutex zone
    // This means that the developer needs to make sure that the implementation is concurrency-safe,
    // and try not to access any of the internal Backend class fields without proper mutex locking
 
    // Now call the proper fence, as implemented by the backend
    fenceImpl(slot, expectedSent, expectedRecvd);
  }
 
  __INLINE__ void fence(const std::shared_ptr<GlobalMemorySlot> &slot, size_t expectedSent, size_t expectedRecvd)
  {
    // To enable concurrent fence operations, the implementation is executed outside the mutex zone
    // This means that the developer needs to make sure that the implementation is concurrency-safe,
    // and try not to access any of the internal Backend class fields without proper mutex locking
 
    // Now call the proper fence, as implemented by the backend
    fenceImpl(slot, expectedSent, expectedRecvd);
  }
 
  __INLINE__ bool acquireGlobalLock(const std::shared_ptr<GlobalMemorySlot> &memorySlot)
  {
    // Getting memory slot global information
    const auto memorySlotTag       = memorySlot->getGlobalTag();
    const auto memorySlotGlobalKey = memorySlot->getGlobalKey();
 
    // Checking whether the memory slot is correctly registered as global
    if (_globalMemorySlotTagKeyMap.contains(memorySlotTag) == false)
      HICR_THROW_LOGIC("Attempting to lock a global memory slot but its tag/key pair is not registered in this backend");
 
    // Checking whether the memory slot is correctly registered as global
    if (_globalMemorySlotTagKeyMap.at(memorySlotTag).contains(memorySlotGlobalKey) == false)
      HICR_THROW_LOGIC("Attempting to lock a global memory slot but its tag/key pair is not registered in this backend");
 
    // Calling internal implementation
    return acquireGlobalLockImpl(memorySlot);
  }
 
  __INLINE__ void releaseGlobalLock(const std::shared_ptr<GlobalMemorySlot> &memorySlot)
  {
    // Getting memory slot global information
    const auto memorySlotTag       = memorySlot->getGlobalTag();
    const auto memorySlotGlobalKey = memorySlot->getGlobalKey();
 
    // Checking whether the memory slot is correctly registered as global
    if (_globalMemorySlotTagKeyMap.contains(memorySlotTag) == false)
    {
      HICR_THROW_LOGIC("Attempting to release a global memory slot but its tag/key pair is not registered in this backend");
    }
 
    // Checking whether the memory slot is correctly registered as global
    if (_globalMemorySlotTagKeyMap.at(memorySlotTag).contains(memorySlotGlobalKey) == false)
    {
      HICR_THROW_LOGIC("Attempting to release a global memory slot but its tag/key pair is not registered in this backend");
    }
 
    // Calling internal implementation
    releaseGlobalLockImpl(memorySlot);
  }
 
  __INLINE__ virtual void flushSent() {}
 
  __INLINE__ virtual void flushReceived() {}
 
  protected:
 
  __INLINE__ void registerGlobalMemorySlot(const std::shared_ptr<GlobalMemorySlot> &memorySlot)
  {
    // Getting memory slot information
    const auto tag       = memorySlot->getGlobalTag();
    const auto globalKey = memorySlot->getGlobalKey();
 
    // Adding memory slot to the global map (based on tag and key)
    _globalMemorySlotTagKeyMap[tag][globalKey] = memorySlot;
  }
 
  virtual std::shared_ptr<GlobalMemorySlot> getGlobalMemorySlotImpl(GlobalMemorySlot::tag_t tag, GlobalMemorySlot::globalKey_t globalKey) = 0;
 
  virtual void exchangeGlobalMemorySlotsImpl(GlobalMemorySlot::tag_t tag, const std::vector<globalKeyMemorySlotPair_t> &memorySlots) = 0;
 
  virtual void queryMemorySlotUpdatesImpl(std::shared_ptr<LocalMemorySlot> memorySlot) = 0;
 
  virtual void deregisterGlobalMemorySlotImpl(const std::shared_ptr<GlobalMemorySlot> &memorySlot) {};
 
  virtual void destroyGlobalMemorySlotImpl(std::shared_ptr<GlobalMemorySlot> memorySlot) = 0;
 
  virtual void memcpyImpl(const std::shared_ptr<LocalMemorySlot> &destination, size_t dst_offset, const std::shared_ptr<LocalMemorySlot> &source, size_t src_offset, size_t size)
  {
    HICR_THROW_LOGIC("Local->Local memcpy operations are unsupported by the given backend");
  };
 
  virtual void memcpyImpl(const std::shared_ptr<GlobalMemorySlot> &destination, size_t dst_offset, const std::shared_ptr<LocalMemorySlot> &source, size_t src_offset, size_t size)
  {
    HICR_THROW_LOGIC("Local->Global memcpy operations are unsupported by the given backend");
  };
 
  virtual void memcpyImpl(const std::shared_ptr<LocalMemorySlot> &destination, size_t dst_offset, const std::shared_ptr<GlobalMemorySlot> &source, size_t src_offset, size_t size)
  {
    HICR_THROW_LOGIC("Global->Local memcpy operations are unsupported by the given backend");
  };
 
  virtual void fenceImpl(const std::shared_ptr<LocalMemorySlot> &slot, size_t expectedSent, size_t expectedRcvd) {}
  virtual void fenceImpl(const std::shared_ptr<GlobalMemorySlot> &slot, size_t expectedSent, size_t expectedRcvd) {}
  virtual void fenceImpl(GlobalMemorySlot::tag_t tag) = 0;
 
  virtual bool acquireGlobalLockImpl(std::shared_ptr<GlobalMemorySlot> memorySlot) = 0;
 
  virtual void releaseGlobalLockImpl(std::shared_ptr<GlobalMemorySlot> memorySlot) = 0;
 
  [[nodiscard]] __INLINE__ auto &getGlobalMemorySlotsToDestroyPerTag() { return _globalMemorySlotsToDestroyPerTag; }
 
  [[nodiscard]] __INLINE__ auto &getGlobalMemorySlotTagKeyMap() { return _globalMemorySlotTagKeyMap; }
 
  protected:
 
  __INLINE__ void increaseMessageRecvCounter(HiCR::LocalMemorySlot &memorySlot) noexcept { memorySlot.increaseMessagesRecv(); }
 
  __INLINE__ void increaseMessageSentCounter(HiCR::LocalMemorySlot &memorySlot) noexcept { memorySlot.increaseMessagesSent(); }
 
  __INLINE__ void setMessagesRecv(HiCR::LocalMemorySlot &memorySlot, const size_t count) noexcept { memorySlot.setMessagesRecv(count); }
 
  __INLINE__ void setMessagesSent(HiCR::LocalMemorySlot &memorySlot, const size_t count) noexcept { memorySlot.setMessagesSent(count); }
 
  private:
 
  std::mutex _mutex;
 
  globalMemorySlotTagKeyMap_t _globalMemorySlotTagKeyMap;
 
  std::map<GlobalMemorySlot::tag_t, std::vector<std::shared_ptr<GlobalMemorySlot>>> _globalMemorySlotsToDestroyPerTag;
};
 
} // namespace HiCR