Interprocess communication via pipes (#2009)

* #include an absolute path didn't work - because paths were taken to always be relative.

* Use 'Process' to communicate with an command line tool.

* Remove slang-win-stream

* Tidy up windows ProcessUtil.

* First version of BufferedReadStream.

* Windows working IPC for steams.

* Test proxy count option.

* Split Process/ProcessUtil. Process is platform dependant. ProcessUtil are functions that are platform independent.

* First implementation of Unix Process interface.

* Unix process compiles on cygwin.

* Fix typo in unix process.

* Separate unix pipe stream error of invalid access, from pipe availability.

* Fix in standard line extraction.

* Make fd non blocking.

* Fix issues with Windows Process streams.

* Added UnixPipe.

* Some fixes around UnixPipeStream.

* Make a unix stream closed explicit.

* Hack to debug linux process/stream.

* Revert to old linux pipe handling.

* Pass executable path for unit tests.
Split out CommandLine into own source.

* Small improvements in process/command line.

* Check process behavior with crash.

* Make stderr and stdout unbuffered for crash testing.

* Only turn disable buffering in crash test.

* Disable crash test on CI.

* Fix crash on clang/linux.

* Enable crash test.
Remove _appendBuffer as can use StreamUtil functionality.

5 files changed, 256 insertions, 20 deletions

diff --git a/tools/render-test/render-test-main.cpp b/tools/render-test/render-test-main.cpp
index 689794cef..a3cde1c0a 100644
--- a/tools/render-test/render-test-main.cpp
+++ b/tools/render-test/render-test-main.cpp
@@ -63,7 +63,7 @@ using namespace Slang;
 static void _outputProfileTime(uint64_t startTicks, uint64_t endTicks)
 {
     WriterHelper out = StdWriters::getOut();
-    double time = double(endTicks - startTicks) / ProcessUtil::getClockFrequency();
+    double time = double(endTicks - startTicks) / Process::getClockFrequency();
     out.print("profile-time=%g\n", time);
 }
 
@@ -991,7 +991,7 @@ Result RenderTestApp::update()
     }
     commandBuffer->close();
 
-    m_startTicks = ProcessUtil::getClockTick();
+    m_startTicks = Process::getClockTick();
     m_queue->executeCommandBuffer(commandBuffer);
     m_queue->wait();
 
@@ -1030,7 +1030,7 @@ Result RenderTestApp::update()
             // Note we don't do the same with screen rendering -> as that will do a lot of work, which may swamp any computation
             // so can only really profile compute shaders at the moment
 
-            const uint64_t endTicks = ProcessUtil::getClockTick();
+            const uint64_t endTicks = Process::getClockTick();
 
             _outputProfileTime(m_startTicks, endTicks);
         }
diff --git a/tools/slang-test/slang-test-main.cpp b/tools/slang-test/slang-test-main.cpp
index ea2c322ad..e007d1978 100644
--- a/tools/slang-test/slang-test-main.cpp
+++ b/tools/slang-test/slang-test-main.cpp
@@ -536,7 +536,7 @@ Result spawnAndWaitExe(TestContext* context, const String& testPath, const Comma
 
     if (options.shouldBeVerbose)
     {
-        String commandLine = ProcessUtil::getCommandLineString(cmdLine);
+        String commandLine = cmdLine.toString();
         context->reporter->messageFormat(TestMessageType::Info, "%s\n", commandLine.begin());
     }
 
@@ -570,8 +570,7 @@ Result spawnAndWaitSharedLibrary(TestContext* context, const String& testPath, c
         testCmdLine.addArg(exeName);
         testCmdLine.m_args.addRange(cmdLine.m_args);
 
-        String testCmdLineString = ProcessUtil::getCommandLineString(testCmdLine);
-        context->reporter->messageFormat(TestMessageType::Info, "%s\n", testCmdLineString.getBuffer());
+        context->reporter->messageFormat(TestMessageType::Info, "%s\n", testCmdLine.toString().getBuffer());
     }
 
     auto func = context->getInnerMainFunc(context->options.binDir, exeName);
@@ -636,14 +635,12 @@ Result spawnAndWaitProxy(TestContext* context, const String& testPath, const Com
 
     // Make the first arg the name of the tool to invoke
     cmdLine.m_args.insert(0, exeName);
-
-    auto exePath = Path::combine(Path::getParentDirectory(inCmdLine.m_executable), String("test-proxy") + ProcessUtil::getExecutableSuffix());
-    cmdLine.setExecutablePath(exePath);
+    cmdLine.setExecutable(context->exeDirectoryPath, "test-proxy");
 
     const auto& options = context->options;
     if (options.shouldBeVerbose)
     {
-        String commandLine = ProcessUtil::getCommandLineString(cmdLine);
+        String commandLine = cmdLine.toString();
         context->reporter->messageFormat(TestMessageType::Info, "%s\n", commandLine.begin());
     }
 
@@ -942,7 +939,7 @@ static RenderApiFlags _getAvailableRenderApiFlags(TestContext* context)
                 }
                 // Try starting up the device
                 CommandLine cmdLine;
-                cmdLine.setExecutablePath(Path::combine(context->options.binDir,  String("render-test") + ProcessUtil::getExecutableSuffix()));
+                cmdLine.setExecutable(context->options.binDir, "render-test");
                 _addRenderTestOptions(context->options, cmdLine);
                 // We just want to see if the device can be started up
                 cmdLine.addArg("-only-startup");
@@ -1086,7 +1083,7 @@ String findExpectedPath(const TestInput& input, const char* postFix)
 
 static void _initSlangCompiler(TestContext* context, CommandLine& ioCmdLine)
 {
-    ioCmdLine.setExecutablePath(Path::combine(context->options.binDir, String("slangc") + ProcessUtil::getExecutableSuffix()));
+    ioCmdLine.setExecutable(context->options.binDir, "slangc");
 
     if (context->options.verbosePaths)
     {
@@ -1123,7 +1120,7 @@ static SlangResult _executeBinary(const UnownedStringSlice& hexDump, ExecuteResu
     String fileName;
     SLANG_RETURN_ON_FAIL(File::generateTemporary(UnownedStringSlice("slang-test"), fileName));
 
-    fileName.append(ProcessUtil::getExecutableSuffix());
+    fileName.append(Process::getExecutableSuffix());
 
     TemporaryFileSet temporaryFileSet;
     temporaryFileSet.add(fileName);
@@ -1535,7 +1532,7 @@ TestResult runReflectionTest(TestContext* context, TestInput& input)
 
     CommandLine cmdLine;
     
-    cmdLine.setExecutablePath(Path::combine(options.binDir, String("slang-reflection-test") + ProcessUtil::getExecutableSuffix()));
+    cmdLine.setExecutable(options.binDir, "slang-reflection-test");
     cmdLine.addArg(filePath);
 
     for( auto arg : input.testOptions->args )
@@ -1820,7 +1817,7 @@ static TestResult runCPPCompilerExecute(TestContext* context, TestInput& input)
     {
         StringBuilder moduleExePath;
         moduleExePath << modulePath;
-        moduleExePath << ProcessUtil::getExecutableSuffix();
+        moduleExePath << Process::getExecutableSuffix();
         File::remove(moduleExePath);
     }
 
@@ -1855,7 +1852,7 @@ static TestResult runCPPCompilerExecute(TestContext* context, TestInput& input)
         CommandLine cmdLine;
 
         StringBuilder exePath;
-        exePath << modulePath << ProcessUtil::getExecutableSuffix();
+        exePath << modulePath << Process::getExecutableSuffix();
 
         cmdLine.setExecutablePath(exePath);
 
@@ -2321,7 +2318,7 @@ TestResult runPerformanceProfile(TestContext* context, TestInput& input)
 
     CommandLine cmdLine;
 
-    cmdLine.setExecutablePath(Path::combine(context->options.binDir, String("render-test") + ProcessUtil::getExecutableSuffix()));
+    cmdLine.setExecutable(context->options.binDir, "render-test");
     
     cmdLine.addArg(input.filePath);
     cmdLine.addArg("-performance-profile");
@@ -2474,7 +2471,7 @@ TestResult runComputeComparisonImpl(TestContext* context, TestInput& input, cons
 
 	CommandLine cmdLine;
 
-    cmdLine.setExecutablePath(Path::combine(context->options.binDir, String("render-test") + ProcessUtil::getExecutableSuffix()));
+    cmdLine.setExecutable(context->options.binDir, "render-test");
     cmdLine.addArg(filePath999);
 
     _addRenderTestOptions(context->options, cmdLine);
@@ -2582,7 +2579,7 @@ TestResult doRenderComparisonTestRun(TestContext* context, TestInput& input, cha
 
     CommandLine cmdLine;
 
-    cmdLine.setExecutablePath(Path::combine(context->options.binDir, String("render-test") + ProcessUtil::getExecutableSuffix()));
+    cmdLine.setExecutable(context->options.binDir, "render-test");
     cmdLine.addArg(filePath);
 
     _addRenderTestOptions(context->options, cmdLine);
@@ -3424,6 +3421,8 @@ static SlangResult runUnitTestModule(TestContext* context, TestOptions& testOpti
     unitTestContext.slangGlobalSession = context->getSession();
     unitTestContext.workDirectory = "";
     unitTestContext.enabledApis = context->options.enabledApis;
+    unitTestContext.executableDirectory = context->exeDirectoryPath.getBuffer();
+
     auto testCount = testModule->getTestCount();
 
     for (SlangInt i = 0; i < testCount; i++)
diff --git a/tools/slang-unit-test/unit-test-process.cpp b/tools/slang-unit-test/unit-test-process.cpp
new file mode 100644
index 000000000..3dccf1eaf
--- /dev/null
+++ b/tools/slang-unit-test/unit-test-process.cpp
@@ -0,0 +1,125 @@
+// unit-test-process.cpp
+
+#include "../../source/core/slang-string-util.h"
+#include "../../source/core/slang-process-util.h"
+#include "../../source/core/slang-io.h"
+
+#include "tools/unit-test/slang-unit-test.h"
+
+using namespace Slang;
+
+static SlangResult _countTest(UnitTestContext* context, Index size, Index crashIndex = -1)
+{
+    RefPtr<Process> process;
+
+    /* Here we are trying to test what happens if the server produces a large amount of data, and
+    we just wait for termination. Do we receive all of the data irrespective of how much there is? */
+
+    {
+        CommandLine cmdLine;
+        
+        cmdLine.setExecutable(context->executableDirectory, "test-proxy");
+        cmdLine.addArg("count");
+        
+        StringBuilder buf;
+        buf << size;
+
+        cmdLine.addArg(buf);
+
+        if (crashIndex >= 0)
+        {
+            buf.Clear();
+            buf << crashIndex;
+            cmdLine.addArg(buf);
+        }
+
+        SLANG_RETURN_ON_FAIL(Process::create(cmdLine, Process::Flag::AttachDebugger, process));
+    }
+
+    ExecuteResult exeRes;
+
+#if 0
+    /* It does block on ~4k of data which matches up with the buffer size, using this mechanism only works up to 4k on windows
+    which matches the default pipe buffer size */
+    process->waitForTermination();
+#endif
+
+    SLANG_RETURN_ON_FAIL(ProcessUtil::readUntilTermination(process, exeRes));
+
+    Index v = 0;
+    for (auto line : LineParser(exeRes.standardOutput.getUnownedSlice()))
+    {
+        if (line.getLength() == 0)
+        {
+            continue;
+        }
+
+        Index value;
+        StringUtil::parseInt(line, value);
+
+        if (value != v)
+        {
+            return SLANG_FAIL;
+        }
+
+        v++;
+    }
+
+    const Index endIndex = (crashIndex >= 0) ? (crashIndex + 1) : size;
+
+    return v == endIndex ? SLANG_OK : SLANG_FAIL;
+}
+
+static SlangResult _countTests(UnitTestContext* context)
+{
+    const Index sizes[] = { 1, 10, 1000, 1000, 10000, 100000 };
+    for (auto size : sizes)
+    {
+        SLANG_RETURN_ON_FAIL(_countTest(context, size));
+        SLANG_RETURN_ON_FAIL(_countTest(context, size, size / 2));
+    }
+
+    return SLANG_OK;
+}
+
+static SlangResult _reflectTest(UnitTestContext* context)
+{
+    RefPtr<Process> process;
+    {
+        CommandLine cmdLine;
+        cmdLine.setExecutable(context->executableDirectory, "test-proxy");
+        cmdLine.addArg("reflect");
+        SLANG_RETURN_ON_FAIL(Process::create(cmdLine, Process::Flag::AttachDebugger, process));
+    }
+
+    // Write a bunch of stuff to the stream
+    Stream* readStream = process->getStream(Process::StreamType::StdOut);
+    Stream* writeStream = process->getStream(Process::StreamType::StdIn);
+
+    List<Byte> readBuffer;
+
+    for (Index i = 0; i < 10000; i++)
+    {
+        SLANG_RETURN_ON_FAIL(StreamUtil::read(readStream, 0, readBuffer));
+
+        StringBuilder buf;
+
+        buf << i << " Hello " << i << "\n";
+        SLANG_RETURN_ON_FAIL(writeStream->write(buf.getBuffer(), buf.getLength()));
+    }
+
+    const char end[] = "end\n";
+    SLANG_RETURN_ON_FAIL(writeStream->write(end, SLANG_COUNT_OF(end) - 1));
+    writeStream->flush();
+
+    SLANG_RETURN_ON_FAIL(StreamUtil::readAll(readStream, 0, readBuffer));
+
+    return SLANG_OK;
+}
+
+SLANG_UNIT_TEST(CommandLineProcess)
+{
+    SLANG_CHECK(SLANG_SUCCEEDED(_countTests(unitTestContext)));
+
+    SLANG_CHECK(SLANG_SUCCEEDED(_reflectTest(unitTestContext)));
+}
diff --git a/tools/test-proxy/test-proxy-main.cpp b/tools/test-proxy/test-proxy-main.cpp
index 2b6c59047..70efc8907 100644
--- a/tools/test-proxy/test-proxy-main.cpp
+++ b/tools/test-proxy/test-proxy-main.cpp
@@ -12,6 +12,7 @@
 #include "../../source/core/slang-io.h"
 #include "../../source/core/slang-writer.h"
 #include "../../source/core/slang-string-util.h"
+#include "../../source/core/slang-process-util.h"
 
 #include "../../source/core/slang-shared-library.h"
 
@@ -111,10 +112,109 @@ static SlangResult _createSlangSession(const char* exePath, int argc, const char
     return SLANG_OK;
 }
 
-static SlangResult execute(int argc, const char* const* argv)
+static void _makeStdStreamsUnbuffered()
+{
+    // Make these streams unbuffered
+    setvbuf(stdout, nullptr, _IONBF, 0);
+    setvbuf(stderr, nullptr, _IONBF, 0);
+}
+
+static SlangResult _outputCount(int argc, const char* const* argv)
+{
+    if (argc < 3)
+    {
+        return SLANG_FAIL;
+    }
+    // Get the count
+    const Index count = StringToInt(argv[2]);
+
+    // If we want to crash
+    Index crashIndex = -1;
+    if (argc > 3)
+    {
+        // When we crash, we want to make sure everything is flushed...
+        _makeStdStreamsUnbuffered();
+        crashIndex = StringToInt(argv[3]);
+    }
+
+    FILE* fileOut = stdout;
+
+    StringBuilder buf;
+    for (Index i = 0; i < count; ++i)
+    {
+        buf.Clear();
+        buf << i << "\n";
+
+        fwrite(buf.getBuffer(), 1, buf.getLength(), fileOut);
+
+        if (i == crashIndex)
+        {
+            // Use to simulate a crash.
+            SLANG_BREAKPOINT(0);
+            throw;
+        }
+    }
+
+    // NOTE! There is no flush here, we want to see if everything works if we just stream out.
+    return SLANG_OK;
+}
+
+static SlangResult _outputReflect()
+{
+    // Read lines from std input.
+    // When hit line with just 'end', terminate
+
+    // Get in as Stream
+
+    RefPtr<Stream> stdinStream;
+    Process::getStdStream(Process::StreamType::StdIn, stdinStream);
+
+    FILE* fileOut = stdout;
+
+    List<Byte> buffer;
+
+    Index lineCount = 0;
+    Index startIndex = 0; 
+
+    while (true)
+    {
+        SLANG_RETURN_ON_FAIL(StreamUtil::read(stdinStream, 0, buffer));
+
+        while (true)
+        {
+            UnownedStringSlice slice((const char*)buffer.begin() + startIndex, (const char*)buffer.end());
+            UnownedStringSlice line;
+
+            if (!StringUtil::extractLine(slice, line) || slice.begin() == nullptr)
+            {
+                break;
+            }
+
+            // Process the line
+            if (line == UnownedStringSlice::fromLiteral("end"))
+            {
+                return SLANG_OK;
+            }
+
+            // Write the text to the output stream
+            fwrite(line.begin(), 1, line.getLength(), fileOut);
+            fputc('\n', fileOut);
+            
+            lineCount++;
+
+            // Move the start index forward
+            const Index newStartIndex = slice.begin() ? Index(slice.begin() - (const char*)buffer.getBuffer()) : buffer.getCount();
+            SLANG_ASSERT(newStartIndex > startIndex);
+            startIndex = newStartIndex;
+        }
+    }
+}
+
+static SlangResult execute(int argc, const char*const* argv)
 {
     typedef Slang::TestToolUtil::InnerMainFunc InnerMainFunc;
 
+    
     if (argc < 2)
     {
         return SLANG_FAIL;
@@ -125,6 +225,16 @@ static SlangResult execute(int argc, const char* const* argv)
     // The 'exeName' of the tool. 
     const String toolName = argv[1];
 
+    if (toolName == "reflect")
+    {
+        return _outputReflect();
+    }
+
+    if (toolName == "count")
+    {
+        return _outputCount(argc, argv);
+    }
+
     {
         StringBuilder sharedLibToolBuilder;
         sharedLibToolBuilder.append(toolName);
@@ -203,6 +313,7 @@ static SlangResult execute(int argc, const char* const* argv)
         unitTestContext.slangGlobalSession = session;
         unitTestContext.workDirectory = "";
         unitTestContext.enabledApis = RenderApiFlags(enabledApis);
+        unitTestContext.executableDirectory = exePath.getBuffer();
 
         auto testCount = testModule->getTestCount();
         SLANG_ASSERT(testIndex >= 0 && testIndex < testCount);
diff --git a/tools/unit-test/slang-unit-test.h b/tools/unit-test/slang-unit-test.h
index 9e437f072..035f67e33 100644
--- a/tools/unit-test/slang-unit-test.h
+++ b/tools/unit-test/slang-unit-test.h
@@ -37,6 +37,7 @@ struct UnitTestContext
 {
     slang::IGlobalSession* slangGlobalSession;
     const char* workDirectory;
+    const char* executableDirectory;
     Slang::RenderApiFlags enabledApis;
 };