[mlgo] Make InteractiveModelRunner actually work with named pipes

[mtrofin]

Summary

Turns out raw_fd_stream doesn't work with named pipes, so we just need
to lower the abstraction. Updated the unittest accordingly. Because
mkfifo's path argument requires a certain naming pattern on Windows
(IIUC), restricted the test to Linux only.

[kazu]

LGTM. Just minor comments about sticking to the FileSystem.h API and removing a couple of #includes.

llvm/include/llvm/Analysis/InteractiveModelRunner.h

19

If you are lowering things down to sys::fs::file_t, you probably don't need this!?

llvm/lib/Analysis/InteractiveModelRunner.cpp

18

You should be able to remove this if you can switch ::read to sys::fs::readNativeFile below.

76

Can you use sys::fs::readNativeFile for consistency with sys::fs::openFileForRead and sys::fs::closeFile?

llvm/unittests/Analysis/MLModelRunnerTest.cpp

179

Likewise, can you use sys::fs::readNativeFile for consistency with sys::fs::openFileForRead?

202–203

Likewise.

214–215

Likewise.

238–239

Likewise.

249

Can you use sys::fs::closeFile here?

[mtrofin]

feedback

[mtrofin]

(addressed feedback)

[hctim]

Hey, looks like this might've broken the windows sanitizer bots. Would you be able to take a look? https://lab.llvm.org/buildbot/#/builders/127/builds/43077

Thanks.

[mtrofin]

In D143110#4098060, @hctim wrote:

Hey, looks like this might've broken the windows sanitizer bots. Would you be able to take a look? https://lab.llvm.org/buildbot/#/builders/127/builds/43077

Thanks.

Yup, looking.

llvm/include/llvm/Analysis/InteractiveModelRunner.h

 //===- InteractiveModelRunner.h ---- "gym" ML model runner  -----*- C++ -*-===//
 //
 // Part of the LLVM Project, under the Apache License v2.0 with LLVM Exceptions.
 // See https://llvm.org/LICENSE.txt for license information.
 // SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
 //
 //===----------------------------------------------------------------------===//
 //
 
 #ifndef LLVM_ANALYSIS_INTERACTIVEMODELRUNNER_H
 #define LLVM_ANALYSIS_INTERACTIVEMODELRUNNER_H
 
 #include "llvm/Analysis/MLModelRunner.h"
 #include "llvm/Analysis/TensorSpec.h"
 #include "llvm/Analysis/Utils/TrainingLogger.h"
 #include "llvm/Config/llvm-config.h"
+#include "llvm/Support/FileSystem.h"
 #include "llvm/Support/raw_ostream.h"
 #include <system_error>
 
 namespace llvm {
 
 /// A MLModelRunner that asks for advice from an external agent, or host. It
 /// uses 2 files - ideally named pipes - one to send data to that agent, and
 /// one to receive advice.
 /// The data exchange uses the training logger (Utils/TrainingLogger.h) format.
 /// Specifically, the compiler will send the log header, set the context, and
 /// send observations; the host is expected to reply with a tensor value after
 /// each observation as a binary buffer that's conforming to the shape of the
 /// advice. Interleaved, the data closely resembles the training log for a
 /// log where we don't capture the reward signal.
 ///
 /// Note that the correctness of the received data is the responsibility of the
 /// host. In particular, if insufficient data were sent, the compiler will block
 /// when waiting for an advice.
+///
+/// Note that the host can either open the pipes RW, or open first the pipe to
+/// the compiler - i.e. the "Inbound" - and then the "Outbound", to avoid
+/// deadlock. This is because the compiler first tries to open the inbound
+/// (which will hang until there's a writer on the other end).
 class InteractiveModelRunner : public MLModelRunner {
 public:
   InteractiveModelRunner(LLVMContext &Ctx,
                          const std::vector<TensorSpec> &Inputs,
                          const TensorSpec &Advice, StringRef OutboundName,
                          StringRef InboundName);
 
   static bool classof(const MLModelRunner *R) {
     return R->getKind() == MLModelRunner::Kind::Interactive;
   }
   void switchContext(StringRef Name) {
-    Log.switchContext(Name);
-    Log.flush();
+    Log->switchContext(Name);
+    Log->flush();
   }
 
+  virtual ~InteractiveModelRunner();
+
 private:
   void *evaluateUntyped() override;
   const std::vector<TensorSpec> InputSpecs;
   const TensorSpec OutputSpec;
   std::error_code OutEC;
   std::error_code InEC;
-  raw_fd_stream Inbound;
+  sys::fs::file_t Inbound;
   std::vector<char> OutputBuffer;
-  Logger Log;
+  std::unique_ptr<Logger> Log;
 };
 } // namespace llvm
 #endif // LLVM_ANALYSIS_INTERACTIVEMODELRUNNER_H

llvm/lib/Analysis/InteractiveModelRunner.cpp

 //===- InteractiveModelRunner.cpp - noop ML model runner   ----------------===//
 //
 // Part of the LLVM Project, under the Apache License v2.0 with LLVM Exceptions.
 // See https://llvm.org/LICENSE.txt for license information.
 // SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
 //
 //===----------------------------------------------------------------------===//
 //
 // A runner that communicates with an external agent via 2 file descriptors.
 //===----------------------------------------------------------------------===//
 #include "llvm/Analysis/InteractiveModelRunner.h"
 #include "llvm/Analysis/MLModelRunner.h"
 #include "llvm/Analysis/TensorSpec.h"
 #include "llvm/Support/CommandLine.h"
 #include "llvm/Support/ErrorHandling.h"
+#include "llvm/Support/FileSystem.h"
 #include "llvm/Support/raw_ostream.h"
 
 using namespace llvm;
 
 #define _IMR_CL_VALS(T, N) clEnumValN(TensorType::N, #T, #T),
 
 static cl::opt<TensorType> DebugReply(
     "interactive-model-runner-echo-type", cl::init(TensorType::Invalid),
     cl::Hidden,
     cl::desc("The InteractiveModelRunner will echo back to stderr "
              "the data received "
              "from the host as the specified type (for debugging purposes)."),
     cl::values(SUPPORTED_TENSOR_TYPES(_IMR_CL_VALS)
                    clEnumValN(TensorType::Invalid, "disable", "Don't echo")));
 
 #undef _IMR_CL_VALS
 
 InteractiveModelRunner::InteractiveModelRunner(
     LLVMContext &Ctx, const std::vector<TensorSpec> &Inputs,
     const TensorSpec &Advice, StringRef OutboundName, StringRef InboundName)
     : MLModelRunner(Ctx, MLModelRunner::Kind::Interactive, Inputs.size()),
-      InputSpecs(Inputs), OutputSpec(Advice), Inbound(InboundName, InEC),
-      OutputBuffer(OutputSpec.getTotalTensorBufferSize()),
-      Log(std::make_unique<raw_fd_ostream>(OutboundName, OutEC), InputSpecs,
-          Advice, /*IncludeReward=*/false, Advice) {
+      InputSpecs(Inputs), OutputSpec(Advice),
+      InEC(sys::fs::openFileForRead(InboundName, Inbound)),
+      OutputBuffer(OutputSpec.getTotalTensorBufferSize()) {
   if (InEC) {
     Ctx.emitError("Cannot open inbound file: " + InEC.message());
     return;
   }
+  {
+    auto OutStream = std::make_unique<raw_fd_ostream>(OutboundName, OutEC);
     if (OutEC) {
       Ctx.emitError("Cannot open outbound file: " + OutEC.message());
       return;
     }
+    Log = std::make_unique<Logger>(std::move(OutStream), InputSpecs, Advice,
+                                   /*IncludeReward=*/false, Advice);
+  }
   // Just like in the no inference case, this will allocate an appropriately
   // sized buffer.
   for (size_t I = 0; I < InputSpecs.size(); ++I)
     setUpBufferForTensor(I, InputSpecs[I], nullptr);
-  Log.flush();
+  Log->flush();
+}
+
+InteractiveModelRunner::~InteractiveModelRunner() {
+  sys::fs::closeFile(Inbound);
 }
 
 void *InteractiveModelRunner::evaluateUntyped() {
-  Log.startObservation();
+  Log->startObservation();
   for (size_t I = 0; I < InputSpecs.size(); ++I)
-    Log.logTensorValue(I, reinterpret_cast<const char *>(getTensorUntyped(I)));
-  Log.endObservation();
-  Log.flush();
+    Log->logTensorValue(I, reinterpret_cast<const char *>(getTensorUntyped(I)));
+  Log->endObservation();
+  Log->flush();
 
   size_t InsPoint = 0;
   char *Buff = OutputBuffer.data();
   const size_t Limit = OutputBuffer.size();
   while (InsPoint < Limit) {
-    auto Read = Inbound.read(Buff + InsPoint, OutputBuffer.size() - InsPoint);
-    if (Read < 0) {
+    auto ReadOrErr = ::sys::fs::readNativeFile(
+        Inbound, {Buff + InsPoint, OutputBuffer.size() - InsPoint});
+    if (ReadOrErr.takeError()) {
       Ctx.emitError("Failed reading from inbound file");
       break;
     }
-    InsPoint += Read;
+    InsPoint += *ReadOrErr;
   }
   if (DebugReply != TensorType::Invalid)
     dbgs() << tensorValueToString(OutputBuffer.data(), OutputSpec);
   return OutputBuffer.data();
-}
+}
 

llvm/unittests/Analysis/MLModelRunnerTest.cpp

 //===- MLModelRunnerTest.cpp - test for MLModelRunner ---------------------===//
 //
 // Part of the LLVM Project, under the Apache License v2.0 with LLVM Exceptions.
 // See https://llvm.org/LICENSE.txt for license information.
 // SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
 //
 //===----------------------------------------------------------------------===//
 
 #include "llvm/Analysis/MLModelRunner.h"
 #include "llvm/Analysis/InteractiveModelRunner.h"
 #include "llvm/Analysis/NoInferenceModelRunner.h"
 #include "llvm/Analysis/ReleaseModeModelRunner.h"
 #include "llvm/Support/BinaryByteStream.h"
+#include "llvm/Support/FileSystem.h"
 #include "llvm/Support/FileUtilities.h"
 #include "llvm/Support/JSON.h"
+#include "llvm/Support/Path.h"
 #include "llvm/Support/raw_ostream.h"
+#include "llvm/Testing/Support/SupportHelpers.h"
 #include "gtest/gtest.h"
 
 #include <atomic>
 #include <thread>
 
 using namespace llvm;
 
 namespace llvm {
TEST(ReleaseModeRunner, ExtraFeaturesOutOfOrder) {
   *Evaluator->getTensor<int64_t>(1) = 2;         // c
   *Evaluator->getTensor<int64_t>(2) = -3;        // b
   EXPECT_EQ(Evaluator->evaluate<int64_t>(), -2); // a + b
   EXPECT_EQ(*Evaluator->getTensor<int64_t>(0), 1);
   EXPECT_EQ(*Evaluator->getTensor<int64_t>(1), 2);
   EXPECT_EQ(*Evaluator->getTensor<int64_t>(2), -3);
 }
 
+#if defined(LLVM_ON_UNIX)
 TEST(InteractiveModelRunner, Evaluation) {
   LLVMContext Ctx;
   // Test the interaction with an external advisor by asking for advice twice.
   // Use simple values, since we use the Logger underneath, that's tested more
   // extensively elsewhere.
   std::vector<TensorSpec> Inputs{
       TensorSpec::createSpec<int64_t>("a", {1}),
       TensorSpec::createSpec<int64_t>("b", {1}),
       TensorSpec::createSpec<int64_t>("c", {1}),
   };
   TensorSpec AdviceSpec = TensorSpec::createSpec<float>("advice", {1});
 
   // Create the 2 files. Ideally we'd create them as named pipes, but that's not
   // quite supported by the generic API.
   std::error_code EC;
-  SmallString<64> FromCompilerName;
-  SmallString<64> ToCompilerName;
-  int FromCompilerFD = 0;
-  int ToCompilerFD = 0;
-  ASSERT_EQ(sys::fs::createTemporaryFile("InteractiveModelRunner_Evaluation",
-                                         "temp", FromCompilerFD,
-                                         FromCompilerName),
-            std::error_code());
-
-  ASSERT_EQ(sys::fs::createTemporaryFile("InteractiveModelRunner_Evaluation",
-                                         "temp", ToCompilerFD, ToCompilerName),
-            std::error_code());
+  llvm::unittest::TempDir Tmp("tmpdir", /*Unique=*/true);
+  SmallString<128> FromCompilerName(Tmp.path().begin(), Tmp.path().end());
+  SmallString<128> ToCompilerName(Tmp.path().begin(), Tmp.path().end());
+  sys::path::append(FromCompilerName, "InteractiveModelRunner_Evaluation.out");
+  sys::path::append(ToCompilerName, "InteractiveModelRunner_Evaluation.in");
+  EXPECT_EQ(::mkfifo(FromCompilerName.c_str(), 0666), 0);
+  EXPECT_EQ(::mkfifo(ToCompilerName.c_str(), 0666), 0);
 
-  raw_fd_stream FromCompiler(FromCompilerName, EC);
-  EXPECT_FALSE(EC);
-  raw_fd_ostream ToCompiler(ToCompilerName, EC);
-  EXPECT_FALSE(EC);
   FileRemover Cleanup1(FromCompilerName);
   FileRemover Cleanup2(ToCompilerName);
-  InteractiveModelRunner Evaluator(Ctx, Inputs, AdviceSpec, FromCompilerName,
-                                   ToCompilerName);
-
-  Evaluator.switchContext("hi");
 
+  // Since the evaluator sends the features over and then blocks waiting for
+  // an answer, we must spawn a thread playing the role of the advisor / host:
+  std::atomic<int> SeenObservations = 0;
+  // Start the host first to make sure the pipes are being prepared. Otherwise
+  // the evaluator will hang.
+  std::thread Advisor([&]() {
+    // Open the writer first. This is because the evaluator will try opening
+    // the "input" pipe first. An alternative that avoids ordering is for the
+    // host to open the pipes RW.
+    raw_fd_ostream ToCompiler(ToCompilerName, EC);
+    EXPECT_FALSE(EC);
+    sys::fs::file_t FromCompiler = {};
+    EXPECT_FALSE(sys::fs::openFileForRead(FromCompilerName, FromCompiler));
+    EXPECT_EQ(SeenObservations, 0);
     // Helper to read headers and other json lines.
     SmallVector<char, 1024> Buffer;
     auto ReadLn = [&]() {
       Buffer.clear();
       while (true) {
         char Chr = 0;
-      auto Read = FromCompiler.read(&Chr, 1);
-      EXPECT_GE(Read, 0);
-      if (!Read)
+        auto ReadOrErr = sys::fs::readNativeFile(FromCompiler, {&Chr, 1});
+        EXPECT_FALSE(ReadOrErr.takeError());
+        if (!*ReadOrErr)
           continue;
         if (Chr == '\n')
           return StringRef(Buffer.data(), Buffer.size());
         Buffer.push_back(Chr);
       }
     };
     // See include/llvm/Analysis/Utils/TrainingLogger.h
     // First comes the header
     auto Header = json::parse(ReadLn());
     EXPECT_FALSE(Header.takeError());
     EXPECT_NE(Header->getAsObject()->getArray("features"), nullptr);
     EXPECT_NE(Header->getAsObject()->getObject("advice"), nullptr);
     // Then comes the context
     EXPECT_FALSE(json::parse(ReadLn()).takeError());
 
-  // Since the evaluator sends the features over and then blocks waiting for
-  // an answer, we must spawn a thread playing the role of the advisor / host:
-  std::atomic<int> SeenObservations = 0;
-  std::thread Advisor([&]() {
-    EXPECT_EQ(SeenObservations, 0);
     int64_t Features[3] = {0};
     auto FullyRead = [&]() {
       size_t InsPt = 0;
       const size_t ToRead = 3 * Inputs[0].getTotalTensorBufferSize();
       char *Buff = reinterpret_cast<char *>(Features);
       while (InsPt < ToRead) {
-        auto Read = FromCompiler.read(Buff + InsPt, ToRead - InsPt);
-        EXPECT_GE(Read, 0);
-        InsPt += Read;
+        auto ReadOrErr = sys::fs::readNativeFile(
+            FromCompiler, {Buff + InsPt, ToRead - InsPt});
+        EXPECT_FALSE(ReadOrErr.takeError());
+        InsPt += *ReadOrErr;
       }
     };
     // Observation
     EXPECT_FALSE(json::parse(ReadLn()).takeError());
     // Tensor values
     FullyRead();
     // a "\n"
     char Chr = 0;
-    while (FromCompiler.read(&Chr, 1) == 0) {
-    }
+    auto ReadNL = [&]() {
+      do {
+        auto ReadOrErr = sys::fs::readNativeFile(FromCompiler, {&Chr, 1});
+        EXPECT_FALSE(ReadOrErr.takeError());
+        if (*ReadOrErr == 1)
+          break;
+      } while (true);
+    };
+    ReadNL();
     EXPECT_EQ(Chr, '\n');
     EXPECT_EQ(Features[0], 42);
     EXPECT_EQ(Features[1], 43);
     EXPECT_EQ(Features[2], 100);
     ++SeenObservations;
 
     // Send the advice
     float Advice = 42.0012;
     ToCompiler.write(reinterpret_cast<const char *>(&Advice),
                      AdviceSpec.getTotalTensorBufferSize());
     ToCompiler.flush();
 
     // Second observation, and same idea as above
     EXPECT_FALSE(json::parse(ReadLn()).takeError());
     FullyRead();
-    while (FromCompiler.read(&Chr, 1) == 0) {
-    }
+    ReadNL();
     EXPECT_EQ(Chr, '\n');
     EXPECT_EQ(Features[0], 10);
     EXPECT_EQ(Features[1], -2);
     EXPECT_EQ(Features[2], 1);
     ++SeenObservations;
     Advice = 50.30;
     ToCompiler.write(reinterpret_cast<const char *>(&Advice),
                      AdviceSpec.getTotalTensorBufferSize());
     ToCompiler.flush();
+    sys::fs::closeFile(FromCompiler);
   });
 
+  InteractiveModelRunner Evaluator(Ctx, Inputs, AdviceSpec, FromCompilerName,
+                                   ToCompilerName);
+
+  Evaluator.switchContext("hi");
+
   EXPECT_EQ(SeenObservations, 0);
   *Evaluator.getTensor<int64_t>(0) = 42;
   *Evaluator.getTensor<int64_t>(1) = 43;
   *Evaluator.getTensor<int64_t>(2) = 100;
   float Ret = Evaluator.evaluate<float>();
   EXPECT_EQ(SeenObservations, 1);
   EXPECT_FLOAT_EQ(Ret, 42.0012);
 
   *Evaluator.getTensor<int64_t>(0) = 10;
   *Evaluator.getTensor<int64_t>(1) = -2;
   *Evaluator.getTensor<int64_t>(2) = 1;
   Ret = Evaluator.evaluate<float>();
   EXPECT_EQ(SeenObservations, 2);
   EXPECT_FLOAT_EQ(Ret, 50.30);
   Advisor.join();
-}
+}
+#endif