CIRCT 24.0.0git
Loading...
Searching...
No Matches
TypedPorts.h
Go to the documentation of this file.
1//===- TypedPorts.h - Strongly-typed ESI port wrappers ----------*- C++ -*-===//
2//
3// Part of the LLVM Project, under the Apache License v2.0 with LLVM Exceptions.
4// See https://llvm.org/LICENSE.txt for license information.
5// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
6//
7//===----------------------------------------------------------------------===//
8//
9// DO NOT EDIT!
10// This file is distributed as part of an ESI package. The source for this file
11// should always be modified within CIRCT.
12//
13//===----------------------------------------------------------------------===//
14//
15// Thin, non-owning wrappers around WriteChannelPort / ReadChannelPort that
16// verify type compatibility at connect() time and provide strongly-typed
17// write/read APIs. Purely additive — no changes to the untyped port classes.
18//
19//===----------------------------------------------------------------------===//
20
21// NOLINTNEXTLINE(llvm-header-guard)
22#ifndef ESI_TYPED_PORTS_H
23#define ESI_TYPED_PORTS_H
24
25#include "esi/Design.h"
26#include "esi/Ports.h"
27#include "esi/Services.h"
28#include "esi/Types.h"
29
30#include <algorithm>
31#include <array>
32#include <cstdint>
33#include <cstring>
34#include <functional>
35#include <future>
36#include <map>
37#include <memory>
38#include <mutex>
39#include <optional>
40#include <queue>
41#include <stdexcept>
42#include <string>
43#include <string_view>
44#include <type_traits>
45#include <typeinfo>
46
47namespace esi {
48
49//===----------------------------------------------------------------------===//
50// AcceleratorMismatchError — thrown for type mismatches and port-not-found.
51//===----------------------------------------------------------------------===//
52
53class AcceleratorMismatchError : public std::runtime_error {
54public:
55 using std::runtime_error::runtime_error;
56};
57
58//===----------------------------------------------------------------------===//
59// Helpers: unwrap TypeAliasType and width-aware serialization.
60//===----------------------------------------------------------------------===//
61
62/// Compute the wire byte count for a port type. Returns 0 if not a
63/// BitVectorType (meaning sizeof(T) should be used instead).
64/// Wire format info for a port type, cached at connect() time.
65struct WireInfo {
66 size_t bytes = 0; // (bitWidth+7)/8, or 0 if not a BitVectorType.
67 size_t bitWidth = 0;
68};
69
70inline WireInfo getWireInfo(const Type *portType) {
71 const Type *inner = unwrapTypeAlias(portType);
72 if (auto *bv = dynamic_cast<const BitVectorType *>(inner))
73 return {(bv->getWidth() + 7) / 8, bv->getWidth()};
74 return {};
75}
76
77/// Pack a C++ integral value into a MessageData with the given wire byte count.
78/// If wi.bytes is 0, falls back to sizeof(T).
79template <typename T>
81 if constexpr (std::is_integral_v<T> && !std::is_same_v<T, bool>) {
82 if (wi.bytes > 0 && wi.bytes != sizeof(T)) {
83 std::vector<uint8_t> buf(wi.bytes, 0);
84 std::memcpy(buf.data(), &data, std::min(wi.bytes, sizeof(T)));
85 return MessageData(std::move(buf));
86 }
87 } else if constexpr (std::is_base_of_v<SegmentedMessageData, T>) {
88 return data.toMessageData();
89 }
90 return MessageData(reinterpret_cast<const uint8_t *>(&data), sizeof(T));
91}
92
93template <typename T>
94MessageData toMessageData(const T &data, const Type *portType) {
95 return toMessageData(data, getWireInfo(portType));
96}
97
98/// Unpack a MessageData into a C++ integral value with the given wire info.
99/// If the wire size differs from sizeof(T), copies available bytes into
100/// a zero-initialized value and sign-extends for signed types using the
101/// actual bit width to locate the sign bit.
102template <typename T>
104 if constexpr (std::is_integral_v<T> && !std::is_same_v<T, bool>) {
105 if (wi.bytes > 0 && msg.getSize() == wi.bytes &&
106 wi.bitWidth < sizeof(T) * 8) {
107 // Copy wire bytes into a zero-initialized value.
108 T val = 0;
109 std::memcpy(&val, msg.getBytes(), std::min(wi.bytes, sizeof(T)));
110 // Sign-extend for signed types if the sign bit is set.
111 if constexpr (std::is_signed_v<T>) {
112 size_t signBit = wi.bitWidth - 1;
113 size_t signByte = signBit / 8;
114 uint8_t signMask = uint8_t(1) << (signBit % 8);
115 if (signByte < wi.bytes && (msg.getBytes()[signByte] & signMask)) {
116 // Set all bits above the sign bit to 1.
117 if (wi.bitWidth < sizeof(T) * 8)
118 val |= static_cast<T>(~T(0)) << wi.bitWidth;
119 }
120 }
121 return val;
122 }
123 }
124 return *msg.as<T>();
125}
126
127template <typename T>
128T fromMessageData(const MessageData &msg, const Type *portType) {
129 return fromMessageData<T>(msg, getWireInfo(portType));
130}
131
132namespace detail {
133
134/// Owning callback used by typed read deserializers.
135///
136/// Returning `false` means the callee did not accept the object and wants the
137/// exact same owned value retried later.
138template <typename T>
139using TypedReadOwnedCallback = std::function<bool(std::unique_ptr<T> &)>;
140
141template <typename T, typename = void>
142struct has_type_deserializer : std::false_type {};
143
144template <typename T>
145struct has_type_deserializer<T, std::void_t<typename T::TypeDeserializer>>
146 : std::true_type {};
147
148template <typename T>
150
151template <typename T>
153 MessageData &scratch) {
154 if (auto *flat = dynamic_cast<const MessageData *>(&msg))
155 return *flat;
156 scratch = msg.toMessageData();
157 return scratch;
158}
159
160/// Default deserializer for simple 1:1 typed reads.
161///
162/// This path converts one raw message into one typed value and forwards it to
163/// the typed callback. If the callback rejects the decoded object, preserve it
164/// until the raw message is retried so the same owned object is presented
165/// again.
166template <typename T>
168public:
170
173
174 bool push(std::unique_ptr<SegmentedMessageData> &msg) {
175 std::scoped_lock<std::mutex> lock(mutex);
176 if (!msg)
177 throw std::runtime_error("PODTypeDeserializer::push: null message");
178
179 // Flush the buffer, bail if still full.
180 pokeLocked();
181 if (pendingOutput)
182 return false;
183
184 // Translate the raw message into a typed object. This always consumes the
185 // raw message.
186 MessageData scratch;
187 const MessageData &flat = getMessageDataRef<T>(*msg, scratch);
188 pendingOutput = std::make_unique<T>(fromMessageData<T>(flat, wireInfo));
189 msg.reset();
190 pokeLocked();
191 return true;
192 }
193
194 bool poke() {
195 std::scoped_lock<std::mutex> lock(mutex);
196 return pokeLocked();
197 }
198
199private:
200 bool pokeLocked() {
202 pendingOutput.reset();
203 return true;
204 }
205 return false;
206 }
207
210 std::mutex mutex;
211 std::unique_ptr<T> pendingOutput;
212};
213
214template <typename T, typename = void>
218
219template <typename T>
220struct DeserializerSelector<T, std::void_t<typename T::TypeDeserializer>> {
221 using type = typename T::TypeDeserializer;
222};
223
224template <typename T>
226
227template <typename T>
229
230} // namespace detail
231
232/// Helper base class for stateful deserializers which may emit zero, one, or
233/// many typed outputs for each raw input message.
234///
235/// Derived classes implement `decode()` and must consume the raw input message
236/// before returning. This base class handles retrying blocked typed outputs and
237/// preserving them in FIFO order until the client accepts them.
238template <typename T>
240public:
242 using DecodedOutputs = std::vector<std::unique_ptr<T>>;
243
246
247 virtual ~QueuedDecodeTypeDeserializer() = default;
248
249 /// Push one raw message into the deserializer.
250 ///
251 /// Returns `false` only when previously decoded typed outputs are still
252 /// blocked on delivery. In that case `msg` is left untouched so the caller
253 /// can retry it later.
254 bool push(std::unique_ptr<SegmentedMessageData> &msg) {
255 std::scoped_lock<std::mutex> lock(mutex);
256 if (!msg)
257 throw std::runtime_error(
258 "QueuedDecodeTypeDeserializer::push: null message");
259
260 if (!pokeLocked())
261 return false;
262
263 DecodedOutputs decoded = decode(msg);
264 if (msg)
265 throw std::runtime_error(
266 "QueuedDecodeTypeDeserializer::push: decode must consume the "
267 "message");
268 for (std::unique_ptr<T> &value : decoded) {
269 if (!value)
270 throw std::runtime_error(
271 "QueuedDecodeTypeDeserializer::push: null decoded output");
272 pendingOutputs.push(std::move(value));
273 }
274
275 // Once decode() has consumed the raw message, keep any blocked typed
276 // outputs in the pending queue and report success. pokeLocked() still
277 // opportunistically drains what it can before returning.
278 pokeLocked();
279 return true;
280 }
281
282 /// Retry delivery of any typed outputs which were previously blocked by the
283 /// client callback.
284 bool poke() {
285 std::scoped_lock<std::mutex> lock(mutex);
286 return pokeLocked();
287 }
288
289protected:
290 /// Decode one raw message into zero or more typed outputs.
291 ///
292 /// Implementations must consume `msg` before returning, even when zero
293 /// outputs are produced.
294 virtual DecodedOutputs decode(std::unique_ptr<SegmentedMessageData> &msg) = 0;
295
296private:
297 bool pokeLocked() {
298 while (!pendingOutputs.empty()) {
299 std::unique_ptr<T> &value = pendingOutputs.front();
300 if (!value)
301 throw std::runtime_error(
302 "QueuedDecodeTypeDeserializer::poke: null pending output");
303 if (!output(value))
304 return false;
305 pendingOutputs.pop();
306 }
307 return true;
308 }
309
311 std::queue<std::unique_ptr<T>> pendingOutputs;
312 std::mutex mutex;
313};
314
315/// Reusable serial-list window deserializer.
316///
317/// Walks the serial-list multi-burst protocol used by codegen'd window
318/// helpers -- zero or more `header(N>0) + N data` bursts terminated by a
319/// `header(0)` footer -- and emits a fully-formed `T` instance built from the
320/// accumulated header fields and data frames.
321///
322/// `T` must be a generated window helper that exposes:
323/// - nested `header_frame` / `data_frame` types of identical size,
324/// - `count_type`,
325/// - `static count_type T::_headerCount(const header_frame &)`,
326/// - `static std::unique_ptr<T> T::_fromFrames(const header_frame &,
327/// std::vector<data_frame> &&)`.
328///
329/// `T` typically declares `SerialListTypeDeserializer<T>` as a friend so the
330/// template can reach the (private) `header_frame` definition; the helpers
331/// themselves can stay private as well.
332template <typename T>
334public:
338
341
342private:
343 using header_frame = typename T::header_frame;
344 using data_frame = typename T::data_frame;
345 using count_type = typename T::count_type;
346
347 static_assert(sizeof(header_frame) == sizeof(data_frame),
348 "header and data frames must be the same width");
349 static constexpr size_t kFrameSize = sizeof(data_frame);
350
351 DecodedOutputs decode(std::unique_ptr<SegmentedMessageData> &msg) override {
352 DecodedOutputs out;
353 MessageData scratch;
354 const MessageData &flat = detail::getMessageDataRef<T>(*msg, scratch);
355 const uint8_t *bytes = flat.getBytes();
356 size_t size = flat.getSize();
357
358 size_t offset = 0;
359 while (offset < size) {
360 size_t needed = kFrameSize - partial_.size();
361 size_t chunk = std::min(needed, size - offset);
362 partial_.insert(partial_.end(), bytes + offset, bytes + offset + chunk);
363 offset += chunk;
364 if (partial_.size() != kFrameSize)
365 break;
366
367 if (remaining_ == 0) {
368 // Header or footer frame. Decode into a local frame first so we can
369 // inspect the count without committing. Only the first header of a
370 // transaction is guaranteed to carry valid static fields; the static
371 // slots of continuation and footer headers may be garbage.
372 header_frame frame{};
373 std::memcpy(&frame, partial_.data(), kFrameSize);
374 partial_.clear();
375 count_type batchCount = T::_headerCount(frame);
376 if (batchCount == 0) {
377 // Footer: emit the accumulated value using the first header's
378 // static fields.
379 if (!pending_header_)
380 throw std::runtime_error(
381 "SerialListTypeDeserializer: footer received before any "
382 "header");
383 out.push_back(
384 T::_fromFrames(*pending_header_, std::move(pending_frames_)));
385 pending_frames_.clear();
386 pending_header_.reset();
387 continue;
388 }
389
390 if (!pending_header_)
391 pending_header_ = frame;
392 remaining_ = batchCount;
393 continue;
394 }
395
396 // Data frame.
397 auto &frame = pending_frames_.emplace_back();
398 std::memcpy(&frame, partial_.data(), kFrameSize);
399 partial_.clear();
400 --remaining_;
401 }
402
403 msg.reset();
404 return out;
405 }
406
407 std::vector<uint8_t> partial_;
408 std::optional<header_frame> pending_header_;
409 std::vector<data_frame> pending_frames_;
411};
412
413//===----------------------------------------------------------------------===//
414// Type-trait: detect T::_ESI_ID (a static constexpr std::string_view).
415//===----------------------------------------------------------------------===//
416
417template <typename T, typename = void>
418struct has_esi_id : std::false_type {};
419
420template <typename T>
421struct has_esi_id<T, std::void_t<decltype(T::_ESI_ID)>>
422 : std::is_convertible<decltype(T::_ESI_ID), std::string_view> {};
423
424template <typename T>
425inline constexpr bool has_esi_id_v = has_esi_id<T>::value;
426
427// Detect T::_ESI_WINDOW_ID (a static constexpr std::string_view) for
428// generated SegmentedMessageData subclasses bound to a specific WindowType.
429template <typename T, typename = void>
430struct has_esi_window_id : std::false_type {};
431
432template <typename T>
433struct has_esi_window_id<T, std::void_t<decltype(T::_ESI_WINDOW_ID)>>
434 : std::is_convertible<decltype(T::_ESI_WINDOW_ID), std::string_view> {};
435
436template <typename T>
438
439//===----------------------------------------------------------------------===//
440// verifyTypeCompatibility<T>(const Type *portType)
441//
442// Checks that the ESI runtime Type is compatible with the C++ type T.
443// Dispatch order: _ESI_ID → void → bool → std::array → signed int →
444// unsigned int → error.
445//===----------------------------------------------------------------------===//
446
447// Detect ``std::array<T, N>``.
448template <typename T>
449struct is_std_array : std::false_type {};
450template <typename T, std::size_t N>
451struct is_std_array<std::array<T, N>> : std::true_type {};
452template <typename T>
454
455template <typename T>
456void verifyTypeCompatibility(const Type *portType) {
457 if (!portType)
458 throw AcceleratorMismatchError("Port type is null");
459
460 // Unwrap TypeAliasType to get the inner type for verification.
461 portType = unwrapTypeAlias(portType);
462
463 // If the port is a windowed type, verify the window id (if T declares one)
464 // and then continue checking the inner ('into') type.
465 if (auto *windowType = dynamic_cast<const WindowType *>(portType)) {
466 if constexpr (has_esi_window_id_v<T>) {
467 if (std::string_view(windowType->getID()) != T::_ESI_WINDOW_ID)
469 "ESI window mismatch: C++ type has _ESI_WINDOW_ID '" +
470 std::string(T::_ESI_WINDOW_ID) + "' but port window type is '" +
471 windowType->toString(/*oneLine=*/true) + "'");
472 } else {
474 "ESI type mismatch: port is a window type ('" +
475 windowType->toString(/*oneLine=*/true) +
476 "') but C++ type has no _ESI_WINDOW_ID");
477 }
478 portType = unwrapTypeAlias(windowType->getIntoType());
479 } else if constexpr (has_esi_window_id_v<T>) {
481 std::string("ESI type mismatch: C++ type has _ESI_WINDOW_ID '") +
482 std::string(T::_ESI_WINDOW_ID) + "' but port type is not a window: '" +
483 portType->toString(/*oneLine=*/true) + "'");
484 }
485
486 if constexpr (has_esi_id_v<T>) {
487 // Highest priority: user-defined ESI ID string comparison.
488 if (std::string_view(portType->getID()) != T::_ESI_ID)
490 "ESI type mismatch: C++ type has _ESI_ID '" +
491 std::string(T::_ESI_ID) + "' but port type is '" +
492 portType->toString(/*oneLine=*/true) + "'");
493 } else if constexpr (std::is_void_v<T>) {
494 if (!dynamic_cast<const VoidType *>(portType))
495 throw AcceleratorMismatchError("ESI type mismatch: expected VoidType for "
496 "void, but port type is '" +
497 portType->toString(/*oneLine=*/true) +
498 "'");
499 } else if constexpr (std::is_same_v<T, bool>) {
500 // bool maps to signless i1, which is BitsType with width <= 1.
501 auto *bits = dynamic_cast<const BitsType *>(portType);
502 if (!bits || bits->getWidth() > 1)
504 "ESI type mismatch: expected BitsType with width <= 1 for "
505 "bool, but port type is '" +
506 portType->toString(/*oneLine=*/true) + "'");
507 } else if constexpr (is_std_array_v<T>) {
508 // std::array<U, N> maps to ESI ArrayType with matching size and element
509 // type. The ArrayType's `reverse` flag controls only wire-byte ordering
510 // during serialization (handled in toMessageData/fromMessageData) and
511 // does not affect compatibility.
512 using Element = typename T::value_type;
513 constexpr size_t N = std::tuple_size<T>::value;
514 auto *arr = dynamic_cast<const ArrayType *>(portType);
515 if (!arr)
517 "ESI type mismatch: expected ArrayType for std::array, but port "
518 "type is '" +
519 portType->toString(/*oneLine=*/true) + "'");
520 if (arr->getSize() != N)
521 throw AcceleratorMismatchError("ESI type mismatch: ArrayType size " +
522 std::to_string(arr->getSize()) +
523 " does not match std::array size " +
524 std::to_string(N));
525 verifyTypeCompatibility<Element>(arr->getElementType());
526 } else if constexpr (std::is_integral_v<T> && std::is_signed_v<T>) {
527 auto *sint = dynamic_cast<const SIntType *>(portType);
528 if (!sint)
530 "ESI type mismatch: expected SIntType for signed integer, "
531 "but port type is '" +
532 portType->toString(/*oneLine=*/true) + "'");
533 if (sint->getWidth() > sizeof(T) * 8)
535 "ESI type mismatch: SIntType width " +
536 std::to_string(sint->getWidth()) + " does not fit in " +
537 std::to_string(sizeof(T) * 8) + "-bit signed integer");
538 // Require closest-size match: reject if a smaller C++ type would suffice.
539 if (sizeof(T) > 1 && sint->getWidth() <= (sizeof(T) / 2) * 8)
540 throw AcceleratorMismatchError("ESI type mismatch: SIntType width " +
541 std::to_string(sint->getWidth()) +
542 " should use a smaller C++ type than " +
543 std::to_string(sizeof(T) * 8) + "-bit");
544 } else if constexpr (std::is_integral_v<T> && std::is_unsigned_v<T>) {
545 // Accept UIntType (uiM) or BitsType (iM, signless).
546 auto *uintPort = dynamic_cast<const UIntType *>(portType);
547 auto *bits = dynamic_cast<const BitsType *>(portType);
548 if (!uintPort && !bits)
550 "ESI type mismatch: expected UIntType or BitsType for unsigned "
551 "integer, but port type is '" +
552 portType->toString(/*oneLine=*/true) + "'");
553 uint64_t width = uintPort ? uintPort->getWidth() : bits->getWidth();
554 if (width > sizeof(T) * 8)
556 "ESI type mismatch: bit width " + std::to_string(width) +
557 " does not fit in " + std::to_string(sizeof(T) * 8) +
558 "-bit unsigned integer");
559 // Require closest-size match: reject if a smaller C++ type would suffice.
560 if (sizeof(T) > 1 && width <= (sizeof(T) / 2) * 8)
561 throw AcceleratorMismatchError("ESI type mismatch: bit width " +
562 std::to_string(width) +
563 " should use a smaller C++ type than " +
564 std::to_string(sizeof(T) * 8) + "-bit");
565 } else {
567 std::string("Cannot verify type compatibility for C++ type '") +
568 typeid(T).name() + "' against ESI port type '" +
569 portType->toString(/*oneLine=*/true) + "'");
570 }
571}
572
573// Port-aware overload: when checking against a ChannelPort, also reconcile
574// the windowed wrapper. ChannelPort::getType() returns the unwrapped 'into'
575// type for windowed ports, so we have to consult getWindowType() directly to
576// detect the window and forward the original WindowType into the Type-based
577// overload above.
578template <typename T>
580 if (!port)
581 throw AcceleratorMismatchError("Port is null");
582 const WindowType *windowType = port->getWindowType();
583 if constexpr (has_esi_window_id_v<T>) {
584 if (!windowType)
586 std::string("ESI type mismatch: C++ type has _ESI_WINDOW_ID '") +
587 std::string(T::_ESI_WINDOW_ID) +
588 "' but port is not a window type ('" +
589 (port->getType() ? port->getType()->toString(/*oneLine=*/true)
590 : std::string("<null>")) +
591 "')");
592 } else {
593 if (windowType)
595 "ESI type mismatch: port is a window type ('" +
596 windowType->toString(/*oneLine=*/true) +
597 "') but C++ type has no _ESI_WINDOW_ID");
598 }
599 const Type *forwardType =
600 windowType ? static_cast<const Type *>(windowType) : port->getType();
601 verifyTypeCompatibility<T>(forwardType);
602}
603
604//===----------------------------------------------------------------------===//
605// TypedWritePort<T, SkipTypeCheck = false>
606//
607// When SkipTypeCheck is false, the `connect` method runs the type check via
608// `verifyTypeCompatibility<T>()`. When SkipTypeCheck is true, `connect`
609// skips that verification.
610//===----------------------------------------------------------------------===//
611
612template <typename T, bool SkipTypeCheck = false>
614public:
615 explicit TypedWritePort(WriteChannelPort &port) : inner(&port) {}
616 // NOLINTNEXTLINE(google-explicit-constructor)
618
619 void connect(const ChannelPort::ConnectOptions &opts = {std::nullopt,
620 false}) {
621 if (!inner)
622 throw AcceleratorMismatchError("TypedWritePort: null port pointer");
623 if (!SkipTypeCheck)
626 inner->connect(opts);
627 }
628
629 void write(const T &data) { inner->write(toMessageData(data, wireInfo_)); }
630
631 /// Write by taking ownership. If T is a SegmentedMessageData, this hands
632 /// the message directly to the port's segmented write path.
633 void write(std::unique_ptr<T> &data) {
634 if (!data)
635 throw std::runtime_error("TypedWritePort::write: null unique_ptr");
636 if constexpr (std::is_base_of_v<SegmentedMessageData, T>) {
637 inner->write(std::move(data));
638 } else {
639 write(*data);
640 data.reset();
641 }
642 }
643
644 bool tryWrite(const T &data) {
645 return inner->tryWrite(toMessageData(data, wireInfo_));
646 }
647
648 /// Non-blocking write by taking ownership. On success, `data` is consumed
649 /// (left null) and true is returned. On failure, `data` is left untouched so
650 /// the caller can retry. If T is a SegmentedMessageData, this hands the
651 /// message directly to the port's segmented tryWrite path.
652 bool tryWrite(std::unique_ptr<T> &data) {
653 if (!data)
654 throw std::runtime_error("TypedWritePort::tryWrite: null unique_ptr");
655 if constexpr (std::is_base_of_v<SegmentedMessageData, T>) {
656 std::unique_ptr<SegmentedMessageData> msg(data.release());
657 bool written = inner->tryWrite(msg);
658 // On failure, the port did not take ownership; hand it back.
659 if (!written)
660 data.reset(static_cast<T *>(msg.release()));
661 return written;
662 } else {
663 if (!tryWrite(*data))
664 return false;
665 data.reset();
666 return true;
667 }
668 }
669
670 bool flush() { return inner->flush(); }
671 void disconnect() { inner->disconnect(); }
672 bool isConnected() const { return inner && inner->isConnected(); }
673
674 WriteChannelPort &raw() { return *inner; }
675 const WriteChannelPort &raw() const { return *inner; }
676
677private:
680};
681
682/// Specialization for void — write takes no data argument.
683template <>
684class TypedWritePort<void> {
685public:
686 explicit TypedWritePort(WriteChannelPort &port) : inner(&port) {}
687 // NOLINTNEXTLINE(google-explicit-constructor)
689
690 void connect(const ChannelPort::ConnectOptions &opts = {std::nullopt,
691 false}) {
692 if (!inner)
693 throw AcceleratorMismatchError("TypedWritePort: null port pointer");
695 inner->connect(opts);
696 }
697
698 void write() {
699 uint8_t zero = 0;
700 inner->write(MessageData(&zero, 1));
701 }
702
703 bool tryWrite() {
704 uint8_t zero = 0;
705 return inner->tryWrite(MessageData(&zero, 1));
706 }
707
708 bool flush() { return inner->flush(); }
709 void disconnect() { inner->disconnect(); }
710 bool isConnected() const { return inner && inner->isConnected(); }
711
712 WriteChannelPort &raw() { return *inner; }
713 const WriteChannelPort &raw() const { return *inner; }
714
715private:
717};
718
719//===----------------------------------------------------------------------===//
720// TypedReadPort<T>
721//===----------------------------------------------------------------------===//
722
723/// Strongly typed wrapper around a raw read channel.
724///
725/// For scalar/POD-like `T`, this performs a 1:1 conversion from raw messages.
726/// If `T` defines a nested `TypeDeserializer`, one instance is created per
727/// connection and drives both callback and polling reads through that
728/// deserializer.
729///
730/// Polling reads return `std::unique_ptr<T>` so complex decoded values can be
731/// delivered without an extra copy.
732template <typename T, bool SkipTypeCheck = false>
734public:
735 explicit TypedReadPort(ReadChannelPort &port) : inner(&port) {}
736 // NOLINTNEXTLINE(google-explicit-constructor)
738 TypedReadPort(const TypedReadPort &) = delete;
742
744 if (inner && mode != Mode::Disconnected)
745 disconnect();
746 }
747
748 /// Connect in polling mode.
749 ///
750 /// The port installs an internal typed output queue. `read()` and
751 /// `readAsync()` consume from that queue.
752 void connect(const ChannelPort::ConnectOptions &opts = {std::nullopt,
753 false}) {
754 if (!inner)
755 throw AcceleratorMismatchError("TypedReadPort: null port pointer");
758 emplaceDeserializer([this](std::unique_ptr<T> &value) -> bool {
759 return pollingState->enqueue(value);
760 });
761 try {
762 inner->connect(
763 [this](std::unique_ptr<SegmentedMessageData> &msg) -> bool {
764 assert(deserializer && "Deserializer should be connected");
765 return deserializer->push(msg);
766 },
767 opts);
768 } catch (...) {
770 throw;
771 }
773 }
774
775 /// Connect a non-owning typed callback.
776 ///
777 /// The callback sees the decoded value by reference. Returning `false`
778 /// requests that the same decoded value be retried later.
779 void connect(std::function<bool(const T &)> callback,
780 const ChannelPort::ConnectOptions &opts = {std::nullopt,
781 false}) {
782 connect([cb = std::move(callback)](
783 std::unique_ptr<T> &value) -> bool { return cb(*value); },
784 opts);
785 }
786
787 /// Connect an owning typed callback.
788 ///
789 /// This is the typed analogue of `ReadChannelPort::ReadCallback`: the
790 /// callback may take ownership of the decoded value or return `false` to
791 /// retry delivery later with the same object.
793 const ChannelPort::ConnectOptions &opts = {std::nullopt,
794 false}) {
795 if (!inner)
796 throw AcceleratorMismatchError("TypedReadPort: null port pointer");
798 emplaceDeserializer(std::move(callback));
799 try {
800 inner->connect(
801 [this](std::unique_ptr<SegmentedMessageData> &msg) -> bool {
802 assert(deserializer && "Deserializer should be connected");
803 return deserializer->push(msg);
804 },
805 opts);
806 } catch (...) {
808 throw;
809 }
810 // TODO: Hook callback-mode custom-deserializer poke() retries into the
811 // existing periodic poll/background-worker path.
813 }
814
815 /// Blocking typed read in polling mode.
816 std::unique_ptr<T> read() {
817 std::future<std::unique_ptr<T>> f = readAsync();
818 f.wait();
819 return f.get();
820 }
821
822 /// Asynchronous typed read in polling mode.
823 ///
824 /// The returned future yields ownership of the next decoded value.
825 std::future<std::unique_ptr<T>> readAsync() {
826 if (mode == Mode::Callback)
827 throw std::runtime_error(
828 "Cannot read from a callback channel. `connect()` without a "
829 "callback specified to use polling mode.");
831 throw std::runtime_error(
832 "Cannot read from a disconnected channel. `connect()` first.");
833
834 if (!pollingState)
835 throw std::runtime_error(
836 "Cannot read from a disconnected channel. `connect()` first.");
837
838 std::future<std::unique_ptr<T>> future = pollingState->readAsync();
839
840 if (deserializer)
841 deserializer->poke();
842 return future;
843 }
844
845 /// Set the maximum number of decoded typed values buffered in polling mode.
846 /// `0` means unbounded.
847 void setMaxDataQueueMsgs(uint64_t maxMsgs) {
848 maxDataQueueMsgs = maxMsgs;
849 if (pollingState)
850 pollingState->setMaxQueued(maxMsgs);
851 }
852
853 /// Disconnect the typed port and abandon any pending polling reads.
854 void disconnect() {
855 inner->disconnect();
858 }
859 bool isConnected() const { return inner && inner->isConnected(); }
860
861 ReadChannelPort &raw() { return *inner; }
862 const ReadChannelPort &raw() const { return *inner; }
863
864private:
868
870 deserializer.reset();
871 pollingState.reset();
872 }
873
876 deserializer.emplace(std::move(callback));
877 } else {
878 deserializer.emplace(std::move(callback), wireInfo_);
879 }
880 }
881
884 throw std::runtime_error("Channel already connected");
885 if constexpr (SkipTypeCheck) {
886 // Skip verification, but still compute wireInfo for the POD path so
887 // small / non-byte-aligned wire widths still encode correctly.
889 wireInfo_ = getWireInfo(inner->getType());
890 } else if constexpr (has_esi_id_v<T>) {
892 } else if constexpr (!detail::has_type_deserializer_v<T>) {
894 wireInfo_ = getWireInfo(inner->getType());
895 }
896 }
897
902 std::optional<Deserializer> deserializer;
903 std::optional<PollingState> pollingState;
904};
905
906/// Specialization for void — read discards data and returns nothing.
907template <>
908class TypedReadPort<void> {
909public:
910 explicit TypedReadPort(ReadChannelPort &port) : inner(&port) {}
911 // NOLINTNEXTLINE(google-explicit-constructor)
913 TypedReadPort(const TypedReadPort &) = delete;
917
919 if (inner && inner->isConnected())
920 disconnect();
921 }
922
923 void connect(const ChannelPort::ConnectOptions &opts = {std::nullopt,
924 false}) {
925 if (!inner)
926 throw AcceleratorMismatchError("TypedReadPort: null port pointer");
928 inner->connect(opts);
929 }
930
931 void connect(std::function<bool()> callback,
932 const ChannelPort::ConnectOptions &opts = {std::nullopt,
933 false}) {
934 if (!inner)
935 throw AcceleratorMismatchError("TypedReadPort: null port pointer");
937 inner->connect(
938 [cb = std::move(callback)](MessageData) -> bool { return cb(); }, opts);
939 }
940
941 void read() {
942 MessageData outData;
943 inner->read(outData);
944 }
945
946 std::future<void> readAsync() {
947 auto innerFuture = inner->readAsync();
948 return std::async(
949 std::launch::deferred,
950 [f = std::move(innerFuture)]() mutable -> void { f.get(); });
951 }
952
953 void disconnect() { inner->disconnect(); }
954 bool isConnected() const { return inner && inner->isConnected(); }
955
956 ReadChannelPort &raw() { return *inner; }
957 const ReadChannelPort &raw() const { return *inner; }
958
959private:
961};
962
963//===----------------------------------------------------------------------===//
964// TypedFunction<ArgT, ResultT>
965//
966// Strongly-typed function-call wrapper built on top of TypedWritePort and
967// TypedReadPort. Implicitly constructible from `FuncService::Function *` (the
968// return type of `BundlePort::getAs<FuncService::Function>()`); construction
969// resolves the underlying raw "arg" / "result" channels but does not connect
970// them until `connect()` is called.
971//
972// We intentionally bypass `FuncService::Function::call`. That helper returns
973// the future from a single `result->readAsync()`, which only sees one raw
974// frame and would race when the response is multi-frame or when calls are
975// pipelined. Driving a `TypedReadPort<ResultT>` instead reuses its persistent
976// per-port deserializer (so partial frames / pending outputs survive between
977// calls) and its FIFO polling buffer (so pipelined `readAsync()` futures
978// hand back per-call decoded values in call order, even when consumers
979// `.get()` them out of order).
980//===----------------------------------------------------------------------===//
981
982namespace detail {
983
984/// Standard ConnectOptions for typed function ports: untranslated frames so
985/// the deserializer can see raw frame boundaries.
987 return ChannelPort::ConnectOptions(/*bufferSize=*/std::nullopt,
988 /*translateMessage=*/false);
989}
990
991/// Convert a `std::future<std::unique_ptr<T>>` (as returned by
992/// `TypedReadPort::readAsync()`) into a `std::future<T>`. Uses a deferred
993/// async so `.get()` blocks only when the caller actually waits for the
994/// value, preserving the per-call FIFO ordering of the underlying polling
995/// buffer. A null `unique_ptr` from a misbehaving deserializer is reported
996/// as a runtime error rather than dereferenced.
997template <typename T>
998std::future<T> awaitDecoded(std::future<std::unique_ptr<T>> inner) {
999 return std::async(std::launch::deferred,
1000 [fut = std::move(inner)]() mutable -> T {
1001 std::unique_ptr<T> v = fut.get();
1002 if (!v)
1003 throw std::runtime_error(
1004 "TypedFunction: deserializer produced a null "
1005 "value");
1006 return std::move(*v);
1007 });
1008}
1009
1010/// Throw the standard "null Function pointer" error used by every
1011/// TypedFunction specialization.
1012[[noreturn]] inline void throwNullFunction() {
1014 "TypedFunction: null Function pointer (getAs failed or wrong type)");
1015}
1016
1017/// Throw a clear "not connected" error from `call()` paths.
1018[[noreturn]] inline void throwNotConnected() {
1019 throw std::runtime_error("TypedFunction: must be 'connect'ed before "
1020 "calling");
1021}
1022
1023/// Throw a clear "already connected" error from `connect()` paths.
1024[[noreturn]] inline void throwAlreadyConnected() {
1025 throw std::runtime_error("TypedFunction is already connected");
1026}
1027
1028} // namespace detail
1029
1030template <typename ArgT, typename ResultT, bool SkipTypeCheck = false>
1032public:
1033 /// Implicit conversion from Function* (returned by getAs<>()).
1034 // NOLINTNEXTLINE(google-explicit-constructor)
1036 TypedFunction(const TypedFunction &) = delete;
1038
1039 void connect() {
1040 if (!inner)
1042 if (argPort)
1044 argPort.emplace(&inner->getRawWrite("arg"));
1045 resultPort.emplace(&inner->getRawRead("result"));
1047 argPort->connect(opts);
1048 resultPort->connect(opts);
1049 }
1050
1051 std::future<ResultT> call(const ArgT &arg) {
1052 if (!argPort)
1054 // Serialize the per-call write+readAsync pair so the polling buffer's
1055 // FIFO matches the call FIFO. Pipelined calls are still allowed -- the
1056 // shared deserializer drains responses in wire order and the FIFO
1057 // ensures call N's future resolves to call N's result, regardless of
1058 // the order in which callers `.get()` the futures.
1059 std::scoped_lock<std::mutex> lock(callMutex);
1060 argPort->write(arg);
1061 return detail::awaitDecoded<ResultT>(resultPort->readAsync());
1062 }
1063
1064 /// Emplace-style call: constructs `ArgT` in-place from the forwarded
1065 /// arguments and forwards to `call(const ArgT &)`. SFINAE-disabled for the
1066 /// single-`ArgT`-argument case so it does not shadow the lvalue overload.
1067 template <
1068 typename First, typename... Rest,
1069 typename = std::enable_if_t<
1070 std::is_constructible_v<ArgT, First, Rest...> &&
1071 (!std::is_same_v<std::decay_t<First>, ArgT> || sizeof...(Rest) != 0)>>
1072 std::future<ResultT> call(First &&first, Rest &&...rest) {
1073 return call(ArgT(std::forward<First>(first), std::forward<Rest>(rest)...));
1074 }
1075
1076 /// Function-call operator overloads: forward to `call()`.
1077 template <typename... Args>
1078 auto operator()(Args &&...args)
1079 -> decltype(this->call(std::forward<Args>(args)...)) {
1080 return call(std::forward<Args>(args)...);
1081 }
1082
1084 const services::FuncService::Function &raw() const { return *inner; }
1085
1086private:
1088 std::optional<TypedWritePort<ArgT, SkipTypeCheck>> argPort;
1089 std::optional<TypedReadPort<ResultT, SkipTypeCheck>> resultPort;
1090 std::mutex callMutex;
1091};
1092
1093/// Partial specialization: void argument, typed result.
1094template <typename ResultT, bool SkipTypeCheck>
1095class TypedFunction<void, ResultT, SkipTypeCheck> {
1096public:
1097 // NOLINTNEXTLINE(google-explicit-constructor)
1099 TypedFunction(const TypedFunction &) = delete;
1101
1102 void connect() {
1103 if (!inner)
1105 if (argPort)
1107 argPort.emplace(&inner->getRawWrite("arg"));
1108 resultPort.emplace(&inner->getRawRead("result"));
1110 argPort->connect(opts);
1111 resultPort->connect(opts);
1112 }
1113
1114 std::future<ResultT> call() {
1115 if (!argPort)
1117 std::scoped_lock<std::mutex> lock(callMutex);
1118 argPort->write();
1119 return detail::awaitDecoded<ResultT>(resultPort->readAsync());
1120 }
1121
1122 /// Function-call operator overload: forwards to `call()`.
1123 std::future<ResultT> operator()() { return call(); }
1124
1126 const services::FuncService::Function &raw() const { return *inner; }
1127
1128private:
1130 std::optional<TypedWritePort<void>> argPort;
1131 std::optional<TypedReadPort<ResultT, SkipTypeCheck>> resultPort;
1132 std::mutex callMutex;
1133};
1134
1135/// Partial specialization: typed argument, void result.
1136template <typename ArgT, bool SkipTypeCheck>
1137class TypedFunction<ArgT, void, SkipTypeCheck> {
1138public:
1139 // NOLINTNEXTLINE(google-explicit-constructor)
1141 TypedFunction(const TypedFunction &) = delete;
1143
1144 void connect() {
1145 if (!inner)
1147 if (argPort)
1149 argPort.emplace(&inner->getRawWrite("arg"));
1150 resultPort.emplace(&inner->getRawRead("result"));
1152 argPort->connect(opts);
1153 resultPort->connect(opts);
1154 }
1155
1156 std::future<void> call(const ArgT &arg) {
1157 if (!argPort)
1159 std::scoped_lock<std::mutex> lock(callMutex);
1160 argPort->write(arg);
1161 return resultPort->readAsync();
1162 }
1163
1164 /// Emplace-style call: constructs `ArgT` in-place from the forwarded
1165 /// arguments and forwards to `call(const ArgT &)`. SFINAE-disabled for the
1166 /// single-`ArgT`-argument case so it does not shadow the lvalue overload.
1167 template <
1168 typename First, typename... Rest,
1169 typename = std::enable_if_t<
1170 std::is_constructible_v<ArgT, First, Rest...> &&
1171 (!std::is_same_v<std::decay_t<First>, ArgT> || sizeof...(Rest) != 0)>>
1172 std::future<void> call(First &&first, Rest &&...rest) {
1173 return call(ArgT(std::forward<First>(first), std::forward<Rest>(rest)...));
1174 }
1175
1176 /// Function-call operator overloads: forward to `call()`.
1177 template <typename... Args>
1178 auto operator()(Args &&...args)
1179 -> decltype(this->call(std::forward<Args>(args)...)) {
1180 return call(std::forward<Args>(args)...);
1181 }
1182
1184 const services::FuncService::Function &raw() const { return *inner; }
1185
1186private:
1188 std::optional<TypedWritePort<ArgT, SkipTypeCheck>> argPort;
1189 std::optional<TypedReadPort<void>> resultPort;
1190 std::mutex callMutex;
1191};
1192
1193/// Full specialization: void argument, void result.
1194template <bool SkipTypeCheck>
1195class TypedFunction<void, void, SkipTypeCheck> {
1196public:
1197 // NOLINTNEXTLINE(google-explicit-constructor)
1199 TypedFunction(const TypedFunction &) = delete;
1201
1202 void connect() {
1203 if (!inner)
1205 if (argPort)
1207 argPort.emplace(&inner->getRawWrite("arg"));
1208 resultPort.emplace(&inner->getRawRead("result"));
1210 argPort->connect(opts);
1211 resultPort->connect(opts);
1212 }
1213
1214 std::future<void> call() {
1215 if (!argPort)
1217 std::scoped_lock<std::mutex> lock(callMutex);
1218 argPort->write();
1219 return resultPort->readAsync();
1220 }
1221
1222 /// Function-call operator overload: forwards to `call()`.
1223 std::future<void> operator()() { return call(); }
1224
1226 const services::FuncService::Function &raw() const { return *inner; }
1227
1228private:
1230 std::optional<TypedWritePort<void>> argPort;
1231 std::optional<TypedReadPort<void>> resultPort;
1232 std::mutex callMutex;
1233};
1234
1235//===----------------------------------------------------------------------===//
1236// TypedCallback<ArgT, ResultT>
1237//
1238// Strongly-typed accelerator-callback wrapper built on top of TypedReadPort
1239// and TypedWritePort. Implicitly constructible from `CallService::Callback *`
1240// (the return type of `BundlePort::getAs<CallService::Callback>()`);
1241// construction resolves the underlying raw "arg" / "result" channels but
1242// does not connect them until `connect()` is called.
1243//
1244// We bypass `CallService::Callback::connect` for the same reason we bypass
1245// `FuncService::Function::call`: typed args may span multiple raw frames and
1246// we need the persistent stateful deserializer that `TypedReadPort<ArgT>`
1247// already provides.
1248//
1249// Note: the `quick` parameter is kept for API compatibility but has no
1250// effect in this design. The typed user callback is always dispatched inline
1251// from the read-channel callback thread (matching the previous
1252// custom-deserializer behavior). If service-thread dispatch is needed in the
1253// future, that can be layered on top by using `TypedReadPort` polling mode
1254// plus a worker.
1255//
1256// Lifetime: the TypedReadPort callback installed on `connect()` captures
1257// `this`. The wrapper is non-copyable and non-movable, so its address is
1258// stable for its entire lifetime. Destruction of the embedded TypedReadPort
1259// synchronously revokes the callback and waits for any in-flight callback
1260// dispatch to complete (see ReadChannelPort::disconnect), so the captured
1261// `this` cannot dangle. Callers must ensure the TypedCallback outlives any
1262// expected callback dispatch -- using a temporary like
1263// `TypedCallback<...>(cb).connect(...)` is safe only because the destructor
1264// runs at end-of-full-expression and disconnects before returning.
1265//===----------------------------------------------------------------------===//
1266
1267namespace detail {
1268
1269/// Throw the standard "null Callback pointer" error used by every
1270/// TypedCallback specialization.
1271[[noreturn]] inline void throwNullCallback() {
1273 "TypedCallback: null Callback pointer (getAs failed or wrong type)");
1274}
1275
1276/// Throw a clear "already connected" error from TypedCallback `connect()`
1277/// paths.
1278[[noreturn]] inline void throwCallbackAlreadyConnected() {
1279 throw std::runtime_error("TypedCallback is already connected");
1280}
1281
1282} // namespace detail
1283
1284template <typename ArgT, typename ResultT, bool SkipTypeCheck = false>
1286public:
1287 // NOLINTNEXTLINE(google-explicit-constructor)
1289 TypedCallback(const TypedCallback &) = delete;
1291
1292 void connect(std::function<ResultT(const ArgT &)> callback,
1293 bool quick = false) {
1294 (void)quick; // See class header.
1295 if (!inner)
1297 if (argPort)
1299 argPort.emplace(&inner->getRawRead("arg"));
1300 resultPort.emplace(&inner->getRawWrite("result"));
1302 resultPort->connect(opts);
1303 userCallback = std::move(callback);
1304 // The TypedReadPort callback runs on the read-channel callback thread
1305 // for every decoded ArgT. We forward to the user's callback and then
1306 // serialize the result through the TypedWritePort.
1307 argPort->connect(
1308 [this](const ArgT &arg) -> bool {
1309 ResultT result = userCallback(arg);
1310 resultPort->write(result);
1311 return true;
1312 },
1313 opts);
1314 }
1315
1317 const services::CallService::Callback &raw() const { return *inner; }
1318
1319private:
1321 std::optional<TypedReadPort<ArgT, SkipTypeCheck>> argPort;
1322 std::optional<TypedWritePort<ResultT, SkipTypeCheck>> resultPort;
1323 std::function<ResultT(const ArgT &)> userCallback;
1324};
1325
1326/// Partial specialization: void argument, typed result.
1327template <typename ResultT, bool SkipTypeCheck>
1328class TypedCallback<void, ResultT, SkipTypeCheck> {
1329public:
1330 // NOLINTNEXTLINE(google-explicit-constructor)
1332 TypedCallback(const TypedCallback &) = delete;
1334
1335 void connect(std::function<ResultT()> callback, bool quick = false) {
1336 (void)quick;
1337 if (!inner)
1339 if (argPort)
1341 argPort.emplace(&inner->getRawRead("arg"));
1342 resultPort.emplace(&inner->getRawWrite("result"));
1344 resultPort->connect(opts);
1345 userCallback = std::move(callback);
1346 argPort->connect(
1347 [this]() -> bool {
1348 ResultT result = userCallback();
1349 resultPort->write(result);
1350 return true;
1351 },
1352 opts);
1353 }
1354
1356 const services::CallService::Callback &raw() const { return *inner; }
1357
1358private:
1360 std::optional<TypedReadPort<void>> argPort;
1361 std::optional<TypedWritePort<ResultT, SkipTypeCheck>> resultPort;
1362 std::function<ResultT()> userCallback;
1363};
1364
1365/// Partial specialization: typed argument, void result.
1366template <typename ArgT, bool SkipTypeCheck>
1367class TypedCallback<ArgT, void, SkipTypeCheck> {
1368public:
1369 // NOLINTNEXTLINE(google-explicit-constructor)
1371 TypedCallback(const TypedCallback &) = delete;
1373
1374 void connect(std::function<void(const ArgT &)> callback, bool quick = false) {
1375 (void)quick;
1376 if (!inner)
1378 if (argPort)
1380 argPort.emplace(&inner->getRawRead("arg"));
1381 resultPort.emplace(&inner->getRawWrite("result"));
1383 resultPort->connect(opts);
1384 userCallback = std::move(callback);
1385 argPort->connect(
1386 [this](const ArgT &arg) -> bool {
1387 userCallback(arg);
1388 resultPort->write();
1389 return true;
1390 },
1391 opts);
1392 }
1393
1395 const services::CallService::Callback &raw() const { return *inner; }
1396
1397private:
1399 std::optional<TypedReadPort<ArgT, SkipTypeCheck>> argPort;
1400 std::optional<TypedWritePort<void>> resultPort;
1401 std::function<void(const ArgT &)> userCallback;
1402};
1403
1404/// Full specialization: void argument, void result.
1405template <bool SkipTypeCheck>
1406class TypedCallback<void, void, SkipTypeCheck> {
1407public:
1408 // NOLINTNEXTLINE(google-explicit-constructor)
1410 TypedCallback(const TypedCallback &) = delete;
1412
1413 void connect(std::function<void()> callback, bool quick = false) {
1414 (void)quick;
1415 if (!inner)
1417 if (argPort)
1419 argPort.emplace(&inner->getRawRead("arg"));
1420 resultPort.emplace(&inner->getRawWrite("result"));
1422 resultPort->connect(opts);
1423 userCallback = std::move(callback);
1424 argPort->connect(
1425 [this]() -> bool {
1426 userCallback();
1427 resultPort->write();
1428 return true;
1429 },
1430 opts);
1431 }
1432
1434 const services::CallService::Callback &raw() const { return *inner; }
1435
1436private:
1438 std::optional<TypedReadPort<void>> argPort;
1439 std::optional<TypedWritePort<void>> resultPort;
1440 std::function<void()> userCallback;
1441};
1442
1443//===----------------------------------------------------------------------===//
1444// IndexedPorts<T> — wrapper over std::map<int, T>.
1445//
1446// Constructed once from a moved std::map (populated via `try_emplace` so T
1447// need not be movable or default-constructible). Supports both mutable and
1448// const iteration / indexing so callers can invoke non-const methods on the
1449// stored ports (e.g. `connect()` on a typed port wrapper).
1450//===----------------------------------------------------------------------===//
1451
1452template <typename T>
1454public:
1455 explicit IndexedPorts(std::map<int, T> &&ports) : ports_(std::move(ports)) {}
1458 IndexedPorts(const IndexedPorts &) = delete;
1460
1461 T &operator[](int idx) { return ports_.at(idx); }
1462 const T &operator[](int idx) const { return ports_.at(idx); }
1463 auto begin() { return ports_.begin(); }
1464 auto end() { return ports_.end(); }
1465 auto begin() const { return ports_.cbegin(); }
1466 auto end() const { return ports_.cend(); }
1467 size_t size() const { return ports_.size(); }
1468 bool contains(int idx) const { return ports_.count(idx) > 0; }
1469
1470private:
1471 std::map<int, T> ports_;
1472};
1473
1474//===----------------------------------------------------------------------===//
1475// Port-lookup helpers: findPortOrThrow, findPortAsOrThrow, findPortIndices.
1476//
1477// `findPortOrThrow` and `findPortAsOrThrow` are thin wrappers over
1478// `HWModule::resolvePort` that translate "not found" into a thrown
1479// `AcceleratorMismatchError` (instead of a nullptr return) and the typed
1480// variant additionally `dynamic_cast`s to a service-specific subclass. They
1481// exist so generated facade code does not have to repeat the same
1482// boilerplate (build a single-element AppIDPath, check for null, format an
1483// error string) for every port lookup it performs.
1484//===----------------------------------------------------------------------===//
1485
1486/// Look up a BundlePort by AppID in `module`. Throws AcceleratorMismatchError
1487/// if the port is not found. The returned reference is mutable: even though
1488/// `HWModule::resolvePort` is a const method, the underlying port map stores
1489/// `BundlePort&` references, so callers can invoke non-const methods on the
1490/// returned port (e.g. `getRawRead`, `connect`).
1491inline BundlePort &findPortOrThrow(HWModule *module, const AppID &id) {
1492 AppIDPath lastLookup;
1493 BundlePort *port = module->resolvePort(AppIDPath{id}, lastLookup);
1494 if (!port)
1495 throw AcceleratorMismatchError("Expected port '" + id.toString() +
1496 "' not found in module");
1497 return *port;
1498}
1499
1500/// Look up a BundlePort by AppID and cast it to `T`. Throws
1501/// AcceleratorMismatchError if the port is missing or has the wrong runtime
1502/// type. The returned pointer is mutable.
1503template <typename T>
1504T *findPortAsOrThrow(HWModule *module, const AppID &id) {
1505 BundlePort &port = findPortOrThrow(module, id);
1506 T *result = port.getAs<T>();
1507 if (!result)
1508 throw AcceleratorMismatchError("Port '" + id.toString() +
1509 "' has unexpected type (expected " +
1510 typeid(T).name() + ")");
1511 return result;
1512}
1513
1514/// Return a sorted vector of the `idx` values for every port whose AppID name
1515/// matches `name`. Ports without an index are ignored.
1516inline std::vector<uint32_t> findPortIndices(HWModule *module,
1517 const std::string &name) {
1518 std::vector<uint32_t> indices;
1519 for (const auto &[appid, port] : module->getPorts())
1520 if (appid.name == name && appid.idx.has_value())
1521 indices.push_back(appid.idx.value());
1522 std::sort(indices.begin(), indices.end());
1523 return indices;
1524}
1525
1526} // namespace esi
1527
1528#endif // ESI_TYPED_PORTS_H
assert(baseType &&"element must be base type")
Arrays have a compile time specified (static) size and an element type.
Definition Types.h:301
Bit vectors include signed, unsigned, and signless integers.
Definition Types.h:211
Bits are just an array of bits.
Definition Types.h:224
Services provide connections to 'bundles' – collections of named, unidirectional communication channe...
Definition Ports.h:637
T * getAs() const
Cast this Bundle port to a subclass which is actually useful.
Definition Ports.h:665
Unidirectional channels are the basic communication primitive between the host and accelerator.
Definition Ports.h:115
const Type * getType() const
Definition Ports.h:219
const WindowType * getWindowType() const
If this port carries a windowed type, return the original WindowType (whose intoType is what getType(...
Definition Ports.h:224
Represents either the top level or an instance of a hardware module.
Definition Design.h:47
const std::map< AppID, BundlePort & > & getPorts() const
Access the module's ports by ID.
Definition Design.h:80
const T & operator[](int idx) const
std::map< int, T > ports_
IndexedPorts(std::map< int, T > &&ports)
auto end() const
auto begin() const
IndexedPorts & operator=(const IndexedPorts &)=delete
IndexedPorts(IndexedPorts &&)=default
IndexedPorts & operator=(IndexedPorts &&)=default
bool contains(int idx) const
T & operator[](int idx)
size_t size() const
IndexedPorts(const IndexedPorts &)=delete
A concrete flat message backed by a single vector of bytes.
Definition Common.h:197
const uint8_t * getBytes() const
Definition Common.h:208
const T * as() const
Cast to a type.
Definition Common.h:232
size_t getSize() const
Get the size of the data in bytes.
Definition Common.h:222
detail::TypedReadOwnedCallback< T > OutputCallback
Definition TypedPorts.h:241
QueuedDecodeTypeDeserializer(OutputCallback output)
Definition TypedPorts.h:244
virtual ~QueuedDecodeTypeDeserializer()=default
std::queue< std::unique_ptr< SerialCoordOutputBatch > > pendingOutputs
Definition TypedPorts.h:311
bool poke()
Retry delivery of any typed outputs which were previously blocked by the client callback.
Definition TypedPorts.h:284
bool push(std::unique_ptr< SegmentedMessageData > &msg)
Push one raw message into the deserializer.
Definition TypedPorts.h:254
virtual DecodedOutputs decode(std::unique_ptr< SegmentedMessageData > &msg)=0
Decode one raw message into zero or more typed outputs.
std::vector< std::unique_ptr< T > > DecodedOutputs
Definition TypedPorts.h:242
A ChannelPort which reads data from the accelerator.
Definition Ports.h:479
virtual void connect(ReadCallback callback, const ConnectOptions &options={})
Definition Ports.cpp:140
static constexpr uint64_t DefaultMaxDataQueueMsgs
Default max data queue size set at connect time.
Definition Ports.h:530
Signed integer.
Definition Types.h:242
Abstract multi-segment message.
Definition Common.h:151
virtual MessageData toMessageData() const
Flatten all segments into a standard MessageData.
Definition Common.cpp:61
std::vector< data_frame > pending_frames_
Definition TypedPorts.h:409
typename Base::DecodedOutputs DecodedOutputs
Definition TypedPorts.h:337
DecodedOutputs decode(std::unique_ptr< SegmentedMessageData > &msg) override
Decode one raw message into zero or more typed outputs.
Definition TypedPorts.h:351
typename Base::OutputCallback OutputCallback
Definition TypedPorts.h:336
std::optional< header_frame > pending_header_
Definition TypedPorts.h:408
SerialListTypeDeserializer(OutputCallback output)
Definition TypedPorts.h:339
QueuedDecodeTypeDeserializer< T > Base
Definition TypedPorts.h:335
typename T::header_frame header_frame
Definition TypedPorts.h:343
static constexpr size_t kFrameSize
Definition TypedPorts.h:349
typename T::count_type count_type
Definition TypedPorts.h:345
typename T::data_frame data_frame
Definition TypedPorts.h:344
std::vector< uint8_t > partial_
Definition TypedPorts.h:407
Root class of the ESI type system.
Definition Types.h:36
std::string toString(bool oneLine=false) const
Definition Types.cpp:120
ID getID() const
Definition Types.h:42
services::CallService::Callback * inner
std::optional< TypedReadPort< ArgT, SkipTypeCheck > > argPort
std::optional< TypedWritePort< void > > resultPort
const services::CallService::Callback & raw() const
std::function< void(const ArgT &)> userCallback
void connect(std::function< void(const ArgT &)> callback, bool quick=false)
TypedCallback(const TypedCallback &)=delete
services::CallService::Callback & raw()
TypedCallback & operator=(const TypedCallback &)=delete
TypedCallback(services::CallService::Callback *cb)
std::optional< TypedWritePort< ResultT, SkipTypeCheck > > resultPort
services::CallService::Callback & raw()
std::optional< TypedReadPort< void > > argPort
void connect(std::function< ResultT()> callback, bool quick=false)
TypedCallback(services::CallService::Callback *cb)
const services::CallService::Callback & raw() const
services::CallService::Callback * inner
TypedCallback & operator=(const TypedCallback &)=delete
std::optional< TypedWritePort< void > > resultPort
std::optional< TypedReadPort< void > > argPort
TypedCallback(services::CallService::Callback *cb)
services::CallService::Callback & raw()
void connect(std::function< void()> callback, bool quick=false)
TypedCallback & operator=(const TypedCallback &)=delete
TypedCallback(const TypedCallback &)=delete
const services::CallService::Callback & raw() const
services::CallService::Callback * inner
services::CallService::Callback * inner
std::optional< TypedReadPort< ArgT, SkipTypeCheck > > argPort
services::CallService::Callback & raw()
void connect(std::function< ResultT(const ArgT &)> callback, bool quick=false)
TypedCallback(services::CallService::Callback *cb)
std::optional< TypedWritePort< ResultT, SkipTypeCheck > > resultPort
TypedCallback & operator=(const TypedCallback &)=delete
std::function< ResultT(const ArgT &)> userCallback
const services::CallService::Callback & raw() const
TypedCallback(const TypedCallback &)=delete
auto operator()(Args &&...args) -> decltype(this->call(std::forward< Args >(args)...))
Function-call operator overloads: forward to call().
TypedFunction & operator=(const TypedFunction &)=delete
std::optional< TypedWritePort< ArgT, SkipTypeCheck > > argPort
std::future< void > call(First &&first, Rest &&...rest)
Emplace-style call: constructs ArgT in-place from the forwarded arguments and forwards to call(const ...
TypedFunction(const TypedFunction &)=delete
std::optional< TypedReadPort< void > > resultPort
const services::FuncService::Function & raw() const
services::FuncService::Function & raw()
std::future< void > call(const ArgT &arg)
services::FuncService::Function * inner
TypedFunction(services::FuncService::Function *func)
std::optional< TypedReadPort< ResultT, SkipTypeCheck > > resultPort
TypedFunction & operator=(const TypedFunction &)=delete
std::optional< TypedWritePort< void > > argPort
services::FuncService::Function & raw()
services::FuncService::Function * inner
std::future< ResultT > operator()()
Function-call operator overload: forwards to call().
const services::FuncService::Function & raw() const
TypedFunction(services::FuncService::Function *func)
const services::FuncService::Function & raw() const
TypedFunction(const TypedFunction &)=delete
TypedFunction & operator=(const TypedFunction &)=delete
services::FuncService::Function * inner
TypedFunction(services::FuncService::Function *func)
std::optional< TypedWritePort< void > > argPort
std::optional< TypedReadPort< void > > resultPort
services::FuncService::Function & raw()
std::future< void > operator()()
Function-call operator overload: forwards to call().
std::optional< TypedWritePort< ArgT, SkipTypeCheck > > argPort
std::optional< TypedReadPort< ResultT, SkipTypeCheck > > resultPort
std::future< ResultT > call(First &&first, Rest &&...rest)
Emplace-style call: constructs ArgT in-place from the forwarded arguments and forwards to call(const ...
TypedFunction(const TypedFunction &)=delete
services::FuncService::Function & raw()
std::mutex callMutex
auto operator()(Args &&...args) -> decltype(this->call(std::forward< Args >(args)...))
Function-call operator overloads: forward to call().
std::future< ResultT > call(const ArgT &arg)
TypedFunction & operator=(const TypedFunction &)=delete
const services::FuncService::Function & raw() const
TypedFunction(services::FuncService::Function *func)
Implicit conversion from Function* (returned by getAs<>()).
services::FuncService::Function * inner
TypedReadPort(const TypedReadPort &)=delete
TypedReadPort & operator=(TypedReadPort &&)=delete
TypedReadPort & operator=(const TypedReadPort &)=delete
ReadChannelPort * inner
Definition TypedPorts.h:960
TypedReadPort(ReadChannelPort &port)
Definition TypedPorts.h:910
TypedReadPort(ReadChannelPort *port)
Definition TypedPorts.h:912
TypedReadPort(TypedReadPort &&)=delete
std::future< void > readAsync()
Definition TypedPorts.h:946
void connect(const ChannelPort::ConnectOptions &opts={std::nullopt, false})
Definition TypedPorts.h:923
ReadChannelPort & raw()
Definition TypedPorts.h:956
void connect(std::function< bool()> callback, const ChannelPort::ConnectOptions &opts={std::nullopt, false})
Definition TypedPorts.h:931
const ReadChannelPort & raw() const
Definition TypedPorts.h:957
TypedReadPort(ReadChannelPort &port)
Definition TypedPorts.h:735
ReadChannelPort & raw()
Definition TypedPorts.h:861
detail::DeserializerFor< T > Deserializer
Definition TypedPorts.h:865
ReadChannelPort * inner
Definition TypedPorts.h:898
TypedReadPort & operator=(const TypedReadPort &)=delete
std::unique_ptr< T > read()
Blocking typed read in polling mode.
Definition TypedPorts.h:816
void connect(const ChannelPort::ConnectOptions &opts={std::nullopt, false})
Connect in polling mode.
Definition TypedPorts.h:752
uint64_t maxDataQueueMsgs
Definition TypedPorts.h:901
void disconnect()
Disconnect the typed port and abandon any pending polling reads.
Definition TypedPorts.h:854
void setMaxDataQueueMsgs(uint64_t maxMsgs)
Set the maximum number of decoded typed values buffered in polling mode.
Definition TypedPorts.h:847
void emplaceDeserializer(detail::DeserializerOutputCallback< T > callback)
Definition TypedPorts.h:874
void connect(std::function< bool(const T &)> callback, const ChannelPort::ConnectOptions &opts={std::nullopt, false})
Connect a non-owning typed callback.
Definition TypedPorts.h:779
const ReadChannelPort & raw() const
Definition TypedPorts.h:862
TypedReadPort & operator=(TypedReadPort &&)=delete
std::optional< PollingState > pollingState
Definition TypedPorts.h:903
TypedReadPort(const TypedReadPort &)=delete
std::optional< Deserializer > deserializer
Definition TypedPorts.h:902
void connect(detail::TypedReadOwnedCallback< T > callback, const ChannelPort::ConnectOptions &opts={std::nullopt, false})
Connect an owning typed callback.
Definition TypedPorts.h:792
detail::PollingBuffer< std::unique_ptr< T > > PollingState
Definition TypedPorts.h:867
std::future< std::unique_ptr< T > > readAsync()
Asynchronous typed read in polling mode.
Definition TypedPorts.h:825
TypedReadPort(ReadChannelPort *port)
Definition TypedPorts.h:737
TypedReadPort(TypedReadPort &&)=delete
bool isConnected() const
Definition TypedPorts.h:859
WriteChannelPort * inner
Definition TypedPorts.h:716
TypedWritePort(WriteChannelPort &port)
Definition TypedPorts.h:686
TypedWritePort(WriteChannelPort *port)
Definition TypedPorts.h:688
WriteChannelPort & raw()
Definition TypedPorts.h:712
void connect(const ChannelPort::ConnectOptions &opts={std::nullopt, false})
Definition TypedPorts.h:690
const WriteChannelPort & raw() const
Definition TypedPorts.h:713
const WriteChannelPort & raw() const
Definition TypedPorts.h:675
void connect(const ChannelPort::ConnectOptions &opts={std::nullopt, false})
Definition TypedPorts.h:619
WriteChannelPort & raw()
Definition TypedPorts.h:674
void write(std::unique_ptr< T > &data)
Write by taking ownership.
Definition TypedPorts.h:633
bool tryWrite(const T &data)
Definition TypedPorts.h:644
bool isConnected() const
Definition TypedPorts.h:672
bool tryWrite(std::unique_ptr< T > &data)
Non-blocking write by taking ownership.
Definition TypedPorts.h:652
TypedWritePort(WriteChannelPort &port)
Definition TypedPorts.h:615
void write(const T &data)
Definition TypedPorts.h:629
TypedWritePort(WriteChannelPort *port)
Definition TypedPorts.h:617
WriteChannelPort * inner
Definition TypedPorts.h:678
Unsigned integer.
Definition Types.h:253
The "void" type is a special type which can be used to represent no type.
Definition Types.h:141
Windows represent a fixed-size sliding window over a stream of data.
Definition Types.h:334
A ChannelPort which sends data to the accelerator.
Definition Ports.h:308
virtual void connect(const ConnectOptions &options={}) override
Set up a connection to the accelerator.
Definition Ports.h:312
Default deserializer for simple 1:1 typed reads.
Definition TypedPorts.h:167
bool push(std::unique_ptr< SegmentedMessageData > &msg)
Definition TypedPorts.h:174
TypedReadOwnedCallback< T > OutputCallback
Definition TypedPorts.h:169
std::unique_ptr< T > pendingOutput
Definition TypedPorts.h:211
PODTypeDeserializer(OutputCallback output, WireInfo wireInfo)
Definition TypedPorts.h:171
Shared queue/promise helper for polling-style read APIs.
Definition Ports.h:43
A function call which gets attached to a service port.
Definition Services.h:437
A function call which gets attached to a service port.
Definition Services.h:385
constexpr bool has_type_deserializer_v
Definition TypedPorts.h:149
std::function< bool(std::unique_ptr< T > &)> TypedReadOwnedCallback
Owning callback used by typed read deserializers.
Definition TypedPorts.h:139
ChannelPort::ConnectOptions typedFunctionConnectOptions()
Standard ConnectOptions for typed function ports: untranslated frames so the deserializer can see raw...
Definition TypedPorts.h:986
void throwAlreadyConnected()
Throw a clear "already connected" error from connect() paths.
typename DeserializerSelector< T >::type DeserializerFor
Definition TypedPorts.h:225
void throwNullFunction()
Throw the standard "null Function pointer" error used by every TypedFunction specialization.
void throwNotConnected()
Throw a clear "not connected" error from call() paths.
void throwCallbackAlreadyConnected()
Throw a clear "already connected" error from TypedCallback connect() paths.
void throwNullCallback()
Throw the standard "null Callback pointer" error used by every TypedCallback specialization.
typename DeserializerFor< T >::OutputCallback DeserializerOutputCallback
Definition TypedPorts.h:228
std::future< T > awaitDecoded(std::future< std::unique_ptr< T > > inner)
Convert a std::future<std::unique_ptr<T>> (as returned by TypedReadPort::readAsync()) into a std::fut...
Definition TypedPorts.h:998
const MessageData & getMessageDataRef(const SegmentedMessageData &msg, MessageData &scratch)
Definition TypedPorts.h:152
Definition esi.py:1
std::string toString(const std::any &a)
'Stringify' a std::any. This is used to log std::any values by some loggers.
Definition Logging.cpp:132
void verifyTypeCompatibility(const Type *portType)
Definition TypedPorts.h:456
MessageData toMessageData(const T &data, WireInfo wi)
Pack a C++ integral value into a MessageData with the given wire byte count.
Definition TypedPorts.h:80
std::vector< uint32_t > findPortIndices(HWModule *module, const std::string &name)
Return a sorted vector of the idx values for every port whose AppID name matches name.
WireInfo getWireInfo(const Type *portType)
Definition TypedPorts.h:70
BundlePort & findPortOrThrow(HWModule *module, const AppID &id)
Look up a BundlePort by AppID in module.
constexpr bool has_esi_id_v
Definition TypedPorts.h:425
T * findPortAsOrThrow(HWModule *module, const AppID &id)
Look up a BundlePort by AppID and cast it to T.
constexpr bool has_esi_window_id_v
Definition TypedPorts.h:437
constexpr bool is_std_array_v
Definition TypedPorts.h:453
T fromMessageData(const MessageData &msg, WireInfo wi)
Unpack a MessageData into a C++ integral value with the given wire info.
Definition TypedPorts.h:103
const Type * unwrapTypeAlias(const Type *type)
Unwrap TypeAliasType (possibly recursively) to get the underlying type.
Definition Types.h:193
Compute the wire byte count for a port type.
Definition TypedPorts.h:65
size_t bitWidth
Definition TypedPorts.h:67
PODTypeDeserializer< T > type
Definition TypedPorts.h:216