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).

3 files changed, 46 insertions, 7 deletions

diff --git a/pkg/flatrpc/flatrpc.fbs b/pkg/flatrpc/flatrpc.fbs
index 690286681..51066d9a4 100644
--- a/pkg/flatrpc/flatrpc.fbs
+++ b/pkg/flatrpc/flatrpc.fbs
@@ -243,6 +243,9 @@ table ExecResultRaw {
 	id			:int64;
 	proc			:int32;
 	output			:[uint8];
+	// The program has hanged and we were not able to properly join it.
+	// So in some sense it's still running (e.g. can trigger a delayed kernel hang report).
+	hanged			:bool;
 	error			:string;
 	info			:ProgInfoRaw;
 }
diff --git a/pkg/flatrpc/flatrpc.go b/pkg/flatrpc/flatrpc.go
index ff2f4c623..0a9c0e0d0 100644
--- a/pkg/flatrpc/flatrpc.go
+++ b/pkg/flatrpc/flatrpc.go
@@ -3023,6 +3023,7 @@ type ExecResultRawT struct {
 	Id     int64         `json:"id"`
 	Proc   int32         `json:"proc"`
 	Output []byte        `json:"output"`
+	Hanged bool          `json:"hanged"`
 	Error  string        `json:"error"`
 	Info   *ProgInfoRawT `json:"info"`
 }
@@ -3041,6 +3042,7 @@ func (t *ExecResultRawT) Pack(builder *flatbuffers.Builder) flatbuffers.UOffsetT
 	ExecResultRawAddId(builder, t.Id)
 	ExecResultRawAddProc(builder, t.Proc)
 	ExecResultRawAddOutput(builder, outputOffset)
+	ExecResultRawAddHanged(builder, t.Hanged)
 	ExecResultRawAddError(builder, errorOffset)
 	ExecResultRawAddInfo(builder, infoOffset)
 	return ExecResultRawEnd(builder)
@@ -3050,6 +3052,7 @@ func (rcv *ExecResultRaw) UnPackTo(t *ExecResultRawT) {
 	t.Id = rcv.Id()
 	t.Proc = rcv.Proc()
 	t.Output = rcv.OutputBytes()
+	t.Hanged = rcv.Hanged()
 	t.Error = string(rcv.Error())
 	t.Info = rcv.Info(nil).UnPack()
 }
@@ -3148,16 +3151,28 @@ func (rcv *ExecResultRaw) MutateOutput(j int, n byte) bool {
 	return false
 }
 
-func (rcv *ExecResultRaw) Error() []byte {
+func (rcv *ExecResultRaw) Hanged() bool {
 	o := flatbuffers.UOffsetT(rcv._tab.Offset(10))
 	if o != 0 {
+		return rcv._tab.GetBool(o + rcv._tab.Pos)
+	}
+	return false
+}
+
+func (rcv *ExecResultRaw) MutateHanged(n bool) bool {
+	return rcv._tab.MutateBoolSlot(10, n)
+}
+
+func (rcv *ExecResultRaw) Error() []byte {
+	o := flatbuffers.UOffsetT(rcv._tab.Offset(12))
+	if o != 0 {
 		return rcv._tab.ByteVector(o + rcv._tab.Pos)
 	}
 	return nil
 }
 
 func (rcv *ExecResultRaw) Info(obj *ProgInfoRaw) *ProgInfoRaw {
-	o := flatbuffers.UOffsetT(rcv._tab.Offset(12))
+	o := flatbuffers.UOffsetT(rcv._tab.Offset(14))
 	if o != 0 {
 		x := rcv._tab.Indirect(o + rcv._tab.Pos)
 		if obj == nil {
@@ -3170,7 +3185,7 @@ func (rcv *ExecResultRaw) Info(obj *ProgInfoRaw) *ProgInfoRaw {
 }
 
 func ExecResultRawStart(builder *flatbuffers.Builder) {
-	builder.StartObject(5)
+	builder.StartObject(6)
 }
 func ExecResultRawAddId(builder *flatbuffers.Builder, id int64) {
 	builder.PrependInt64Slot(0, id, 0)
@@ -3184,11 +3199,14 @@ func ExecResultRawAddOutput(builder *flatbuffers.Builder, output flatbuffers.UOf
 func ExecResultRawStartOutputVector(builder *flatbuffers.Builder, numElems int) flatbuffers.UOffsetT {
 	return builder.StartVector(1, numElems, 1)
 }
+func ExecResultRawAddHanged(builder *flatbuffers.Builder, hanged bool) {
+	builder.PrependBoolSlot(3, hanged, false)
+}
 func ExecResultRawAddError(builder *flatbuffers.Builder, error flatbuffers.UOffsetT) {
-	builder.PrependUOffsetTSlot(3, flatbuffers.UOffsetT(error), 0)
+	builder.PrependUOffsetTSlot(4, flatbuffers.UOffsetT(error), 0)
 }
 func ExecResultRawAddInfo(builder *flatbuffers.Builder, info flatbuffers.UOffsetT) {
-	builder.PrependUOffsetTSlot(4, flatbuffers.UOffsetT(info), 0)
+	builder.PrependUOffsetTSlot(5, flatbuffers.UOffsetT(info), 0)
 }
 func ExecResultRawEnd(builder *flatbuffers.Builder) flatbuffers.UOffsetT {
 	return builder.EndObject()
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);
 }