Fix file handle starvation linux (#15333)

* Fixes https://github.com/o3de/o3de-extras/issues/118

You cannot use FD_SET and select() on file handle numbers greater
than 1023, as the fd set structure is actually a bit mask and has
a fixed size.  Using file handles greater than 1023 will cause it to
buffer overflow.

This fixes the problem in 2 ways
1. It reduces the default number of keep-open handles to a more
   reasonable number (128) across the board on all platforms, as it is
   generally not neccessary for O3DE applications to keep 1024 files
   open all the time for any real reason.
2. It uses epoll instead of select and FD_SET to wait for data to be
   ready, future-proofing this on linux in case it occurs again or
   for whatever reason we use a lot of file handles again.

Note that the file handle count max 1024 is a global system wide
problem, so any time any application uses a large number of file handles
other applications may die, if they use FD_SET in their code.

This is why the action taken is to both future proof O3DE applications
but ALSO reduce the amount of handles the applications themselves
claim so that other programs are not affected.

Signed-off-by: Nicholas Lawson <[email protected]>

* Fixes windows deadlock

It turns out windows deadlocks if a child process writes enough data
to both stderr (unbuffered) and stdout (buffered) because it is incorrectly
using the WaitForMultipleObjects api on objects it cannot wait on (pipes).

This causes WaitForMultipleObjects to return immediately with 0.
This causes it to believe that the stdout handle is ready for reading.
This causes it to call ReadFile on stdout handle and block forever.

This means that it was always busy looping before, the busy loop is just
super obvious now with comments.
I leave it up to future self or others to refactor this code to use
asynch io functions and named pipes instead of WaitForMultipleObjects API.

Signed-off-by: Nicholas Lawson <[email protected]>

Code/Framework/AzFramework/AzFramework/Process/ProcessCommunicator.cpp

@@ -45,13 +45,15 @@ namespace AzFramework
     void ProcessCommunicator::ReadIntoProcessOutput(ProcessOutput& processOutput)
     {
         OutputStatus status;
-        char readBuffer[s_readBufferSize];
 
         // read from the process until the handle is no longer valid
         while (true)
         {
             WaitForReadyOutputs(status);
-            ReadFromOutputs(processOutput, status, readBuffer, s_readBufferSize);
+            if (status.shouldReadErrors || status.shouldReadOutput)
+            {
+                ReadFromOutputs(processOutput, status);
+            }
 
             if (!status.outputDeviceReady && !status.errorsDeviceReady)
             {
@@ -60,27 +62,26 @@ namespace AzFramework
         }
     }
 
-    void ProcessCommunicator::ReadFromOutputs(ProcessOutput& processOutput, OutputStatus& status, char* buffer, AZ::u32 bufferSize)
+    void ProcessCommunicator::ReadFromOutputs(ProcessOutput& processOutput, OutputStatus& status)
     {
         AZ::u32 bytesRead = 0;
+        // Send in the size + 1 to leave room for us to write out the 0 in
+        char readBuffer[s_readBufferSize+1];
 
         if (status.shouldReadOutput)
         {
-            // Send in the size - 1 to leave room for us to write out the 0 in
-            // bytesRead position on the next line
-            bytesRead = ReadOutput(buffer, bufferSize - 1);
-            buffer[bytesRead] = 0;
-            processOutput.outputResult.append(buffer, bytesRead);
+            bytesRead = ReadOutput(readBuffer, s_readBufferSize);
+            readBuffer[bytesRead] = 0;
+            processOutput.outputResult.append(readBuffer, bytesRead);
         }
 
         if (status.shouldReadErrors)
         {
-            // Send in the size - 1 to leave room for us to write out the 0 in
-            // bytesRead position on the next line
-            bytesRead = ReadError(buffer, bufferSize - 1);
-            buffer[bytesRead] = 0;
-            processOutput.errorResult.append(buffer, bytesRead);
+            bytesRead = ReadError(readBuffer, s_readBufferSize);
+            readBuffer[bytesRead] = 0;
+            processOutput.errorResult.append(readBuffer, bytesRead);
         }
+
     }
 
     AZ::u32 ProcessCommunicatorForChildProcess::BlockUntilInputAvailable(AZStd::string& readBuffer)
@@ -141,6 +142,7 @@ namespace AzFramework
         m_stdInWrite->Close();
         m_stdOutRead->Close();
         m_stdErrRead->Close();
+        m_communicatorData.reset();
         m_initialized = false;
     }
 

Code/Framework/AzFramework/AzFramework/Process/ProcessCommunicator.h

@@ -85,11 +85,11 @@ namespace AzFramework
     protected:
         AZ_DISABLE_COPY(ProcessCommunicator);
         
-        // Waits for output or error to be ready for reading
-        virtual void WaitForReadyOutputs(OutputStatus& outputStatus) const = 0;
+        // Waits for stdout or stderr to be ready for reading.  Note that its non-const
+        // because it can detect if the outputs break and update them to be "broken".
+        virtual void WaitForReadyOutputs(OutputStatus& outputStatus) = 0;
 
-        void ReadFromOutputs(ProcessOutput& processOutput,
-            OutputStatus& status, char* buffer, AZ::u32 bufferSize);
+        void ReadFromOutputs(ProcessOutput& processOutput, OutputStatus& status);
 
     private:
         static const size_t s_readBufferSize = 16 * 1024;
@@ -149,6 +149,15 @@ namespace AzFramework
         virtual bool CreatePipesForProcess(AzFramework::ProcessData* processData) = 0;
     };
 
+    //! Platform-specific class, default does nothing, but you can derive from it
+    //! and supply it in your platform-specific implementation.  
+    class StdInOutProcessCommunicatorData
+    {
+        public:
+            StdInOutProcessCommunicatorData() = default;
+            virtual ~StdInOutProcessCommunicatorData() = default;
+    };
+
     /**
     * Communicator to talk to processes via std::in and std::out
     *
@@ -184,12 +193,16 @@ namespace AzFramework
 
         //////////////////////////////////////////////////////////////////////////
         // AzFramework::ProcessCommunicator overrides
-        void WaitForReadyOutputs(OutputStatus& outputStatus) const override;
+        void WaitForReadyOutputs(OutputStatus& outputStatus) override;
         //////////////////////////////////////////////////////////////////////////
 
         AZStd::unique_ptr<CommunicatorHandleImpl> m_stdInWrite;
         AZStd::unique_ptr<CommunicatorHandleImpl> m_stdOutRead;
         AZStd::unique_ptr<CommunicatorHandleImpl> m_stdErrRead;
+
+        //! OPTIONAL - platform-specific classes can plug in additional arbitrary platform
+        //! specific data in here.  or leave it nullptr.
+        AZStd::unique_ptr<StdInOutProcessCommunicatorData> m_communicatorData;
         bool m_initialized = false;
     };
 

Code/Framework/AzFramework/Platform/Android/AzFramework/Process/ProcessCommunicator_Android.cpp

@@ -31,7 +31,7 @@ namespace AzFramework
         return false;
     }
 
-    void StdInOutProcessCommunicator::WaitForReadyOutputs([[maybe_unused]] OutputStatus& status) const
+    void StdInOutProcessCommunicator::WaitForReadyOutputs([[maybe_unused]] OutputStatus& status)
     {
         status.outputDeviceReady = false;
         status.errorsDeviceReady = false;

Code/Framework/AzFramework/Platform/Common/Default/AzFramework/Process/ProcessCommunicator_Default.cpp

@@ -32,7 +32,7 @@ namespace AzFramework
         return false;
     }
 
-    void StdInOutProcessCommunicator::WaitForReadyOutputs([[maybe_unused]] OutputStatus& status) const
+    void StdInOutProcessCommunicator::WaitForReadyOutputs([[maybe_unused]] OutputStatus& status)
     {
         status.outputDeviceReady = false;
         status.errorsDeviceReady = false;

Code/Framework/AzFramework/Platform/Linux/AzFramework/Process/ProcessCommon.h

@@ -15,16 +15,18 @@ namespace AzFramework
     public:
         ~CommunicatorHandleImpl() = default;
 
-        bool IsValid() const;
-        bool IsBroken() const;
-        int GetHandle() const;
+        [[nodiscard]] bool IsValid() const;
+        [[nodiscard]] bool IsBroken() const;
+        [[nodiscard]] int GetHandle() const;
+        [[nodiscard]] int GetEPollHandle() const;
 
         void Break();
         void Close();
-        void SetHandle(int handle);
+        [[nodiscard]] bool SetHandle(int handle, bool createEPollHandle);
 
     protected:
         int m_handle = -1;
+        int m_epollHandle = -1;
         bool m_broken = false;
     };
 

Code/Framework/AzFramework/Platform/Linux/AzFramework/Process/ProcessCommunicator_Linux.cpp

@@ -11,7 +11,32 @@
 #include <fcntl.h>
 #include <errno.h>
 #include <sys/ioctl.h>
+#include <sys/epoll.h>
+
 #include <AzFramework/Process/ProcessCommunicator.h>
+#include <AzCore/Memory/SystemAllocator.h>
+
+namespace ProcessCommunicatorLinuxInternal
+{
+
+    class CommunicatorDataImpl : public AzFramework::StdInOutProcessCommunicatorData
+    {
+    public:
+        AZ_CLASS_ALLOCATOR(CommunicatorDataImpl, AZ::SystemAllocator);
+        CommunicatorDataImpl() = default;
+        virtual ~CommunicatorDataImpl()
+        {
+            if (m_stdInAndOutPollHandle != -1)
+            {
+                close(m_stdInAndOutPollHandle);
+            }
+        }
+        int m_stdInAndOutPollHandle = -1;
+        
+        AZ_DISABLE_COPY_MOVE(CommunicatorDataImpl);
+    };
+
+}
 
 namespace AzFramework
 {
@@ -30,6 +55,11 @@ namespace AzFramework
         return m_handle;
     }
 
+    int CommunicatorHandleImpl::GetEPollHandle() const
+    {
+        return m_epollHandle;
+    }
+
     void CommunicatorHandleImpl::Break()
     {
         m_broken = true;
@@ -37,13 +67,45 @@ namespace AzFramework
 
     void CommunicatorHandleImpl::Close()
     {
-        close(m_handle);
-        m_handle = -1;
+        if (m_handle != -1)
+        {
+            close(m_handle);
+            m_handle = -1;
+        }
+
+        if (m_epollHandle != -1)
+        {
+            close(m_epollHandle);
+            m_epollHandle = -1;
+        }
     }
 
-    void CommunicatorHandleImpl::SetHandle(int handle)
+    bool CommunicatorHandleImpl::SetHandle(int handle, bool createEPollHandle)
     {
         m_handle = handle;
+        // also create an epoll handle
+        if ( (handle != -1) && (createEPollHandle) )
+        {
+            int m_epollHandle = epoll_create1(0);
+
+            AZ_Assert(m_epollHandle -1, "Failed to create epoll handle. errno: %d", errno);
+            if (m_epollHandle == -1) 
+            {
+                return false;
+            }
+            struct epoll_event event;
+
+            event.events = EPOLLIN;
+            event.data.fd = GetHandle();
+            if(epoll_ctl(m_epollHandle, EPOLL_CTL_ADD, event.data.fd, &event))
+            {
+                AZ_Assert(false, "Unable to add a file descriptor to epoll. errno: %d", errno);
+                close(m_epollHandle);
+                m_epollHandle = -1;
+                return false;
+            }
+        }
+        return true;
     }
 
     AZ::u32 StdInOutCommunication::PeekHandle(StdProcessCommunicatorHandle& handle)
@@ -71,25 +133,15 @@ namespace AzFramework
             return 0;
         }
 
-        //Block if buffersize == 0
+        // Block until data is ready if buffersize == 0
         if (bufferSize == 0)
         {
-            fd_set set;
-            FD_ZERO(&set);
-            FD_SET(handle->GetHandle(), &set);
-
-            [[maybe_unused]] int numReady = select(handle->GetHandle() + 1, &set, nullptr, nullptr, nullptr);
-
-            // if numReady == -1 and errno == EINTR then the child process died unexpectedly and
-            // the handle was closed. Not something to assert about in regards to trying to read
-            // data from the child as there is not anything useful we can say or do in that case.
-            // Normal code/data flow will work and we as the parent will know that the child is
-            // dead and return any error codes the child may have written to the error stream.
-            AZ_Assert(numReady != -1 || errno == EINTR, "Could not determine if any data is available for reading due to an error. Errno: %d", errno);
-
-            [[maybe_unused]] const bool wasSet = FD_ISSET(handle->GetHandle(), &set);
-            AZ_Assert(wasSet, "handle was not set when we selected it for read");
-
+            if (handle->GetEPollHandle() == -1)
+            {
+                return 0;
+            }
+            struct epoll_event event;
+            epoll_wait(handle->GetEPollHandle(), &event, 1, -1);
             return 0;
         }
 
@@ -97,7 +149,7 @@ namespace AzFramework
         bytesRead = read(handle->GetHandle(), readBuffer, bufferSize);
         if (bytesRead < 0)
         {
-            AZ_Assert(errno != EIO, "ReadFile performed unexpected async io");
+            AZ_Assert(errno != EIO, "ReadFile performed unexpected async io. errno: %d", errno);
             if (errno == EBADF || errno == EINVAL)
             {
                 // Child process exited, we may have read something, so return amount
@@ -128,7 +180,7 @@ namespace AzFramework
         const ssize_t bytesWritten = write(handle->GetHandle(), writeBuffer, bytesToWrite);
         if (bytesWritten < 0)
         {
-            AZ_Assert(errno != EIO, "Parent performed unexpected async io when trying to write to child process.");
+            AZ_Assert(errno != EIO, "Parent performed unexpected async io when trying to write to child process (errno=%d EIO).", errno);
             if (errno == EPIPE)
             {
                 // Child process exited, may have written something, so return amount
@@ -144,6 +196,7 @@ namespace AzFramework
 
     bool StdInOutProcessCommunicator::CreatePipesForProcess(ProcessData* processData)
     {
+        using namespace ProcessCommunicatorLinuxInternal;
         int pipeFileDescriptors[2] = { 0 };
 
         // Create a pipe to monitor process std in (output from us)
@@ -155,7 +208,14 @@ namespace AzFramework
         }
 
         processData->m_startupInfo.m_inputHandleForChild = pipeFileDescriptors[0];
-        m_stdInWrite->SetHandle(pipeFileDescriptors[1]);
+        // we don't need to create an epoll handle for the child's stdin, since we don't use
+        // epoll for it.
+        if (!m_stdInWrite->SetHandle(pipeFileDescriptors[1], /*create epoll handle*/false))
+        {
+            processData->m_startupInfo.CloseAllHandles();
+            CloseAllHandles();
+            return false;
+        }
 
         // Create a pipe to monitor process std out (input to us)
         result = pipe(pipeFileDescriptors);
@@ -164,12 +224,16 @@ namespace AzFramework
         {
             processData->m_startupInfo.CloseAllHandles();
             CloseAllHandles();
-
             return false;
         }
 
-        m_stdOutRead->SetHandle(pipeFileDescriptors[0]);
         processData->m_startupInfo.m_outputHandleForChild = pipeFileDescriptors[1];
+        if (!m_stdOutRead->SetHandle(pipeFileDescriptors[0], /*create epoll handle*/true))
+        {
+            processData->m_startupInfo.CloseAllHandles();
+            CloseAllHandles();
+            return false;
+        }
 
         // Create a pipe to monitor process std error (input to us)
         result = pipe(pipeFileDescriptors);
@@ -178,61 +242,104 @@ namespace AzFramework
         {
             processData->m_startupInfo.CloseAllHandles();
             CloseAllHandles();
-
             return false;
         }
 
-        m_stdErrRead->SetHandle(pipeFileDescriptors[0]);
         processData->m_startupInfo.m_errorHandleForChild = pipeFileDescriptors[1];
+        if (!m_stdErrRead->SetHandle(pipeFileDescriptors[0], /*create epoll handle*/true))
+        {
+            processData->m_startupInfo.CloseAllHandles();
+            CloseAllHandles();
+            return false;
+        }
+
+        // create an epoll file descriptor for both inputs (stderr and stdout)
+        m_communicatorData.reset(aznew CommunicatorDataImpl());
+        CommunicatorDataImpl* platformImpl = static_cast<CommunicatorDataImpl*>(m_communicatorData.get());
+        platformImpl->m_stdInAndOutPollHandle = epoll_create1(0);
+
+        AZ_Assert(platformImpl->m_stdInAndOutPollHandle != -1, "Failed to create epoll handle to monitor output process, errno = %d", errno);
+        if (platformImpl->m_stdInAndOutPollHandle == -1) 
+        {
+            processData->m_startupInfo.CloseAllHandles();
+            CloseAllHandles();
+            return false;
+        }
+        
+        struct epoll_event event;
+        event.events = EPOLLIN;
+        event.data.fd = m_stdOutRead->GetHandle();
+        if(epoll_ctl(platformImpl->m_stdInAndOutPollHandle, EPOLL_CTL_ADD, event.data.fd, &event))
+        {
+            AZ_Assert(false, "Unable to add a file descriptor for stdOut to epoll, errno = %d", errno);
+            processData->m_startupInfo.CloseAllHandles();
+            CloseAllHandles();
+            return false;
+        }
+
+        event.events = EPOLLIN;
+        event.data.fd = m_stdErrRead->GetHandle();
+        if(epoll_ctl(platformImpl->m_stdInAndOutPollHandle, EPOLL_CTL_ADD, event.data.fd, &event))
+        {
+            AZ_Assert(false, "Unable to add a file descriptor for stdErr to epoll, errno = %d", errno);
+            processData->m_startupInfo.CloseAllHandles();
+            CloseAllHandles();
+            return false;
+        }
 
         m_initialized = true;
         return true;
     }
 
-    void StdInOutProcessCommunicator::WaitForReadyOutputs(OutputStatus& status) const
+    void StdInOutProcessCommunicator::WaitForReadyOutputs(OutputStatus& status)
     {
+        using namespace ProcessCommunicatorLinuxInternal;
+
         status.outputDeviceReady = m_stdOutRead->IsValid() && !m_stdOutRead->IsBroken();
         status.errorsDeviceReady = m_stdErrRead->IsValid() && !m_stdErrRead->IsBroken();
         status.shouldReadOutput = status.shouldReadErrors = false;
 
+        CommunicatorDataImpl* platformImpl = static_cast<CommunicatorDataImpl*>(m_communicatorData.get());
         if (status.outputDeviceReady || status.errorsDeviceReady)
         {
-            fd_set readSet;
-            int maxHandle = 0;
-
-            FD_ZERO(&readSet);
-
-            if (status.outputDeviceReady)
-            {
-                int currentHandle = m_stdOutRead->GetHandle();
-                FD_SET(currentHandle, &readSet);
-                maxHandle = currentHandle > maxHandle ? currentHandle : maxHandle;
-            }
-
-            if (status.errorsDeviceReady)
-            {
-                int currentHandle = m_stdErrRead->GetHandle();
-                FD_SET(currentHandle, &readSet);
-                maxHandle = currentHandle > maxHandle ? currentHandle : maxHandle;
-            }
-
-            if (select(maxHandle + 1, &readSet, nullptr, nullptr, nullptr) != -1)
+            struct epoll_event events[2];
+            if (int numberSignalled = epoll_wait(platformImpl->m_stdInAndOutPollHandle, events, 1, -1) > -1)
             {
-                status.shouldReadOutput = (status.outputDeviceReady && FD_ISSET(m_stdOutRead->GetHandle(), &readSet));
-                status.shouldReadErrors = (status.errorsDeviceReady && FD_ISSET(m_stdErrRead->GetHandle(), &readSet));
+                for (int readEventIndex = 0; readEventIndex < numberSignalled; ++ readEventIndex)
+                {
+                    if (events[readEventIndex].data.fd == m_stdOutRead->GetHandle())
+                    {
+                        status.shouldReadOutput = true;
+                    }
+                    else if (events[readEventIndex].data.fd == m_stdErrRead->GetHandle())
+                    {
+                        status.shouldReadErrors = true;
+                    }
+                }
             }
         }
     }
 
     bool StdInOutProcessCommunicatorForChildProcess::AttachToExistingPipes()
     {
-        m_stdInRead->SetHandle(STDIN_FILENO);
+        m_initialized = false;
+
+        if (!m_stdInRead->SetHandle(STDIN_FILENO,/*create epoll handle*/false))
+        {
+            return false;
+        }
         AZ_Assert(m_stdInRead->IsValid(), "In read handle is invalid");
 
-        m_stdOutWrite->SetHandle(STDOUT_FILENO);
+        if (!m_stdOutWrite->SetHandle(STDOUT_FILENO, /*create epoll handle*/ true))
+        {
+            return false;
+        }
         AZ_Assert(m_stdOutWrite->IsValid(), "Output write handle is invalid");
 
-        m_stdErrWrite->SetHandle(STDERR_FILENO);
+        if (!m_stdErrWrite->SetHandle(STDERR_FILENO, /*create epoll handle*/ true))
+        {
+            return false;
+        }
         AZ_Assert(m_stdErrWrite->IsValid(), "Error write handle is invalid");
 
         m_initialized = m_stdInRead->IsValid() && m_stdOutWrite->IsValid() && m_stdErrWrite->IsValid();

Code/Framework/AzFramework/Platform/Mac/AzFramework/Process/ProcessCommunicator_Mac.cpp

@@ -186,7 +186,7 @@ namespace AzFramework
         return true;
     }
 
-    void StdInOutProcessCommunicator::WaitForReadyOutputs(OutputStatus& status) const
+    void StdInOutProcessCommunicator::WaitForReadyOutputs(OutputStatus& status)
     {
         status.outputDeviceReady = m_stdOutRead->IsValid() && !m_stdOutRead->IsBroken();
         status.errorsDeviceReady = m_stdErrRead->IsValid() && !m_stdErrRead->IsBroken();

Code/Framework/AzFramework/Platform/Windows/AzFramework/Process/ProcessCommunicator_Win.cpp

@@ -210,15 +210,19 @@ namespace AzFramework
         return true;
     }
 
-    void StdInOutProcessCommunicator::WaitForReadyOutputs(OutputStatus& status) const
+    void StdInOutProcessCommunicator::WaitForReadyOutputs(OutputStatus& status)
     {
+        // you cannot, in Windows, WaitForMultipleObjects on a pipe.
+        // we only call it if either one of them is not a pipe...
         status.outputDeviceReady = m_stdOutRead->IsValid() && !m_stdOutRead->IsBroken();
         status.errorsDeviceReady = m_stdErrRead->IsValid() && !m_stdErrRead->IsBroken();
         status.shouldReadOutput = status.shouldReadErrors = false;
 
-        if (status.outputDeviceReady || status.errorsDeviceReady)
+        bool areAnyInputsPipes = m_stdOutRead->IsPipe() || m_stdErrRead->IsPipe();
+
+        // if neither of them is a pipe, we can use WaitForMultipleObjects on that one:
+        if ((status.outputDeviceReady || status.errorsDeviceReady) && !areAnyInputsPipes)
         {
-            DWORD waitResult = 0;
             HANDLE waitHandles[2];
             AZ::u32 handleCount = 0;
 
@@ -226,28 +230,18 @@ namespace AzFramework
             {
                 waitHandles[handleCount++] = m_stdOutRead->GetHandle();
             }
-
             if (status.errorsDeviceReady)
             {
                 waitHandles[handleCount++] = m_stdErrRead->GetHandle();
             }
-
-            waitResult = WaitForMultipleObjects(handleCount, waitHandles, false, INFINITE);
-            switch (waitResult)
-            {
-            case WAIT_OBJECT_0:
-                // If output handle was present, that's the one that signaled, otherwise it was stdError
-                status.shouldReadOutput = status.outputDeviceReady;
-                status.shouldReadErrors = !status.shouldReadOutput;
-                break;
-            case WAIT_OBJECT_0 + 1:
-                // this can only ever be stdError
-                status.shouldReadErrors = true;
-                break;
-            default:
-                break;
-            };
+            WaitForMultipleObjects(handleCount, waitHandles, false, INFINITE);
         }
+        // Don't trust the result of WaitForMultipleObjects!
+        // Not only does it fail to work when any of them are pipes (it is not supported to call WaitForXXXX on pipes),
+        // but it also doesn't handle the case where more than one object is signaled...
+        // So regardless of what happened, check if maybe BOTH are ready:
+        status.shouldReadErrors = (PeekHandle(m_stdErrRead) > 0);
+        status.shouldReadOutput = (PeekHandle(m_stdOutRead) > 0);
     }
 
     bool StdInOutProcessCommunicatorForChildProcess::AttachToExistingPipes()

Code/Framework/AzFramework/Platform/iOS/AzFramework/Process/ProcessCommunicator_iOS.cpp

@@ -31,7 +31,7 @@ namespace AzFramework
         return false;
     }
 
-    void StdInOutProcessCommunicator::WaitForReadyOutputs([[maybe_unused]] OutputStatus& status) const
+    void StdInOutProcessCommunicator::WaitForReadyOutputs([[maybe_unused]] OutputStatus& status)
     {
         status.outputDeviceReady = false;
         status.errorsDeviceReady = false;

Code/Framework/AzFramework/Tests/ProcessLaunchMain.cpp

@@ -5,11 +5,13 @@
  * SPDX-License-Identifier: Apache-2.0 OR MIT
  *
  */
-
-#include <AzCore/Memory/AllocatorManager.h>
+#include <iostream>
+#include <AzCore/std/utility/charconv.h>
 #include <AzFramework/CommandLine/CommandLine.h>
 
-#include <iostream>
+// this is an intentional relative local include so that it can be shared between two unrelated projects
+// namely this one shot test executable, and the tests which run it in a different project with no relation.
+#include "ProcessLaunchTestTokens.h"
 
 void OutputArgs(const AzFramework::CommandLine& commandLine)
 {
@@ -29,12 +31,56 @@ void OutputArgs(const AzFramework::CommandLine& commandLine)
 
 int main(int argc, char** argv)
 {
+    using namespace ProcessLaunchTestInternal;
+
     const AZ::Debug::Trace tracer;
+    int exitCode = 0;
     {
         AzFramework::CommandLine commandLine;
 
         commandLine.Parse(argc, argv);
-        OutputArgs(commandLine);
+
+        if (commandLine.GetNumSwitchValues("exitCode") > 0)
+        {
+            exitCode = atoi(commandLine.GetSwitchValue("exitCode").c_str());
+            const AZStd::string& exitCodeStr = commandLine.GetSwitchValue("exitCode");
+            AZStd::from_chars(exitCodeStr.begin(), exitCodeStr.end(), exitCode);
+        }
+
+        if (commandLine.HasSwitch("plentyOfOutput"))
+        {
+            // special case - plenty of output mode requires that it output an easily recognizable begin
+            // then plentiful output (enough to saturate any buffers) then output a recognizable end.
+            // this makes sure that if there are any buffers that get filled up or overwritten we can detect
+            // deadlock or buffer starvation or buffer overflows.
+            size_t midLength = strlen(s_midToken);
+            size_t beginLength = strlen(s_beginToken);
+            size_t endLength = strlen(s_endToken);
+            auto spamABunchOfLines = [&](FILE *fileDescriptor)
+            {
+                fwrite(s_beginToken, beginLength, 1, fileDescriptor);
+                size_t currentBytesOutput = beginLength;
+                while (currentBytesOutput < s_numOutputBytesInPlentyMode - endLength)
+                {
+                    fwrite(s_midToken, midLength, 1, fileDescriptor);
+                    currentBytesOutput += midLength;
+                }
+                fwrite(s_endToken, endLength, 1, fileDescriptor);
+            };
+
+            spamABunchOfLines(stdout);
+
+            if (exitCode != 0)
+            {
+                // note that stderr is unbuffered so this will take longer!
+                spamABunchOfLines(stderr);
+            }
+        }
+        else
+        {
+            OutputArgs(commandLine);
+        }
+        
     }
-    return 0;
+    return exitCode;
 }

Code/Framework/AzFramework/Tests/ProcessLaunchParseTests.cpp

@@ -15,6 +15,9 @@
 #include <AzCore/StringFunc/StringFunc.h>
 #include <AzCore/IO/Path/Path.h>
 
+// this is an intentional relative local include so that it can be shared between two unrelated projects
+// namely this one test file as well as the one shot executable that these tests invoke.
+#include "ProcessLaunchTestTokens.h"
 
 namespace UnitTest
 {
@@ -82,6 +85,58 @@ namespace UnitTest
         EXPECT_EQ(launchReturn, true);
         EXPECT_EQ(processOutput.outputResult.empty(), false);
     }
+
+    TEST_F(ProcessLaunchParseTests, LaunchProcessAndRetrieveOutput_LargeDataNoError_ContainsEntireOutput)
+    {
+        using namespace ProcessLaunchTestInternal;
+
+        AzFramework::ProcessOutput processOutput;
+        AzFramework::ProcessLauncher::ProcessLaunchInfo processLaunchInfo;
+
+        processLaunchInfo.m_commandlineParameters.emplace<AZStd::vector<AZStd::string>>(
+            AZStd::vector<AZStd::string>{(AZ::IO::Path(AZ::Test::GetCurrentExecutablePath()) / "ProcessLaunchTest").Native(), "-plentyOfOutput"});
+
+        processLaunchInfo.m_workingDirectory = AZ::Test::GetCurrentExecutablePath();
+        processLaunchInfo.m_showWindow = false;
+        bool launchReturn = AzFramework::ProcessWatcher::LaunchProcessAndRetrieveOutput(processLaunchInfo, AzFramework::ProcessCommunicationType::COMMUNICATOR_TYPE_STDINOUT, processOutput);
+
+        EXPECT_TRUE(launchReturn);
+        EXPECT_FALSE(processOutput.outputResult.empty());
+        EXPECT_FALSE(processOutput.HasError());
+
+        const auto& output = processOutput.outputResult;
+        EXPECT_EQ(output.length(), s_numOutputBytesInPlentyMode);
+        EXPECT_TRUE(output.starts_with(s_beginToken));
+        EXPECT_TRUE(output.ends_with(s_endToken));
+    }
+    
+    // this also tests that it can separately read error and stdout, and that the stream way of doing it works.
+    TEST_F(ProcessLaunchParseTests, LaunchProcessAndRetrieveOutput_LargeDataWithError_ContainsEntireOutput)
+    {
+        using namespace ProcessLaunchTestInternal;
+
+        AzFramework::ProcessOutput processOutput;
+        AzFramework::ProcessLauncher::ProcessLaunchInfo processLaunchInfo;
+
+        processLaunchInfo.m_commandlineParameters.emplace<AZStd::vector<AZStd::string>>(
+            AZStd::vector<AZStd::string>{(AZ::IO::Path(AZ::Test::GetCurrentExecutablePath()) / "ProcessLaunchTest").Native(), "-plentyOfOutput", "-exitCode", "1"});
+
+        processLaunchInfo.m_workingDirectory = AZ::Test::GetCurrentExecutablePath();
+        processLaunchInfo.m_showWindow = false;
+        bool launchReturn = AzFramework::ProcessWatcher::LaunchProcessAndRetrieveOutput(processLaunchInfo, AzFramework::ProcessCommunicationType::COMMUNICATOR_TYPE_STDINOUT, processOutput);
+
+        EXPECT_TRUE(launchReturn);
+        EXPECT_TRUE(processOutput.HasOutput());
+        EXPECT_TRUE(processOutput.HasError());
+
+        EXPECT_EQ(processOutput.outputResult.length(), s_numOutputBytesInPlentyMode);
+        EXPECT_TRUE(processOutput.outputResult.starts_with(s_beginToken));
+        EXPECT_TRUE(processOutput.outputResult.ends_with(s_endToken));
+
+        EXPECT_EQ(processOutput.errorResult.length(), s_numOutputBytesInPlentyMode);
+        EXPECT_TRUE(processOutput.errorResult.starts_with(s_beginToken));
+        EXPECT_TRUE(processOutput.errorResult.ends_with(s_endToken));
+    }
    
     TEST_F(ProcessLaunchParseTests, ProcessLauncher_BasicParameter_Success)
     {

Code/Framework/AzFramework/Tests/ProcessLaunchTestTokens.h

@@ -0,0 +1,22 @@
+/*
+ * Copyright (c) Contributors to the Open 3D Engine Project.
+ * For complete copyright and license terms please see the LICENSE at the root of this distribution.
+ *
+ * SPDX-License-Identifier: Apache-2.0 OR MIT
+ *
+ */
+
+// shared between the tests (ProcessLaunchParseTests.cpp) and the simple exe that they use to test
+// themselves (ProcessLaunchMain.cpp).  The targets are unrelated to each other in CMake but still
+// need to include this file in both.
+
+namespace ProcessLaunchTestInternal
+{
+    // the default buffer size for stdout/stdin on some platforms is 4k.  This number
+    // should just be higher than a couple multiples of buffer sizes to ensure that it overwhelms
+    // any buffer and reveals any deadlock caused by not reading from stdout/stderr when its full.
+    constexpr const size_t s_numOutputBytesInPlentyMode = 16 * 1024 ; 
+    constexpr const char* s_beginToken = "BEGIN";
+    constexpr const char* s_endToken = "END";
+    constexpr const char* s_midToken = "x";
+}

Code/Framework/AzFramework/Tests/process_launch_test_files.cmake

@@ -8,4 +8,5 @@
 
 set(FILES
     ProcessLaunchMain.cpp
+    ProcessLaunchTestTokens.h
 )

Registry/Platform/Mac/streamer.editor.setreg

@@ -19,7 +19,7 @@
                             {
                                 "$type": "AzFramework::RemoteStorageDriveConfig",
                                 "$stack_after": "Drive",
-                                "MaxFileHandles": 1024
+                                "MaxFileHandles": 128
                             }
                         }
                     }

Registry/Platform/Windows/streamer.editor.setreg

@@ -15,7 +15,7 @@
                             {
                                 "$type": "AZ::IO::WindowsStorageDriveConfig",
                                 "$stack_after": "Drive",
-                                "MaxFileHandles": 1024,
+                                "MaxFileHandles": 128,
                                 "MaxMetaDataCache": 1024,
                                 "Overcommit": 8,
                                 "EnableFileSharing": true,

Registry/Platform/Windows/streamer.game.debug.setreg

@@ -24,7 +24,7 @@
                             "Drive":
                             {
                                 "$type": "AZ::IO::WindowsStorageDriveConfig",
-                                "MaxFileHandles": 1024,
+                                "MaxFileHandles": 128,
                                 "MaxMetaDataCache": 1024,
                                 "Overcommit": 8,
                                 "EnableFileSharing": true,

Registry/Platform/Windows/streamer.game.profile.setreg

@@ -24,7 +24,7 @@
                             "Drive":
                             {
                                 "$type": "AZ::IO::WindowsStorageDriveConfig",
-                                "MaxFileHandles": 1024,
+                                "MaxFileHandles": 128,
                                 "MaxMetaDataCache": 1024,
                                 "Overcommit": 8,
                                 "EnableFileSharing": true,

Registry/Platform/Windows/streamer.test.setreg

@@ -16,7 +16,7 @@
                             {
                                 "$type": "AZ::IO::WindowsStorageDriveConfig",
                                 "$stack_after": "Drive",
-                                "MaxFileHandles": 1024,
+                                "MaxFileHandles": 128,
                                 "MaxMetaDataCache": 1024,
                                 "Overcommit": 8,
                                 "EnableFileSharing": true,

Registry/streamer.editor.setreg

@@ -16,7 +16,8 @@
                                 "$type": "AzFramework::RemoteStorageDriveConfig",
                                 "$stack_after": "Drive",
                                 // The maximum number of file handles that the drive will cache.
-                                "MaxFileHandles": 1024
+                                // The Streamer will not close handles until it has at least this many handles open
+                                "MaxFileHandles": 128
                             }
                         }
                     }

Registry/streamer.game.debug.setreg

@@ -15,7 +15,7 @@
                             {
                                 "$type": "AzFramework::RemoteStorageDriveConfig",
                                 "$stack_after": "Splitter",
-                                "MaxFileHandles": 1024 
+                                "MaxFileHandles": 128 
                             }
                         }
                     },
@@ -26,12 +26,12 @@
                             "Drive":
                             {
                                 "$type": "AZ::IO::StorageDriveConfig",
-                                "MaxFileHandles": 1024 
+                                "MaxFileHandles": 128 
                             },
                             "Remote drive":
                             {
                                 "$type": "AzFramework::RemoteStorageDriveConfig",
-                                "MaxFileHandles": 1024 
+                                "MaxFileHandles": 128 
                             }
                         }
                     }

Registry/streamer.game.profile.setreg

@@ -15,7 +15,7 @@
                             {
                                 "$type": "AzFramework::RemoteStorageDriveConfig",
                                 "$stack_after": "Splitter",
-                                "MaxFileHandles": 1024 
+                                "MaxFileHandles": 128 
                             }
                         }
                     },
@@ -26,12 +26,12 @@
                             "Drive":
                             {
                                 "$type": "AZ::IO::StorageDriveConfig",
-                                "MaxFileHandles": 1024 
+                                "MaxFileHandles": 128 
                             },
                             "Remote drive":
                             {
                                 "$type": "AzFramework::RemoteStorageDriveConfig",
-                                "MaxFileHandles": 1024 
+                                "MaxFileHandles": 128 
                             }
                         }
                     }

Registry/streamer.game.setreg

@@ -15,6 +15,7 @@
                             {
                                 "$type": "AZ::IO::StorageDriveConfig",
                                 // The maximum number of file handles that the drive will cache.
+                                // The Streamer will not close handles until it has at least this many handles open
                                 "MaxFileHandles": 32 
                             },
                             "Splitter":

Registry/streamer.setreg

@@ -19,7 +19,8 @@
                             {
                                 "$type": "AZ::IO::StorageDriveConfig",
                                 // The maximum number of file handles that the drive will cache.
-                                "MaxFileHandles": 1024 
+                                // The Streamer will not close handles until it has at least this many handles open
+                                "MaxFileHandles": 128 
                             }
                         }
                     }

Registry/streamer.test.setreg

@@ -15,7 +15,7 @@
                             "Drive":
                             {
                                 "$type": "AZ::IO::StorageDriveConfig",
-                                "MaxFileHandles": 1024
+                                "MaxFileHandles": 128
                             },
                             "Splitter":
                             {