Major functionality implemented (missing reconfig).

2026-04-14 01:40:43 +02:00
parent 96d98dfc3d
commit 3a1ecd3aba
6 changed files with 1173 additions and 107 deletions
@@ -9,6 +9,7 @@
 #include "MemoryMapBroker.h"
 #include "ObjectBuilder.h"
 #include "ObjectRegistryDatabase.h"
+#include "Threads.h"
 #include "Vec.h"

 // Original broker headers
@@ -55,17 +56,62 @@ static Reference FindByNameRecursive(ReferenceContainer *container,
 */
 class DebugBrokerHelper {
 public:
+  // Evaluate a break condition against the current live value of a signal.
+  // Returns true if the condition is met and execution should be paused.
+  // Only called when signal->breakOp != BREAK_OFF; reads memoryAddress directly.
+  static bool EvaluateBreak(const DebugSignalInfo *s) {
+    if (s->memoryAddress == NULL_PTR(void *)) return false;
+    float64 val = 0.0;
+    const TypeDescriptor &t = s->type;
+    if      (t == Float64Bit)           val = *static_cast<const float64 *>(s->memoryAddress);
+    else if (t == Float32Bit)           val = *static_cast<const float32 *>(s->memoryAddress);
+    else if (t == UnsignedInteger32Bit) val = *static_cast<const uint32  *>(s->memoryAddress);
+    else if (t == SignedInteger32Bit)   val = *static_cast<const int32   *>(s->memoryAddress);
+    else if (t == UnsignedInteger64Bit) val = static_cast<float64>(*static_cast<const uint64 *>(s->memoryAddress));
+    else if (t == SignedInteger64Bit)   val = static_cast<float64>(*static_cast<const int64  *>(s->memoryAddress));
+    else if (t == UnsignedInteger16Bit) val = *static_cast<const uint16  *>(s->memoryAddress);
+    else if (t == SignedInteger16Bit)   val = *static_cast<const int16   *>(s->memoryAddress);
+    else if (t == UnsignedInteger8Bit)  val = *static_cast<const uint8   *>(s->memoryAddress);
+    else if (t == SignedInteger8Bit)    val = *static_cast<const int8    *>(s->memoryAddress);
+    else return false; // unsupported type — skip
+    const float64 thr = s->breakThreshold;
+    switch (s->breakOp) {
+      case BREAK_GT:  return val >  thr;
+      case BREAK_LT:  return val <  thr;
+      case BREAK_EQ:  return val == thr;
+      case BREAK_GEQ: return val >= thr;
+      case BREAK_LEQ: return val <= thr;
+      case BREAK_NEQ: return val != thr;
+      default: return false;
+    }
+  }
+
+  // Spin-wait point for output brokers — called from Execute() AFTER Process().
+  // Spins while paused, then consumes one step counter tick (if stepping).
+  // Input brokers must NOT call this — they complete normally to avoid blocking
+  // cross-thread EventSem posts (RealTimeThreadSynchBroker would time out).
+  static void OutputPauseAndStep(DebugService *service, const char8 *gamName) {
+    if (service == NULL_PTR(DebugService *)) return;
+    // Wait if already paused (manual PAUSE or breakpoint from a previous cycle)
+    while (service->IsPaused()) Sleep::MSec(10);
+    // Pass the OS thread name so per-thread step filtering works.
+    const char8 *tName = Threads::Name(Threads::Id());
+    service->ConsumeStepIfNeeded(gamName, tName);
+    while (service->IsPaused()) Sleep::MSec(10);
+  }
+
  static void Process(DebugService *service,
                      DebugSignalInfo **signalInfoPointers,
                      Vec<uint32> &activeIndices, Vec<uint32> &activeSizes,
-                      FastPollingMutexSem &activeMutex) {
+                      FastPollingMutexSem &activeMutex,
+                      volatile bool *anyBreakFlag,
+                      Vec<uint32> *breakIndices) {
    if (service == NULL_PTR(DebugService *))
      return;

-    // Re-establish break logic
-    while (service->IsPaused()) {
-      Sleep::MSec(10);
-    }
+    // NOTE: No spin here. Spinning for paused state is handled in Execute() of
+    // OUTPUT brokers only (see OutputPauseAndStep). Input brokers must not block
+    // because that prevents cross-thread EventSem posts from completing.

    activeMutex.FastLock();
    uint32 n = activeIndices.Size();
@@ -83,6 +129,23 @@ public:
        }
      }
    }
+
+    // Conditional break check — zero cost when anyBreakFlag is false
+    // (single volatile read; the RT loop never pays for this when no break is set).
+    if (*anyBreakFlag && !service->IsPaused() &&
+        breakIndices != NULL_PTR(Vec<uint32> *)) {
+      uint32 nb = breakIndices->Size();
+      for (uint32 i = 0; i < nb; i++) {
+        uint32 idx = (*breakIndices)[i];
+        DebugSignalInfo *s = signalInfoPointers[idx];
+        if (s != NULL_PTR(DebugSignalInfo *) && s->breakOp != BREAK_OFF &&
+            EvaluateBreak(s)) {
+          service->SetPaused(true);
+          break;
+        }
+      }
+    }
+
    activeMutex.FastUnLock();
  }

@@ -93,7 +156,8 @@ public:
              uint32 numCopies, MemoryMapBrokerCopyTableEntry *copyTable,
              const char8 *functionName, SignalDirection direction,
              volatile bool *anyActiveFlag, Vec<uint32> *activeIndices,
-              Vec<uint32> *activeSizes, FastPollingMutexSem *activeMutex) {
+              Vec<uint32> *activeSizes, FastPollingMutexSem *activeMutex,
+              volatile bool *anyBreakFlag, Vec<uint32> *breakIndices) {
    if (numCopies > 0) {
      signalInfoPointers = new DebugSignalInfo *[numCopies];
      for (uint32 i = 0; i < numCopies; i++)
@@ -181,8 +245,12 @@ public:
      }

      // Register broker in DebugService for optimized control
+      bool isOutputBroker = (direction == OutputSignals);
      service->RegisterBroker(signalInfoPointers, numCopies, mmb, anyActiveFlag,
-                              activeIndices, activeSizes, activeMutex);
+                              activeIndices, activeSizes, activeMutex,
+                              anyBreakFlag, breakIndices,
+                              (functionName != NULL_PTR(const char8 *)) ? functionName : dsPath.Buffer(),
+                              isOutputBroker);
    }
  }
 };
@@ -197,6 +265,9 @@ public:
    signalInfoPointers = NULL_PTR(DebugSignalInfo **);
    numSignals = 0;
    anyActive = false;
+    anyBreakActive = false;
+    isOutput = false;
+    gamName[0] = '\0';
  }

  virtual ~DebugBrokerWrapper() {
@@ -206,9 +277,15 @@ public:

  virtual bool Execute() {
    bool ret = BaseClass::Execute();
-    if (ret && (anyActive || (service && service->IsPaused()))) {
+    if (ret && (anyActive || anyBreakActive)) {
      DebugBrokerHelper::Process(service, signalInfoPointers, activeIndices,
-                                 activeSizes, activeMutex);
+                                 activeSizes, activeMutex,
+                                 &anyBreakActive, &breakIndices);
+    }
+    // Output brokers are the safe pause point: base Execute has already
+    // committed data / posted any cross-thread EventSems.
+    if (ret && isOutput) {
+      DebugBrokerHelper::OutputPauseAndStep(service, gamName);
    }
    return ret;
  }
@@ -220,10 +297,13 @@ public:
            direction == InputSignals ? "In" : "Out");
    if (ret) {
      numSignals = this->GetNumberOfCopies();
+      isOutput = (direction == OutputSignals);
+      StringHelper::CopyN(gamName, name, 255u);
      DebugBrokerHelper::InitSignals(this, ds, service, signalInfoPointers,
                                     numSignals, this->copyTable, name,
                                     direction, &anyActive, &activeIndices,
-                                     &activeSizes, &activeMutex);
+                                     &activeSizes, &activeMutex,
+                                     &anyBreakActive, &breakIndices);
    }
    return ret;
  }
@@ -235,10 +315,13 @@ public:
            direction == InputSignals ? "In" : "Out");
    if (ret) {
      numSignals = this->GetNumberOfCopies();
+      isOutput = (direction == OutputSignals);
+      StringHelper::CopyN(gamName, name, 255u);
      DebugBrokerHelper::InitSignals(this, ds, service, signalInfoPointers,
                                     numSignals, this->copyTable, name,
                                     direction, &anyActive, &activeIndices,
-                                     &activeSizes, &activeMutex);
+                                     &activeSizes, &activeMutex,
+                                     &anyBreakActive, &breakIndices);
    }
    return ret;
  }
@@ -247,8 +330,12 @@ public:
  DebugSignalInfo **signalInfoPointers;
  uint32 numSignals;
  volatile bool anyActive;
+  volatile bool anyBreakActive;
+  bool isOutput;
+  char8 gamName[256];
  Vec<uint32> activeIndices;
  Vec<uint32> activeSizes;
+  Vec<uint32> breakIndices;
  FastPollingMutexSem activeMutex;
 };

@@ -260,6 +347,9 @@ public:
    signalInfoPointers = NULL_PTR(DebugSignalInfo **);
    numSignals = 0;
    anyActive = false;
+    anyBreakActive = false;
+    isOutput = false;
+    gamName[0] = '\0';
  }

  virtual ~DebugBrokerWrapperNoOptim() {
@@ -269,9 +359,13 @@ public:

  virtual bool Execute() {
    bool ret = BaseClass::Execute();
-    if (ret && (anyActive || (service && service->IsPaused()))) {
+    if (ret && (anyActive || anyBreakActive)) {
      DebugBrokerHelper::Process(service, signalInfoPointers, activeIndices,
-                                 activeSizes, activeMutex);
+                                 activeSizes, activeMutex,
+                                 &anyBreakActive, &breakIndices);
+    }
+    if (ret && isOutput) {
+      DebugBrokerHelper::OutputPauseAndStep(service, gamName);
    }
    return ret;
  }
@@ -281,10 +375,13 @@ public:
    bool ret = BaseClass::Init(direction, ds, name, gamMem);
    if (ret) {
      numSignals = this->GetNumberOfCopies();
+      isOutput = (direction == OutputSignals);
+      StringHelper::CopyN(gamName, name, 255u);
      DebugBrokerHelper::InitSignals(this, ds, service, signalInfoPointers,
                                     numSignals, this->copyTable, name,
                                     direction, &anyActive, &activeIndices,
-                                     &activeSizes, &activeMutex);
+                                     &activeSizes, &activeMutex,
+                                     &anyBreakActive, &breakIndices);
    }
    return ret;
  }
@@ -293,8 +390,12 @@ public:
  DebugSignalInfo **signalInfoPointers;
  uint32 numSignals;
  volatile bool anyActive;
+  volatile bool anyBreakActive;
+  bool isOutput;
+  char8 gamName[256];
  Vec<uint32> activeIndices;
  Vec<uint32> activeSizes;
+  Vec<uint32> breakIndices;
  FastPollingMutexSem activeMutex;
 };

@@ -305,6 +406,8 @@ public:
    signalInfoPointers = NULL_PTR(DebugSignalInfo **);
    numSignals = 0;
    anyActive = false;
+    anyBreakActive = false;
+    gamName[0] = '\0';
  }
  virtual ~DebugMemoryMapAsyncOutputBroker() {
    if (signalInfoPointers)
@@ -312,9 +415,14 @@ public:
  }
  virtual bool Execute() {
    bool ret = MemoryMapAsyncOutputBroker::Execute();
-    if (ret && (anyActive || (service && service->IsPaused()))) {
+    if (ret && (anyActive || anyBreakActive)) {
      DebugBrokerHelper::Process(service, signalInfoPointers, activeIndices,
-                                 activeSizes, activeMutex);
+                                 activeSizes, activeMutex,
+                                 &anyBreakActive, &breakIndices);
+    }
+    // Async output brokers are always output direction
+    if (ret) {
+      DebugBrokerHelper::OutputPauseAndStep(service, gamName);
    }
    return ret;
  }
@@ -330,10 +438,11 @@ public:
        numberOfBuffersIn, cpuMaskIn, stackSizeIn);
    if (ret) {
      numSignals = this->GetNumberOfCopies();
+      StringHelper::CopyN(gamName, (functionName != NULL_PTR(const char8 *)) ? functionName : "", 255u);
      DebugBrokerHelper::InitSignals(
          this, dataSourceIn, service, signalInfoPointers, numSignals,
          this->copyTable, functionName, direction, &anyActive, &activeIndices,
-          &activeSizes, &activeMutex);
+          &activeSizes, &activeMutex, &anyBreakActive, &breakIndices);
    }
    return ret;
  }
@@ -341,8 +450,11 @@ public:
  DebugSignalInfo **signalInfoPointers;
  uint32 numSignals;
  volatile bool anyActive;
+  volatile bool anyBreakActive;
+  char8 gamName[256];
  Vec<uint32> activeIndices;
  Vec<uint32> activeSizes;
+  Vec<uint32> breakIndices;
  FastPollingMutexSem activeMutex;
 };

@@ -355,6 +467,8 @@ public:
    signalInfoPointers = NULL_PTR(DebugSignalInfo **);
    numSignals = 0;
    anyActive = false;
+    anyBreakActive = false;
+    gamName[0] = '\0';
  }
  virtual ~DebugMemoryMapAsyncTriggerOutputBroker() {
    if (signalInfoPointers)
@@ -362,9 +476,13 @@ public:
  }
  virtual bool Execute() {
    bool ret = MemoryMapAsyncTriggerOutputBroker::Execute();
-    if (ret && (anyActive || (service && service->IsPaused()))) {
+    if (ret && (anyActive || anyBreakActive)) {
      DebugBrokerHelper::Process(service, signalInfoPointers, activeIndices,
-                                 activeSizes, activeMutex);
+                                 activeSizes, activeMutex,
+                                 &anyBreakActive, &breakIndices);
+    }
+    if (ret) {
+      DebugBrokerHelper::OutputPauseAndStep(service, gamName);
    }
    return ret;
  }
@@ -380,10 +498,11 @@ public:
        stackSizeIn);
    if (ret) {
      numSignals = this->GetNumberOfCopies();
+      StringHelper::CopyN(gamName, (functionName != NULL_PTR(const char8 *)) ? functionName : "", 255u);
      DebugBrokerHelper::InitSignals(
          this, dataSourceIn, service, signalInfoPointers, numSignals,
          this->copyTable, functionName, direction, &anyActive, &activeIndices,
-          &activeSizes, &activeMutex);
+          &activeSizes, &activeMutex, &anyBreakActive, &breakIndices);
    }
    return ret;
  }
@@ -391,8 +510,11 @@ public:
  DebugSignalInfo **signalInfoPointers;
  uint32 numSignals;
  volatile bool anyActive;
+  volatile bool anyBreakActive;
+  char8 gamName[256];
  Vec<uint32> activeIndices;
  Vec<uint32> activeSizes;
+  Vec<uint32> breakIndices;
  FastPollingMutexSem activeMutex;
 };

@@ -8,6 +8,17 @@

 namespace MARTe {

+// Break condition operators stored in DebugSignalInfo::breakOp
+enum BreakOp {
+    BREAK_OFF = 0,
+    BREAK_GT  = 1,  // >
+    BREAK_LT  = 2,  // <
+    BREAK_EQ  = 3,  // ==
+    BREAK_GEQ = 4,  // >=
+    BREAK_LEQ = 5,  // <=
+    BREAK_NEQ = 6   // !=
+};
+
 struct DebugSignalInfo {
    void* memoryAddress;
    TypeDescriptor type;
@@ -16,10 +27,13 @@ struct DebugSignalInfo {
    uint32 numberOfElements;
    volatile bool isTracing;
    volatile bool isForcing;
-    uint8 forcedValue[1024]; 
+    uint8 forcedValue[1024];
    uint32 internalID;
    volatile uint32 decimationFactor;
    volatile uint32 decimationCounter;
+    // Conditional break fields (zero-cost when breakOp == BREAK_OFF)
+    volatile uint8 breakOp;         // BreakOp enum value
+    float64 breakThreshold;         // comparison threshold
 };

 #pragma pack(push, 1)
@@ -97,6 +97,7 @@ DebugService::DebugService()
  isServer = false;
  suppressTimeoutLogs = true;
  isPaused = false;
+  stepRemaining = 0u;
  manualConfigSet = false;
  activeClient = NULL_PTR(BasicTCPSocket *);
  streamerPacketOffset = 0u;
@@ -431,6 +432,8 @@ DebugSignalInfo *DebugService::RegisterSignal(void *memoryAddress,
    res->internalID = sigIdx;
    res->decimationFactor = 1;
    res->decimationCounter = 0;
+    res->breakOp = BREAK_OFF;
+    res->breakThreshold = 0.0;
    signals.Push(res);
  }
  if (sigIdx != 0xFFFFFFFF) {
@@ -474,7 +477,10 @@ void DebugService::RegisterBroker(DebugSignalInfo **signalPointers,
                                  volatile bool *anyActiveFlag,
                                  Vec<uint32> *activeIndices,
                                  Vec<uint32> *activeSizes,
-                                  FastPollingMutexSem *activeMutex) {
+                                  FastPollingMutexSem *activeMutex,
+                                  volatile bool *anyBreakFlag,
+                                  Vec<uint32> *breakIndices,
+                                  const char8 *gamName, bool isOutput) {
  mutex.FastLock();
  BrokerInfo b;
  b.signalPointers = signalPointers;
@@ -484,6 +490,11 @@ void DebugService::RegisterBroker(DebugSignalInfo **signalPointers,
  b.activeIndices = activeIndices;
  b.activeSizes = activeSizes;
  b.activeMutex = activeMutex;
+  b.anyBreakFlag = anyBreakFlag;
+  b.breakIndices = breakIndices;
+  b.isOutput = isOutput;
+  if (gamName != NULL_PTR(const char8 *))
+    b.gamName = gamName;
  brokers.Push(b);
  mutex.FastUnLock();
 }
@@ -525,6 +536,28 @@ void DebugService::UpdateBrokersActiveStatus() {
  }
 }

+void DebugService::UpdateBrokersBreakStatus() {
+  for (uint32 i = 0; i < brokers.Size(); i++) {
+    Vec<uint32> tempBreak;
+    uint32 count = 0;
+    for (uint32 j = 0; j < brokers[i].numSignals; j++) {
+      DebugSignalInfo *s = brokers[i].signalPointers[j];
+      if (s != NULL_PTR(DebugSignalInfo *) && s->breakOp != BREAK_OFF) {
+        tempBreak.Push(j);
+        count++;
+      }
+    }
+    if (brokers[i].activeMutex)
+      brokers[i].activeMutex->FastLock();
+    if (brokers[i].breakIndices)
+      *(brokers[i].breakIndices) = tempBreak;
+    if (brokers[i].anyBreakFlag)
+      *(brokers[i].anyBreakFlag) = (count > 0);
+    if (brokers[i].activeMutex)
+      brokers[i].activeMutex->FastUnLock();
+  }
+}
+
 ErrorManagement::ErrorType DebugService::Execute(ExecutionInfo &info) {
  return ErrorManagement::FatalError;
 }
@@ -731,6 +764,72 @@ void DebugService::HandleCommand(StreamString cmd, BasicTCPSocket *client) {
          (void)client->Write(resp.Buffer(), s);
        }
      }
+    } else if (token == "STEP") {
+      StreamString nStr;
+      uint32 n = 1u;
+      if (cmd.GetToken(nStr, delims, term)) {
+        AnyType nVal(UnsignedInteger32Bit, 0u, &n);
+        AnyType nS(CharString, 0u, nStr.Buffer());
+        (void)TypeConvert(nVal, nS);
+      }
+      // Optional thread name: STEP <n> [<threadName>]
+      StreamString threadStr;
+      const char8 *threadArg = NULL_PTR(const char8 *);
+      if (cmd.GetToken(threadStr, delims, term) && threadStr.Size() > 0u) {
+        threadArg = threadStr.Buffer();
+      }
+      Step(n, threadArg);
+      if (client) {
+        StreamString resp;
+        resp.Printf("OK STEP %u\n", n);
+        uint32 s = resp.Size();
+        (void)client->Write(resp.Buffer(), s);
+      }
+    } else if (token == "STEP_STATUS") {
+      GetStepStatus(client);
+    } else if (token == "VALUE") {
+      StreamString sigName;
+      if (cmd.GetToken(sigName, delims, term)) {
+        GetSignalValue(sigName.Buffer(), client);
+      } else if (client) {
+        const char8 *errResp = "{\"Error\": \"Missing signal name\"}\nOK VALUE\n";
+        uint32 s = StringHelper::Length(errResp);
+        (void)client->Write(errResp, s);
+      }
+    } else if (token == "BREAK") {
+      // BREAK <signal> <op> <threshold>   — set break condition
+      // BREAK <signal> OFF                — clear break condition
+      StreamString name, opStr;
+      if (cmd.GetToken(name, delims, term) && cmd.GetToken(opStr, delims, term)) {
+        uint32 count = 0;
+        if (opStr == "OFF") {
+          count = ClearBreak(name.Buffer());
+        } else {
+          StreamString threshStr;
+          if (cmd.GetToken(threshStr, delims, term)) {
+            uint8 op = BREAK_OFF;
+            if      (opStr == ">")  op = BREAK_GT;
+            else if (opStr == "<")  op = BREAK_LT;
+            else if (opStr == "==") op = BREAK_EQ;
+            else if (opStr == ">=") op = BREAK_GEQ;
+            else if (opStr == "<=") op = BREAK_LEQ;
+            else if (opStr == "!=") op = BREAK_NEQ;
+            if (op != BREAK_OFF) {
+              float64 threshold = 0.0;
+              AnyType thrVal(Float64Bit, 0u, &threshold);
+              AnyType thrStr(CharString, 0u, threshStr.Buffer());
+              (void)TypeConvert(thrVal, thrStr);
+              count = SetBreak(name.Buffer(), op, threshold);
+            }
+          }
+        }
+        if (client) {
+          StreamString resp;
+          resp.Printf("OK BREAK %u\n", count);
+          uint32 s = resp.Size();
+          (void)client->Write(resp.Buffer(), s);
+        }
+      }
    } else if (token == "DISCOVER")
      Discover(client);
        else if (token == "MSG") {
@@ -1369,6 +1468,159 @@ uint32 DebugService::TraceSignal(const char8 *name, bool enable,
  return count;
 }

+void DebugService::ConsumeStepIfNeeded(const char8 *gamName,
+                                       const char8 *threadName) {
+  if (stepRemaining == 0u) return;
+  mutex.FastLock();
+  // If a thread filter is set, only the matching OS thread consumes step credits.
+  if (stepThreadFilter.Size() > 0u &&
+      (threadName == NULL_PTR(const char8 *) || stepThreadFilter != threadName)) {
+    mutex.FastUnLock();
+    return;
+  }
+  if (stepRemaining > 0u) {
+    stepRemaining--;
+    if (stepRemaining == 0u) {
+      isPaused = true;
+      pausedAtGam = (gamName != NULL_PTR(const char8 *)) ? gamName : "";
+    }
+  }
+  mutex.FastUnLock();
+}
+
+void DebugService::Step(uint32 n, const char8 *threadName) {
+  mutex.FastLock();
+  stepRemaining = n;
+  isPaused = false;
+  stepThreadFilter = (threadName != NULL_PTR(const char8 *)) ? threadName : "";
+  mutex.FastUnLock();
+}
+
+void DebugService::GetStepStatus(BasicTCPSocket *client) {
+  if (client == NULL_PTR(BasicTCPSocket *)) return;
+  mutex.FastLock();
+  bool paused = isPaused;
+  uint32 remaining = stepRemaining;
+  StreamString gam = pausedAtGam;
+  StreamString threadFilter = stepThreadFilter;
+  mutex.FastUnLock();
+  StreamString resp;
+  resp.Printf("{\"Paused\": %s, \"PausedAtGam\": \"%s\", \"StepRemaining\": %u, \"StepThread\": \"%s\"}\nOK STEP_STATUS\n",
+              paused ? "true" : "false",
+              gam.Buffer(),
+              remaining,
+              threadFilter.Buffer());
+  uint32 s = resp.Size();
+  (void)client->Write(resp.Buffer(), s);
+}
+
+void DebugService::GetSignalValue(const char8 *name, BasicTCPSocket *client) {
+  if (client == NULL_PTR(BasicTCPSocket *)) return;
+  mutex.FastLock();
+  DebugSignalInfo *sig = NULL_PTR(DebugSignalInfo *);
+  for (uint32 i = 0u; i < aliases.Size(); i++) {
+    if (aliases[i].name == name ||
+        SuffixMatch(aliases[i].name.Buffer(), name)) {
+      sig = signals[aliases[i].signalIndex];
+      break;
+    }
+  }
+  if (sig == NULL_PTR(DebugSignalInfo *)) {
+    mutex.FastUnLock();
+    StreamString resp;
+    resp.Printf("{\"Error\": \"Signal not found: %s\"}\nOK VALUE\n", name);
+    uint32 s = resp.Size();
+    (void)client->Write(resp.Buffer(), s);
+    return;
+  }
+  TypeDescriptor td = sig->type;
+  uint32 nElem = sig->numberOfElements;
+  uint32 byteSize = (td.numberOfBits > 0u) ? (td.numberOfBits / 8u) : 1u;
+  uint32 totalBytes = byteSize * nElem;
+  if (totalBytes > 1024u) { totalBytes = 1024u; nElem = totalBytes / byteSize; }
+  // Copy bytes while holding the mutex to avoid data races with the RT thread
+  uint8 localBuf[1024];
+  memset(localBuf, 0, sizeof(localBuf));
+  if (sig->memoryAddress != NULL_PTR(void *)) {
+    memcpy(localBuf, sig->memoryAddress, totalBytes);
+  }
+  mutex.FastUnLock();
+
+  // Build the value string via CharString TypeConvert — the same path used by
+  // EnrichWithConfig/JsonifyDatabase, which is known to work for all types.
+  StreamString valueStr;
+  if (nElem > 1u) {
+    // Array or matrix: produce comma-separated text
+    for (uint32 i = 0u; i < nElem; i++) {
+      char8 elemBuf[128] = {'\0'};
+      AnyType srcElem(td, 0u, (void *)(localBuf + i * byteSize));
+      AnyType dstStr(CharString, 0u, elemBuf);
+      dstStr.SetNumberOfElements(0u, 128u);
+      (void)TypeConvert(dstStr, srcElem);
+      if (i > 0u) valueStr += ", ";
+      valueStr += elemBuf;
+    }
+  } else {
+    char8 elemBuf[256] = {'\0'};
+    AnyType srcElem(td, 0u, (void *)localBuf);
+    AnyType dstStr(CharString, 0u, elemBuf);
+    dstStr.SetNumberOfElements(0u, 256u);
+    (void)TypeConvert(dstStr, srcElem);
+    valueStr = elemBuf;
+  }
+
+  StreamString resp;
+  resp += "{\"Name\": \"";
+  resp += name;
+  resp += "\", \"Value\": \"";
+  // Escape the value text (quotes inside a string value)
+  const char8 *vp = valueStr.Buffer();
+  while (vp != NULL_PTR(const char8 *) && *vp != '\0') {
+    if (*vp == '"') resp += "\\\"";
+    else if (*vp == '\\') resp += "\\\\";
+    else { char8 tmp[2] = { *vp, '\0' }; resp += tmp; }
+    vp++;
+  }
+  resp += "\", \"Elements\": ";
+  resp.Printf("%u}\nOK VALUE\n", nElem);
+  uint32 s = resp.Size();
+  (void)client->Write(resp.Buffer(), s);
+}
+
+uint32 DebugService::SetBreak(const char8 *name, uint8 op, float64 threshold) {
+  mutex.FastLock();
+  uint32 count = 0;
+  for (uint32 i = 0; i < aliases.Size(); i++) {
+    if (aliases[i].name == name ||
+        SuffixMatch(aliases[i].name.Buffer(), name)) {
+      DebugSignalInfo *s = signals[aliases[i].signalIndex];
+      s->breakThreshold = threshold;
+      s->breakOp = op;
+      count++;
+    }
+  }
+  if (count > 0)
+    UpdateBrokersBreakStatus();
+  mutex.FastUnLock();
+  return count;
+}
+
+uint32 DebugService::ClearBreak(const char8 *name) {
+  mutex.FastLock();
+  uint32 count = 0;
+  for (uint32 i = 0; i < aliases.Size(); i++) {
+    if (aliases[i].name == name ||
+        SuffixMatch(aliases[i].name.Buffer(), name)) {
+      signals[aliases[i].signalIndex]->breakOp = BREAK_OFF;
+      count++;
+    }
+  }
+  if (count > 0)
+    UpdateBrokersBreakStatus();
+  mutex.FastUnLock();
+  return count;
+}
+
 bool DebugService::IsInstrumented(const char8 *fullPath, bool &traceable,
                                 bool &forcable) {
  mutex.FastLock();
@@ -32,6 +32,12 @@ struct BrokerInfo {
  Vec<uint32> *activeIndices;
  Vec<uint32> *activeSizes;
  FastPollingMutexSem *activeMutex;
+  // Conditional break — mirrors activeIndices but for break-enabled signals
+  volatile bool *anyBreakFlag;
+  Vec<uint32> *breakIndices;
+  // GAM association for step-by-GAM
+  StreamString gamName;
+  bool isOutput;
 };

 class DebugService : public ReferenceContainer,
@@ -55,7 +61,10 @@ public:
  void RegisterBroker(DebugSignalInfo **signalPointers, uint32 numSignals,
                      MemoryMapBroker *broker, volatile bool *anyActiveFlag,
                      Vec<uint32> *activeIndices, Vec<uint32> *activeSizes,
-                      FastPollingMutexSem *activeMutex);
+                      FastPollingMutexSem *activeMutex,
+                      volatile bool *anyBreakFlag, Vec<uint32> *breakIndices,
+                      const char8 *gamName = NULL_PTR(const char8 *),
+                      bool isOutput = false);

  virtual ErrorManagement::ErrorType Execute(ExecutionInfo &info);

@@ -64,11 +73,27 @@ public:
  bool IsPaused() const { return isPaused; }
  void SetPaused(bool paused) { isPaused = paused; }

+  // Step-by-GAM / per-thread: called by each output broker.
+  // threadName is the OS thread name (from Threads::Name(Threads::Id())).
+  // Decrements stepRemaining only when stepThreadFilter is empty or matches;
+  // when stepRemaining reaches 0, sets isPaused = true and records gamName.
+  void ConsumeStepIfNeeded(const char8 *gamName,
+                           const char8 *threadName = NULL_PTR(const char8 *));
+
+  void GetStepStatus(BasicTCPSocket *client);
+  // Step n cycles; if threadName is non-null/non-empty, only that thread advances.
+  void Step(uint32 n, const char8 *threadName = NULL_PTR(const char8 *));
+
+  // Read the current raw value of a signal from its memory address.
+  void GetSignalValue(const char8 *name, BasicTCPSocket *client);
+
  static bool GetFullObjectName(const Object &obj, StreamString &fullPath);

  uint32 ForceSignal(const char8 *name, const char8 *valueStr);
  uint32 UnforceSignal(const char8 *name);
  uint32 TraceSignal(const char8 *name, bool enable, uint32 decimation = 1);
+  uint32 SetBreak(const char8 *name, uint8 op, float64 threshold);
+  uint32 ClearBreak(const char8 *name);
  bool IsInstrumented(const char8 *fullPath, bool &traceable, bool &forcable);
  void Discover(BasicTCPSocket *client);
  void InfoNode(const char8 *path, BasicTCPSocket *client);
@@ -93,6 +118,7 @@ public:
 private:
  void HandleCommand(StreamString cmd, BasicTCPSocket *client);
  void UpdateBrokersActiveStatus();
+  void UpdateBrokersBreakStatus();
  void InjectTcpLoggerIfNeeded();

  uint32 ExportTree(ReferenceContainer *container, StreamString &json, const char8 *pathPrefix);
@@ -111,6 +137,9 @@ private:
  bool isServer;
  bool suppressTimeoutLogs;
  volatile bool isPaused;
+  volatile uint32 stepRemaining;
+  StreamString pausedAtGam;
+  StreamString stepThreadFilter;  // empty = all threads; non-empty = only this OS thread name

  BasicTCPSocket tcpServer;
  BasicUDPSocket udpSocket;