Ports.cpp

Go to the documentation of this file.

//===- Ports.cpp - ESI communication channels  -------------------------===//
//
// Part of the LLVM Project, under the Apache License v2.0 with LLVM Exceptions.
// See https://llvm.org/LICENSE.txt for license information.
// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
//
//===----------------------------------------------------------------------===//
//
// DO NOT EDIT!
// This file is distributed as part of an ESI package. The source for this file
// should always be modified within CIRCT (lib/dialect/ESI/runtime/cpp/).
//
//===----------------------------------------------------------------------===//
 
#include "esi/Ports.h"
#include "esi/Types.h"
 
#include <algorithm>
#include <chrono>
#include <cstring>
#include <map>
#include <stdexcept>
 
using namespace esi;
 
ChannelPort::ChannelPort(const Type *type) {
  if (auto chanType = dynamic_cast<const ChannelType *>(type))
    type = chanType->getInner();
  auto translationType = dynamic_cast<const WindowType *>(type);
  if (translationType) {
    this->type = translationType->getIntoType();
    this->translationInfo = std::make_unique<TranslationInfo>(translationType);
  } else {
    this->type = type;
    this->translationInfo = nullptr;
  }
}
 
BundlePort::BundlePort(AppID id, const BundleType *type, PortMap channels)
    : id(id), type(type), channels(channels) {}
 
WriteChannelPort &BundlePort::getRawWrite(const std::string &name) const {
  auto f = channels.find(name);
  if (f == channels.end())
    throw std::runtime_error("Channel '" + name + "' not found");
  auto *write = dynamic_cast<WriteChannelPort *>(&f->second);
  if (!write)
    throw std::runtime_error("Channel '" + name + "' is not a write channel");
  return *write;
}
 
ReadChannelPort &BundlePort::getRawRead(const std::string &name) const {
  auto f = channels.find(name);
  if (f == channels.end())
    throw std::runtime_error("Channel '" + name + "' not found");
  auto *read = dynamic_cast<ReadChannelPort *>(&f->second);
  if (!read)
    throw std::runtime_error("Channel '" + name + "' is not a read channel");
  return *read;
}
 
void ReadChannelPort::resetTranslationState() {
  nextFrameIndex = 0;
  accumulatingListData = false;
  translationBuffer.clear();
  listDataBuffer.clear();
  translatedMessage.reset();
}
 
void ReadChannelPort::disconnect() {
  {
    std::unique_lock<std::mutex> lock(callbackMutex);
    // Revoke the callback under the same mutex used by invokeCallback(). New
    // deliveries need to observe the disconnected state and the callback
    // nullification at the same time.
    callback = nullptr;
    mode = Mode::Disconnected;
    // Wait for any callback which already snapped the old function to finish
    // before clearing per-connection state below.
    callbackCv.wait(lock, [this]() { return activeCallbacks == 0; });
  }
 
  pollingState.reset();
  resetTranslationState();
}
 
bool ReadChannelPort::invokeCallback(
    std::unique_ptr<SegmentedMessageData> &msg) {
  ReadCallback activeCallback;
  {
    std::lock_guard<std::mutex> lock(callbackMutex);
    // Check the disconnected state and snapshot the callback while holding the
    // same mutex as disconnect(). That prevents disconnect() from clearing the
    // callback after we decide to invoke it but before we've retained our own
    // safe local copy.
    if (mode == Mode::Disconnected)
      return false;
    assert(callback && "Callback should be set in non-disconnected mode");
    activeCallback = callback;
    // Count only callbacks which successfully captured a callable target.
    // disconnect() waits on this count so it can safely tear down state which
    // the callback path may still reference.
    ++activeCallbacks;
  }
 
  // Release the in-flight slot on every exit path, including exceptions from
  // the callback itself.
  auto finish = [this]() {
    std::lock_guard<std::mutex> lock(callbackMutex);
    assert(activeCallbacks > 0 && "Callback count underflow");
    --activeCallbacks;
    if (activeCallbacks == 0)
      callbackCv.notify_all();
  };
 
  // For zero-width port types (e.g. `VoidType`), transports send a single
  // placeholder byte to satisfy the "every message is at least one byte"
  // invariant. Only strip the payload when it matches that encoding so
  // malformed transport data is surfaced instead of silently discarded. The
  // matching pad happens in `WriteChannelPort::maybePadEmptyMessage`.
  if (msg && type && type->getBitWidth() == 0) {
    if (msg->totalSize() == 1) {
      msg = std::make_unique<MessageData>();
    } else {
      throw std::runtime_error(
          "zero-width message must use a single-byte placeholder payload");
    }
  }
 
  try {
    bool consumed = activeCallback(msg);
    finish();
    return consumed;
  } catch (...) {
    finish();
    throw;
  }
}
 
void ReadChannelPort::connect(ReadCallback callback,
                              const ConnectOptions &options) {
  if (mode != Mode::Disconnected)
    throw std::runtime_error("Channel already connected");
 
  resetTranslationState();
 
  if (translationInfo)
    translationInfo->precomputeFrameInfo();
 
  if (options.translateMessage && translationInfo) {
    translationInfo->requireTranslationSupported();
    this->callback = [this, cb = std::move(callback)](
                         std::unique_ptr<SegmentedMessageData> &data) {
      if (!translatedMessage) {
        MessageData flat = data->toMessageData();
        if (!translateIncoming(flat))
          return true;
        translatedMessage =
            std::make_unique<MessageData>(std::move(translationBuffer));
      }
 
      if (!cb(translatedMessage))
        return false;
 
      translatedMessage.reset();
      return true;
    };
  } else {
    this->callback = callback;
  }
  connectImpl(options);
  mode = Mode::Callback;
}
 
void ReadChannelPort::connect(FlatReadCallback callback,
                              const ConnectOptions &options) {
  connect(
      [cb = std::move(callback)](std::unique_ptr<SegmentedMessageData> &data)
          -> bool { return cb(data->toMessageData()); },
      options);
}
 
void ReadChannelPort::connect(const ConnectOptions &options) {
  if (mode != Mode::Disconnected)
    throw std::runtime_error("Channel already connected");
 
  if (options.bufferSize.has_value())
    maxDataQueueMsgs = options.bufferSize.value();
  pollingState.emplace(maxDataQueueMsgs);
 
  resetTranslationState();
 
  if (translationInfo)
    translationInfo->precomputeFrameInfo();
 
  bool translate = options.translateMessage && translationInfo;
  if (translate)
    translationInfo->requireTranslationSupported();
  this->callback = [this,
                    translate](std::unique_ptr<SegmentedMessageData> &msg) {
    MessageData data;
    if (translate) {
      if (!translatedMessage) {
        MessageData flat = msg->toMessageData();
        if (!translateIncoming(flat))
          return true;
        translatedMessage =
            std::make_unique<MessageData>(std::move(translationBuffer));
      }
      data = translatedMessage->toMessageData();
    } else {
      data = msg->toMessageData();
    }
 
    if (!pollingState->enqueue(data))
      return false;
 
    translatedMessage.reset();
    return true;
  };
  connectImpl(options);
  mode = Mode::Polling;
}
 
std::future<MessageData> ReadChannelPort::readAsync() {
  if (mode == Mode::Callback)
    throw std::runtime_error(
        "Cannot read from a callback channel. Call `connect()` without a "
        "callback specified to use polling mode.");
  if (mode == Mode::Disconnected)
    throw std::runtime_error(
        "Cannot read from a disconnected channel. `connect()` the channel "
        "without a callback before calling `readAsync()`.");
 
  assert(mode == Mode::Polling && "Channel must be in polling mode to read");
  assert(pollingState && "Polling state should be initialized in polling mode");
 
  return pollingState->readAsync();
}
 
//===----------------------------------------------------------------------===//
// Window translation support
//===----------------------------------------------------------------------===//
 
void ChannelPort::TranslationInfo::requireTranslationSupported() const {
  // Reject serial-encoded windows: any field with bulkCountWidth > 0 indicates
  // serial (bulk) encoding, which the translator does not yet support. Surface
  // this clearly at connect()-time rather than silently buffering forever in
  // translateIncoming().
  for (const auto &frame : windowType->getFrames()) {
    for (const auto &field : frame.fields) {
      if (field.bulkCountWidth > 0)
        throw std::runtime_error(
            "Window translation for serial (bulk) list encoding is not "
            "implemented (window '" +
            windowType->getName() + "', frame '" + frame.name + "', field '" +
            field.name +
            "'). Connect the port with ConnectOptions{translateMessage=false} "
            "and decode the frames manually.");
    }
  }
}
 
void ChannelPort::TranslationInfo::precomputeFrameInfo() {
  const Type *intoType = windowType->getIntoType();
 
  const StructType *intoStruct = dynamic_cast<const StructType *>(intoType);
  if (!intoStruct)
    throw std::runtime_error(
        "Window intoType must be a struct for translation");
 
  const auto &intoFields = intoStruct->getFields();
 
  // Build a map from field name to (offset, type, isListField).
  // For list fields, the offset is where the list_length field will be stored,
  // and we also track the element type.
  struct FieldInfo {
    size_t offset;
    const Type *type;
    bool isList;
    const Type *listElementType; // Only valid if isList is true
    size_t listElementSize;      // Only valid if isList is true
  };
  std::map<std::string, FieldInfo> fieldMap;
  size_t currentOffset = 0;
  hasListField = false;
 
  auto processField = [&](const std::string &name, const Type *fieldType) {
    auto *listType = dynamic_cast<const ListType *>(fieldType);
    if (listType) {
      hasListField = true;
      // For list fields in the intoType:
      // - We store a list_length (size_t / 8 bytes) followed by the list data
      // - The offset here is where the list_length goes
      const Type *elemType = listType->getElementType();
      std::ptrdiff_t elemBits = elemType->getBitWidth();
      if (elemBits < 0)
        throw std::runtime_error(
            "Cannot translate list with dynamically-sized element type: " +
            name);
      if (elemBits == 0)
        throw std::runtime_error(
            "Cannot translate list with zero-width element type: " + name);
      if (elemBits % 8 != 0)
        throw std::runtime_error(
            "Cannot translate list element with non-byte-aligned size: " +
            name);
      size_t elemBytes = (static_cast<size_t>(elemBits) + 7) / 8;
      fieldMap[name] = {currentOffset, fieldType, true, elemType, elemBytes};
      // Reserve space for list_length. We use size_t (8 bytes on 64-bit
      // platforms) for consistency with standard C/C++ container sizes.
      // Note: This means the translated message format is platform-dependent.
      currentOffset += sizeof(size_t);
      // List data will be appended dynamically after the fixed header
    } else {
      std::ptrdiff_t fieldBits = fieldType->getBitWidth();
      if (fieldBits < 0)
        throw std::runtime_error("Cannot translate field with dynamic size: " +
                                 name);
      if (fieldBits % 8 != 0)
        throw std::runtime_error(
            "Cannot translate field with non-byte-aligned size: " + name);
      size_t fieldBytes = (static_cast<size_t>(fieldBits) + 7) / 8;
      fieldMap[name] = {currentOffset, fieldType, false, nullptr, 0};
      currentOffset += fieldBytes;
    }
  };
 
  if (intoStruct->isReverse()) {
    for (auto it = intoFields.rbegin(); it != intoFields.rend(); ++it)
      processField(it->first, it->second);
  } else {
    for (const auto &[name, fieldType] : intoFields)
      processField(name, fieldType);
  }
 
  // intoTypeBytes is now the size of the fixed header portion
  // (for types with lists, this excludes the variable-length list data)
  intoTypeBytes = currentOffset;
 
  const auto &windowFrames = windowType->getFrames();
  frames.clear();
  frames.reserve(windowFrames.size());
 
  for (const auto &frame : windowFrames) {
    FrameInfo frameInfo;
    size_t frameOffset = 0;
 
    // Calculate frame layout in SV memory order.
    // SV structs are laid out with the last declared field at the lowest
    // address. So when we iterate fields in reverse order (rbegin to rend),
    // we get the memory layout order.
    //
    // For list fields, the lowered struct in CIRCT adds fields in this order:
    //   1. list_data[numItems] - array of list elements
    //   2. list_size - (if numItems > 1) count of valid items
    //   3. last - indicates end of list
    //
    // In SV memory layout (reversed), this becomes:
    //   offset 0: last (1 byte)
    //   offset 1: list_size (if present)
    //   offset after size: list_data[numItems]
    //   offset after list: header fields...
    struct FrameFieldLayout {
      std::string name;
      size_t frameOffset;
      size_t size;
      bool isList;
      size_t numItems;             // From window spec
      size_t listElementSize;      // Size of each list element
      size_t bufferOffset;         // Offset in translation buffer
      const Type *listElementType; // Element type for list fields
    };
    std::vector<FrameFieldLayout> fieldLayouts;
 
    for (auto fieldIt = frame.fields.rbegin(); fieldIt != frame.fields.rend();
         ++fieldIt) {
      const WindowType::Field &field = *fieldIt;
      auto it = fieldMap.find(field.name);
      if (it == fieldMap.end())
        throw std::runtime_error("Frame field '" + field.name +
                                 "' not found in intoType");
 
      const FieldInfo &fieldInfo = it->second;
      FrameFieldLayout layout;
      layout.name = field.name;
      layout.bufferOffset = fieldInfo.offset;
      layout.isList = fieldInfo.isList;
      layout.numItems = field.numItems;
      layout.listElementType = fieldInfo.listElementType;
      layout.listElementSize = fieldInfo.listElementSize;
 
      if (fieldInfo.isList) {
        // For list fields, the frame layout is (in memory order):
        //   1. 'last' field (1 byte)
        //   2. list_data (one element per frame)
        // Note: numItems > 1 is not yet supported.
        size_t numItems = field.numItems > 0 ? field.numItems : 1;
        if (numItems != 1)
          throw std::runtime_error(
              "List translation with numItems > 1 is not yet supported. "
              "Field '" +
              field.name + "' has numItems=" + std::to_string(numItems));
 
        // 'last' field comes first in memory
        size_t lastOffset = frameOffset;
        frameOffset += 1;
 
        // List data comes after 'last'
        layout.frameOffset = frameOffset;
        layout.size = fieldInfo.listElementSize;
        fieldLayouts.push_back(layout);
        frameOffset += layout.size;
 
        // Create ListFieldInfo for this list field
        ListFieldInfo listInfo;
        listInfo.fieldName = layout.name;
        listInfo.dataOffset = layout.frameOffset;
        listInfo.elementSize = fieldInfo.listElementSize;
        listInfo.listLengthBufferOffset = layout.bufferOffset;
        listInfo.listDataBufferOffset = intoTypeBytes;
        listInfo.lastFieldOffset = lastOffset;
        frameInfo.listField = listInfo;
      } else {
        // Regular (non-list) field
        std::ptrdiff_t fieldBits = fieldInfo.type->getBitWidth();
        layout.frameOffset = frameOffset;
        layout.size = (static_cast<size_t>(fieldBits) + 7) / 8;
        fieldLayouts.push_back(layout);
        frameOffset += layout.size;
      }
    }
 
    frameInfo.expectedSize = frameOffset;
 
    // Second pass: build copy ops for non-list fields
    for (const auto &layout : fieldLayouts) {
      if (!layout.isList) {
        // Regular field - add copy op
        frameInfo.copyOps.push_back(
            {layout.frameOffset, layout.bufferOffset, layout.size});
      }
      // List fields were already handled in the first pass
    }
 
    // Sort copy ops by frameOffset to ensure processing order matches frame
    // layout.
    std::sort(frameInfo.copyOps.begin(), frameInfo.copyOps.end(),
              [](const CopyOp &a, const CopyOp &b) {
                return a.frameOffset < b.frameOffset;
              });
 
    // Merge adjacent copy ops
    if (!frameInfo.copyOps.empty()) {
      std::vector<CopyOp> mergedOps;
      mergedOps.push_back(frameInfo.copyOps[0]);
 
      for (size_t i = 1; i < frameInfo.copyOps.size(); ++i) {
        CopyOp &last = mergedOps.back();
        CopyOp current = frameInfo.copyOps[i];
 
        if (last.frameOffset + last.size == current.frameOffset &&
            last.bufferOffset + last.size == current.bufferOffset) {
          last.size += current.size;
        } else {
          mergedOps.push_back(current);
        }
      }
 
      frameInfo.copyOps = std::move(mergedOps);
    }
 
    frames.push_back(std::move(frameInfo));
  }
 
  // Set frameBytes from the lowered type's bit width. Fall back to the
  // maximum computed frame expectedSize if the type system can't report width
  // (e.g. opaque union types).
  std::ptrdiff_t loweredBits = windowType->getLoweredType()->getBitWidth();
  if (loweredBits > 0) {
    frameBytes = (loweredBits + 7) / 8;
  } else {
    frameBytes = 0;
    for (const auto &f : frames)
      frameBytes = std::max(frameBytes, f.expectedSize);
  }
}
 
bool ReadChannelPort::translateIncoming(MessageData &data) {
  assert(translationInfo &&
         "Translation type must be set for window translation.");
 
  // Get the frame data directly.
  const uint8_t *frameData = data.getBytes();
  size_t frameDataSize = data.size();
 
  // Get the frame metadata for the current expected frame.
  const auto &frameInfo = translationInfo->frames[nextFrameIndex];
 
  // Check size
  if (frameDataSize < frameInfo.expectedSize)
    throw std::runtime_error("Frame data too small: expected at least " +
                             std::to_string(frameInfo.expectedSize) +
                             " bytes, got " + std::to_string(frameDataSize) +
                             " bytes");
 
  // Check if this is the first frame of a new message.
  // For list frames, we use accumulatingListData to track whether we're
  // continuing to accumulate list items.
  bool isFirstFrame = (nextFrameIndex == 0) && !accumulatingListData;
 
  if (isFirstFrame) {
    // Initialize the translation buffer to hold the fixed header portion.
    translationBuffer.resize(translationInfo->intoTypeBytes, 0);
    // Clear list data buffer for types with lists
    if (translationInfo->hasListField)
      listDataBuffer.clear();
  }
 
  // Execute copy ops for non-list fields
  for (const auto &op : frameInfo.copyOps)
    std::memcpy(translationBuffer.data() + op.bufferOffset,
                frameData + op.frameOffset, op.size);
 
  // Handle list field if present in this frame
  if (frameInfo.listField.has_value()) {
    const auto &listInfo = frameInfo.listField.value();
 
    // With numItems == 1, each frame contains exactly one list element
    size_t bytesToCopy = listInfo.elementSize;
    // Additional check: dataOffset must not be beyond frameDataSize
    if (listInfo.dataOffset > frameDataSize)
      throw std::runtime_error("List data offset is beyond frame bounds");
    // Bounds check to prevent buffer overflow from corrupted _size field
    if (listInfo.dataOffset + bytesToCopy > frameDataSize)
      throw std::runtime_error("List data extends beyond frame bounds");
    size_t oldSize = listDataBuffer.size();
    listDataBuffer.resize(oldSize + bytesToCopy);
    std::memcpy(listDataBuffer.data() + oldSize,
                frameData + listInfo.dataOffset, bytesToCopy);
 
    // Check if this is the last frame of the list
    uint8_t lastFlag = frameData[listInfo.lastFieldOffset];
    if (lastFlag) {
      // List is complete. Build the final message:
      // [fixed header with list_length (size_t)][list data...]
      // Write the actual length to the listLengthBufferOffset position.
      size_t listLength = listDataBuffer.size() / listInfo.elementSize;
      size_t *listLengthPtr = reinterpret_cast<size_t *>(
          translationBuffer.data() + listInfo.listLengthBufferOffset);
      *listLengthPtr = listLength;
 
      // Append list data to translation buffer
      size_t headerSize = translationBuffer.size();
      translationBuffer.resize(headerSize + listDataBuffer.size());
      std::memcpy(translationBuffer.data() + headerSize, listDataBuffer.data(),
                  listDataBuffer.size());
 
      // Reset for next message
      nextFrameIndex = 0;
      listDataBuffer.clear();
      accumulatingListData = false;
      return true;
    }
 
    // Not the last frame - stay on this frame index for list accumulation
    // (list frames repeat until last=true)
    accumulatingListData = true;
    return false;
  }
 
  // No list field in this frame - advance to next frame
  nextFrameIndex++;
  size_t numFrames = translationInfo->frames.size();
 
  // Check if all frames have been received.
  if (nextFrameIndex >= numFrames) {
    // Reset for the next message.
    nextFrameIndex = 0;
    return true;
  }
 
  return false;
}
 
void WriteChannelPort::translateOutgoing(const MessageData &data) {
  assert(translationInfo &&
         "Translation type must be set for window translation.");
 
  const uint8_t *srcData = data.getBytes();
  size_t srcDataSize = data.size();
 
  if (srcDataSize < translationInfo->intoTypeBytes)
    throw std::runtime_error("Source data too small: expected at least " +
                             std::to_string(translationInfo->intoTypeBytes) +
                             " bytes, got " + std::to_string(srcDataSize) +
                             " bytes");
 
  // Check if we have list fields
  if (!translationInfo->hasListField) {
    // No list fields - simple fixed-size translation
    for (const auto &frameInfo : translationInfo->frames) {
      std::vector<uint8_t> frameBuffer(frameInfo.expectedSize, 0);
      for (const auto &op : frameInfo.copyOps)
        std::memcpy(frameBuffer.data() + op.frameOffset,
                    srcData + op.bufferOffset, op.size);
      translationBuffer.emplace_back(std::move(frameBuffer));
    }
    return;
  }
 
  // Handle list fields - need to split list data into multiple frames
  for (const auto &frameInfo : translationInfo->frames) {
    if (!frameInfo.listField.has_value()) {
      // Non-list frame - copy as normal
      std::vector<uint8_t> frameBuffer(frameInfo.expectedSize, 0);
      for (const auto &op : frameInfo.copyOps)
        std::memcpy(frameBuffer.data() + op.frameOffset,
                    srcData + op.bufferOffset, op.size);
      translationBuffer.emplace_back(std::move(frameBuffer));
    } else {
      // List frame - need to generate multiple frames
      const auto &listInfo = frameInfo.listField.value();
 
      // Read the list length from the source data
      size_t listLength = 0;
      std::memcpy(&listLength, srcData + listInfo.listLengthBufferOffset,
                  sizeof(size_t));
 
      // Check that the buffer is large enough for the list data
      if (translationInfo->intoTypeBytes + (listLength * listInfo.elementSize) >
          srcDataSize) {
        throw std::runtime_error(
            "Source buffer too small for list data: possible corrupted or "
            "inconsistent list length field");
      }
      // Get pointer to list data (after the fixed header)
      const uint8_t *listData = srcData + translationInfo->intoTypeBytes;
 
      // Generate frames for the list
      size_t itemsRemaining = listLength;
      size_t listDataOffset = 0;
 
      // Handle empty list case - still need to send one frame with last=true
      if (listLength == 0)
        throw std::runtime_error(
            "Cannot send empty lists - parallel ESI list encoding requires at "
            "least one frame to be sent, and each frame must contain at least "
            "one element.");
 
      while (itemsRemaining > 0) {
        std::vector<uint8_t> frameBuffer(frameInfo.expectedSize, 0);
 
        // Copy non-list fields (header data) to each frame
        for (const auto &op : frameInfo.copyOps)
          std::memcpy(frameBuffer.data() + op.frameOffset,
                      srcData + op.bufferOffset, op.size);
 
        // With numItems == 1, each frame contains exactly one list element
        size_t bytesInThisFrame = listInfo.elementSize;
 
        // Copy list data
        std::memcpy(frameBuffer.data() + listInfo.dataOffset,
                    listData + listDataOffset, bytesInThisFrame);
 
        // Update remaining count
        itemsRemaining -= 1;
        listDataOffset += bytesInThisFrame;
 
        // Set last field
        frameBuffer[listInfo.lastFieldOffset] = (itemsRemaining == 0) ? 1 : 0;
 
        translationBuffer.emplace_back(std::move(frameBuffer));
      }
    }
  }
}
 
std::vector<MessageData>
WriteChannelPort::getMessageFrames(const MessageData &data) {
  size_t frameBytes = getFrameSizeBytes();
  assert(frameBytes > 0 && "Frame size must be greater than 0");
  if (data.getSize() % frameBytes != 0)
    throw std::runtime_error(
        "write message size (" + std::to_string(data.getSize()) +
        ") is not a multiple of frame size (" + std::to_string(frameBytes) +
        ") on type " + type->toString());
  std::vector<MessageData> frames;
  size_t numFrames = data.getSize() / frameBytes;
  const uint8_t *ptr = data.getBytes();
  for (size_t i = 0; i < numFrames; ++i) {
    frames.emplace_back(ptr, frameBytes);
    ptr += frameBytes;
  }
  return frames;
}