executor: better handling for hanged test processes

Currently we kill hanged processes and consider the corresponding test finished.
We don't kill/wait for the actual test subprocess (we don't know its pid to kill,
and waiting will presumably hang). This has 2 problems:
1. If the hanged process causes "task hung" report, we can't reproduce it,
since the test finished too long ago (manager thinks its finished and
discards the request).
2. The test process still consumed per-pid resources.

Explicitly detect and handle such cases:
Manager keeps these hanged tests forever,
and we assign a new proc id for future processes
(don't reuse the hanged one).

1 files changed, 20 insertions, 2 deletions

diff --git a/pkg/flatrpc/flatrpc.h b/pkg/flatrpc/flatrpc.h
index d1496630e..5232114b3 100644
--- a/pkg/flatrpc/flatrpc.h
+++ b/pkg/flatrpc/flatrpc.h
@@ -2368,6 +2368,7 @@ struct ExecResultRawT : public flatbuffers::NativeTable {
   int64_t id = 0;
   int32_t proc = 0;
   std::vector<uint8_t> output{};
+  bool hanged = false;
   std::string error{};
   std::unique_ptr<rpc::ProgInfoRawT> info{};
   ExecResultRawT() = default;
@@ -2383,8 +2384,9 @@ struct ExecResultRaw FLATBUFFERS_FINAL_CLASS : private flatbuffers::Table {
     VT_ID = 4,
     VT_PROC = 6,
     VT_OUTPUT = 8,
-    VT_ERROR = 10,
-    VT_INFO = 12
+    VT_HANGED = 10,
+    VT_ERROR = 12,
+    VT_INFO = 14
   };
   int64_t id() const {
     return GetField<int64_t>(VT_ID, 0);
@@ -2395,6 +2397,9 @@ struct ExecResultRaw FLATBUFFERS_FINAL_CLASS : private flatbuffers::Table {
   const flatbuffers::Vector<uint8_t> *output() const {
     return GetPointer<const flatbuffers::Vector<uint8_t> *>(VT_OUTPUT);
   }
+  bool hanged() const {
+    return GetField<uint8_t>(VT_HANGED, 0) != 0;
+  }
   const flatbuffers::String *error() const {
     return GetPointer<const flatbuffers::String *>(VT_ERROR);
   }
@@ -2407,6 +2412,7 @@ struct ExecResultRaw FLATBUFFERS_FINAL_CLASS : private flatbuffers::Table {
            VerifyField<int32_t>(verifier, VT_PROC, 4) &&
            VerifyOffset(verifier, VT_OUTPUT) &&
            verifier.VerifyVector(output()) &&
+           VerifyField<uint8_t>(verifier, VT_HANGED, 1) &&
            VerifyOffset(verifier, VT_ERROR) &&
            verifier.VerifyString(error()) &&
            VerifyOffset(verifier, VT_INFO) &&
@@ -2431,6 +2437,9 @@ struct ExecResultRawBuilder {
   void add_output(flatbuffers::Offset<flatbuffers::Vector<uint8_t>> output) {
     fbb_.AddOffset(ExecResultRaw::VT_OUTPUT, output);
   }
+  void add_hanged(bool hanged) {
+    fbb_.AddElement<uint8_t>(ExecResultRaw::VT_HANGED, static_cast<uint8_t>(hanged), 0);
+  }
   void add_error(flatbuffers::Offset<flatbuffers::String> error) {
     fbb_.AddOffset(ExecResultRaw::VT_ERROR, error);
   }
@@ -2453,6 +2462,7 @@ inline flatbuffers::Offset<ExecResultRaw> CreateExecResultRaw(
     int64_t id = 0,
     int32_t proc = 0,
     flatbuffers::Offset<flatbuffers::Vector<uint8_t>> output = 0,
+    bool hanged = false,
     flatbuffers::Offset<flatbuffers::String> error = 0,
     flatbuffers::Offset<rpc::ProgInfoRaw> info = 0) {
   ExecResultRawBuilder builder_(_fbb);
@@ -2461,6 +2471,7 @@ inline flatbuffers::Offset<ExecResultRaw> CreateExecResultRaw(
   builder_.add_error(error);
   builder_.add_output(output);
   builder_.add_proc(proc);
+  builder_.add_hanged(hanged);
   return builder_.Finish();
 }
 
@@ -2469,6 +2480,7 @@ inline flatbuffers::Offset<ExecResultRaw> CreateExecResultRawDirect(
     int64_t id = 0,
     int32_t proc = 0,
     const std::vector<uint8_t> *output = nullptr,
+    bool hanged = false,
     const char *error = nullptr,
     flatbuffers::Offset<rpc::ProgInfoRaw> info = 0) {
   auto output__ = output ? _fbb.CreateVector<uint8_t>(*output) : 0;
@@ -2478,6 +2490,7 @@ inline flatbuffers::Offset<ExecResultRaw> CreateExecResultRawDirect(
       id,
       proc,
       output__,
+      hanged,
       error__,
       info);
 }
@@ -3459,6 +3472,7 @@ inline ExecResultRawT::ExecResultRawT(const ExecResultRawT &o)
       : id(o.id),
         proc(o.proc),
         output(o.output),
+        hanged(o.hanged),
         error(o.error),
         info((o.info) ? new rpc::ProgInfoRawT(*o.info) : nullptr) {
 }
@@ -3467,6 +3481,7 @@ inline ExecResultRawT &ExecResultRawT::operator=(ExecResultRawT o) FLATBUFFERS_N
   std::swap(id, o.id);
   std::swap(proc, o.proc);
   std::swap(output, o.output);
+  std::swap(hanged, o.hanged);
   std::swap(error, o.error);
   std::swap(info, o.info);
   return *this;
@@ -3484,6 +3499,7 @@ inline void ExecResultRaw::UnPackTo(ExecResultRawT *_o, const flatbuffers::resol
   { auto _e = id(); _o->id = _e; }
   { auto _e = proc(); _o->proc = _e; }
   { auto _e = output(); if (_e) { _o->output.resize(_e->size()); std::copy(_e->begin(), _e->end(), _o->output.begin()); } }
+  { auto _e = hanged(); _o->hanged = _e; }
   { auto _e = error(); if (_e) _o->error = _e->str(); }
   { auto _e = info(); if (_e) _o->info = std::unique_ptr<rpc::ProgInfoRawT>(_e->UnPack(_resolver)); }
 }
@@ -3499,6 +3515,7 @@ inline flatbuffers::Offset<ExecResultRaw> CreateExecResultRaw(flatbuffers::FlatB
   auto _id = _o->id;
   auto _proc = _o->proc;
   auto _output = _o->output.size() ? _fbb.CreateVector(_o->output) : 0;
+  auto _hanged = _o->hanged;
   auto _error = _o->error.empty() ? 0 : _fbb.CreateString(_o->error);
   auto _info = _o->info ? CreateProgInfoRaw(_fbb, _o->info.get(), _rehasher) : 0;
   return rpc::CreateExecResultRaw(
@@ -3506,6 +3523,7 @@ inline flatbuffers::Offset<ExecResultRaw> CreateExecResultRaw(flatbuffers::FlatB
       _id,
       _proc,
       _output,
+      _hanged,
       _error,
       _info);
 }