[Orc] Handle hangup messages in SimpleRemoteEPC

[sgraenitz]

Summary

On the controller-side, handle Hangup messages from the executor. The executor passed Error::success() or a failure message as payload.

Hangups cause an immediate disconnect of the transport layer. The disconnect function may be called later again and so implementations should be prepared. FDSimpleRemoteEPCTransport::disconnect() already has a flag to check that:
https://github.com/llvm/llvm-project/blob/cd1bd95d8707371da0e4f75cd01669c427466931/llvm/lib/ExecutionEngine/Orc/Shared/SimpleRemoteEPCUtils.cpp#L112

[lhames]

LGTM. Thanks Stefan!

llvm/include/llvm/ExecutionEngine/Orc/Shared/SimpleRemoteEPCUtils.h

public:
   ///
   /// This function may be called concurrently. Subclasses should implement
   /// locking if required for the underlying transport.
   virtual Error sendMessage(SimpleRemoteEPCOpcode OpC, uint64_t SeqNo,
                             ExecutorAddr TagAddr, ArrayRef<char> ArgBytes) = 0;
 
   /// Trigger disconnection from the transport. The implementation should
   /// respond by calling handleDisconnect on the client once disconnection
-  /// is complete.
+  /// is complete. May be called more than once and from different threads.
   virtual void disconnect() = 0;
 };
 
 /// Uses read/write on FileDescriptors for transport.
 class FDSimpleRemoteEPCTransport : public SimpleRemoteEPCTransport {
 public:
   /// Create a FDSimpleRemoteEPCTransport using the given FDs for
   /// reading (InFD) and writing (OutFD).

llvm/include/llvm/ExecutionEngine/Orc/SimpleRemoteEPC.h

private:
   Error handleSetup(uint64_t SeqNo, ExecutorAddr TagAddr,
                     SimpleRemoteEPCArgBytesVector ArgBytes);
   Error setup();
 
   Error handleResult(uint64_t SeqNo, ExecutorAddr TagAddr,
                      SimpleRemoteEPCArgBytesVector ArgBytes);
   void handleCallWrapper(uint64_t RemoteSeqNo, ExecutorAddr TagAddr,
                          SimpleRemoteEPCArgBytesVector ArgBytes);
+  Error handleHangup(SimpleRemoteEPCArgBytesVector ArgBytes);
 
   uint64_t getNextSeqNo() { return NextSeqNo++; }
   void releaseSeqNo(uint64_t SeqNo) {}
 
   using PendingCallWrapperResultsMap =
     DenseMap<uint64_t, IncomingWFRHandler>;
 
   std::mutex SimpleRemoteEPCMutex;

llvm/lib/ExecutionEngine/Orc/SimpleRemoteEPC.cpp

return make_error<StringError>("Unexpected opcode",
                                    inconvertibleErrorCode());
 
   switch (OpC) {
   case SimpleRemoteEPCOpcode::Setup:
     if (auto Err = handleSetup(SeqNo, TagAddr, std::move(ArgBytes)))
       return std::move(Err);
     break;
   case SimpleRemoteEPCOpcode::Hangup:
-    return make_error<StringError>("Unexpected Hangup opcode",
-                                   inconvertibleErrorCode());
+    T->disconnect();
+    if (auto Err = handleHangup(std::move(ArgBytes)))
+      return std::move(Err);
+    return EndSession;
   case SimpleRemoteEPCOpcode::Result:
     if (auto Err = handleResult(SeqNo, TagAddr, std::move(ArgBytes)))
       return std::move(Err);
     break;
   case SimpleRemoteEPCOpcode::CallWrapper:
     handleCallWrapper(SeqNo, TagAddr, std::move(ArgBytes));
     break;
   }
ES->runJITDispatchHandler(
       [this, RemoteSeqNo](shared::WrapperFunctionResult WFR) {
         if (auto Err = sendMessage(SimpleRemoteEPCOpcode::Result, RemoteSeqNo,
                                    ExecutorAddr(), {WFR.data(), WFR.size()}))
           getExecutionSession().reportError(std::move(Err));
       },
       TagAddr.getValue(), ArgBytes);
 }
 
+Error SimpleRemoteEPC::handleHangup(SimpleRemoteEPCArgBytesVector ArgBytes) {
+  using namespace llvm::orc::shared;
+  auto WFR = WrapperFunctionResult::copyFrom(ArgBytes.data(), ArgBytes.size());
+  if (const char *ErrMsg = WFR.getOutOfBandError())
+    return make_error<StringError>(ErrMsg, inconvertibleErrorCode());
+
+  detail::SPSSerializableError Info;
+  SPSInputBuffer IB(WFR.data(), WFR.size());
+  if (!SPSArgList<SPSError>::deserialize(IB, Info))
+    return make_error<StringError>("Could not deserialize hangup info",
+                                   inconvertibleErrorCode());
+  return fromSPSSerializable(std::move(Info));
+}
+
 } // end namespace orc
 } // end namespace llvm