CodexUI/src/codex/nodegraph/ThreadChannels.h at master · SNodeC/CodexUI · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
// SPDX-License-Identifier: LGPL-3.0-or-later OR MIT

#ifndef CODEXUI_CODEX_NODEGRAPH_THREADCHANNELS_H
#define CODEXUI_CODEX_NODEGRAPH_THREADCHANNELS_H

#include "codex/nodegraph/EventFd.h"
#include "codex/nodegraph/Messages.h"
#include "codex/nodegraph/SpscQueue.h"

#include <atomic>
#include <cstddef>
#include <cstdint>

namespace codexui::nodegraph {

enum class ChannelSendStatus : std::uint8_t {
  Accepted,
  QueueFull,
  CoalescedRescan,
  AcceptedWakeFailed,
  CoalescedRescanWakeFailed,
};

[[nodiscard]] bool messageAdmitted(ChannelSendStatus status) noexcept;
// Normal eventfd delivery and the bounded fallback drains both guarantee that
// an admitted payload is consumed exactly once. Wake failure remains visible
// so callers can report degraded delivery without retrying a mutation.
[[nodiscard]] bool deliveryGuaranteed(ChannelSendStatus status) noexcept;
[[nodiscard]] bool wakeFailed(ChannelSendStatus status) noexcept;

// The application owns exactly one instance. It contains exactly one bounded
// SPSC queue and one eventfd in each direction; eventfds carry wake counts
// only.
class ThreadChannels final {
public:
  static constexpr std::size_t WorkerToQtCapacity = 512;
  static constexpr std::size_t QtToWorkerCapacity = 256;
  // Larger transactions are already committed and are cheaper for Qt to
  // rediscover through its bounded graph scans than to process as one event.
  static constexpr std::size_t MaximumDirectGraphReferences = 64;
  // Graph bursts and non-authoritative diagnostics stop before the final slot,
  // which remains available for terminal WorkerStopped delivery.
  static constexpr std::size_t WorkerToQtReservedSlots = 1;

  ThreadChannels() = default;
  ThreadChannels(const ThreadChannels &) = delete;
  ThreadChannels &operator=(const ThreadChannels &) = delete;

  [[nodiscard]] bool valid() const noexcept;
  [[nodiscard]] int workerToQtEventFd() const noexcept;
  [[nodiscard]] int qtToWorkerEventFd() const noexcept;
  [[nodiscard]] int workerToQtCreationError() const noexcept;
  [[nodiscard]] int qtToWorkerCreationError() const noexcept;

  // Worker-thread producer operations.
  [[nodiscard]] ChannelSendStatus sendGraphChanged(GraphChange change);
  [[nodiscard]] ChannelSendStatus
  sendProtocolDiagnostic(ProtocolDiagnostic &diagnostic);
  [[nodiscard]] ChannelSendStatus sendWorkerStopped(WorkerStopped &stopped);

  // Qt-thread consumer operations.
  [[nodiscard]] EventFd::DrainResult drainWorkerToQtWake() const noexcept;
  [[nodiscard]] bool tryReceiveForQt(WorkerToQtMessage &message);

  // Qt-thread producer operations. The supplied payload remains intact when
  // QueueFull is returned, so the user's text/attachments stay available.
  [[nodiscard]] ChannelSendStatus sendNodeAction(NodeAction &action);
  [[nodiscard]] ChannelSendStatus sendRuntimeAction(RuntimeAction &action);
  [[nodiscard]] ChannelSendStatus sendShutdown(ShutdownRequest &request);

  // Worker-thread consumer operations.
  [[nodiscard]] EventFd::DrainResult drainQtToWorkerWake() const noexcept;
  [[nodiscard]] bool tryReceiveForWorker(QtToWorkerMessage &message);

  [[nodiscard]] std::size_t workerToQtSizeApprox() const noexcept;
  [[nodiscard]] std::size_t qtToWorkerSizeApprox() const noexcept;
  [[nodiscard]] bool rescanPending() const noexcept;

  // Deterministic syscall-failure seams used by the headless recovery tests.
  // They suppress one wake only; payload admission and FIFO storage are real.
  void failNextWorkerToQtWakeForTest() noexcept;
  void failNextQtToWorkerWakeForTest() noexcept;

  // Call only after both event-loop observers are disabled and the worker is
  // joined.
  void close() noexcept;

private:
  [[nodiscard]] ChannelSendStatus wakeWorkerToQt(bool coalesced) const noexcept;
  [[nodiscard]] ChannelSendStatus wakeQtToWorker() const noexcept;
  void requireRescan(std::uint64_t revision) noexcept;

  SpscQueue<WorkerToQtMessage, WorkerToQtCapacity> workerToQt_;
  SpscQueue<QtToWorkerMessage, QtToWorkerCapacity> qtToWorker_;
  EventFd workerToQtWake_;
  EventFd qtToWorkerWake_;
  std::atomic<std::uint64_t> rescanRevision_{0};
  std::atomic<std::uint64_t> pendingProviderAuthorityRevision_{0};
  std::atomic_bool closed_{false};
  mutable std::atomic_bool failNextWorkerToQtWake_{false};
  mutable std::atomic_bool failNextQtToWorkerWake_{false};
  bool queuedTurnAfterRescan_ = false;
};

} // namespace codexui::nodegraph

#endif // CODEXUI_CODEX_NODEGRAPH_THREADCHANNELS_H